This commit is contained in:
shrey 2026-10-29 00:36:26 +00:00 • committed by GitHub
commit 9e5a0b0b5e
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
14 changed files with 901 additions and 45 deletions

View file

@ -423,6 +423,8 @@ 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"
# https://developers.deepgram.com/reference/text-to-speech/speak-streaming, checked 2026-10-28
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"

View file

@ -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"})
@ -34,8 +41,12 @@ class DeepgramException(BaseLLMException):
pass
def deepgram_requested_model(query_string: str, default_model: str) -> str:
return httpx.QueryParams(query_string).get("model") or default_model
def deepgram_listen_requested_model(query_string: str) -> str:
return httpx.QueryParams(query_string).get("model") or DEEPGRAM_LISTEN_DEFAULT_MODEL
return deepgram_requested_model(query_string, DEEPGRAM_LISTEN_DEFAULT_MODEL)
def _first_occurrences(query_string: str) -> httpx.QueryParams:
@ -50,21 +61,29 @@ def _first_occurrences(query_string: str) -> httpx.QueryParams:
)
def deepgram_listen_websocket_target(api_base: str | None, query_string: str) -> str:
listen_url: Final = httpx.URL(join_api_base(api_base or DEEPGRAM_DEFAULT_API_BASE, "listen"))
websocket_url: Final = listen_url.copy_with(scheme=_WEBSOCKET_SCHEMES.get(listen_url.scheme, listen_url.scheme))
def deepgram_websocket_target(api_base: str | None, path: str, query_string: str, default_model: str) -> str:
http_url: Final = httpx.URL(join_api_base(api_base or DEEPGRAM_DEFAULT_API_BASE, path))
websocket_url: Final = http_url.copy_with(scheme=_WEBSOCKET_SCHEMES.get(http_url.scheme, http_url.scheme))
params: Final = _first_occurrences(query_string)
query: Final = params if params.get("model") else params.remove("model").add("model", DEEPGRAM_LISTEN_DEFAULT_MODEL)
query: Final = params if params.get("model") else params.remove("model").add("model", default_model)
return f"{websocket_url}?{query}"
def deepgram_listen_websocket_target(api_base: str | None, query_string: str) -> str:
return deepgram_websocket_target(api_base, "listen", query_string, DEEPGRAM_LISTEN_DEFAULT_MODEL)
def deepgram_listen_callback_params(query_string: str) -> tuple[str, ...]:
return tuple(sorted(DEEPGRAM_LISTEN_CALLBACK_PARAMS.intersection(httpx.QueryParams(query_string).keys())))
def deepgram_listen_model(upstream_url: str) -> str:
def deepgram_upstream_model(upstream_url: str, default_model: str) -> str:
models: Final = parse_qs(urlparse(upstream_url).query).get("model")
return models[0] if models else DEEPGRAM_LISTEN_DEFAULT_MODEL
return models[0] if models else default_model
def deepgram_listen_model(upstream_url: str) -> str:
return deepgram_upstream_model(upstream_url, DEEPGRAM_LISTEN_DEFAULT_MODEL)
def _param_enabled(values: Sequence[str]) -> bool:
@ -89,12 +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."""
registry_key: Final = deepgram_listen_registry_key(upstream_url)
return _exact_registry_entry(deepgram_listen_registry_key(upstream_url)) is not None
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, ...]:
@ -199,3 +221,34 @@ 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_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_upstream_model(upstream_url, DEEPGRAM_SPEAK_DEFAULT_MODEL)
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")
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

View file

@ -22775,6 +22775,30 @@
"output_cost_per_token_batches": 1e-06,
"source": "https://developers.openai.com/api/docs/pricing"
},
"deepgram/aura": {
"input_cost_per_character": 1.5e-05,
"litellm_provider": "deepgram",
"metadata": {
"calculation": "$0.0150/1000 characters = $0.000015 per character",
"note": "Aura-1 text-to-speech voices (aura-<voice>-<lang>), pay as you go"
},
"source": "https://deepgram.com/pricing",
"supported_endpoints": [
"/v1/speak"
]
},
"deepgram/aura-2": {
"input_cost_per_character": 3e-05,
"litellm_provider": "deepgram",
"metadata": {
"calculation": "$0.030/1000 characters = $0.00003 per character",
"note": "Aura-2 text-to-speech voices (aura-2-<voice>-<lang>), pay as you go"
},
"source": "https://deepgram.com/pricing",
"supported_endpoints": [
"/v1/speak"
]
},
"deepgram/base": {
"input_cost_per_second": 0.00020833,
"litellm_provider": "deepgram",

View file

@ -44,6 +44,7 @@ from litellm.constants import (
AZURE_SPEECH_STT_DOMAIN,
AZURE_SPEECH_SUBSCRIPTION_KEY_HEADER,
BEDROCK_AGENT_RUNTIME_PASS_THROUGH_ROUTES,
DEEPGRAM_SPEAK_DEFAULT_MODEL,
PASS_THROUGH_HEADER_PREFIX,
TYPESAFE_DANGER_ZONE_HEADER_PREFIX,
)
@ -63,6 +64,11 @@ from litellm.llms.deepgram.common_utils import (
deepgram_listen_registry_key,
deepgram_listen_requested_model,
deepgram_listen_websocket_target,
deepgram_requested_model,
deepgram_speak_character_rate,
deepgram_speak_characters,
deepgram_speak_registry_key,
deepgram_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 +3360,7 @@ class _WebsocketRelay(Protocol):
forward_headers: bool,
endpoint: str,
accept_websocket: bool,
count_client_text: Callable[[str | bytes], int] | None = None,
) -> None: ...
@ -3469,12 +3476,22 @@ async def deepgram_listen_user_api_key_auth(websocket: WebSocket) -> UserAPIKeyA
)
@router.websocket("/deepgram/v1/listen")
@router.websocket("/deepgram/listen")
async def deepgram_listen_websocket_route(
def _deepgram_listen_refusal(target: str) -> str | None:
callback_params: Final = deepgram_listen_callback_params(httpx.URL(target).query.decode())
if callback_params:
return _DEEPGRAM_WS_CALLBACK_REASON.format(params=", ".join(callback_params))
if deepgram_listen_is_priced(target):
return None
return _DEEPGRAM_WS_UNPRICED_REASON.format(registry_key=deepgram_listen_registry_key(target))
async def _relay_deepgram_websocket(
websocket: WebSocket,
user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(deepgram_listen_user_api_key_auth)],
relay: Annotated[_WebsocketRelay, Depends(_websocket_relay)],
user_api_key_dict: UserAPIKeyAuth,
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,
@ -3485,23 +3502,8 @@ async def deepgram_listen_websocket_route(
return
await websocket.accept(subprotocol=_negotiated_websocket_subprotocol(websocket))
callback_params: Final = deepgram_listen_callback_params(websocket.url.query)
if callback_params:
await websocket.close(
code=1008,
reason=_DEEPGRAM_WS_CALLBACK_REASON.format(params=", ".join(callback_params)),
)
return
target: Final = deepgram_listen_websocket_target(
api_base=get_secret_str("DEEPGRAM_API_BASE"),
query_string=websocket.url.query,
)
if not deepgram_listen_is_priced(target):
await websocket.close(
code=1008,
reason=_DEEPGRAM_WS_UNPRICED_REASON.format(registry_key=deepgram_listen_registry_key(target)),
)
if refusal is not None:
await websocket.close(code=1008, reason=refusal)
return
await relay(
@ -3512,6 +3514,50 @@ async def deepgram_listen_websocket_route(
forward_headers=False,
endpoint=websocket.url.path,
accept_websocket=False,
count_client_text=count_client_text,
)
@router.websocket("/deepgram/v1/listen")
@router.websocket("/deepgram/listen")
async def deepgram_listen_websocket_route(
websocket: WebSocket,
user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(deepgram_listen_user_api_key_auth)],
relay: Annotated[_WebsocketRelay, Depends(_websocket_relay)],
) -> None:
target: Final = deepgram_listen_websocket_target(get_secret_str("DEEPGRAM_API_BASE"), websocket.url.query)
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_requested_model(websocket.url.query, DEEPGRAM_SPEAK_DEFAULT_MODEL)
)
def _deepgram_speak_refusal(target: str) -> str | None:
if deepgram_speak_character_rate(target) is not None:
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_websocket_target(
get_secret_str("DEEPGRAM_API_BASE"), "speak", websocket.url.query, DEEPGRAM_SPEAK_DEFAULT_MODEL
)
await _relay_deepgram_websocket(
websocket,
user_api_key_dict,
relay,
target,
_deepgram_speak_refusal(target),
count_client_text=deepgram_speak_characters,
)

View file

@ -0,0 +1,35 @@
import json
from typing import Final
from urllib.parse import urlparse
import litellm
from litellm.constants import DEEPGRAM_SPEAK_DEFAULT_MODEL
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.llms.deepgram.common_utils import deepgram_speak_character_rate, deepgram_upstream_model
from litellm.types.passthrough_endpoints.pass_through_endpoints import WEBSOCKET_CLIENT_TEXT_COUNT_KEY
from litellm.types.utils import StandardPassThroughResponseObject
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("/speak")
@staticmethod
def deepgram_speak_passthrough_handler(
logging_obj: LiteLLMLoggingObj, upstream_url: str
) -> StandardPassThroughResponseObject:
counted: Final = logging_obj.model_call_details.get(WEBSOCKET_CLIENT_TEXT_COUNT_KEY)
characters: Final = counted if isinstance(counted, int) else 0
rate: Final = deepgram_speak_character_rate(upstream_url)
model: Final = deepgram_upstream_model(upstream_url, DEEPGRAM_SPEAK_DEFAULT_MODEL)
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"] = litellm.LlmProviders.DEEPGRAM.value # rebind-ok: same
logging_obj.model_call_details["response_cost"] = None if rate is None else characters * rate # rebind-ok: same
return StandardPassThroughResponseObject(response=json.dumps({"characters": characters}))

View file

@ -140,6 +140,7 @@ from litellm.types.passthrough_endpoints.pass_through_endpoints import (
LITELLM_PASS_THROUGH_DEPLOYMENT_MODEL_INFO_STATE_KEY,
LITELLM_PASS_THROUGH_ENDPOINT_MARKER,
LITELLM_PASS_THROUGH_RAW_BODY_STATE_KEY,
WEBSOCKET_CLIENT_TEXT_COUNT_KEY,
EndpointType,
PassthroughStandardLoggingPayload,
)
@ -2508,6 +2509,7 @@ async def websocket_passthrough_request(
cost_per_request: float | None = None,
accept_websocket: bool = True,
setup_model_rewriter: Callable[[str], str] | None = None,
count_client_text: Callable[[str | bytes], int] | None = None,
):
"""
WebSocket passthrough request handler.
@ -2521,6 +2523,10 @@ async def websocket_passthrough_request(
endpoint: The endpoint path (for logging purposes)
cost_per_request: Optional field - cost per request to the target endpoint
setup_model_rewriter: Optional rewrite of the setup frame's model before it reaches the upstream
count_client_text: Optional read-only count of each client frame, text or binary, after it reaches the
upstream, such as billable characters. Only the sum reaches the success handler, under
WEBSOCKET_CLIENT_TEXT_COUNT_KEY in logging_obj.model_call_details, so routes whose client text must not be
stored can still bill it
"""
from litellm.litellm_core_utils.litellm_logging import Logging
from litellm.proxy.proxy_server import proxy_config, proxy_logging_obj
@ -2531,6 +2537,7 @@ async def websocket_passthrough_request(
# Initialize tracking variables
start_time: Final = datetime.now()
websocket_messages: Final[list[dict[str, object]]] = []
client_text_count: int = 0 # rebind-ok: forward_client_to_upstream adds each client frame's count
litellm_call_id: Final = str(uuid.uuid4())
verbose_proxy_logger.info("WebSocket passthrough (%s): Starting WebSocket connection to %s", endpoint, target)
@ -2647,6 +2654,16 @@ async def websocket_passthrough_request(
"WebSocket passthrough (%s): Upstream connection established successfully", endpoint
)
async def send_upstream(frame: str | bytes) -> None:
nonlocal client_text_count
billed: Final = 0 if count_client_text is None else count_client_text(frame)
try:
await upstream_ws.send(frame)
except asyncio.CancelledError:
client_text_count += billed # rebind-ok: the frame can already be on the wire when the cancel lands
raise
client_text_count += billed # rebind-ok: the session total grows with each delivered frame
async def forward_client_to_upstream() -> None:
"""Forward messages from client to upstream WebSocket"""
try:
@ -2717,9 +2734,9 @@ async def websocket_passthrough_request(
)
# Not a JSON message or doesn't contain setup data
await upstream_ws.send(_rewrite_vertex_live_setup_model(text_data, setup_model_rewriter))
await send_upstream(_rewrite_vertex_live_setup_model(text_data, setup_model_rewriter))
elif bytes_data is not None:
await upstream_ws.send(bytes_data)
await send_upstream(bytes_data)
except asyncio.CancelledError:
raise
except Exception:
@ -2845,6 +2862,8 @@ async def websocket_passthrough_request(
mock_response: Final = MockWebSocketResponse(target)
if count_client_text is not None:
logging_obj.model_call_details[WEBSOCKET_CLIENT_TEXT_COUNT_KEY] = client_text_count
# Use the same success handler as HTTP passthrough endpoints
GLOBAL_LOGGING_WORKER.ensure_initialized_and_enqueue(
async_coroutine=pass_through_endpoint_logging.pass_through_async_success_handler(

View file

@ -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,12 @@ 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):
standard_logging_response_object = ( # rebind-ok: elif-chain
DeepgramSpeakPassthroughLoggingHandler.deepgram_speak_passthrough_handler(
logging_obj=logging_obj, upstream_url=str(httpx_response.request.url)
)
)
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({}),

View file

@ -27,6 +27,8 @@ LITELLM_PASS_THROUGH_ENDPOINT_MARKER: Final = "__litellm_pass_through_endpoint__
SETTLES_BACKGROUND_RESPONSE_KWARG: Final = "settles_background_response"
WEBSOCKET_CLIENT_TEXT_COUNT_KEY: Final = "litellm_websocket_client_text_count"
class BackgroundResponseSettlement(str, Enum):
SETTLES = "settles"

View file

@ -22775,6 +22775,30 @@
"output_cost_per_token_batches": 1e-06,
"source": "https://developers.openai.com/api/docs/pricing"
},
"deepgram/aura": {
"input_cost_per_character": 1.5e-05,
"litellm_provider": "deepgram",
"metadata": {
"calculation": "$0.0150/1000 characters = $0.000015 per character",
"note": "Aura-1 text-to-speech voices (aura-<voice>-<lang>), pay as you go"
},
"source": "https://deepgram.com/pricing",
"supported_endpoints": [
"/v1/speak"
]
},
"deepgram/aura-2": {
"input_cost_per_character": 3e-05,
"litellm_provider": "deepgram",
"metadata": {
"calculation": "$0.030/1000 characters = $0.00003 per character",
"note": "Aura-2 text-to-speech voices (aura-2-<voice>-<lang>), pay as you go"
},
"source": "https://deepgram.com/pricing",
"supported_endpoints": [
"/v1/speak"
]
},
"deepgram/base": {
"input_cost_per_second": 0.00020833,
"litellm_provider": "deepgram",

View file

@ -1,3 +1,4 @@
import json
import math
from collections.abc import Mapping, Sequence
from typing import Final
@ -6,6 +7,7 @@ from urllib.parse import parse_qs, urlparse
import pytest
import litellm
from litellm.constants import DEEPGRAM_SPEAK_DEFAULT_MODEL
from litellm.llms.deepgram.common_utils import (
deepgram_listen_addon_pricing_models,
deepgram_listen_audio_seconds,
@ -18,6 +20,12 @@ from litellm.llms.deepgram.common_utils import (
deepgram_listen_requested_model,
deepgram_listen_transcript,
deepgram_listen_websocket_target,
deepgram_requested_model,
deepgram_speak_character_rate,
deepgram_speak_characters,
deepgram_speak_registry_key,
deepgram_upstream_model,
deepgram_websocket_target,
)
NOVA_3_URL: Final = "wss://api.deepgram.com/v1/listen?model=nova-3&encoding=linear16&sample_rate=16000"
@ -324,3 +332,133 @@ def test_deepgram_listen_is_priced(
)
def test_deepgram_listen_addon_pricing_models(upstream_url: str, expected: tuple[str, ...]):
assert deepgram_listen_addon_pricing_models(upstream_url) == expected
@pytest.mark.usefixtures("local_model_cost_map")
@pytest.mark.parametrize(
("pricing_model", "usd_per_thousand_characters"),
[
pytest.param("aura-2", 0.030, id="aura-2"),
pytest.param("aura", 0.015, id="aura-1"),
],
)
def test_aura_rows_bill_speech_per_character_at_the_published_rate(
pricing_model: str, usd_per_thousand_characters: float
) -> None:
"""Pay as you go rates from https://deepgram.com/pricing, checked 2026-10-26"""
prompt_cost, completion_cost = litellm.cost_per_token(
model=pricing_model, custom_llm_provider="deepgram", call_type="speech", prompt_characters=1000
)
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) -> None:
assert deepgram_websocket_target(api_base, "speak", query_string, DEEPGRAM_SPEAK_DEFAULT_MODEL) == 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) -> None:
"""The key is checked against deepgram_requested_model, so the upstream must use exactly that voice"""
requested: Final = deepgram_requested_model(query_string, DEEPGRAM_SPEAK_DEFAULT_MODEL)
target: Final = deepgram_websocket_target(None, "speak", query_string, DEEPGRAM_SPEAK_DEFAULT_MODEL)
assert parse_qs(urlparse(target).query)["model"] == [requested]
assert deepgram_upstream_model(target, DEEPGRAM_SPEAK_DEFAULT_MODEL) == requested
@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-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
) -> None:
assert deepgram_speak_registry_key(f"{SPEAK_BASE}?model={model}") == f"deepgram/{expected_family}"
@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) -> None:
"""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
@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) -> None:
assert deepgram_speak_characters(frame) == expected

View file

@ -0,0 +1,89 @@
"""Deepgram ``/v1/speak`` WebSocket passthrough: per-character cost tracking from the relay's client text count."""
import json
from datetime import datetime
from typing import Final
import pytest
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.types.passthrough_endpoints.pass_through_endpoints import WEBSOCKET_CLIENT_TEXT_COUNT_KEY
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) -> None:
assert DeepgramSpeakPassthroughLoggingHandler.is_deepgram_speak_route(url_route, request_method) is expected
def _logging_obj(characters: object) -> LiteLLMLoggingObj:
logging_obj: Final = LiteLLMLoggingObj(
model="unknown",
messages=[{"role": "user", "content": "WebSocket connection"}],
stream=True,
call_type="pass_through_endpoint",
start_time=datetime.now(),
litellm_call_id="call-dg-speak",
function_id="websocket_passthrough",
)
logging_obj.model_call_details[WEBSOCKET_CLIENT_TEXT_COUNT_KEY] = characters
return logging_obj
def _billed_cost(upstream_url: str, characters: object) -> object:
logging_obj: Final = _logging_obj(characters)
DeepgramSpeakPassthroughLoggingHandler.deepgram_speak_passthrough_handler(logging_obj, upstream_url)
return logging_obj.model_call_details["response_cost"]
def test_handler_bills_spoken_characters_at_the_aura_2_rate_and_logs_only_the_count() -> None:
logging_obj: Final = _logging_obj(SPOKEN_CHARACTERS)
result: Final = DeepgramSpeakPassthroughLoggingHandler.deepgram_speak_passthrough_handler(logging_obj, AURA_2_URL)
assert logging_obj.model_call_details["response_cost"] == pytest.approx(
SPOKEN_CHARACTERS * AURA_2_USD_PER_CHARACTER
)
assert logging_obj.model == logging_obj.model_call_details["model"] == "aura-2-thalia-en"
assert logging_obj.model_call_details["custom_llm_provider"] == "deepgram"
assert result == {"response": json.dumps({"characters": SPOKEN_CHARACTERS})}
def test_handler_bills_an_aura_1_voice_at_the_aura_1_rate() -> None:
assert _billed_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) -> None:
assert _billed_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) -> None:
assert _billed_cost(f"wss://api.deepgram.com/v1/speak?model={model}", 1000) is None

