diff --git a/litellm/constants.py b/litellm/constants.py
index 1d5c6404bfd..796e4456bbb 100644
--- a/litellm/constants.py
+++ b/litellm/constants.py
@@ -423,6 +423,7 @@ WEBSOCKET_CLOSE_REASON_MAX_BYTES: Final = 123
 DEEPGRAM_DEFAULT_API_BASE: Final = "https://api.deepgram.com/v1"
 NADIR_DEFAULT_API_BASE: Final = "https://api.getnadir.com/v1"
 DEEPGRAM_LISTEN_DEFAULT_MODEL: Final = "nova-3"
+DEEPGRAM_SPEAK_DEFAULT_MODEL: Final = "aura-asteria-en"
 
 BEDROCK_REALTIME_PENDING_OPENING_MESSAGE_SCOPE_KEY: Final = "litellm.bedrock_realtime.pending_opening_message"
 BEDROCK_REALTIME_SESSION_COMMITTED_SCOPE_KEY: Final = "litellm.bedrock_realtime.session_committed"
diff --git a/litellm/llms/deepgram/common_utils.py b/litellm/llms/deepgram/common_utils.py
index 5638f2e0e4b..a4a413370f7 100644
--- a/litellm/llms/deepgram/common_utils.py
+++ b/litellm/llms/deepgram/common_utils.py
@@ -1,16 +1,18 @@
 import math
+import re
 from collections.abc import Mapping, Sequence
 from types import MappingProxyType
-from typing import Final
+from typing import Final, Literal
 from urllib.parse import parse_qs, urlparse
 
 import httpx
+from pydantic import BaseModel, ConfigDict, ValidationError
 
 import litellm
-from litellm.constants import DEEPGRAM_DEFAULT_API_BASE, DEEPGRAM_LISTEN_DEFAULT_MODEL
+from litellm.constants import DEEPGRAM_DEFAULT_API_BASE, DEEPGRAM_LISTEN_DEFAULT_MODEL, DEEPGRAM_SPEAK_DEFAULT_MODEL
 from litellm.litellm_core_utils.url_utils import join_api_base
 from litellm.llms.base_llm.chat.transformation import BaseLLMException
-from litellm.types.utils import LlmProviders
+from litellm.types.utils import LlmProviders, ModelInfo
 
 _WEBSOCKET_SCHEMES: Final = MappingProxyType({"https": "wss", "http": "ws", "wss": "wss", "ws": "ws"})
 DEEPGRAM_LISTEN_CALLBACK_PARAMS: Final = frozenset({"callback", "callback_method"})
@@ -26,6 +28,11 @@ DEEPGRAM_LISTEN_ADDON_PRICING_PARAMS: Final = MappingProxyType(
         "diarize_model": "diarize",
     }
 )
+# Deepgram bills every voice of an Aura generation at one rate, so each generation has one cost-map row
+DEEPGRAM_SPEAK_PRICING_FAMILIES: Final = (
+    (re.compile(r"aura-2-[a-z]+-[a-z]{2}"), "aura-2"),
+    (re.compile(r"aura-[a-z]+-[a-z]{2}"), "aura"),
+)
 _DISABLED_PARAM_VALUES: Final = frozenset({"", "false"})
 _SINGLE_VALUED_PARAMS: Final = frozenset({"model", "language"})
 
@@ -101,15 +108,15 @@ def deepgram_listen_registry_key(upstream_url: str) -> str:
 def deepgram_listen_is_priced(upstream_url: str) -> bool:
     """Only an exact registry hit counts: the cost calculator resolves a missing ``streaming/<model>`` row to the
     pre-recorded ``<model>`` row, which is not the rate Deepgram bills a WebSocket session at."""
-    return _has_exact_registry_entry(deepgram_listen_registry_key(upstream_url))
+    return _exact_registry_entry(deepgram_listen_registry_key(upstream_url)) is not None
 
 
-def _has_exact_registry_entry(registry_key: str) -> bool:
+def _exact_registry_entry(registry_key: str) -> ModelInfo | None:
     try:
         model_info: Final = litellm.get_model_info(model=registry_key, custom_llm_provider=LlmProviders.DEEPGRAM.value)
     except Exception:
-        return False
-    return model_info["key"] == registry_key
+        return None
+    return model_info if model_info["key"] == registry_key else None
 
 
 def deepgram_listen_addon_pricing_models(upstream_url: str) -> tuple[str, ...]:
@@ -132,10 +139,11 @@ def _channel_count(value: object) -> int | None:
 
 
 def _results_channel_count(frame: Mapping[str, object]) -> int | None:
-    channel_index: Final = frame.get("channel_index")
-    if not isinstance(channel_index, list) or len(channel_index) != 2:
-        return None
-    return _channel_count(channel_index[1])
+    match frame.get("channel_index"):
+        case [object(), int() as channels]:
+            return _channel_count(channels)
+        case _:
+            return None
 
 
 def _declared_channel_count(upstream_url: str) -> int | None:
@@ -214,3 +222,50 @@ def deepgram_listen_transcript(websocket_messages: Sequence[Mapping[str, object]
         if frame.get("type") == "Results"
         if (transcript := _final_transcript(frame)) is not None
     )
+
+
+def deepgram_speak_requested_model(query_string: str) -> str:
+    return _requested_model(query_string, DEEPGRAM_SPEAK_DEFAULT_MODEL)
+
+
+def deepgram_speak_websocket_target(api_base: str | None, query_string: str) -> str:
+    return _deepgram_websocket_target(api_base, "speak", query_string, DEEPGRAM_SPEAK_DEFAULT_MODEL)
+
+
+def deepgram_speak_model(upstream_url: str) -> str:
+    return _upstream_model(upstream_url, DEEPGRAM_SPEAK_DEFAULT_MODEL)
+
+
+def deepgram_speak_registry_key(upstream_url: str) -> str:
+    """The row a speak session is billed at: its Aura generation's row, else a row named after the voice itself"""
+    model: Final = deepgram_speak_model(upstream_url)
+    pricing_model: Final = next(
+        (family for pattern, family in DEEPGRAM_SPEAK_PRICING_FAMILIES if pattern.fullmatch(model)), model
+    )
+    return f"{LlmProviders.DEEPGRAM.value}/{pricing_model}"
+
+
+def deepgram_speak_character_rate(upstream_url: str) -> float | None:
+    model_info: Final = _exact_registry_entry(deepgram_speak_registry_key(upstream_url))
+    return None if model_info is None else model_info.get("input_cost_per_character")
+
+
+def deepgram_speak_is_priced(upstream_url: str) -> bool:
+    return deepgram_speak_character_rate(upstream_url) is not None
+
+
+class _DeepgramSpeakMessage(BaseModel):
+    model_config = ConfigDict(frozen=True)
+
+    type: Literal["Speak"]
+    text: str
+
+
+def deepgram_speak_characters(frame: str | bytes) -> int:
+    """Characters Deepgram bills for one client frame: the length of a Speak message's text, else 0. It never raises,
+    because the relay runs it on every frame and a traceback from a malformed frame would carry text that must not be
+    logged"""
+    try:
+        return len(_DeepgramSpeakMessage.model_validate_json(frame).text)
+    except ValidationError:
+        return 0
diff --git a/litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py b/litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py
index dfbe87bd496..4e4b5de90f8 100644
--- a/litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py
+++ b/litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py
@@ -63,6 +63,11 @@ from litellm.llms.deepgram.common_utils import (
     deepgram_listen_registry_key,
     deepgram_listen_requested_model,
     deepgram_listen_websocket_target,
+    deepgram_speak_characters,
+    deepgram_speak_is_priced,
+    deepgram_speak_registry_key,
+    deepgram_speak_requested_model,
+    deepgram_speak_websocket_target,
 )
 from litellm.llms.fal_ai.cost_calculator import fal_ai_passthrough_cost, fal_ai_queue_base
 from litellm.llms.nvidia_nim.passthrough.transformation import nvidia_nim_model_group_in_path
@@ -3354,6 +3359,7 @@ class _WebsocketRelay(Protocol):
         forward_headers: bool,
         endpoint: str,
         accept_websocket: bool,
+        count_client_text: Callable[[str | bytes], int] | None = None,
     ) -> None: ...
 
 
