From a9adf506566187d0ee9f153e4cb9fed07c4f05ee Mon Sep 17 00:00:00 2001 From: Emerson Gomes Date: Tue, 6 Oct 2026 09:24:20 -0500 Subject: [PATCH] fix(realtime): reconcile audio protocols billing and current CI --- .github/workflows/test-unit.yml | 1 + litellm/cost_calculator.py | 2 +- .../audio_utils/transcription_streaming.py | 8 +-- .../litellm_core_utils/realtime_streaming.py | 66 +++++++++++++------ .../azure/realtime/http_transformation.py | 2 +- litellm/llms/custom_httpx/llm_http_handler.py | 19 ++---- litellm/llms/openai/realtime/handler.py | 6 +- .../transcriptions/gpt_transformation.py | 4 +- litellm/llms/openai/transcriptions/handler.py | 2 +- litellm/main.py | 8 ++- litellm/proxy/_lazy_features.py | 2 +- .../auth/managed_authorization.py | 11 +++- .../proxy/common_utils/http_parsing_utils.py | 6 +- litellm/proxy/proxy_server.py | 4 +- litellm/proxy/realtime_endpoints/endpoints.py | 18 ++--- litellm/realtime_api/main.py | 40 +++++------ .../test_realtime_streaming.py | 43 ++++++++++-- .../realtime/test_openai_realtime_handler.py | 53 +++++++++------ .../auth/test_managed_authorization.py | 17 ++++- 19 files changed, 194 insertions(+), 118 deletions(-) diff --git a/.github/workflows/test-unit.yml b/.github/workflows/test-unit.yml index 6f0da90fd66..87b4a13166d 100644 --- a/.github/workflows/test-unit.yml +++ b/.github/workflows/test-unit.yml @@ -471,6 +471,7 @@ jobs: test-path: >- tests/unit/anthropic_interface tests/unit/compression + tests/unit/cookbook tests/unit/enterprise/enterprise_callbacks/test_callback_controls.py tests/unit/enterprise/enterprise_callbacks/test_llm_guard.py tests/unit/enterprise/enterprise_callbacks/test_secret_detection.py diff --git a/litellm/cost_calculator.py b/litellm/cost_calculator.py index 94a18d98e0e..d81df94367d 100644 --- a/litellm/cost_calculator.py +++ b/litellm/cost_calculator.py @@ -3056,7 +3056,7 @@ def handle_realtime_stream_cost_calculation( ) total_cost: Final = input_cost_per_token + output_cost_per_token + transcription_cost + translation_cost - additional_costs: Final = { # mutable-ok: logging stores a mutable per-request cost breakdown + additional_costs: Final = { key: value for key, value in ( ("transcription_cost", transcription_cost), diff --git a/litellm/litellm_core_utils/audio_utils/transcription_streaming.py b/litellm/litellm_core_utils/audio_utils/transcription_streaming.py index 979a1847362..cf2681f5d36 100644 --- a/litellm/litellm_core_utils/audio_utils/transcription_streaming.py +++ b/litellm/litellm_core_utils/audio_utils/transcription_streaming.py @@ -72,13 +72,7 @@ class _TranscriptionEventCollector: response: Final = TranscriptionResponse( text=done_event.text if done_event is not None else "".join(self.text_deltas), usage=done_event.usage.model_dump() if done_event is not None and done_event.usage is not None else None, - languages=( - [ # mutable-ok: the response model requires a concrete serialized language list - language.model_dump() for language in done_languages - ] - if done_languages is not None - else None - ), + languages=([language.model_dump() for language in done_languages] if done_languages is not None else None), ) if self.duration is not None: response.set_audio_transcription_duration(self.duration) diff --git a/litellm/litellm_core_utils/realtime_streaming.py b/litellm/litellm_core_utils/realtime_streaming.py index 4d2d4dde999..907f6b73250 100644 --- a/litellm/litellm_core_utils/realtime_streaming.py +++ b/litellm/litellm_core_utils/realtime_streaming.py @@ -7,6 +7,7 @@ from dataclasses import dataclass from enum import Enum, auto from typing import TYPE_CHECKING, Any, Final, NoReturn, Protocol, TypedDict, cast +from pydantic import TypeAdapter, ValidationError from typing_extensions import ReadOnly import litellm @@ -112,6 +113,25 @@ def _decode_json_object(payload: str) -> Mapping[str, object]: return json.loads(payload) +_TRANSLATION_AUDIO_FORMAT: Final = TypeAdapter(str | Mapping[str, object] | None) + + +def _translation_audio_bytes_per_second(audio_format: str | Mapping[str, object] | None) -> float | None: + if isinstance(audio_format, str): + return 8000.0 if audio_format in ("g711_ulaw", "g711_alaw") else None + if audio_format is None: + return None + format_type: Final = audio_format.get("type") + rate: Final = audio_format.get("rate") + if format_type in ("audio/pcmu", "audio/pcma") and rate is None: + return 8000.0 + if not isinstance(rate, (int, float)) or rate <= 0: + return None + if format_type == "audio/pcm": + return float(rate) * 2 + return float(rate) if format_type in ("audio/pcmu", "audio/pcma") else None + + class RealtimeEventNormalizer(Protocol): def should_drop(self, event: object) -> bool: ... def normalize(self, event: dict) -> dict: ... @@ -155,13 +175,14 @@ class RealTimeStreaming: self.tool_calls: list[dict] = [] self._is_translation_session = translation_session self._translation_input_seconds = 0.0 + self._translation_input_bytes_per_second = 48000.0 self._translation_output_audio_bytes = 0 self._translation_output_bytes_per_second = 48000.0 self._translation_usage_finalized = False # Detect whether the client is explicitly opting into the beta protocol. self._client_wants_beta = self._detect_beta_header(websocket) - self._backend_uses_beta_protocol = ( + self._backend_uses_beta_protocol = not translation_session and ( self._client_wants_beta if backend_uses_beta_protocol is None else backend_uses_beta_protocol ) @@ -226,8 +247,8 @@ class RealTimeStreaming: _CLIENT_AUDIO_BUFFER_COMMIT_TYPES = frozenset(["input_audio_buffer.commit", "input_audio_buffer.end"]) _AUDIO_FORMAT_MAP: dict[str, dict[str, str | int]] = { "pcm16": {"type": "audio/pcm", "rate": 24000}, - "g711_ulaw": {"type": "audio/G711-ulaw", "rate": 8000}, - "g711_alaw": {"type": "audio/G711-alaw", "rate": 8000}, + "g711_ulaw": {"type": "audio/pcmu"}, + "g711_alaw": {"type": "audio/pcma"}, } # GA name → beta name (when client WebSocket includes OpenAI-Beta: realtime=v1) _GA_TO_BETA_EVENT_TYPES: dict[str, str] = { @@ -492,7 +513,7 @@ class RealTimeStreaming: ) self._translation_usage_finalized = True return - self._capture_translation_output_format(event_obj) + self._capture_translation_audio_formats(event_obj) if event_obj.get("type") not in ( "session.output_audio.delta", "response.output_audio.delta", @@ -515,35 +536,38 @@ class RealTimeStreaming: event: Final = _decode_json_object(message) except (json.JSONDecodeError, TypeError): return + self._capture_translation_audio_formats(event) if event.get("type") != "session.input_audio_buffer.append" or not isinstance(audio := event.get("audio"), str): return try: decoded: Final = base64.b64decode(audio, validate=True) except (ValueError, TypeError): return - self._translation_input_seconds += len(decoded) / 48000.0 + self._translation_input_seconds += len(decoded) / self._translation_input_bytes_per_second - def _capture_translation_output_format(self, event_obj: Mapping[str, object]) -> None: + def _capture_translation_audio_formats(self, event_obj: Mapping[str, object]) -> None: session: Final = event_obj.get("session") if not isinstance(session, dict): return audio: Final = session.get("audio") - output: Final = audio.get("output") if isinstance(audio, dict) else None - audio_format: Final = output.get("format") if isinstance(output, dict) else None - if isinstance(audio_format, str): - if audio_format in ("g711_ulaw", "g711_alaw"): - self._translation_output_bytes_per_second = 8000.0 + if not isinstance(audio, dict): return - if not isinstance(audio_format, dict): - return - format_type: Final = audio_format.get("type") - rate: Final = audio_format.get("rate") - if not isinstance(rate, (int, float)) or rate <= 0: - return - if format_type == "audio/pcm": - self._translation_output_bytes_per_second = float(rate) * 2 - elif format_type in ("audio/pcmu", "audio/pcma"): - self._translation_output_bytes_per_second = float(rate) + for direction, configuration in (("input", audio.get("input")), ("output", audio.get("output"))): + if not isinstance(configuration, dict): + continue + try: + if ( + bytes_per_second := _translation_audio_bytes_per_second( + _TRANSLATION_AUDIO_FORMAT.validate_python(configuration.get("format")) + ) + ) is None: + continue + except ValidationError: + continue + if direction == "input": + self._translation_input_bytes_per_second = bytes_per_second + else: + self._translation_output_bytes_per_second = bytes_per_second def _finalize_translation_usage(self) -> None: if self._translation_usage_finalized: diff --git a/litellm/llms/azure/realtime/http_transformation.py b/litellm/llms/azure/realtime/http_transformation.py index 9b37d1faa95..18fd2e3e53d 100644 --- a/litellm/llms/azure/realtime/http_transformation.py +++ b/litellm/llms/azure/realtime/http_transformation.py @@ -49,7 +49,7 @@ class AzureRealtimeHTTPConfig(BaseRealtimeHTTPConfig): model: str, api_key: str | None = None, ) -> dict: - validated_headers: Final = { # mutable-ok: provider authentication headers are extended before dispatch + validated_headers: Final = { **headers, "Content-Type": "application/json", } diff --git a/litellm/llms/custom_httpx/llm_http_handler.py b/litellm/llms/custom_httpx/llm_http_handler.py index d94913192fc..7fcf99e7c57 100644 --- a/litellm/llms/custom_httpx/llm_http_handler.py +++ b/litellm/llms/custom_httpx/llm_http_handler.py @@ -6421,7 +6421,7 @@ class BaseLLMHTTPHandler: logging_obj.pre_call( input=request_data, api_key="", - additional_args={ # mutable-ok: logging owns a mutable request metadata payload + additional_args={ "complete_input_dict": request_data, "api_base": normalized_api_base, }, @@ -6429,10 +6429,7 @@ class BaseLLMHTTPHandler: try: configured_client: Final = openai_client.with_options( timeout=timeout, - set_default_headers={ # mutable-ok: OpenAI SDK accepts a mutable custom-header mapping - key: str(value) - for key, value in (extra_headers or {}).items() # mutable-ok: SDK requires concrete headers - }, + set_default_headers={key: str(value) for key, value in (extra_headers or {}).items()}, ) raw_response: Final = await configured_client.post( "/realtime/client_secrets", @@ -6504,7 +6501,7 @@ class BaseLLMHTTPHandler: logging_obj.pre_call( input=request_data, api_key="", - additional_args={ # mutable-ok: logging owns a mutable request metadata payload + additional_args={ "complete_input_dict": request_data, "api_base": normalized_api_base, }, @@ -6637,7 +6634,7 @@ class BaseLLMHTTPHandler: logging_obj.pre_call( input="realtime_sdp_offer", api_key="", - additional_args={ # mutable-ok: logging owns a mutable request metadata payload + additional_args={ "api_base": normalized_api_base, "session": session_data, }, @@ -6646,7 +6643,7 @@ class BaseLLMHTTPHandler: if translation: configured_client: Final = openai_client.with_options( timeout=timeout, - set_default_headers={ # mutable-ok: OpenAI SDK accepts a mutable custom-header mapping + set_default_headers={ "Content-Type": "application/sdp", **MappingProxyType( {key: str(value) for key, value in (extra_headers or MappingProxyType({})).items()} @@ -6687,7 +6684,7 @@ class BaseLLMHTTPHandler: @staticmethod def _decoded_realtime_sdk_response(response: httpx.Response) -> httpx.Response: - headers: Final = { # mutable-ok: httpx accepts a concrete response header mapping + headers: Final = { key: value for key, value in response.headers.items() if key.lower() not in ("content-encoding", "content-length", "transfer-encoding") @@ -6732,9 +6729,7 @@ class BaseLLMHTTPHandler: - session: JSON string with {"type": "realtime", "model": "...", ...} """ session_data: Final[dict[str, object]] = { # mutable-ok: model and session type are resolved locally - **( - session_config or {} # mutable-ok: absent session configuration starts from an empty provider payload - ) + **(session_config or {}) } if "type" not in session_data: session_data["type"] = "translation" if translation else "realtime" diff --git a/litellm/llms/openai/realtime/handler.py b/litellm/llms/openai/realtime/handler.py index 484c881a2b3..44cacb2f5cc 100644 --- a/litellm/llms/openai/realtime/handler.py +++ b/litellm/llms/openai/realtime/handler.py @@ -200,7 +200,7 @@ class OpenAIRealtime(OpenAIChatCompletion): model=sdk_model, extra_query=extra_query, extra_headers=headers, - websocket_connection_options={ # mutable-ok: OpenAI SDK forwards a mutable options mapping + websocket_connection_options={ "max_size": REALTIME_WEBSOCKET_MAX_MESSAGE_SIZE_BYTES, **(MappingProxyType({"ssl": ssl_config}) if url.startswith("wss://") else MappingProxyType({})), **(MappingProxyType({"open_timeout": timeout}) if timeout is not None else MappingProxyType({})), @@ -241,7 +241,9 @@ class OpenAIRealtime(OpenAIChatCompletion): # Get provider-specific SSL configuration ssl_config: Final = self._get_ssl_config(url) - openai_beta_realtime: Final = client_sent_openai_beta_realtime_header(websocket) + openai_beta_realtime: Final = realtime_mode != "translation" and client_sent_openai_beta_realtime_header( + websocket + ) if not openai_beta_realtime: verbose_logger.debug( "OpenAI Realtime: connecting with GA protocol (no OpenAI-Beta header). " diff --git a/litellm/llms/openai/transcriptions/gpt_transformation.py b/litellm/llms/openai/transcriptions/gpt_transformation.py index c55ecf1e56c..cbeb7a6d811 100644 --- a/litellm/llms/openai/transcriptions/gpt_transformation.py +++ b/litellm/llms/openai/transcriptions/gpt_transformation.py @@ -14,7 +14,7 @@ class OpenAIGPTAudioTranscriptionConfig(OpenAIWhisperAudioTranscriptionConfig): """ Get the supported OpenAI params for the `gpt-4o-transcribe` models """ - return [ # mutable-ok: base transcription interface requires a mutable supported-parameter list + return [ "language", "prompt", "response_format", @@ -56,7 +56,7 @@ class OpenAIGPTTranscribeAudioTranscriptionConfig(OpenAIGPTAudioTranscriptionCon optional_params: dict, # mutable-ok: base transformation interface supplies a mutable request payload litellm_params: dict, # mutable-ok: base transformation interface supplies mutable provider parameters ) -> AudioTranscriptionRequestData: - data: Final = { # mutable-ok: OpenAI SDK consumes this multipart request mapping + data: Final = { "model": model, "file": audio_file, "response_format": "json", diff --git a/litellm/llms/openai/transcriptions/handler.py b/litellm/llms/openai/transcriptions/handler.py index ac57cc72290..491baa7c403 100644 --- a/litellm/llms/openai/transcriptions/handler.py +++ b/litellm/llms/openai/transcriptions/handler.py @@ -35,7 +35,7 @@ class OpenAIAudioTranscription(OpenAIChatCompletion): sdk_data: Final = sdk_compatible_transcription_request_data(data) if data.get("stream") is True: stream_response: Final = await openai_aclient.audio.transcriptions.create(**sdk_data, timeout=timeout) - return {}, stream_response # mutable-ok: response headers use the existing mutable mapping contract + return {}, stream_response raw_response: Final = await openai_aclient.audio.transcriptions.with_raw_response.create( **sdk_data, timeout=timeout ) # pyright: ignore[reportArgumentType] # SDK TypedDict lags accepted transcription options diff --git a/litellm/main.py b/litellm/main.py index da7d775fc3c..c2cad11b83b 100644 --- a/litellm/main.py +++ b/litellm/main.py @@ -7922,7 +7922,9 @@ async def atranscription( if file is not None: calculated_duration = calculate_request_duration(file) if calculated_duration is not None: - response._litellm_audio_duration = calculated_duration # pyright: ignore[reportAttributeAccess] # SDK streams permit runtime metadata but do not type this field + response._litellm_audio_duration = ( # pyright: ignore[reportAttributeAccessIssue] # SDK omits it + calculated_duration + ) return response if not isinstance(response, TranscriptionResponse): raise ValueError( @@ -8273,7 +8275,9 @@ def transcription( calculated_duration: Final = calculate_request_duration(file) if isinstance(response, (Stream, AsyncStream)): if calculated_duration is not None: - response._litellm_audio_duration = calculated_duration # pyright: ignore[reportAttributeAccess] # SDK streams permit runtime metadata but do not type this field + response._litellm_audio_duration = ( # pyright: ignore[reportAttributeAccessIssue] # SDK omits it + calculated_duration + ) elif getattr(response, "duration", None) is None and calculated_duration is not None: response.set_audio_transcription_duration(calculated_duration) diff --git a/litellm/proxy/_lazy_features.py b/litellm/proxy/_lazy_features.py index 03f6038c4d3..034f42e23b8 100644 --- a/litellm/proxy/_lazy_features.py +++ b/litellm/proxy/_lazy_features.py @@ -43,7 +43,7 @@ def _include_router_prepend(attr_name: str = "router") -> Callable[["FastAPI", o route_count: Final = len(router.routes) app.include_router(getattr(module, attr_name)) new_routes: Final = router.routes[route_count:] - router.routes[:] = [ # mutable-ok: FastAPI route registration requires in-place list replacement + router.routes[:] = [ *new_routes, *router.routes[:route_count], ] diff --git a/litellm/proxy/agent_endpoints/auth/managed_authorization.py b/litellm/proxy/agent_endpoints/auth/managed_authorization.py index 5d8287227a5..7e9526e411f 100644 --- a/litellm/proxy/agent_endpoints/auth/managed_authorization.py +++ b/litellm/proxy/agent_endpoints/auth/managed_authorization.py @@ -11,7 +11,16 @@ from litellm.proxy.agent_endpoints.managed_identity import raise_identity_failur from litellm.types.agents import AgentResponse from litellm.types.proxy.agent_identity import AgentIdentityFailure, ManagedAgentContext -_MANAGED_REALTIME_ROUTES: Final = frozenset(("/realtime", "/v1/realtime", "/openai/v1/realtime")) +_MANAGED_REALTIME_ROUTES: Final = frozenset( + ( + "/realtime", + "/v1/realtime", + "/openai/v1/realtime", + "/realtime/translations", + "/v1/realtime/translations", + "/openai/v1/realtime/translations", + ) +) _MANAGED_MODEL_ROUTES: Final = frozenset( f"{prefix}/{operation}" for prefix, operation in product( diff --git a/litellm/proxy/common_utils/http_parsing_utils.py b/litellm/proxy/common_utils/http_parsing_utils.py index b66c56e6772..2ef4dff9569 100644 --- a/litellm/proxy/common_utils/http_parsing_utils.py +++ b/litellm/proxy/common_utils/http_parsing_utils.py @@ -479,10 +479,8 @@ async def get_form_data(request: Request) -> dict[str, Any]: array_keys: Final = frozenset(key[:-2] for key, _ in form_items if key.endswith("[]")) normalized_items: Final = tuple((key.removesuffix("[]"), value) for key, value in form_items) normalized_keys: Final = frozenset(key for key, _ in normalized_items) - return { # mutable-ok: request parsers expose a mutable form-data mapping to endpoint handlers - key: [ # mutable-ok: repeated multipart values follow the established mutable list contract - value for item_key, value in normalized_items if item_key == key - ] + return { + key: [value for item_key, value in normalized_items if item_key == key] if key in array_keys else next(value for item_key, value in reversed(normalized_items) if item_key == key) for key in normalized_keys diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index a29375bf226..4007b471a94 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -13122,9 +13122,7 @@ async def realtime_websocket_endpoint( ) query_params: Final = cast( # cast-ok: cached tuples contain only the declared realtime query keys RealtimeQueryParams, - dict( # mutable-ok: downstream realtime routing normalizes this request-scoped query mapping - _realtime_query_params_template(query_model, intent) - ), + dict(_realtime_query_params_template(query_model, intent)), ) data: dict[str, object] = { diff --git a/litellm/proxy/realtime_endpoints/endpoints.py b/litellm/proxy/realtime_endpoints/endpoints.py index f358111835b..d202e75dab7 100644 --- a/litellm/proxy/realtime_endpoints/endpoints.py +++ b/litellm/proxy/realtime_endpoints/endpoints.py @@ -282,18 +282,18 @@ def _decode_realtime_token_payload( ) @router.post( "/v1/realtime/translations/client_secrets", - dependencies=[Depends(user_api_key_auth)], # mutable-ok: FastAPI decorator requires dependency lists - tags=["realtime"], # mutable-ok: FastAPI decorator requires tag lists + dependencies=[Depends(user_api_key_auth)], + tags=["realtime"], ) @router.post( "/realtime/translations/client_secrets", - dependencies=[Depends(user_api_key_auth)], # mutable-ok: FastAPI decorator requires dependency lists - tags=["realtime"], # mutable-ok: FastAPI decorator requires tag lists + dependencies=[Depends(user_api_key_auth)], + tags=["realtime"], ) @router.post( "/openai/v1/realtime/translations/client_secrets", - dependencies=[Depends(user_api_key_auth)], # mutable-ok: FastAPI decorator requires dependency lists - tags=["realtime"], # mutable-ok: FastAPI decorator requires tag lists + dependencies=[Depends(user_api_key_auth)], + tags=["realtime"], ) async def create_realtime_client_secret( request: Request, @@ -446,15 +446,15 @@ async def create_realtime_client_secret( ) @router.post( "/v1/realtime/translations/calls", - tags=["realtime"], # mutable-ok: FastAPI decorator requires tag lists + tags=["realtime"], ) @router.post( "/realtime/translations/calls", - tags=["realtime"], # mutable-ok: FastAPI decorator requires tag lists + tags=["realtime"], ) @router.post( "/openai/v1/realtime/translations/calls", - tags=["realtime"], # mutable-ok: FastAPI decorator requires tag lists + tags=["realtime"], ) async def proxy_realtime_calls( request: Request, diff --git a/litellm/realtime_api/main.py b/litellm/realtime_api/main.py index 9ac86177fd7..7870808bd8f 100644 --- a/litellm/realtime_api/main.py +++ b/litellm/realtime_api/main.py @@ -73,24 +73,18 @@ def _model_params_with_stored_credentials(model_params: Mapping[str, object]) -> def _with_resolved_session_model(session: dict[str, object], model_name: str) -> dict[str, object]: if session.get("type") == "transcription": audio = session.get("audio") - audio = audio if isinstance(audio, dict) else {} # mutable-ok: nested session model is rebuilt locally + audio = audio if isinstance(audio, dict) else {} audio_input = audio.get("input") - audio_input = ( - audio_input if isinstance(audio_input, dict) else {} # mutable-ok: nested session model is rebuilt locally - ) + audio_input = audio_input if isinstance(audio_input, dict) else {} transcription = audio_input.get("transcription") - transcription = ( - transcription - if isinstance(transcription, dict) - else {} # mutable-ok: nested session model is rebuilt locally - ) - return { # mutable-ok: provider routing requires an independently mutable session payload + transcription = transcription if isinstance(transcription, dict) else {} + return { **session, - "audio": { # mutable-ok: provider routing rebuilds nested audio configuration + "audio": { **audio, - "input": { # mutable-ok: provider routing rebuilds nested input configuration + "input": { **audio_input, - "transcription": { # mutable-ok: resolved deployment replaces only the transcription model + "transcription": { **transcription, "model": model_name, }, @@ -228,7 +222,7 @@ async def acreate_realtime_client_secret( if isinstance(request_data.get("session"), dict): request_data["session"] = _with_resolved_session_model(request_data["session"], model_name) elif req.model is not None: - request_data["session"] = { # mutable-ok: OpenAI SDK consumes this request-scoped session payload + request_data["session"] = { "type": "realtime", "model": model_name, } @@ -257,7 +251,7 @@ async def acreate_realtime_translation_client_secret( ) -> httpx.Response: requested_model_name: Final = model or (session or {}).get("model") or "gpt-realtime-translate" session_config: Final = RealtimeSessionConfig.model_validate( - { # mutable-ok: Pydantic validates this request-scoped translation session payload + { **(session or {}), "type": "translation", "model": requested_model_name, @@ -293,18 +287,16 @@ async def acreate_realtime_translation_client_secret( litellm_logging_obj.update_from_kwargs( kwargs=kwargs, model=model_name, - optional_params={ # mutable-ok: logging owns a mutable request metadata payload + optional_params={ "expires_after": expires_after, "session": session, }, - litellm_params={ # mutable-ok: logging owns a mutable provider metadata payload - "api_base": resolved_api_base - }, + litellm_params={"api_base": resolved_api_base}, custom_llm_provider=custom_llm_provider, ) request_data: Final = req.model_dump( exclude_none=True, - exclude={"model"}, # mutable-ok: Pydantic requires a mutable field-exclusion set + exclude={"model"}, ) request_data["session"] = _with_resolved_session_model(request_data["session"], model_name) request_data["session"].pop("type", None) @@ -482,7 +474,7 @@ async def arealtime_translation_calls( litellm_params=litellm_params, ) session_config: Final = _with_resolved_session_model( - { # mutable-ok: provider routing requires an independently mutable session payload + { **(session or {}), "type": "translation", "model": model_name, @@ -492,13 +484,11 @@ async def arealtime_translation_calls( litellm_logging_obj.update_from_kwargs( kwargs=kwargs, model=model_name, - optional_params={ # mutable-ok: logging owns a mutable request metadata payload + optional_params={ "realtime_translation_calls": True, "session": session_config, }, - litellm_params={ # mutable-ok: logging owns a mutable provider metadata payload - "api_base": resolved_api_base - }, + litellm_params={"api_base": resolved_api_base}, custom_llm_provider=custom_llm_provider, ) return await base_llm_http_handler.async_realtime_calls_handler( diff --git a/tests/unit/litellm_core_utils/test_realtime_streaming.py b/tests/unit/litellm_core_utils/test_realtime_streaming.py index 94b0d215bf4..27501105188 100644 --- a/tests/unit/litellm_core_utils/test_realtime_streaming.py +++ b/tests/unit/litellm_core_utils/test_realtime_streaming.py @@ -604,9 +604,10 @@ async def test_client_ack_messages_keeps_beta_session_shape_for_beta_backend(): @pytest.mark.asyncio -async def test_translation_session_update_omits_session_type(): +@pytest.mark.parametrize("beta_client", (False, True)) +async def test_translation_session_update_omits_session_type(beta_client: bool) -> None: client_ws = MagicMock() - client_ws.scope = {"headers": []} + client_ws.scope = {"headers": [(b"openai-beta", b"realtime=v1")] if beta_client else []} client_ws.receive_text = AsyncMock( side_effect=[ json.dumps( @@ -614,6 +615,8 @@ async def test_translation_session_update_omits_session_type(): "type": "session.update", "session": { "type": "translation", + "input_audio_format": "g711_ulaw", + "output_audio_format": "pcm16", "audio": {"output": {"language": "fr"}}, }, } @@ -637,6 +640,11 @@ async def test_translation_session_update_omits_session_type(): sent_to_backend = json.loads(backend_ws.send.call_args_list[0].args[0]) assert "type" not in sent_to_backend["session"] assert sent_to_backend["session"]["audio"]["output"]["language"] == "fr" + assert sent_to_backend["session"]["audio"]["input"]["format"]["type"] == "audio/pcmu", ( + "OpenAI Realtime audio formats, checked 2026-10-06: " + "https://github.com/openai/openai-python/blob/v2.33.0/src/openai/types/realtime/realtime_audio_formats.py" + ) + assert sent_to_backend["session"]["audio"]["output"]["format"]["type"] == "audio/pcm" def test_translate_event_to_beta_renames_delta_types(): @@ -2935,8 +2943,22 @@ def test_translation_audio_duration_is_finalized_once(event_type: str): @pytest.mark.asyncio @pytest.mark.parametrize("output_bytes", (0, 48000)) +@pytest.mark.parametrize("format_event_type", ("session.created", "session.update")) +@pytest.mark.parametrize( + "audio_format,bytes_per_second", + ( + ("pcm16", 48000), + ("g711_ulaw", 8000), + ("g711_alaw", 8000), + ({"type": "audio/pcm", "rate": 16000}, 32000), + ({"type": "audio/pcmu", "rate": 8000}, 8000), + ({"type": "audio/pcma", "rate": 8000}, 8000), + ({"type": "audio/pcmu"}, 8000), + ({"type": "audio/pcma"}, 8000), + ), +) async def test_translation_disconnect_bills_sent_input_audio( - output_bytes: int, + output_bytes: int, format_event_type: str, audio_format: str | Mapping[str, object], bytes_per_second: int ) -> None: import base64 @@ -2949,8 +2971,21 @@ async def test_translation_disconnect_bills_sent_input_audio( model="gpt-realtime-translate", translation_session=True, ) + format_event: Final = { + "type": format_event_type, + "session": {"audio": {"input": {"format": audio_format}}}, + } + if format_event_type == "session.update": + await streaming._send_to_backend(json.dumps(format_event)) + else: + streaming._capture_translation_output_audio(format_event) await streaming._send_to_backend( - json.dumps({"type": "session.input_audio_buffer.append", "audio": base64.b64encode(bytes(96000)).decode()}) + json.dumps( + { + "type": "session.input_audio_buffer.append", + "audio": base64.b64encode(bytes(bytes_per_second * 2)).decode(), + } + ) ) streaming._capture_translation_output_audio( {"type": "session.output_audio.delta", "delta": base64.b64encode(bytes(output_bytes)).decode()} diff --git a/tests/unit/llms/openai/realtime/test_openai_realtime_handler.py b/tests/unit/llms/openai/realtime/test_openai_realtime_handler.py index 3d1c953492b..b9597265dd1 100644 --- a/tests/unit/llms/openai/realtime/test_openai_realtime_handler.py +++ b/tests/unit/llms/openai/realtime/test_openai_realtime_handler.py @@ -1,8 +1,12 @@ +import asyncio import json +from collections.abc import Mapping +from typing import Final from unittest.mock import AsyncMock, MagicMock, patch import pytest from openai import AsyncOpenAI, omit +from pydantic import TypeAdapter class DummySDKConnectionManager: @@ -407,24 +411,35 @@ def test_translation_url_uses_dedicated_path(): @pytest.mark.asyncio -async def test_translation_websocket_uses_direct_transport(): +@pytest.mark.parametrize("beta_client", (False, True)) +async def test_translation_websocket_uses_direct_transport(beta_client: bool) -> None: from litellm.llms.openai.realtime.handler import OpenAIRealtime - backend = AsyncMock() + backend: Final = AsyncMock() + pending: Final = asyncio.Event() - class TranslationConnectionManager: - async def __aenter__(self): - return backend + async def receive_backend() -> str: + await pending.wait() + return "" - async def __aexit__(self, exc_type, exc, tb): - return None - - websocket = MagicMock() - websocket.scope = {"headers": []} + backend.recv.side_effect = receive_backend + websocket: Final = MagicMock() + websocket.scope = {"headers": [(b"openai-beta", b"realtime=v1")] if beta_client else []} + websocket.receive_text = AsyncMock( + side_effect=( + json.dumps( + { + "type": "session.update", + "session": {"type": "translation", "audio": {"output": {"language": "fr"}}}, + } + ), + RuntimeError("connection closed"), + ) + ) websocket.close = AsyncMock() - logging_obj = MagicMock() - handler = OpenAIRealtime() - expected_url = "wss://api.openai.com/v1/realtime/translations?model=gpt-realtime-translate" + logging_obj: Final = MagicMock() + handler: Final = OpenAIRealtime() + expected_url: Final = "wss://api.openai.com/v1/realtime/translations?model=gpt-realtime-translate" assert ( handler._construct_url( api_base="https://api.openai.com/v1", @@ -434,13 +449,7 @@ async def test_translation_websocket_uses_direct_transport(): == expected_url ) - with ( - patch("websockets.connect", return_value=TranslationConnectionManager()) as connect, - patch( # test-quality-ok: transport test replaces the unbounded streaming loop - "litellm.llms.openai.realtime.handler.RealTimeStreaming" - ) as streaming, - ): - streaming.return_value.bidirectional_forward = AsyncMock() + with patch("websockets.connect", return_value=DummySDKConnectionManager(backend)) as connect: await handler.async_realtime( model="gpt-realtime-translate", websocket=websocket, @@ -453,7 +462,9 @@ async def test_translation_websocket_uses_direct_transport(): connect.assert_called_once() assert connect.call_args.args[0] == expected_url - assert streaming.call_args.kwargs["translation_session"] is True + assert "OpenAI-Beta" not in connect.call_args.kwargs["additional_headers"] + sent: Final = TypeAdapter(Mapping[str, object]).validate_json(backend.send.call_args_list[0].args[0]) + assert sent["session"] == {"audio": {"output": {"language": "fr"}}} @pytest.mark.parametrize("sdk_client", [True, False]) diff --git a/tests/unit/proxy/agent_endpoints/auth/test_managed_authorization.py b/tests/unit/proxy/agent_endpoints/auth/test_managed_authorization.py index eee985f0aca..8ca11caed8e 100644 --- a/tests/unit/proxy/agent_endpoints/auth/test_managed_authorization.py +++ b/tests/unit/proxy/agent_endpoints/auth/test_managed_authorization.py @@ -364,6 +364,10 @@ async def test_unknown_invocation_target_leaves_billing_unset(monkeypatch: pytes ("/v1beta/models/gemini-model:generateContent", "POST", True), ("/v1/realtime", "GET", True), ("/v1/realtime", "POST", False), + ("/v1/realtime/translations", "GET", True), + ("/v1/realtime/translations", "POST", False), + ("/v1/realtime/translations/client_secrets", "POST", False), + ("/v1/realtime/translations/calls", "POST", False), ("/v1/realtime/client_secrets", "POST", False), ("/mcp/tools/call", "POST", True), ("/a2a/target/message/send", "POST", True), @@ -437,7 +441,17 @@ def test_managed_inference_ignores_unsupported_query_model(): ) -@pytest.mark.parametrize("route", ["/realtime", "/v1/realtime", "/openai/v1/realtime"]) +@pytest.mark.parametrize( + "route", + ( + "/realtime", + "/v1/realtime", + "/openai/v1/realtime", + "/realtime/translations", + "/v1/realtime/translations", + "/openai/v1/realtime/translations", + ), +) def test_managed_realtime_requires_a_model_and_ignores_completion_defaults(route: str) -> None: from litellm.proxy.agent_endpoints.auth.managed_authorization import managed_inference_request @@ -574,6 +588,7 @@ def test_registered_inference_routes_have_an_explicit_managed_access_decision(ro "/vector_stores", "/vector_store/", "/search", "/containers", "/skills", "/claude-code/", "/interactions", "/agents", "/responses/{", "/responses/input_tokens", "/realtime/client_secrets", "/realtime/calls", "/realtime/transcription_sessions", + "/realtime/translations/client_secrets", "/realtime/translations/calls", )) or normalized in ("/models", "/cursor/models", "/cursor/v1/models") concrete: Final = route.split("?")[0].replace("{model}", "model").replace("{model_name:path}", "model") assert managed_agent_route_allowed(concrete, None) is not unsupported, route