View file

@ -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() -> None:
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: str) -> None:
"""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,180 @@ 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: str, monkeypatch: pytest.MonkeyPatch
) -> None:
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: pytest.MonkeyPatch) -> None:
"""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]
async def test_deepgram_speak_honours_server_configured_api_base_and_ignores_the_callers(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""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() -> None:
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: str, missing_key: str, removed_row: str | None, monkeypatch: pytest.MonkeyPatch
) -> None:
"""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() -> None:
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: str, expected_target: str | None, monkeypatch: pytest.MonkeyPatch
) -> None:
"""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 == []

View file

@ -7,7 +7,7 @@ import os
import sys
import uuid
import zlib
from collections.abc import AsyncIterator, Callable, Mapping
from collections.abc import AsyncIterator, Callable, Coroutine, Mapping
from contextlib import ExitStack, contextmanager
from dataclasses import dataclass
from datetime import datetime
@ -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
@ -69,6 +70,7 @@ from litellm.types.passthrough_endpoints.pass_through_endpoints import (
EndpointType,
PassthroughStandardLoggingPayload,
)
from litellm.types.utils import StandardLoggingPayload
from tests._master_key import MASTER_KEY
from tests._vcr_conftest_common import install_live_call_probe, record_vcr_outcome
@ -7715,6 +7717,66 @@ async def test_websocket_passthrough_sends_deepgram_audio_bytes_and_control_text
upstream_ws.close.assert_awaited_once()
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) -> None:
self.payloads.append(kwargs["standard_logging_object"])
@dataclass(frozen=True, slots=True)
class _RelayedSession:
spend_logs: tuple[StandardLoggingPayload, ...]
failure_hook: AsyncMock
async def _relay_and_capture_spend(
monkeypatch: pytest.MonkeyPatch,
upstream_ws: "FakeUpstreamWebSocket | SpeakingUpstreamWebSocket",
websocket: MagicMock,
target: str,
endpoint: str,
count_client_text: Callable[[str | bytes], int] | None = None,
) -> _RelayedSession:
"""Runs the relay with the real success handler and returns what a spend logger receives"""
capturing_logger: Final = _CapturingSpendLogger()
monkeypatch.setattr(litellm, "_async_success_callback", [capturing_logger])
monkeypatch.setattr(litellm, "success_callback", [])
monkeypatch.setattr(litellm, "callbacks", [])
queued: Final[list[Coroutine[object, object, None]]] = [] # mutable-ok: the fake worker keeps each coroutine
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 websocket_passthrough_request(
websocket=websocket,
target=target,
custom_headers={"Authorization": "Token dg-provider-key"},
user_api_key_dict=UserAPIKeyAuth(),
forward_headers=False,
endpoint=endpoint,
accept_websocket=False,
count_client_text=count_client_text,
)
for coroutine in queued:
await coroutine
return _RelayedSession(
spend_logs=tuple(capturing_logger.payloads), failure_hook=proxy_logging.post_call_failure_hook
)
@pytest.mark.asyncio
async def test_websocket_passthrough_vertex_live_setup_ack_names_the_model_but_is_not_billed_as_usage():
"""Vertex Live keeps its special first frame: the setup acknowledgement is forwarded verbatim, read for the
@ -7808,6 +7870,174 @@ async def _track_cost_for_passthrough_kwargs(kwargs: dict) -> AsyncMock:
return increment_spend_counters
DEEPGRAM_SPEAK_TARGET: Final = "wss://api.deepgram.com/v1/speak?model=aura-2-thalia-en"
SPEAK_AUDIO_CHUNK: Final = bytes(range(256)) * 8
SPOKEN_TEXT: Final = "The quick brown fox jumps over the lazy dog."
SPEAK_FRAME: Final = json.dumps({"type": "Speak", "text": SPOKEN_TEXT})
FLUSH_FRAME: Final = json.dumps({"type": "Flush"})
FLUSHED_FRAME: Final = json.dumps({"type": "Flushed", "sequence_id": 0})
SPEAK_CLOSE_FRAME: Final = json.dumps({"type": "Close"})
CLIENT_BINARY_FRAME: Final = b"\x00\xffclient-binary"
SPEAK_SESSION_FRAMES: Final = ({"text": SPEAK_FRAME}, {"bytes": CLIENT_BINARY_FRAME}, {"text": FLUSH_FRAME})
SPEAK_SESSION_CLIENT_LENGTH: Final = sum(map(len, (SPEAK_FRAME, CLIENT_BINARY_FRAME, FLUSH_FRAME, SPEAK_CLOSE_FRAME)))
class SpeakingUpstreamWebSocket:
"""Answers each Speak with audio as soon as it arrives and ends the session on Close, the way Deepgram speak
does, so a relay that held audio back until the end would never see the Close it waits for"""
def __init__(self, close_code: int = 1000, close_reason: str = "") -> None:
self.received: list[str | bytes] = []
self.close = AsyncMock()
self._close_code = close_code
self._close_reason = close_reason
self._replies: asyncio.Queue[str | bytes | None] = asyncio.Queue()
async def send(self, message: str | bytes) -> None:
self.received.append(message)
match json.loads(message).get("type") if isinstance(message, str) else None:
case "Speak":
self._replies.put_nowait(SPEAK_AUDIO_CHUNK)
case "Flush":
self._replies.put_nowait(FLUSHED_FRAME)
case "Close":
self._replies.put_nowait(None)
async def recv(self, decode: bool | None = None) -> str | bytes:
from websockets.exceptions import ConnectionClosedError, ConnectionClosedOK
from websockets.frames import Close
reply: Final = await self._replies.get()
if reply is not None:
return reply
close: Final = Close(self._close_code, self._close_reason)
closed: Final = ConnectionClosedOK if self._close_code == 1000 else ConnectionClosedError
raise closed(rcvd=close, sent=close, rcvd_then_sent=True)
def _speak_client(frames: tuple[dict[str, str | bytes], ...] = SPEAK_SESSION_FRAMES) -> MagicMock:
"""A client socket that sends ``frames``, then Close once the first audio has reached it"""
first_audio: Final = asyncio.Event()
async def script() -> AsyncIterator[dict[str, str | bytes]]:
for frame in frames:
yield {"type": "websocket.receive", **frame}
await first_audio.wait()
yield {"type": "websocket.receive", "text": SPEAK_CLOSE_FRAME}
await asyncio.Event().wait()
steps: Final = script()
async def receive() -> dict[str, str | bytes]:
return await anext(steps)
websocket: Final = _client_websocket(receive)
websocket.send_bytes = AsyncMock(side_effect=lambda data: first_audio.set())
return websocket
def _aura_2_usd_per_character() -> float:
return litellm.model_cost["deepgram/aura-2"]["input_cost_per_character"]
@pytest.mark.asyncio
@pytest.mark.usefixtures("local_model_cost_map")
async def test_websocket_passthrough_counts_client_text_without_changing_what_is_relayed(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Audio reaches the client while the session is open, every client frame reaches the upstream unchanged, and
the session is billed for the summed count of the text and binary frames"""
upstream_ws: Final = SpeakingUpstreamWebSocket()
websocket: Final = _speak_client()
session: Final = await _relay_and_capture_spend(
monkeypatch, upstream_ws, websocket, DEEPGRAM_SPEAK_TARGET, "/deepgram/v1/speak", count_client_text=len
)
assert upstream_ws.received == [SPEAK_FRAME, CLIENT_BINARY_FRAME, FLUSH_FRAME, SPEAK_CLOSE_FRAME]
assert [call.args[0] for call in websocket.send_bytes.await_args_list] == [SPEAK_AUDIO_CHUNK]
assert [call.args[0] for call in websocket.send_text.await_args_list] == [FLUSHED_FRAME]
assert [log["response_cost"] for log in session.spend_logs] == [
pytest.approx(SPEAK_SESSION_CLIENT_LENGTH * _aura_2_usd_per_character())
]
websocket.close.assert_awaited_once_with()
@pytest.mark.asyncio
@pytest.mark.usefixtures("local_model_cost_map")
@pytest.mark.parametrize(
("close_code", "close_reason"),
[
pytest.param(1008, "DATA-0000: invalid text", id="policy"),
pytest.param(1011, "NET-0001: internal", id="internal"),
],
)
async def test_websocket_passthrough_counts_client_text_when_the_upstream_ends_with_an_error_close(
monkeypatch: pytest.MonkeyPatch, close_code: int, close_reason: str
) -> None:
upstream_ws: Final = SpeakingUpstreamWebSocket(close_code=close_code, close_reason=close_reason)
websocket: Final = _speak_client()
session: Final = await _relay_and_capture_spend(
monkeypatch, upstream_ws, websocket, DEEPGRAM_SPEAK_TARGET, "/deepgram/v1/speak", count_client_text=len
)
websocket.close.assert_awaited_once_with(code=close_code, reason=close_reason)
assert [log["response_cost"] for log in session.spend_logs] == [
pytest.approx(SPEAK_SESSION_CLIENT_LENGTH * _aura_2_usd_per_character())
]
class _DrainStallingUpstreamWebSocket(SpeakingUpstreamWebSocket):
"""Takes the client's Close, then stalls in send the way a full write buffer does, so the session ends while the
relay is still inside that send"""
async def send(self, message: str | bytes) -> None:
await super().send(message)
if message == SPEAK_CLOSE_FRAME:
await asyncio.Event().wait()
@pytest.mark.asyncio
@pytest.mark.usefixtures("local_model_cost_map")
async def test_websocket_passthrough_bills_a_client_frame_whose_send_is_cancelled_after_reaching_the_upstream(
monkeypatch: pytest.MonkeyPatch,
) -> None:
session: Final = await _relay_and_capture_spend(
monkeypatch,
_DrainStallingUpstreamWebSocket(),
_speak_client(),
DEEPGRAM_SPEAK_TARGET,
"/deepgram/v1/speak",
count_client_text=len,
)
assert [log["response_cost"] for log in session.spend_logs] == [
pytest.approx(SPEAK_SESSION_CLIENT_LENGTH * _aura_2_usd_per_character())
]
@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: pytest.MonkeyPatch,
) -> None:
session: Final = await _relay_and_capture_spend(
monkeypatch,
SpeakingUpstreamWebSocket(),
_speak_client(),
DEEPGRAM_SPEAK_TARGET,
"/deepgram/v1/speak",
count_client_text=deepgram_speak_characters,
)
assert [(log["model"], log["custom_llm_provider"]) for log in session.spend_logs] == [
("aura-2-thalia-en", "deepgram")
]
assert session.spend_logs[0]["response_cost"] == pytest.approx(len(SPOKEN_TEXT) * _aura_2_usd_per_character())
assert SPOKEN_TEXT not in json.dumps(session.spend_logs[0], default=str)
@pytest.mark.asyncio
async def test_passthrough_success_reconciles_budget_reservation():
"""

View file

@ -1252,6 +1252,7 @@ _COST_MAP_ROW_SCHEMA: Final = {
"/v1/videos",
"/vertex_ai/live",
"/v1/listen",
"/v1/speak",
"/v1/systemone",
"/v1/decisions",
"/v1beta/interactions",