@@ -3484,6 +3490,7 @@ async def _relay_deepgram_websocket(
     relay: _WebsocketRelay,
     target: str,
     refusal: str | None,
+    count_client_text: Callable[[str | bytes], int] | None = None,
 ) -> None:
     deepgram_api_key: Final = passthrough_endpoint_router.get_credentials(
         custom_llm_provider=litellm.LlmProviders.DEEPGRAM.value,
@@ -3506,6 +3513,7 @@ async def _relay_deepgram_websocket(
         forward_headers=False,
         endpoint=websocket.url.path,
         accept_websocket=False,
+        count_client_text=count_client_text,
     )
 
 
@@ -3520,6 +3528,36 @@ async def deepgram_listen_websocket_route(
     await _relay_deepgram_websocket(websocket, user_api_key_dict, relay, target, _deepgram_listen_refusal(target))
 
 
+async def deepgram_speak_user_api_key_auth(websocket: WebSocket) -> UserAPIKeyAuth:
+    return await user_api_key_auth_websocket_for_model(
+        websocket, model=deepgram_speak_requested_model(websocket.url.query)
+    )
+
+
+def _deepgram_speak_refusal(target: str) -> str | None:
+    if deepgram_speak_is_priced(target):
+        return None
+    return _DEEPGRAM_WS_UNPRICED_REASON.format(registry_key=deepgram_speak_registry_key(target))
+
+
+@router.websocket("/deepgram/v1/speak")
+@router.websocket("/deepgram/speak")
+async def deepgram_speak_websocket_route(
+    websocket: WebSocket,
+    user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(deepgram_speak_user_api_key_auth)],
+    relay: Annotated[_WebsocketRelay, Depends(_websocket_relay)],
+) -> None:
+    target: Final = deepgram_speak_websocket_target(get_secret_str("DEEPGRAM_API_BASE"), websocket.url.query)
+    await _relay_deepgram_websocket(
+        websocket,
+        user_api_key_dict,
+        relay,
+        target,
+        _deepgram_speak_refusal(target),
+        count_client_text=deepgram_speak_characters,
+    )
+
+
 class BaseOpenAIPassThroughHandler:
     @staticmethod
     async def base_openai_pass_through_handler(
diff --git a/litellm/proxy/pass_through_endpoints/llm_provider_handlers/deepgram_speak_passthrough_logging_handler.py b/litellm/proxy/pass_through_endpoints/llm_provider_handlers/deepgram_speak_passthrough_logging_handler.py
new file mode 100644
index 00000000000..572d183ecc2
--- /dev/null
+++ b/litellm/proxy/pass_through_endpoints/llm_provider_handlers/deepgram_speak_passthrough_logging_handler.py
@@ -0,0 +1,79 @@
+import json
+from collections.abc import Mapping
+from types import MappingProxyType
+from typing import Final
+from urllib.parse import urlparse
+
+import litellm
+from litellm._logging import verbose_proxy_logger
+from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
+from litellm.llms.deepgram.common_utils import (
+    deepgram_speak_character_rate,
+    deepgram_speak_model,
+    deepgram_speak_registry_key,
+)
+from litellm.proxy._types import PassThroughEndpointLoggingTypedDict
+from litellm.types.passthrough_endpoints.pass_through_endpoints import WEBSOCKET_CLIENT_TEXT_COUNT_KEY
+from litellm.types.utils import StandardPassThroughResponseObject
+
+DEEPGRAM_SPEAK_ROUTE_SUFFIX: Final = "/speak"
+
+
+def _billed_characters(request_body: Mapping[str, object]) -> int:
+    characters: Final = request_body.get(WEBSOCKET_CLIENT_TEXT_COUNT_KEY)
+    return characters if isinstance(characters, int) else 0
+
+
+def _speech_cost(upstream_url: str, characters: int) -> float | None:
+    rate: Final = deepgram_speak_character_rate(upstream_url)
+    if rate is None:
+        verbose_proxy_logger.warning(
+            "Deepgram speak passthrough: no per-character registry entry '%s'",
+            deepgram_speak_registry_key(upstream_url),
+        )
+        return None
+    return characters * rate
+
+
+class DeepgramSpeakPassthroughLoggingHandler:
+    """Bills a Deepgram speak session by the characters of Speak text the client sent. The relay hands over only the
+    count, never the text, so the spend row and every callback see a model, a count and a cost and nothing the caller
+    asked to have spoken"""
+
+    @staticmethod
+    def is_deepgram_speak_route(url_route: str, request_method: str) -> bool:
+        path: Final = urlparse(url_route).path
+        return request_method == "WEBSOCKET" and "/deepgram/" in path and path.endswith(DEEPGRAM_SPEAK_ROUTE_SUFFIX)
+
+    @staticmethod
+    def deepgram_speak_passthrough_handler(
+        request_body: Mapping[str, object],
+        logging_obj: LiteLLMLoggingObj,
+        upstream_url: str,
+        kwargs: Mapping[str, object] = MappingProxyType({}),
+    ) -> PassThroughEndpointLoggingTypedDict:
+        model: Final = deepgram_speak_model(upstream_url)
+        characters: Final = _billed_characters(request_body)
+        response_cost: Final = _speech_cost(upstream_url, characters)
+
+        provider: Final = litellm.LlmProviders.DEEPGRAM.value
+        logging_obj.model = model  # rebind-ok: the spend logger reads model and cost off the shared logging object
+        logging_obj.model_call_details["model"] = model  # rebind-ok: same shared logging object
+        logging_obj.model_call_details["custom_llm_provider"] = provider  # rebind-ok: same shared logging object
+        logging_obj.model_call_details["response_cost"] = response_cost  # rebind-ok: same shared logging object
+        verbose_proxy_logger.debug(
+            "Deepgram speak passthrough cost tracking: model %s, characters %s, cost %s",
+            model,
+            characters,
+            response_cost,
+        )
+        logging_result: Final[PassThroughEndpointLoggingTypedDict] = {
+            "result": StandardPassThroughResponseObject(response=json.dumps({"characters": characters})),
+            "kwargs": {
+                **kwargs,
+                "model": model,
+                "custom_llm_provider": provider,
+                "response_cost": response_cost,
+            },
+        }
+        return logging_result
diff --git a/litellm/proxy/pass_through_endpoints/success_handler.py b/litellm/proxy/pass_through_endpoints/success_handler.py
index 9c6c3111209..8ad1a221f52 100644
--- a/litellm/proxy/pass_through_endpoints/success_handler.py
+++ b/litellm/proxy/pass_through_endpoints/success_handler.py
@@ -29,6 +29,9 @@ from .llm_provider_handlers.cursor_passthrough_logging_handler import (
 from .llm_provider_handlers.deepgram_listen_passthrough_logging_handler import (
     DeepgramListenPassthroughLoggingHandler,
 )
+from .llm_provider_handlers.deepgram_speak_passthrough_logging_handler import (
+    DeepgramSpeakPassthroughLoggingHandler,
+)
 from .llm_provider_handlers.fal_ai_passthrough_logging_handler import (
     FalAIPassthroughLoggingHandler,
 )
@@ -150,7 +153,7 @@ class PassThroughEndpointLogging:
 
     _handle_logging = handle_logging
 
-    def normalize_llm_passthrough_logging_payload(
+    def normalize_llm_passthrough_logging_payload(  # noqa: C901  # provider dispatch, one branch per provider
         self,
         httpx_response: httpx.Response,
         response_body: dict | list[dict[str, object]] | None,
@@ -400,6 +403,17 @@ class PassThroughEndpointLogging:
             )
             standard_logging_response_object = deepgram_handler_result["result"]  # rebind-ok: elif-chain
             kwargs = deepgram_handler_result["kwargs"]  # rebind-ok: elif-chain contract
+        elif DeepgramSpeakPassthroughLoggingHandler.is_deepgram_speak_route(url_route, httpx_response.request.method):
+            deepgram_speak_handler_result: Final = (
+                DeepgramSpeakPassthroughLoggingHandler.deepgram_speak_passthrough_handler(
+                    request_body=request_body,
+                    logging_obj=logging_obj,
+                    upstream_url=str(httpx_response.request.url),
+                    kwargs=kwargs,
+                )
+            )
+            standard_logging_response_object = deepgram_speak_handler_result["result"]  # rebind-ok: elif-chain
+            kwargs = deepgram_speak_handler_result["kwargs"]  # rebind-ok: elif-chain contract
         elif FalAIPassthroughLoggingHandler.is_fal_ai_route(url_route, custom_llm_provider):
             fal_ai_handler_result: Final = FalAIPassthroughLoggingHandler().fal_ai_passthrough_handler(
                 response_body=response_body if isinstance(response_body, dict) else MappingProxyType({}),
diff --git a/tests/unit/llms/deepgram/test_deepgram_common_utils.py b/tests/unit/llms/deepgram/test_deepgram_common_utils.py
index 7f480d3459f..c0ea885aebf 100644
--- a/tests/unit/llms/deepgram/test_deepgram_common_utils.py
+++ b/tests/unit/llms/deepgram/test_deepgram_common_utils.py
@@ -1,3 +1,4 @@
+import json
 import math
 from collections.abc import Mapping, Sequence
 from typing import Final
@@ -18,6 +19,13 @@ from litellm.llms.deepgram.common_utils import (
     deepgram_listen_requested_model,
     deepgram_listen_transcript,
     deepgram_listen_websocket_target,
+    deepgram_speak_character_rate,
+    deepgram_speak_characters,
+    deepgram_speak_is_priced,
+    deepgram_speak_model,
+    deepgram_speak_registry_key,
+    deepgram_speak_requested_model,
+    deepgram_speak_websocket_target,
 )
 
 NOVA_3_URL: Final = "wss://api.deepgram.com/v1/listen?model=nova-3&encoding=linear16&sample_rate=16000"
@@ -344,3 +352,141 @@ def test_aura_rows_bill_speech_per_character_at_the_published_rate(
 
     assert prompt_cost == pytest.approx(usd_per_thousand_characters)
     assert completion_cost == 0.0
+
+
+SPEAK_BASE: Final = "wss://api.deepgram.com/v1/speak"
+
+
+@pytest.mark.parametrize(
+    ("api_base", "query_string", "expected"),
+    [
+        pytest.param(None, "", f"{SPEAK_BASE}?model=aura-asteria-en", id="default model added"),
+        pytest.param(None, "model=", f"{SPEAK_BASE}?model=aura-asteria-en", id="empty model replaced"),
+        pytest.param(
+            None,
+            "model=aura-2-thalia-en&encoding=linear16&sample_rate=24000",
+            f"{SPEAK_BASE}?model=aura-2-thalia-en&encoding=linear16&sample_rate=24000",
+            id="query kept in order",
+        ),
+        pytest.param(
+            None,
+            "model=aura-2-thalia-en&model=aura-2-zeus-en",
+            f"{SPEAK_BASE}?model=aura-2-thalia-en",
+            id="only the first model",
+        ),
+        pytest.param(
+            "http://127.0.0.1:8080/v1", "model=aura-2-thalia-en", "ws://127.0.0.1:8080/v1/speak?model=aura-2-thalia-en"
+        ),
+        pytest.param(
+            "https://dg.internal/v1/",
+            "encoding=mulaw",
+            "wss://dg.internal/v1/speak?encoding=mulaw&model=aura-asteria-en",
+        ),
+    ],
+)
+def test_deepgram_speak_websocket_target(api_base: str | None, query_string: str, expected: str):
+    assert deepgram_speak_websocket_target(api_base=api_base, query_string=query_string) == expected
+
+
+@pytest.mark.parametrize(
+    "query_string",
+    ["", "model=aura-2-thalia-en", "model=aura-2-thalia-en&model=aura-2-zeus-en", "encoding=mulaw&model=aura-luna-en"],
+)
+def test_the_authorized_speak_model_is_the_only_model_the_upstream_target_carries(query_string: str):
+    """The key is checked against deepgram_speak_requested_model, so the upstream must use exactly that voice"""
+    target: Final = deepgram_speak_websocket_target(api_base=None, query_string=query_string)
+
+    assert parse_qs(urlparse(target).query)["model"] == [deepgram_speak_requested_model(query_string)]
+    assert deepgram_speak_model(target) == deepgram_speak_requested_model(query_string)
+
+
+@pytest.mark.parametrize(
+    ("model", "expected_family"),
+    [
+        # Voice names from https://developers.deepgram.com/docs/tts-models, checked 2026-10-26
+        pytest.param("aura-2-thalia-en", "aura-2", id="aura-2 english"),
+        pytest.param("aura-2-celeste-es", "aura-2", id="aura-2 spanish"),
+        pytest.param("aura-2-rhea-nl", "aura-2", id="aura-2 dutch"),
+        pytest.param("aura-2-agathe-fr", "aura-2", id="aura-2 french"),
+        pytest.param("aura-2-viktoria-de", "aura-2", id="aura-2 german"),
+        pytest.param("aura-2-livia-it", "aura-2", id="aura-2 italian"),
+        pytest.param("aura-2-izanami-ja", "aura-2", id="aura-2 japanese"),
+        pytest.param("aura-asteria-en", "aura", id="aura-1"),
+        pytest.param("aura-zeus-en", "aura", id="another aura-1"),
+        pytest.param("aura-3-foo-en", "aura-3-foo-en", id="unknown aura generation"),
+        pytest.param("aura-2", "aura-2", id="bare aura-2"),
+        pytest.param("aura-x", "aura-x", id="aura without a language"),
+        pytest.param("aura-2-thalia-en-extra", "aura-2-thalia-en-extra", id="trailing part"),
+        pytest.param("flux-tts-en", "flux-tts-en", id="other family"),
+        pytest.param("nova-3", "nova-3", id="listen model"),
+    ],
+)
+def test_deepgram_speak_registry_key_names_the_aura_generation_row_else_the_voice(model: str, expected_family: str):
+    target: Final = deepgram_speak_websocket_target(api_base=None, query_string=f"model={model}")
+
+    assert deepgram_speak_registry_key(target) == f"deepgram/{expected_family}"
+
+
+@pytest.mark.usefixtures("local_model_cost_map")
+@pytest.mark.parametrize(
+    ("model", "expected"),
+    [
+        pytest.param("aura-2-thalia-en", True, id="aura-2"),
+        pytest.param("aura-asteria-en", True, id="aura-1"),
+        pytest.param("aura-3-foo-en", False, id="unknown aura generation"),
+        pytest.param("flux-tts-en", False, id="unpriced family"),
+        pytest.param("nova-3", False, id="a listen model has no speech row"),
+    ],
+)
+def test_deepgram_speak_is_priced(model: str, expected: bool):
+    assert deepgram_speak_is_priced(f"{SPEAK_BASE}?model={model}") is expected
+
+
+@pytest.mark.usefixtures("local_model_cost_map")
+def test_deepgram_speak_is_not_priced_once_its_family_row_is_gone(monkeypatch: pytest.MonkeyPatch):
+    monkeypatch.delitem(litellm.model_cost, "deepgram/aura-2")
+
+    assert deepgram_speak_is_priced(f"{SPEAK_BASE}?model=aura-2-thalia-en") is False
+
+
+@pytest.mark.usefixtures("local_model_cost_map")
+def test_a_cost_map_row_named_after_a_new_voice_prices_it_at_that_rate(monkeypatch: pytest.MonkeyPatch):
+    """The 1008 refusal tells the operator to add the voice's row, so adding it must be enough"""
+    url: Final = f"{SPEAK_BASE}?model=aura-3-foo-en"
+    monkeypatch.setitem(
+        litellm.model_cost,
+        "deepgram/aura-3-foo-en",
+        {"input_cost_per_character": 4e-05, "litellm_provider": "deepgram", "mode": "audio_speech"},
+    )
+
+    assert deepgram_speak_character_rate(url) == 4e-05
+    assert deepgram_speak_is_priced(url) is True
+
+
+@pytest.mark.parametrize(
+    ("frame", "expected"),
+    [
+        pytest.param(
+            json.dumps({"type": "Speak", "text": "The quick brown fox jumps over the lazy dog."}), 44, id="speak"
+        ),
+        pytest.param(json.dumps({"type": "Speak", "text": "Gr\u00fc\u00dfe \U0001f44b"}), 7, id="code points"),
+        pytest.param(
+            json.dumps({"type": "Speak", "text": "Sent as a binary frame."}).encode(), 23, id="speak in a binary frame"
+        ),
+        pytest.param(b"\xff\xfe\x00", 0, id="binary frame that is not utf-8"),
+        pytest.param(json.dumps({"type": "Speak", "text": ""}), 0, id="empty text"),
+        pytest.param(json.dumps({"type": "Flush"}), 0, id="flush"),
+        pytest.param(json.dumps({"type": "Clear"}), 0, id="clear"),
+        pytest.param(json.dumps({"type": "Close"}), 0, id="close"),
+        pytest.param(json.dumps({"type": "Speak"}), 0, id="speak without text"),
+        pytest.param(json.dumps({"type": "Speak", "text": 12345}), 0, id="text not a string"),
+        pytest.param(json.dumps({"type": "Speak", "text": ["a", "b"]}), 0, id="text a list"),
+        pytest.param(json.dumps([{"type": "Speak", "text": "hello"}]), 0, id="not an object"),
+        pytest.param("hello there", 0, id="not json"),
+        pytest.param("", 0, id="empty frame"),
+        pytest.param("[" * 100_000, 0, id="nesting deep enough to recurse"),
+        pytest.param("1" * 5_000, 0, id="integer longer than the parser allows"),
+    ],
+)
+def test_deepgram_speak_characters_counts_speak_text_and_never_raises(frame: str | bytes, expected: int):
+    assert deepgram_speak_characters(frame) == expected
diff --git a/tests/unit/proxy/pass_through_endpoints/llm_provider_handlers/test_deepgram_speak_passthrough_logging_handler.py b/tests/unit/proxy/pass_through_endpoints/llm_provider_handlers/test_deepgram_speak_passthrough_logging_handler.py
new file mode 100644
index 00000000000..96247df28f1
--- /dev/null
+++ b/tests/unit/proxy/pass_through_endpoints/llm_provider_handlers/test_deepgram_speak_passthrough_logging_handler.py
@@ -0,0 +1,174 @@
+"""Deepgram ``/v1/speak`` WebSocket passthrough: per-character cost tracking from the relay's client text count."""
+
+import json
+from datetime import datetime
+from types import SimpleNamespace
+from typing import Final
+
+import pytest
+
+import litellm
+from litellm.integrations.custom_logger import CustomLogger
+from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
+from litellm.proxy.pass_through_endpoints.llm_provider_handlers.deepgram_speak_passthrough_logging_handler import (
+    DeepgramSpeakPassthroughLoggingHandler,
+)
+from litellm.proxy.pass_through_endpoints.success_handler import PassThroughEndpointLogging
+from litellm.types.passthrough_endpoints.pass_through_endpoints import (
+    WEBSOCKET_CLIENT_TEXT_COUNT_KEY,
+    PassthroughStandardLoggingPayload,
+)
+from litellm.types.utils import StandardLoggingPayload
+
+AURA_2_URL: Final = "wss://api.deepgram.com/v1/speak?model=aura-2-thalia-en&encoding=linear16&sample_rate=24000"
+AURA_1_URL: Final = "wss://api.deepgram.com/v1/speak?model=aura-asteria-en"
+SPOKEN_CHARACTERS: Final = len("The quick brown fox jumps over the lazy dog.")
+# Pay as you go rates from https://deepgram.com/pricing, checked 2026-10-26
+AURA_2_USD_PER_CHARACTER: Final = 0.030 / 1000
+AURA_1_USD_PER_CHARACTER: Final = 0.0150 / 1000
+
+pytestmark: Final = pytest.mark.usefixtures("local_model_cost_map")
+
+
+@pytest.mark.parametrize(
+    ("url_route", "request_method", "expected"),
+    [
+        ("/deepgram/v1/speak", "WEBSOCKET", True),
+        ("/deepgram/speak", "WEBSOCKET", True),
+        ("/deepgram/v1/speak?model=aura-2-thalia-en", "WEBSOCKET", True),
+        ("/deepgram/v1/listen", "WEBSOCKET", False),
+        ("/deepgram/v1/speak/extra", "WEBSOCKET", False),
+        ("/openai/v1/audio/speech", "WEBSOCKET", False),
+        ("", "WEBSOCKET", False),
+        pytest.param("https://gateway.internal/deepgram/v1/speak", "POST", False, id="http passthrough to a speak url"),
+    ],
+)
+def test_is_deepgram_speak_route(url_route: str, request_method: str, expected: bool):
+    assert DeepgramSpeakPassthroughLoggingHandler.is_deepgram_speak_route(url_route, request_method) is expected
+
+
+def _logging_obj(call_id: str = "call-dg-speak") -> LiteLLMLoggingObj:
+    return LiteLLMLoggingObj(
+        model="unknown",
+        messages=[{"role": "user", "content": "WebSocket connection"}],
+        stream=True,
+        call_type="pass_through_endpoint",
+        start_time=datetime.now(),
+        litellm_call_id=call_id,
+        function_id="websocket_passthrough",
+    )
+
+
+def _handler_cost(upstream_url: str, characters: object) -> float | None:
+    handler_result: Final = DeepgramSpeakPassthroughLoggingHandler.deepgram_speak_passthrough_handler(
+        request_body={WEBSOCKET_CLIENT_TEXT_COUNT_KEY: characters},
+        logging_obj=_logging_obj(),
+        upstream_url=upstream_url,
+    )
+    response_cost: Final = handler_result["kwargs"]["response_cost"]
+    assert response_cost is None or isinstance(response_cost, float)
+    return response_cost
+
+
+def test_handler_bills_spoken_characters_at_the_aura_2_rate_and_names_the_voice():
+    logging_obj: Final = _logging_obj()
+
+    handler_result: Final = DeepgramSpeakPassthroughLoggingHandler.deepgram_speak_passthrough_handler(
+        request_body={WEBSOCKET_CLIENT_TEXT_COUNT_KEY: SPOKEN_CHARACTERS},
+        logging_obj=logging_obj,
+        upstream_url=AURA_2_URL,
+        kwargs={"litellm_params": {"metadata": {}}},
+    )
+
+    assert handler_result["kwargs"]["response_cost"] == pytest.approx(SPOKEN_CHARACTERS * AURA_2_USD_PER_CHARACTER)
+    assert handler_result["kwargs"]["model"] == "aura-2-thalia-en"
+    assert handler_result["kwargs"]["custom_llm_provider"] == "deepgram"
+    assert handler_result["kwargs"]["litellm_params"] == {"metadata": {}}
+    assert handler_result["result"] == {"response": json.dumps({"characters": SPOKEN_CHARACTERS})}
+    assert logging_obj.model == "aura-2-thalia-en"
+    assert logging_obj.model_call_details["response_cost"] == pytest.approx(
+        SPOKEN_CHARACTERS * AURA_2_USD_PER_CHARACTER
+    )
+
+
+def test_handler_bills_an_aura_1_voice_at_the_aura_1_rate():
+    assert _handler_cost(AURA_1_URL, 1000) == pytest.approx(1000 * AURA_1_USD_PER_CHARACTER)
+
+
+@pytest.mark.parametrize(
+    "characters",
+    [pytest.param(0, id="nothing spoken"), pytest.param(None, id="no count")],
+)
+def test_handler_bills_zero_for_a_session_without_speak_text(characters: object):
+    assert _handler_cost(AURA_2_URL, characters) == 0.0
+
+
+@pytest.mark.parametrize("model", ["flux-tts-en", "aura-3-x-en", "aura-x", "nova-3"])
+def test_handler_reports_no_cost_rather_than_a_wrong_one_for_an_unpriced_voice(model: str):
+    assert _handler_cost(f"wss://api.deepgram.com/v1/speak?model={model}", 1000) is None
+
+
+class _CapturingLogger(CustomLogger):
+    def __init__(self) -> None:
+        super().__init__()
+        self.payloads: list[StandardLoggingPayload] = []  # mutable-ok: the logger appends each payload it receives
+
+    async def async_log_success_event(self, kwargs, response_obj, start_time, end_time):
+        self.payloads.append(kwargs["standard_logging_object"])
+
+
+@pytest.mark.parametrize(
+    ("characters", "expected_cost"),
+    [
+        pytest.param(SPOKEN_CHARACTERS, SPOKEN_CHARACTERS * AURA_2_USD_PER_CHARACTER, id="one speak"),
+        pytest.param(0, 0.0, id="zero stays zero and is not recomputed"),
+    ],
+)
+async def test_success_handler_dispatches_deepgram_speak_and_logs_character_based_spend(
+    monkeypatch: pytest.MonkeyPatch, characters: int, expected_cost: float
+):
+    """Drives the shared passthrough success handler the way the relay does at socket close and reads what a spend
+    logger receives"""
+    capturing_logger: Final = _CapturingLogger()
+    monkeypatch.setattr(litellm, "_async_success_callback", [capturing_logger])
+    monkeypatch.setattr(litellm, "success_callback", [])
+    monkeypatch.setattr(litellm, "callbacks", [])
+    logging_obj: Final = _logging_obj("call-dg-speak-e2e")
+    litellm_params: Final = {"metadata": {"user_api_key": "hashed-key", "user_api_key_team_id": "team-tts"}}
+    logging_obj.update_environment_variables(
+        model="unknown",
+        user="unknown",
+        optional_params={},
+        litellm_params=litellm_params,
+        call_type="pass_through_endpoint",
+    )
+
+    await PassThroughEndpointLogging().pass_through_async_success_handler(
+        httpx_response=SimpleNamespace(
+            status_code=200,
+            text="WebSocket connection successful",
+            headers={},
+            request=SimpleNamespace(method="WEBSOCKET", url=AURA_2_URL),
+        ),
+        response_body=[{"type": "Metadata", "request_id": "req-1"}, {"type": "Flushed", "sequence_id": 0}],
+        logging_obj=logging_obj,
+        url_route="/deepgram/v1/speak",
+        result="websocket_connection_successful",
+        start_time=datetime.now(),
+        end_time=datetime.now(),
+        cache_hit=False,
+        request_body={WEBSOCKET_CLIENT_TEXT_COUNT_KEY: characters},
+        passthrough_logging_payload=PassthroughStandardLoggingPayload(
+            url=AURA_2_URL, request_body={}, request_method="WEBSOCKET", cost_per_request=None
+        ),
+        litellm_params=litellm_params,
+    )
+
+    assert len(capturing_logger.payloads) == 1
+    payload: Final = capturing_logger.payloads[0]
+    assert payload["model"] == "aura-2-thalia-en"
+    assert payload["custom_llm_provider"] == "deepgram"
+    assert payload["response_cost"] == pytest.approx(expected_cost)
+    assert payload["response_cost"] is not None
+    assert payload["metadata"]["user_api_key_team_id"] == "team-tts"
+    assert payload["id"] == "call-dg-speak-e2e"
diff --git a/tests/unit/proxy/pass_through_endpoints/test_deepgram_ws_passthrough_routes.py b/tests/unit/proxy/pass_through_endpoints/test_deepgram_ws_passthrough_routes.py
index 404ed0f5800..69106acb7e1 100644
--- a/tests/unit/proxy/pass_through_endpoints/test_deepgram_ws_passthrough_routes.py
+++ b/tests/unit/proxy/pass_through_endpoints/test_deepgram_ws_passthrough_routes.py
@@ -1,7 +1,8 @@
-"""Deepgram ``/v1/listen`` passthrough WebSocket route: registration, auth, credential injection, target URL."""
+"""Deepgram ``/v1/listen`` and ``/v1/speak`` passthrough WebSocket routes: registration, auth, credential injection,
+target URL."""
 
 import asyncio
-from collections.abc import Mapping
+from collections.abc import Callable, Mapping
 from dataclasses import dataclass
 from types import MappingProxyType, SimpleNamespace
 from typing import Final
@@ -15,12 +16,14 @@ from starlette.websockets import WebSocketDisconnect
 
 import litellm
 from litellm.caching.dual_cache import DualCache
+from litellm.llms.deepgram.common_utils import deepgram_speak_characters
 from litellm.proxy._lazy_features import LAZY_FEATURES
 from litellm.proxy._types import LiteLLMRoutes, UserAPIKeyAuth
 from litellm.proxy.auth.auth_checks import cache_key_object
 from litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints import (
     _websocket_relay,
     deepgram_listen_websocket_route,
+    deepgram_speak_websocket_route,
     router,
 )
 from litellm.proxy.utils import hash_token
@@ -30,6 +33,7 @@ GET_CREDENTIALS: Final = (
 )
 USER_API_KEY_AUTH: Final = "litellm.proxy.auth.user_api_key_auth.user_api_key_auth"
 LISTEN_PATHS: Final = ("/deepgram/v1/listen", "/deepgram/listen")
+SPEAK_PATHS: Final = ("/deepgram/v1/speak", "/deepgram/speak")
 NOVA_2_STREAMING_KEY: Final = "deepgram/streaming/nova-2"
 
 pytestmark: Final = pytest.mark.usefixtures("local_model_cost_map")
@@ -62,6 +66,7 @@ class _RelayCall:
     forward_headers: bool
     endpoint: str
     accept_websocket: bool
+    count_client_text: Callable[[str | bytes], int] | None = None
 
 
 class _FakeRelay:
@@ -78,6 +83,7 @@ class _FakeRelay:
         forward_headers: bool,
         endpoint: str,
         accept_websocket: bool,
+        count_client_text: Callable[[str | bytes], int] | None = None,
     ) -> None:
         self.calls.append(
             _RelayCall(
@@ -87,6 +93,7 @@ class _FakeRelay:
                 forward_headers=forward_headers,
                 endpoint=endpoint,
                 accept_websocket=accept_websocket,
+                count_client_text=count_client_text,
             )
         )
 
@@ -101,13 +108,13 @@ async def _serve(websocket: _FakeWebSocket, user_api_key_dict: UserAPIKeyAuth |
     return relay
 
 
-def test_deepgram_listen_websocket_routes_registered():
+def test_deepgram_websocket_routes_registered():
     ws_paths = {route.path for route in router.routes if isinstance(route, WebSocketRoute)}
-    assert set(LISTEN_PATHS) <= ws_paths
+    assert {*LISTEN_PATHS, *SPEAK_PATHS} <= ws_paths
 
 
-@pytest.mark.parametrize("path", LISTEN_PATHS)
-def test_deepgram_listen_is_a_lazily_loaded_mapped_pass_through_route(path):
+@pytest.mark.parametrize("path", (*LISTEN_PATHS, *SPEAK_PATHS))
+def test_deepgram_websocket_routes_are_lazily_loaded_mapped_pass_through_routes(path):
     """The route must be reachable before the passthrough module is imported and must be authed and
     billed as a mapped pass-through route like the other provider prefixes."""
     feature = next(feature for feature in LAZY_FEATURES if feature.name == "llm_passthrough")
@@ -472,3 +479,198 @@ def test_deepgram_listen_echoes_the_browser_subprotocol_that_carries_the_litellm
         MappingProxyType({"Authorization": "Token dg-provider-key"})
     ]
     assert [call.forward_headers for call in relay.calls] == [False]
+
+
+async def _serve_speak(websocket: _FakeWebSocket, user_api_key_dict: UserAPIKeyAuth | None = None) -> _FakeRelay:
+    relay: Final = _FakeRelay()
+    await deepgram_speak_websocket_route(
+        websocket=websocket,
+        user_api_key_dict=user_api_key_dict or UserAPIKeyAuth(),
+        relay=relay,
+    )
+    return relay
+
+
+@pytest.mark.parametrize("path", SPEAK_PATHS)
+async def test_deepgram_speak_forwards_query_injects_only_provider_auth_and_counts_characters(path, monkeypatch):
+    monkeypatch.delenv("DEEPGRAM_API_BASE", raising=False)
+    websocket: Final = _FakeWebSocket(path, "model=aura-2-thalia-en&encoding=linear16&sample_rate=24000")
+    caller: Final = UserAPIKeyAuth(api_key="sk-litellm-virtual", team_id="team-tts")
+
+    with patch(GET_CREDENTIALS, return_value="dg-provider-key") as get_credentials:
+        relay: Final = await _serve_speak(websocket, caller)
+
+    assert get_credentials.call_args.kwargs == {"custom_llm_provider": "deepgram", "region_name": None}
+    assert relay.calls == [
+        _RelayCall(
+            target="wss://api.deepgram.com/v1/speak?model=aura-2-thalia-en&encoding=linear16&sample_rate=24000",
+            custom_headers=MappingProxyType({"Authorization": "Token dg-provider-key"}),
+            user_api_key_dict=caller,
+            forward_headers=False,
+            endpoint=path,
+            accept_websocket=False,
+            count_client_text=deepgram_speak_characters,
+        )
+    ]
+    assert websocket.accepts == [None]
+    assert websocket.closed is None
+
+
+async def test_deepgram_listen_does_not_count_client_text(monkeypatch):
+    """Listen bills audio seconds from upstream frames, so it must keep handing the relay no counter"""
+    monkeypatch.delenv("DEEPGRAM_API_BASE", raising=False)
+
+    with patch(GET_CREDENTIALS, return_value="dg-provider-key"):
+        relay: Final = await _serve(_FakeWebSocket("/deepgram/v1/listen", "model=nova-3"))
+
+    assert [call.count_client_text for call in relay.calls] == [None]
+
+
+@pytest.mark.parametrize(
+    ("query", "expected_target"),
+    [
+        pytest.param("", "wss://api.deepgram.com/v1/speak?model=aura-asteria-en", id="no query"),
+        pytest.param("model=", "wss://api.deepgram.com/v1/speak?model=aura-asteria-en", id="blank model"),
+        pytest.param(
+            "encoding=mulaw&sample_rate=8000",
+            "wss://api.deepgram.com/v1/speak?encoding=mulaw&sample_rate=8000&model=aura-asteria-en",
+            id="other params kept",
+        ),
+        pytest.param(
+            "model=aura-2-thalia-en&model=aura-asteria-en",
+            "wss://api.deepgram.com/v1/speak?model=aura-2-thalia-en",
+            id="only the first model",
+        ),
+    ],
+)
+async def test_deepgram_speak_names_exactly_one_model_upstream(query, expected_target, monkeypatch):
+    monkeypatch.delenv("DEEPGRAM_API_BASE", raising=False)
+
+    with patch(GET_CREDENTIALS, return_value="dg-provider-key"):
+        relay: Final = await _serve_speak(_FakeWebSocket("/deepgram/speak", query))
+
+    assert [call.target for call in relay.calls] == [expected_target]
+
+
+async def test_deepgram_speak_honours_server_configured_api_base_and_ignores_the_callers(monkeypatch):
+    """The server's Deepgram key must only ever go to the server-configured host"""
+    monkeypatch.setenv("DEEPGRAM_API_BASE", "https://api.eu.deepgram.com/v1/")
+    websocket: Final = _FakeWebSocket(
+        "/deepgram/v1/speak", "api_base=wss%3A%2F%2Fattacker.example%2Fv1&model=aura-2-thalia-en"
+    )
+
+    with patch(GET_CREDENTIALS, return_value="dg-provider-key"):
+        relay: Final = await _serve_speak(websocket)
+
+    assert [call.target for call in relay.calls] == [
+        "wss://api.eu.deepgram.com/v1/speak?api_base=wss%3A%2F%2Fattacker.example%2Fv1&model=aura-2-thalia-en"
+    ]
+
+
+async def test_deepgram_speak_closes_before_accepting_when_provider_credentials_missing():
+    websocket: Final = _FakeWebSocket("/deepgram/v1/speak", "model=aura-2-thalia-en")
+
+    with patch(GET_CREDENTIALS, return_value=None):
+        relay: Final = await _serve_speak(websocket)
+
+    assert websocket.closed is not None
+    assert websocket.closed[0] == 1011
+    assert "DEEPGRAM_API_KEY" in websocket.closed[1]
+    assert websocket.accepts == []
+    assert relay.calls == []
+
+
+@pytest.mark.parametrize(
+    ("query", "missing_key", "removed_row"),
+    [
+        pytest.param("model=flux-tts-en", "deepgram/flux-tts-en", None, id="voice outside the priced families"),
+        pytest.param("model=aura-3-x-en", "deepgram/aura-3-x-en", None, id="unknown aura generation"),
+        pytest.param("model=aura-x", "deepgram/aura-x", None, id="aura voice without a language"),
+        pytest.param("model=nova-3", "deepgram/nova-3", None, id="transcription model with a per-second row"),
+        pytest.param("model=aura-2-thalia-en", "deepgram/aura-2", "deepgram/aura-2", id="family row removed"),
+    ],
+)
+async def test_deepgram_speak_refuses_sessions_it_cannot_price(query, missing_key, removed_row, monkeypatch):
+    """A session without a per-character price would be logged at no cost, so the proxy closes it before Deepgram is
+    contacted and names the registry row to add"""
+    monkeypatch.delenv("DEEPGRAM_API_BASE", raising=False)
+    if removed_row is not None:
+        monkeypatch.delitem(litellm.model_cost, removed_row)
+    websocket: Final = _FakeWebSocket("/deepgram/v1/speak", query)
+
+    with patch(GET_CREDENTIALS, return_value="dg-provider-key"):
+        relay: Final = await _serve_speak(websocket)
+
+    assert relay.calls == []
+    assert websocket.accepts == [None]
+    assert websocket.closed is not None
+    assert websocket.closed[0] == 1008
+    assert missing_key in websocket.closed[1]
+    assert "dg-provider-key" not in websocket.closed[1]
+
+
+def test_deepgram_speak_rejects_connections_without_a_litellm_key():
+    relay: Final = _FakeRelay()
+    client: Final = TestClient(_app_with_relay(relay))
+
+    with patch(GET_CREDENTIALS, return_value="dg-provider-key") as get_credentials:
+        with pytest.raises(WebSocketDisconnect) as disconnect:
+            with client.websocket_connect("/deepgram/v1/speak?model=aura-2-thalia-en"):
+                pass
+
+    assert disconnect.value.code == 1008
+    assert relay.calls == []
+    get_credentials.assert_not_called()
+
+
+@pytest.mark.parametrize(
+    ("query", "expected_target"),
+    [
+        pytest.param(
+            "model=aura-2-thalia-en",
+            "wss://api.deepgram.com/v1/speak?model=aura-2-thalia-en",
+            id="allowed voice named",
+        ),
+        pytest.param(
+            "model=aura-2-thalia-en&model=aura-2-zeus-en",
+            "wss://api.deepgram.com/v1/speak?model=aura-2-thalia-en",
+            id="second voice stripped",
+        ),
+        pytest.param("model=aura-2-zeus-en", None, id="denied voice named"),
+        pytest.param("", None, id="voice omitted, default denied"),
+        pytest.param("model=&encoding=mulaw", None, id="voice blank, default denied"),
+    ],
+)
+def test_deepgram_speak_authorizes_the_voice_it_will_actually_send_upstream(query, expected_target, monkeypatch):
+    """A key allowed only one voice must not reach another by leaving ``model`` out or naming a second one, because
+    the real key auth path must see the same model the upstream target carries"""
+    monkeypatch.delenv("DEEPGRAM_API_BASE", raising=False)
+    monkeypatch.setattr(litellm, "max_budget", 0.0)
+    cache: Final = asyncio.run(_cache_restricted_key("sk-only-thalia", ["aura-2-thalia-en"]))
+    relay: Final = _FakeRelay()
+    client: Final = TestClient(_app_with_relay(relay))
+    url: Final = f"/deepgram/v1/speak?{query}"
+    headers: Final = {"Authorization": "Bearer sk-only-thalia"}
+
+    with (
+        patch(GET_CREDENTIALS, return_value="dg-provider-key"),
+        patch.multiple(  # test-quality-ok: the real key auth path reads these proxy_server globals and has no injection seam
+            "litellm.proxy.proxy_server",
+            master_key="sk-master",
+            prisma_client=MagicMock(),
+            user_api_key_cache=cache,
+            llm_model_list=None,
+            llm_router=None,
+        ),
+    ):
+        if expected_target is not None:
+            with client.websocket_connect(url, headers=headers):
+                pass
+            assert [call.target for call in relay.calls] == [expected_target]
+            return
+        with pytest.raises(WebSocketDisconnect) as disconnect:
+            with client.websocket_connect(url, headers=headers):
+                pass
+
+    assert disconnect.value.code == 1008
+    assert relay.calls == []
diff --git a/tests/unit/proxy/pass_through_endpoints/test_pass_through_endpoints.py b/tests/unit/proxy/pass_through_endpoints/test_pass_through_endpoints.py
index a797f8084b9..abb59bc3b69 100644
--- a/tests/unit/proxy/pass_through_endpoints/test_pass_through_endpoints.py
+++ b/tests/unit/proxy/pass_through_endpoints/test_pass_through_endpoints.py
@@ -34,6 +34,7 @@ from litellm.constants import DEFAULT_REQUEST_TIMEOUT_SECONDS
 from litellm.integrations.custom_guardrail import CustomGuardrail
 from litellm.integrations.custom_logger import CustomLogger
 from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
+from litellm.llms.deepgram.common_utils import deepgram_speak_characters
 from litellm.proxy._lazy_features import LazyFeature, attach_lazy_features
 from litellm.proxy._types import ProxyException, UserAPIKeyAuth
 from litellm.proxy.litellm_pre_call_utils import LiteLLMProxyRequestSetup
@@ -7954,6 +7955,66 @@ async def test_websocket_passthrough_counts_client_text_when_the_upstream_ends_w
     }
 
 
+class _CapturingSpendLogger(CustomLogger):
+    def __init__(self) -> None:
+        super().__init__()
+        self.payloads: list[StandardLoggingPayload] = []  # mutable-ok: the logger appends each payload it receives
+
+    async def async_log_success_event(self, kwargs, response_obj, start_time, end_time):
+        self.payloads.append(kwargs["standard_logging_object"])
+
+
+@pytest.mark.asyncio
+@pytest.mark.usefixtures("local_model_cost_map")
+async def test_deepgram_speak_session_logs_character_spend_once_and_never_the_spoken_text(monkeypatch):
+    """Runs the relay with the speak counter and the real success handler, so the spend logger sees what production
+    writes for one Speak message"""
+    capturing_logger = _CapturingSpendLogger()
+    monkeypatch.setattr(litellm, "_async_success_callback", [capturing_logger])
+    monkeypatch.setattr(litellm, "success_callback", [])
+    monkeypatch.setattr(litellm, "callbacks", [])
+    queued: list[Coroutine[object, object, None]] = []  # mutable-ok: the fake logging worker keeps each coroutine
+    upstream_ws = SpeakingUpstreamWebSocket()
+
+    with (
+        patch("litellm.proxy.proxy_server.proxy_logging_obj") as proxy_logging,
+        patch(
+            "litellm.proxy.pass_through_endpoints.pass_through_endpoints.connect",
+            return_value=FakeUpstreamConnect(upstream_ws),
+        ),
+        patch("litellm.proxy.pass_through_endpoints.pass_through_endpoints.GLOBAL_LOGGING_WORKER") as worker,
+    ):
+        proxy_logging.pre_call_hook = AsyncMock(return_value={})
+        proxy_logging.post_call_success_hook = AsyncMock()
+        proxy_logging.post_call_failure_hook = AsyncMock()
+        worker.ensure_initialized_and_enqueue = MagicMock(
+            side_effect=lambda async_coroutine: queued.append(async_coroutine)
+        )
+        await asyncio.wait_for(
+            websocket_passthrough_request(
+                websocket=_speak_client(),
+                target=DEEPGRAM_SPEAK_TARGET,
+                custom_headers={"Authorization": "Token dg-provider-key"},
+                user_api_key_dict=UserAPIKeyAuth(api_key="hashed-key", team_id="team-tts"),
+                forward_headers=False,
+                endpoint="/deepgram/v1/speak",
+                accept_websocket=False,
+                count_client_text=deepgram_speak_characters,
+            ),
+            timeout=RELAY_HANG_GUARD_SECONDS,
+        )
+    for coroutine in queued:
+        await coroutine
+
+    assert len(capturing_logger.payloads) == 1
+    payload = capturing_logger.payloads[0]
+    assert payload["model"] == "aura-2-thalia-en"
+    assert payload["custom_llm_provider"] == "deepgram"
+    # Aura-2 pay as you go rate, $0.030 per 1,000 characters, from https://deepgram.com/pricing, checked 2026-10-26
+    assert payload["response_cost"] == pytest.approx(len(SPOKEN_TEXT) * 0.030 / 1000)
+    assert SPOKEN_TEXT not in json.dumps(payload, default=str)
+
+
 @pytest.mark.asyncio
 async def test_passthrough_success_reconciles_budget_reservation():
     """