From 515f6da5980c8b757f68885714590aa3e6eb0280 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 + .../audio_utils/transcription_streaming.py | 8 +-- .../litellm_core_utils/realtime_streaming.py | 67 +++++++++++++------ .../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 ++++- 18 files changed, 193 insertions(+), 118 deletions(-) diff --git a/.github/workflows/test-unit.yml b/.github/workflows/test-unit.yml index 2276efc4e34..1f7c170d329 100644 --- a/.github/workflows/test-unit.yml +++ b/.github/workflows/test-unit.yml @@ -422,6 +422,7 @@ jobs: tests/unit/ocr tests/unit/passthrough tests/unit/rag + tests/unit/cookbook tests/unit/realtime_api tests/unit/rerank_api tests/unit/search 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 9ed257d9f42..04fed504038 100644 --- a/litellm/litellm_core_utils/realtime_streaming.py +++ b/litellm/litellm_core_utils/realtime_streaming.py @@ -8,7 +8,7 @@ from enum import Enum, auto from types import MappingProxyType from typing import TYPE_CHECKING, Any, Final, NoReturn, Protocol, TypeAlias, TypedDict, cast, runtime_checkable -from pydantic import TypeAdapter +from pydantic import TypeAdapter, ValidationError from typing_extensions import ReadOnly import litellm @@ -207,6 +207,25 @@ def _gate_session_update( return {**client_session, "turn_detection": gate(None)} +_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: ... @@ -261,13 +280,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 ) @@ -349,8 +369,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] = { @@ -616,7 +636,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", @@ -639,35 +659,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 285d7f487ff..732b0946c6d 100644 --- a/litellm/llms/azure/realtime/http_transformation.py +++ b/litellm/llms/azure/realtime/http_transformation.py @@ -52,7 +52,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 cdca6ec3d4e..1365b9d30ab 100644 --- a/litellm/llms/custom_httpx/llm_http_handler.py +++ b/litellm/llms/custom_httpx/llm_http_handler.py @@ -6865,7 +6865,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, }, @@ -6873,10 +6873,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", @@ -6948,7 +6945,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, }, @@ -7081,7 +7078,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, }, @@ -7090,7 +7087,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()} @@ -7131,7 +7128,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") @@ -7176,9 +7173,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 1faebd4bb64..259a920b975 100644 --- a/litellm/llms/openai/realtime/handler.py +++ b/litellm/llms/openai/realtime/handler.py @@ -212,7 +212,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({})), @@ -253,7 +253,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 16b8e919fc5..8f020901714 100644 --- a/litellm/llms/openai/transcriptions/handler.py +++ b/litellm/llms/openai/transcriptions/handler.py @@ -36,7 +36,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 1f12b933bee..09880ec5768 100644 --- a/litellm/main.py +++ b/litellm/main.py @@ -8218,7 +8218,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( @@ -8567,7 +8569,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 9942d6ea113..0feb32592d8 100644 --- a/litellm/proxy/_lazy_features.py +++ b/litellm/proxy/_lazy_features.py @@ -47,7 +47,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 cdbbd6029fa..229c5d23d74 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_LIVE_ROUTES: Final = frozenset(("/live/sessions", "/v1/live/sessions", "/openai/v1/live/sessions")) _MANAGED_MODEL_ROUTES: Final = frozenset( f"{prefix}/{operation}" diff --git a/litellm/proxy/common_utils/http_parsing_utils.py b/litellm/proxy/common_utils/http_parsing_utils.py index 56a46de67f2..5d4dcda1e83 100644 --- a/litellm/proxy/common_utils/http_parsing_utils.py +++ b/litellm/proxy/common_utils/http_parsing_utils.py @@ -527,10 +527,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 adda6f1ded0..d2c1d5c9268 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -13754,9 +13754,7 @@ async def _route_realtime_websocket_session( ) 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 99d02fc7f4c..c28439a6395 100644 --- a/litellm/proxy/realtime_endpoints/endpoints.py +++ b/litellm/proxy/realtime_endpoints/endpoints.py @@ -289,18 +289,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, @@ -453,15 +453,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 b9e0b655cc3..de807867bed 100644 --- a/litellm/realtime_api/main.py +++ b/litellm/realtime_api/main.py @@ -81,24 +81,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, }, @@ -248,7 +242,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, } @@ -279,7 +273,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, @@ -315,18 +309,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) @@ -508,7 +500,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, @@ -518,13 +510,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 267e95d2773..8ee61c45cec 100644 --- a/tests/unit/litellm_core_utils/test_realtime_streaming.py +++ b/tests/unit/litellm_core_utils/test_realtime_streaming.py @@ -648,9 +648,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( @@ -658,6 +659,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"}}, }, } @@ -681,6 +684,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(): @@ -3068,8 +3076,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 @@ -3082,8 +3104,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 704efe8323a..14e8bb34801 100644 --- a/tests/unit/llms/openai/realtime/test_openai_realtime_handler.py +++ b/tests/unit/llms/openai/realtime/test_openai_realtime_handler.py @@ -1,4 +1,7 @@ +import asyncio import json +from collections.abc import Mapping +from typing import Final from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -9,6 +12,7 @@ from litellm.types.realtime import RealtimeQueryParams from openai import AsyncOpenAI, omit +from pydantic import TypeAdapter class DummySDKConnectionManager: @@ -487,24 +491,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", @@ -514,13 +529,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, @@ -533,7 +542,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 fb676ae9a61..14f57d95632 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 @@ -588,6 +602,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