From 63d754d27b961b518e9a48cd95494075ba5ac581 Mon Sep 17 00:00:00 2001 From: Emerson Gomes Date: Fri, 5 Jun 2026 07:08:11 -0500 Subject: [PATCH 1/9] Add gpt-realtime-whisper Realtime transcription support (OpenAI + Azure) Adds first-class support for the gpt-realtime-whisper streaming speech-to-text model, which uses the Realtime transcription session API rather than the file-based /audio/transcriptions path. Model registration: registers gpt-realtime-whisper and azure/gpt-realtime-whisper with audio-duration pricing (input_cost_per_second = 0.017/60, matching the published $0.017/minute input audio rate). REST endpoint: implements POST /v1/realtime/transcription_sessions (plus /realtime and /openai/v1 aliases) to mint an ephemeral transcription session for the WebRTC flow. Adds request/response types, OpenAI and Azure URL builders, a shared base handler (refactored from the client_secrets handler), the acreate_realtime_transcription_session SDK function, and route registration. The proxy encrypts the ephemeral key returned under client_secret.value and records the session type in the token so the follow-up /realtime/calls replays type=transcription rather than type=realtime. WebSocket: forwards intent=transcription through to the Azure handler (OpenAI already received it) with URL-encoding, so gpt-realtime-whisper opens a transcription session. Transcription-only sessions no longer trigger an erroneous response.create. Cost tracking: transcription sessions emit no response.done events; their usage arrives on conversation.item.input_audio_transcription.completed as {type: duration, seconds}. That usage is captured out-of-band (usage only, no transcript duplication) and billed by input_cost_per_second, with a token-billed fallback for token-priced transcription models. Adds tests for pricing math, URL builders, request/response types, the proxy route and SDK function, WebSocket intent forwarding, transcription-session streaming behavior, and the /realtime/calls session-type replay. --- litellm/__init__.py | 1 + litellm/cost_calculator.py | 97 +++++++++ .../litellm_core_utils/realtime_streaming.py | 67 ++++++- litellm/llms/azure/realtime/handler.py | 19 +- .../azure/realtime/http_transformation.py | 7 + .../base_llm/realtime/http_transformation.py | 9 + litellm/llms/custom_httpx/llm_http_handler.py | 76 ++++++- .../openai/realtime/http_transformation.py | 8 + ...odel_prices_and_context_window_backup.json | 34 ++++ litellm/proxy/realtime_endpoints/endpoints.py | 144 +++++++++++++- litellm/proxy/route_llm_request.py | 3 + litellm/realtime_api/README.md | 62 +++++- litellm/realtime_api/main.py | 68 +++++++ litellm/types/realtime.py | 37 ++++ litellm/types/utils.py | 1 + model_prices_and_context_window.json | 34 ++++ .../test_realtime_streaming.py | 111 +++++++++++ .../realtime/test_azure_realtime_handler.py | 72 +++++++ .../realtime/test_transcription_sessions.py | 174 ++++++++++++++++ .../test_realtime_webrtc_endpoints.py | 188 ++++++++++++++++++ tests/test_litellm/test_cost_calculator.py | 123 +++++++++++- 21 files changed, 1324 insertions(+), 11 deletions(-) create mode 100644 tests/test_litellm/llms/openai/realtime/test_transcription_sessions.py diff --git a/litellm/__init__.py b/litellm/__init__.py index e49f4a4699d9..70224072ae63 100644 --- a/litellm/__init__.py +++ b/litellm/__init__.py @@ -1350,6 +1350,7 @@ def add_known_models(model_cost_map: Optional[Dict] = None): from .realtime_api.main import ( _arealtime, acreate_realtime_client_secret, + acreate_realtime_transcription_session, arealtime_calls, ) from .responses.main import _aresponses_websocket diff --git a/litellm/cost_calculator.py b/litellm/cost_calculator.py index 9a4b158b6228..aea1ba3f2201 100644 --- a/litellm/cost_calculator.py +++ b/litellm/cost_calculator.py @@ -2534,4 +2534,101 @@ def handle_realtime_stream_cost_calculation( break # exit if we find a valid model total_cost = input_cost_per_token + output_cost_per_token + total_cost += handle_realtime_transcription_cost_calculation( + results=results, + custom_llm_provider=custom_llm_provider, + litellm_model_name=litellm_model_name, + ) + + return total_cost + + +_TRANSCRIPTION_COMPLETED_EVENT_TYPE = ( + "conversation.item.input_audio_transcription.completed" +) + + +def handle_realtime_transcription_cost_calculation( + results: OpenAIRealtimeStreamList, + custom_llm_provider: str, + litellm_model_name: str, +) -> float: + """ + Cost for realtime transcription sessions (e.g. gpt-realtime-whisper). + + Transcription sessions emit no `response.done` events; instead each + `conversation.item.input_audio_transcription.completed` event carries a + `usage` object billed by the ASR model. The usage is one of: + - {"type": "duration", "seconds": } → priced via input_cost_per_second + - {"type": "tokens", "input_tokens": ...} → priced via input/audio token cost + """ + completed_events = [ + cast(dict, result) + for result in results + if result.get("type") == _TRANSCRIPTION_COMPLETED_EVENT_TYPE + ] + if not completed_events: + return 0.0 + + model_name = ( + _get_transcription_model_name_from_results(results) or litellm_model_name + ) + try: + model_info = litellm.get_model_info( + model=model_name, custom_llm_provider=custom_llm_provider + ) + except Exception: + model_info = {} + + total_cost = 0.0 + for event in completed_events: + usage = event.get("usage") or {} + total_cost += _transcription_usage_cost(usage, model_info) return total_cost + + +def _get_transcription_model_name_from_results( + results: OpenAIRealtimeStreamList, +) -> Optional[str]: + """Resolve the ASR model from a transcription_session.* / session.* event.""" + for result in results: + if result.get("type") in ( + "transcription_session.created", + "transcription_session.updated", + "session.created", + "session.updated", + ): + session = cast(dict, result).get("session", {}) or {} + transcription = ( + (session.get("audio", {}) or {}).get("input", {}) or {} + ).get("transcription", {}) or session.get("input_audio_transcription", {}) + model = (transcription or {}).get("model") + if model: + return model + return None + + +def _transcription_usage_cost(usage: dict, model_info: dict) -> float: + usage_type = usage.get("type") + if usage_type == "duration": + seconds = usage.get("seconds") or 0.0 + per_second = model_info.get("input_cost_per_second") or 0.0 + return float(seconds) * float(per_second) + if usage_type == "tokens": + input_token_details = usage.get("input_token_details") or {} + audio_tokens = input_token_details.get("audio_tokens") or 0 + text_tokens = input_token_details.get("text_tokens") or 0 + output_tokens = usage.get("output_tokens") or 0 + audio_cost = float(audio_tokens) * float( + model_info.get("input_cost_per_audio_token") + or model_info.get("input_cost_per_token") + or 0.0 + ) + text_cost = float(text_tokens) * float( + model_info.get("input_cost_per_token") or 0.0 + ) + output_cost = float(output_tokens) * float( + model_info.get("output_cost_per_token") or 0.0 + ) + return audio_cost + text_cost + output_cost + return 0.0 diff --git a/litellm/litellm_core_utils/realtime_streaming.py b/litellm/litellm_core_utils/realtime_streaming.py index 772f058d9bbd..cfabb6e5e1a5 100644 --- a/litellm/litellm_core_utils/realtime_streaming.py +++ b/litellm/litellm_core_utils/realtime_streaming.py @@ -100,6 +100,10 @@ def __init__( self._flushing_pending_messages_until_setup: bool = False self._pending_messages_until_setup: List[str] = [] self._pending_messages_byte_total: int = 0 + # Whether this is a transcription-only session (session.type == "transcription", + # e.g. gpt-realtime-whisper). Such sessions must not be sent response.create and + # their input_audio_transcription.completed usage drives duration-based cost. + self._is_transcription_session: bool = False # Per-connection caps for pre-setup audio frames (message count + total bytes). _MAX_BUFFERED_MESSAGES: int = 200 @@ -209,6 +213,8 @@ def _collect_user_input_from_client_event(self, message: Union[str, dict]) -> No self.session_tools = tools # GA: session.type is required; log it for traceability but no action needed verbose_logger.debug(f"Realtime session.type: {session.get('type')}") + if session.get("type") == "transcription": + self._is_transcription_session = True except (json.JSONDecodeError, AttributeError, TypeError): pass @@ -225,6 +231,55 @@ def _collect_user_input_from_backend_event( except (AttributeError, TypeError): pass + def _detect_transcription_session_from_backend( + self, event_obj: Union[dict, OpenAIRealtimeEvents] + ) -> None: + """Flag transcription-only sessions from backend session events.""" + try: + event_type = event_obj.get("type", "") + if event_type in ( + "transcription_session.created", + "transcription_session.updated", + ): + self._is_transcription_session = True + elif event_type in ("session.created", "session.updated"): + session = cast(dict, event_obj).get("session", {}) or {} + if session.get("type") == "transcription": + self._is_transcription_session = True + except (AttributeError, TypeError): + pass + + def _capture_transcription_usage( + self, event_obj: Union[dict, OpenAIRealtimeEvents] + ) -> None: + """ + Append a usage-only transcription completed event to the logged results so + the cost calculator can bill it by audio duration. The default logged event + types exclude this event, so it is captured here directly for transcription + sessions rather than widening logging for every realtime session. Only the + type and usage are kept — the transcript is already captured separately in + input_messages, so it is not duplicated into the response log here. + """ + try: + usage = event_obj.get("usage") + if usage is None: + return + # If this event type is already captured by store_message (e.g. the user + # logs all realtime events), don't append a second copy. + if self._should_store_message(event_obj): + return + self.messages.append( + cast( + OpenAIRealtimeEvents, + { + "type": "conversation.item.input_audio_transcription.completed", + "usage": usage, + }, + ) + ) + except (AttributeError, TypeError): + pass + def _collect_tool_calls_from_response_done( self, event_obj: Union[dict, OpenAIRealtimeEvents] ) -> None: @@ -713,6 +768,8 @@ async def _handle_raw_backend_message(self, raw_response) -> bool: try: event_obj = json.loads(raw_response) + self._detect_transcription_session_from_backend(event_obj) + # For audio/VAD guardrail path: once the session is ready, tell the backend # not to auto-respond after VAD detects end-of-speech. We send the # session.created to the client FIRST so the client is always in sync, then @@ -731,12 +788,20 @@ async def _handle_raw_backend_message(self, raw_response) -> bool: event_obj.get("type") == "conversation.item.input_audio_transcription.completed" ): - transcript = event_obj.get("transcript", "") self._collect_user_input_from_backend_event(event_obj) ## LOGGING — must happen before continue below self.store_message(raw_response) # Forward transcript to client so user sees what they said await self.websocket.send_text(raw_response) + + # Transcription-only sessions (e.g. gpt-realtime-whisper) have no + # assistant turn: capture audio-duration usage for cost and never + # trigger response.create. + if self._is_transcription_session: + self._capture_transcription_usage(event_obj) + return True + + transcript = event_obj.get("transcript", "") blocked = await self.run_realtime_guardrails( transcript, item_id=event_obj.get("item_id"), diff --git a/litellm/llms/azure/realtime/handler.py b/litellm/llms/azure/realtime/handler.py index 1f3357fd7883..8f0037570b07 100644 --- a/litellm/llms/azure/realtime/handler.py +++ b/litellm/llms/azure/realtime/handler.py @@ -5,6 +5,7 @@ """ from typing import Any, Optional, cast +from urllib.parse import quote from litellm._logging import _redact_string, verbose_proxy_logger from litellm.constants import REALTIME_WEBSOCKET_MAX_MESSAGE_SIZE_BYTES @@ -35,6 +36,7 @@ def _construct_url( model: str, api_version: Optional[str], realtime_protocol: Optional[str] = None, + query_params: Optional[dict] = None, ) -> str: """ Construct Azure realtime WebSocket URL. @@ -46,6 +48,7 @@ def _construct_url( realtime_protocol: Protocol version to use: - "GA" or "v1": Uses /openai/v1/realtime (GA path) - "beta" or None: Uses /openai/realtime (beta path, default) + query_params: Extra query params to forward (e.g. intent=transcription). Returns: WebSocket URL string @@ -63,11 +66,16 @@ def _construct_url( ) if _is_ga: path = "/openai/v1/realtime" - return f"{api_base}{path}?model={model}" + url = f"{api_base}{path}?model={model}" else: # Default to beta path for backwards compatibility path = "/openai/realtime" - return f"{api_base}{path}?api-version={api_version}&deployment={model}" + url = f"{api_base}{path}?api-version={api_version}&deployment={model}" + + intent = (query_params or {}).get("intent") + if intent: + url = f"{url}&intent={quote(str(intent), safe='')}" + return url async def async_realtime( self, @@ -81,6 +89,7 @@ async def async_realtime( client: Optional[Any] = None, timeout: Optional[float] = None, realtime_protocol: Optional[str] = None, + query_params: Optional[dict] = None, user_api_key_dict: Optional[Any] = None, litellm_metadata: Optional[dict] = None, ): @@ -96,7 +105,11 @@ async def async_realtime( raise ValueError("api_version is required for Azure OpenAI calls") url = self._construct_url( - api_base, model, api_version, realtime_protocol=realtime_protocol + api_base, + model, + api_version, + realtime_protocol=realtime_protocol, + query_params=query_params, ) try: diff --git a/litellm/llms/azure/realtime/http_transformation.py b/litellm/llms/azure/realtime/http_transformation.py index df1e2707af20..d6bdbd24db43 100644 --- a/litellm/llms/azure/realtime/http_transformation.py +++ b/litellm/llms/azure/realtime/http_transformation.py @@ -40,6 +40,13 @@ def get_realtime_calls_url( version = api_version or get_secret_str("AZURE_API_VERSION") or "2024-12-17" return f"{base}/openai/realtime/calls?api-version={version}" + def get_transcription_session_url( + self, api_base: Optional[str], model: str, api_version: Optional[str] = None + ) -> str: + base = self.get_api_base(api_base).rstrip("/") + version = api_version or get_secret_str("AZURE_API_VERSION") or "2024-12-17" + return f"{base}/openai/realtime/transcription_sessions?api-version={version}" + def get_realtime_calls_headers(self, ephemeral_key: str) -> dict: return { "api-key": ephemeral_key, diff --git a/litellm/llms/base_llm/realtime/http_transformation.py b/litellm/llms/base_llm/realtime/http_transformation.py index 712ec42380ff..be1413a3c0b9 100644 --- a/litellm/llms/base_llm/realtime/http_transformation.py +++ b/litellm/llms/base_llm/realtime/http_transformation.py @@ -59,6 +59,15 @@ def get_complete_url( ) -> str: """Return the full URL for POST /realtime/client_secrets.""" + def get_transcription_session_url( + self, api_base: Optional[str], model: str, api_version: Optional[str] = None + ) -> str: + """Return the full URL for POST /realtime/transcription_sessions.""" + base = (api_base or "").rstrip("/") + if base.endswith("/v1"): + base = base[:-3] + return f"{base}/v1/realtime/transcription_sessions" + @abstractmethod def validate_environment( self, diff --git a/litellm/llms/custom_httpx/llm_http_handler.py b/litellm/llms/custom_httpx/llm_http_handler.py index eedab7fc36cf..377faf1400c8 100644 --- a/litellm/llms/custom_httpx/llm_http_handler.py +++ b/litellm/llms/custom_httpx/llm_http_handler.py @@ -5382,6 +5382,69 @@ async def async_realtime_client_secret_handler( """ Forward POST /v1/realtime/client_secrets to upstream provider. + Uses provider_config (BaseRealtimeHTTPConfig) for URL construction and + header auth when available; falls back to the legacy OpenAI-style defaults. + """ + return await self._async_realtime_session_post( + endpoint="client_secrets", + api_base=api_base, + api_key=api_key, + request_data=request_data, + logging_obj=logging_obj, + timeout=timeout, + provider_config=provider_config, + model=model, + extra_headers=extra_headers, + client=client, + api_version=api_version, + ) + + async def async_realtime_transcription_session_handler( + self, + api_base: str, + api_key: str, + request_data: Dict[str, Any], + logging_obj: LiteLLMLoggingObj, + timeout: Union[float, httpx.Timeout], + provider_config: Optional[Any] = None, + model: Optional[str] = None, + extra_headers: Optional[Dict[str, Any]] = None, + client: Optional[Union[HTTPHandler, AsyncHTTPHandler]] = None, + api_version: Optional[str] = None, + ) -> httpx.Response: + """Forward POST /v1/realtime/transcription_sessions to upstream provider.""" + return await self._async_realtime_session_post( + endpoint="transcription_sessions", + api_base=api_base, + api_key=api_key, + request_data=request_data, + logging_obj=logging_obj, + timeout=timeout, + provider_config=provider_config, + model=model, + extra_headers=extra_headers, + client=client, + api_version=api_version, + ) + + async def _async_realtime_session_post( + self, + endpoint: Literal["client_secrets", "transcription_sessions"], + api_base: str, + api_key: str, + request_data: Dict[str, Any], + logging_obj: LiteLLMLoggingObj, + timeout: Union[float, httpx.Timeout], + provider_config: Optional[Any] = None, + model: Optional[str] = None, + extra_headers: Optional[Dict[str, Any]] = None, + client: Optional[Union[HTTPHandler, AsyncHTTPHandler]] = None, + api_version: Optional[str] = None, + ) -> httpx.Response: + """ + Shared POST flow for the realtime HTTP session endpoints + (client_secrets and transcription_sessions). + Uses provider_config (BaseRealtimeHTTPConfig) for URL construction and header auth when available; falls back to the legacy OpenAI-style defaults. """ @@ -5393,14 +5456,19 @@ async def async_realtime_client_secret_handler( async_httpx_client = client if provider_config is not None: - url = provider_config.get_complete_url( - api_base=api_base, model=model or "", api_version=api_version - ) + if endpoint == "transcription_sessions": + url = provider_config.get_transcription_session_url( + api_base=api_base, model=model or "", api_version=api_version + ) + else: + url = provider_config.get_complete_url( + api_base=api_base, model=model or "", api_version=api_version + ) headers: Dict[str, Any] = provider_config.validate_environment( headers={}, model=model or "", api_key=api_key ) else: - url = f"{api_base.rstrip('/')}/v1/realtime/client_secrets" + url = f"{api_base.rstrip('/')}/v1/realtime/{endpoint}" headers = { "Authorization": f"Bearer {api_key}", "Content-Type": "application/json", diff --git a/litellm/llms/openai/realtime/http_transformation.py b/litellm/llms/openai/realtime/http_transformation.py index 1663fcd1fcd9..7a6af39ba655 100644 --- a/litellm/llms/openai/realtime/http_transformation.py +++ b/litellm/llms/openai/realtime/http_transformation.py @@ -41,6 +41,14 @@ def get_realtime_calls_url( base = base[:-3] return f"{base}/v1/realtime/calls" + def get_transcription_session_url( + self, api_base: Optional[str], model: str, api_version: Optional[str] = None + ) -> str: + base = self.get_api_base(api_base).rstrip("/") + if base.endswith("/v1"): + base = base[:-3] + return f"{base}/v1/realtime/transcription_sessions" + def validate_environment( self, headers: dict, diff --git a/litellm/model_prices_and_context_window_backup.json b/litellm/model_prices_and_context_window_backup.json index 4eec27f29bd2..6b7e4511d7e1 100644 --- a/litellm/model_prices_and_context_window_backup.json +++ b/litellm/model_prices_and_context_window_backup.json @@ -4204,6 +4204,23 @@ "/v1/audio/transcriptions" ] }, + "azure/gpt-realtime-whisper": { + "input_cost_per_second": 0.0002833333333333333, + "litellm_provider": "azure", + "mode": "audio_transcription", + "source": "https://learn.microsoft.com/en-us/azure/ai-foundry/openai/concepts/gpt-realtime-whisper", + "supported_endpoints": [ + "/v1/realtime", + "/v1/realtime/transcription_sessions" + ], + "supported_modalities": [ + "audio" + ], + "supported_output_modalities": [ + "text" + ], + "supports_audio_input": true + }, "azure/gpt-5.1-2025-11-13": { "cache_read_input_token_cost": 1.25e-07, "cache_read_input_token_cost_priority": 2.5e-07, @@ -40507,6 +40524,23 @@ "supports_system_messages": true, "supports_tool_choice": true }, + "gpt-realtime-whisper": { + "input_cost_per_second": 0.0002833333333333333, + "litellm_provider": "openai", + "mode": "audio_transcription", + "source": "https://platform.openai.com/docs/models/gpt-realtime-whisper", + "supported_endpoints": [ + "/v1/realtime", + "/v1/realtime/transcription_sessions" + ], + "supported_modalities": [ + "audio" + ], + "supported_output_modalities": [ + "text" + ], + "supports_audio_input": true + }, "sora-2": { "litellm_provider": "openai", "mode": "video_generation", diff --git a/litellm/proxy/realtime_endpoints/endpoints.py b/litellm/proxy/realtime_endpoints/endpoints.py index 14d004d977ee..a2afd868ec40 100644 --- a/litellm/proxy/realtime_endpoints/endpoints.py +++ b/litellm/proxy/realtime_endpoints/endpoints.py @@ -19,6 +19,8 @@ from litellm.types.realtime import ( RealtimeClientSecretRequest, RealtimeClientSecretResponse, + RealtimeTranscriptionSessionRequest, + RealtimeTranscriptionSessionResponse, ) router = APIRouter() @@ -32,6 +34,7 @@ def _encode_realtime_token_payload( user_id: Optional[str], team_id: Optional[str], expires_at: Optional[int], + session_type: str = "realtime", ) -> str: """ Encode metadata with the upstream ephemeral key so /realtime/calls can @@ -44,6 +47,7 @@ def _encode_realtime_token_payload( "user_id": user_id or "", "team_id": team_id or "", "expires_at": expires_at, + "session_type": session_type, } return json.dumps(payload, separators=(",", ":")) @@ -199,6 +203,9 @@ async def create_realtime_client_secret( user_id=getattr(user_api_key_dict, "user_id", None), team_id=getattr(user_api_key_dict, "team_id", None), expires_at=expires_at if isinstance(expires_at, int) else None, + session_type=( + req.session.type if req.session and req.session.type else "realtime" + ), ) encrypted_token: str = encrypt_value_helper(token_payload) upstream_json["value"] = encrypted_token @@ -283,12 +290,14 @@ async def proxy_realtime_calls( ) user_id = decoded_payload.get("user_id") or None team_id = decoded_payload.get("team_id") or None + session_type = decoded_payload.get("session_type") or "realtime" else: # Backward compatibility: older tokens contained only encrypted upstream key. openai_ephemeral_key = decrypted_token_value model = request.query_params.get("model", "gpt-4o-realtime-preview") user_id = None team_id = None + session_type = "realtime" # Build a minimal UserAPIKeyAuth with user/team IDs from the token # so spend tracking and budget enforcement work correctly. @@ -301,7 +310,7 @@ async def proxy_realtime_calls( try: # Build session config for the multipart form data session_config = { - "type": "realtime", + "type": session_type, "model": model, } @@ -366,3 +375,136 @@ async def proxy_realtime_calls( status_code=upstream_resp.status_code, media_type=upstream_resp.headers.get("content-type", "application/sdp"), ) + + +@router.post( + "/v1/realtime/transcription_sessions", + dependencies=[Depends(user_api_key_auth)], + tags=["realtime"], +) +@router.post( + "/realtime/transcription_sessions", + dependencies=[Depends(user_api_key_auth)], + tags=["realtime"], +) +@router.post( + "/openai/v1/realtime/transcription_sessions", + dependencies=[Depends(user_api_key_auth)], + tags=["realtime"], +) +async def create_realtime_transcription_session( + request: Request, + fastapi_response: Response, + user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth), +) -> RealtimeTranscriptionSessionResponse: + """ + Create an ephemeral Realtime transcription session + (POST /v1/realtime/transcription_sessions) for the WebRTC/WebSocket flow. + + Mirrors the client_secrets route but targets the transcription_sessions + endpoint and encrypts the ephemeral key returned under `client_secret.value`. + """ + from litellm.proxy.proxy_server import ( + add_litellm_data_to_request, + general_settings, + llm_router, + proxy_config, + proxy_logging_obj, + route_request, + user_model, + version, + ) + + data: dict = {} + try: + body = await _read_request_body(request=request) + req = RealtimeTranscriptionSessionRequest(**body) + + model: str = req.resolved_model() or "gpt-realtime-whisper" + + transcription_session = {k: v for k, v in body.items() if k != "model"} + data = {"model": model, "transcription_session": transcription_session} + + data = await add_litellm_data_to_request( + data=data, + request=request, + general_settings=general_settings, + user_api_key_dict=user_api_key_dict, + version=version, + proxy_config=proxy_config, + ) + + data = await proxy_logging_obj.pre_call_hook( + user_api_key_dict=user_api_key_dict, + data=data, + call_type="acreate_realtime_transcription_session", + ) + + verbose_proxy_logger.debug( + "Realtime: /v1/realtime/transcription_sessions (model=%s)", model + ) + + llm_call = await route_request( + data=data, + route_type="acreate_realtime_transcription_session", + llm_router=llm_router, + user_model=user_model, + ) + upstream_resp: httpx.Response = await llm_call # type: ignore + + except Exception as e: + await proxy_logging_obj.post_call_failure_hook( + user_api_key_dict=user_api_key_dict, + original_exception=e, + request_data=data, + ) + verbose_proxy_logger.error( + "litellm.proxy.realtime_endpoints.create_realtime_transcription_session(): Exception - %s", + str(e), + ) + if isinstance(e, HTTPException): + raise ProxyException( + message=getattr(e, "message", str(e)), + type=getattr(e, "type", "None"), + param=getattr(e, "param", "None"), + code=getattr(e, "status_code", http_status.HTTP_400_BAD_REQUEST), + ) + raise ProxyException( + message=getattr(e, "message", str(e)), + type=getattr(e, "type", "None"), + param=getattr(e, "param", "None"), + code=getattr(e, "status_code", 500), + ) + + if upstream_resp.status_code != 200: + verbose_proxy_logger.error( + "Realtime transcription_sessions upstream error %s: %s", + upstream_resp.status_code, + upstream_resp.text, + ) + return Response( # type: ignore[return-value] + content=upstream_resp.content, + status_code=upstream_resp.status_code, + media_type="application/json", + ) + + upstream_json: dict = upstream_resp.json() + + # Encrypt the ephemeral key (returned under client_secret.value) with routing + # metadata so the follow-up /realtime/calls request can recover the model. + client_secret = upstream_json.get("client_secret") + if isinstance(client_secret, dict) and "value" in client_secret: + raw_value: str = client_secret.get("value", "") + expires_at = client_secret.get("expires_at") + token_payload = _encode_realtime_token_payload( + ephemeral_key=raw_value, + model_id=model, + user_id=getattr(user_api_key_dict, "user_id", None), + team_id=getattr(user_api_key_dict, "team_id", None), + expires_at=expires_at if isinstance(expires_at, int) else None, + session_type="transcription", + ) + client_secret["value"] = encrypt_value_helper(token_payload) + upstream_json["client_secret"] = client_secret + + return RealtimeTranscriptionSessionResponse(**upstream_json) diff --git a/litellm/proxy/route_llm_request.py b/litellm/proxy/route_llm_request.py index 8f6f7084a0c9..d573839b013b 100644 --- a/litellm/proxy/route_llm_request.py +++ b/litellm/proxy/route_llm_request.py @@ -74,6 +74,7 @@ def _is_a2a_agent_model(model_name: Any) -> bool: "avideo_extension": "/videos/extensions", "acreate_realtime_client_secret": "/realtime/client_secrets", "arealtime_calls": "/realtime/calls", + "acreate_realtime_transcription_session": "/realtime/transcription_sessions", "acreate_container": "/containers", "alist_containers": "/containers", "aretrieve_container": "/containers/{container_id}", @@ -261,6 +262,7 @@ async def route_request( # noqa: PLR0915 - Complex routing function, refactorin "_arealtime", # private function for realtime API "acreate_realtime_client_secret", "arealtime_calls", + "acreate_realtime_transcription_session", "_aresponses_websocket", # private function for responses WebSocket mode "aimage_edit", "agenerate_content", @@ -427,6 +429,7 @@ async def route_request( # noqa: PLR0915 - Complex routing function, refactorin "adelete_run", "acreate_realtime_client_secret", "arealtime_calls", + "acreate_realtime_transcription_session", ]: # If a model is provided, get its credentials from the router model = data.get("model") diff --git a/litellm/realtime_api/README.md b/litellm/realtime_api/README.md index 6b467c056a60..cf3d2973cfc6 100644 --- a/litellm/realtime_api/README.md +++ b/litellm/realtime_api/README.md @@ -1 +1,61 @@ -Abstraction / Routing logic for OpenAI's `/v1/realtime` endpoint. \ No newline at end of file +Abstraction / Routing logic for OpenAI's `/v1/realtime` endpoints. + +## Realtime transcription (`gpt-realtime-whisper`) + +`gpt-realtime-whisper` is the low-latency streaming speech-to-text model. It is a +Realtime transcription session, not the file-based `/audio/transcriptions` path. Use +the standard `gpt-4o-transcribe` / `whisper-1` models for request/response or file +transcription; use `gpt-realtime-whisper` for live streaming transcript deltas. + +Both OpenAI and Azure OpenAI (Microsoft Foundry) are supported. Cost is tracked by input +audio duration (OpenAI: $0.017/minute), derived from the +`conversation.item.input_audio_transcription.completed` usage events. + +### WebSocket + +Connect to the proxy realtime WebSocket with `intent=transcription`, then send a +`session.update` configuring a transcription session: + +``` +wss:///v1/realtime?model=gpt-realtime-whisper&intent=transcription +``` + +```json +{ + "type": "session.update", + "session": { + "type": "transcription", + "audio": { + "input": { + "format": { "type": "audio/pcm", "rate": 24000 }, + "transcription": { "model": "gpt-realtime-whisper", "language": "en" } + } + } + } +} +``` + +Append audio with `input_audio_buffer.append`, then `input_audio_buffer.commit` (when not +using server VAD). Listen for `conversation.item.input_audio_transcription.delta` and +`.completed` events. The proxy does not auto-trigger `response.create` for transcription +sessions. + +### Ephemeral transcription session (WebRTC) + +`POST /v1/realtime/transcription_sessions` mints an ephemeral session for browser/WebRTC +clients. The returned `client_secret.value` is encrypted by the proxy and exchanged via +`POST /v1/realtime/calls`. + +```bash +curl https:///v1/realtime/transcription_sessions \ + -H "Authorization: Bearer $LITELLM_KEY" \ + -H "Content-Type: application/json" \ + -d '{ + "input_audio_format": "pcm16", + "input_audio_transcription": { "model": "gpt-realtime-whisper", "language": "en" } + }' +``` + +For Azure, route to an `azure/gpt-realtime-whisper` deployment; the proxy targets +`/openai/realtime/transcription_sessions?api-version=...` and forwards +`intent=transcription` on the WebSocket. \ No newline at end of file diff --git a/litellm/realtime_api/main.py b/litellm/realtime_api/main.py index 95d6f7c3e034..4ec00e1f65bc 100644 --- a/litellm/realtime_api/main.py +++ b/litellm/realtime_api/main.py @@ -15,6 +15,7 @@ RealtimeExpiresAfter, RealtimeQueryParams, RealtimeSessionConfig, + RealtimeTranscriptionSessionRequest, ) from litellm.types.router import GenericLiteLLMParams from litellm.types.utils import LlmProviders @@ -159,6 +160,72 @@ async def acreate_realtime_client_secret( ) +@wrapper_client +async def acreate_realtime_transcription_session( + model: Optional[str] = None, + transcription_session: Optional[Dict[str, Any]] = None, + timeout: Optional[float] = None, + **kwargs, +): + """ + Create an ephemeral transcription session via POST + /v1/realtime/transcription_sessions. + + ``transcription_session`` is the upstream request body (input_audio_format, + input_audio_transcription, turn_detection, …). ``model`` is a LiteLLM-only + routing hint; the provider model lives in + ``transcription_session.input_audio_transcription.model``. + """ + req = RealtimeTranscriptionSessionRequest( + model=model, + **(transcription_session or {}), + ) + model_name = req.resolved_model() or "gpt-realtime-whisper" + litellm_logging_obj: LiteLLMLogging = kwargs.get("litellm_logging_obj") # type: ignore + litellm_params = GenericLiteLLMParams(**kwargs) + + ( + model_name, + custom_llm_provider, + dynamic_api_key, + dynamic_api_base, + ) = get_llm_provider( + model=model_name, + api_base=litellm_params.api_base, + api_key=litellm_params.api_key, + ) + ( + provider_config, + resolved_api_base, + resolved_api_key, + ) = _get_realtime_http_provider_config( + custom_llm_provider=custom_llm_provider, + dynamic_api_base=dynamic_api_base, + dynamic_api_key=dynamic_api_key, + litellm_params=litellm_params, + ) + litellm_logging_obj.update_from_kwargs( + kwargs=kwargs, + model=model_name, + optional_params={"transcription_session": transcription_session}, + litellm_params={"api_base": resolved_api_base}, + custom_llm_provider=custom_llm_provider, + ) + request_data = req.model_dump(exclude_none=True, exclude={"model"}) + return await base_llm_http_handler.async_realtime_transcription_session_handler( + api_base=resolved_api_base, + api_key=resolved_api_key, + request_data=request_data, + logging_obj=litellm_logging_obj, + timeout=timeout or request_timeout, + provider_config=provider_config, + model=model_name, + extra_headers=kwargs.get("extra_headers"), + client=kwargs.get("client"), + api_version=litellm_params.api_version, + ) + + @wrapper_client async def arealtime_calls( openai_ephemeral_key: str, @@ -313,6 +380,7 @@ async def _arealtime( # noqa: PLR0915 timeout=timeout, logging_obj=litellm_logging_obj, realtime_protocol=realtime_protocol, + query_params=query_params, user_api_key_dict=kwargs.get("user_api_key_dict"), litellm_metadata=_build_litellm_metadata(kwargs), ) diff --git a/litellm/types/realtime.py b/litellm/types/realtime.py index 62e4044061bd..0db8232a54d3 100644 --- a/litellm/types/realtime.py +++ b/litellm/types/realtime.py @@ -115,3 +115,40 @@ class RealtimeClientSecretResponse(BaseModel): expires_at: Optional[int] = None value: str session: Optional[Dict[str, Any]] = None + + +class RealtimeTranscriptionSessionRequest(BaseModel): + """ + Request body for POST /v1/realtime/transcription_sessions. + + Mirrors OpenAI's RealtimeTranscriptionSessionCreateRequest. The model used + for routing is taken from the LiteLLM-only top-level `model` hint, falling + back to `input_audio_transcription.model`. All other fields pass through + unchanged to the provider. + """ + + model_config = {"extra": "allow"} + + # LiteLLM-only routing hint — stripped before forwarding upstream. + model: Optional[str] = None + input_audio_transcription: Optional[Dict[str, Any]] = None + + def resolved_model(self) -> Optional[str]: + if self.model: + return self.model + if self.input_audio_transcription: + return self.input_audio_transcription.get("model") + return None + + +class RealtimeTranscriptionSessionResponse(BaseModel): + """ + Response from POST /v1/realtime/transcription_sessions. + + `client_secret.value` contains the encrypted token instead of the raw + ephemeral key. Unknown fields pass through unchanged. + """ + + model_config = {"extra": "allow"} + + client_secret: Optional[Dict[str, Any]] = None diff --git a/litellm/types/utils.py b/litellm/types/utils.py index 3dcff2be689d..d4c306309e57 100644 --- a/litellm/types/utils.py +++ b/litellm/types/utils.py @@ -527,6 +527,7 @@ class CallTypes(str, Enum): "acreate_skill", "acreate_realtime_client_secret", "arealtime_calls", + "acreate_realtime_transcription_session", ] # Mapping of API routes to their corresponding call types diff --git a/model_prices_and_context_window.json b/model_prices_and_context_window.json index c50b3def6510..54d963e82d29 100644 --- a/model_prices_and_context_window.json +++ b/model_prices_and_context_window.json @@ -4204,6 +4204,23 @@ "/v1/audio/transcriptions" ] }, + "azure/gpt-realtime-whisper": { + "input_cost_per_second": 0.0002833333333333333, + "litellm_provider": "azure", + "mode": "audio_transcription", + "source": "https://learn.microsoft.com/en-us/azure/ai-foundry/openai/concepts/gpt-realtime-whisper", + "supported_endpoints": [ + "/v1/realtime", + "/v1/realtime/transcription_sessions" + ], + "supported_modalities": [ + "audio" + ], + "supported_output_modalities": [ + "text" + ], + "supports_audio_input": true + }, "azure/gpt-5.1-2025-11-13": { "cache_read_input_token_cost": 1.25e-07, "cache_read_input_token_cost_priority": 2.5e-07, @@ -40542,6 +40559,23 @@ "supports_system_messages": true, "supports_tool_choice": true }, + "gpt-realtime-whisper": { + "input_cost_per_second": 0.0002833333333333333, + "litellm_provider": "openai", + "mode": "audio_transcription", + "source": "https://platform.openai.com/docs/models/gpt-realtime-whisper", + "supported_endpoints": [ + "/v1/realtime", + "/v1/realtime/transcription_sessions" + ], + "supported_modalities": [ + "audio" + ], + "supported_output_modalities": [ + "text" + ], + "supports_audio_input": true + }, "sora-2": { "litellm_provider": "openai", "mode": "video_generation", diff --git a/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py b/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py index 3424bfd801c8..870b0051f786 100644 --- a/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py +++ b/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py @@ -521,6 +521,117 @@ async def test_transcription_captured_in_backend_to_client(): assert logging_obj.model_call_details["messages"] == streaming.input_messages +@pytest.mark.asyncio +async def test_transcription_session_captures_usage_and_skips_response_create(): + """ + For a transcription-only session (session.type == "transcription", e.g. + gpt-realtime-whisper), the completed event's audio-duration usage must be + captured for cost and response.create must NOT be sent to the backend. + """ + client_ws = MagicMock() + client_ws.send_text = AsyncMock() + + session_created = json.dumps( + { + "type": "session.created", + "session": { + "type": "transcription", + "audio": { + "input": {"transcription": {"model": "gpt-realtime-whisper"}} + }, + }, + } + ).encode() + completed = json.dumps( + { + "type": "conversation.item.input_audio_transcription.completed", + "transcript": "hello world", + "item_id": "item_1", + "usage": {"type": "duration", "seconds": 12.0}, + } + ).encode() + + backend_ws = MagicMock() + backend_ws.recv = AsyncMock( + side_effect=[session_created, completed, ConnectionClosed(None, None)] + ) + backend_ws.send = AsyncMock() + + logging_obj = MagicMock() + logging_obj.model_call_details = {} + logging_obj.async_success_handler = AsyncMock() + logging_obj.success_handler = MagicMock() + + streaming = RealTimeStreaming(client_ws, backend_ws, logging_obj) + await streaming.backend_to_client_send_messages() + + assert streaming._is_transcription_session is True + + captured = [ + m + for m in streaming.messages + if m.get("type") == "conversation.item.input_audio_transcription.completed" + ] + assert len(captured) == 1, "completed usage event must be captured for cost" + assert captured[0]["usage"]["seconds"] == 12.0 + + # Transcript still forwarded to the client. + client_ws.send_text.assert_any_call(completed.decode()) + + # No response.create — transcription sessions have no assistant turn. + sent_to_backend = [ + json.loads(c.args[0]) for c in backend_ws.send.call_args_list if c.args + ] + assert all( + e.get("type") != "response.create" for e in sent_to_backend + ), f"transcription session must not trigger response.create, got: {sent_to_backend}" + + +@pytest.mark.asyncio +async def test_non_transcription_completed_event_still_triggers_response_create(): + """ + Regression guard: a normal (non-transcription) session with no guardrails must + keep triggering response.create on a completed transcription event. + """ + client_ws = MagicMock() + client_ws.send_text = AsyncMock() + + completed = json.dumps( + { + "type": "conversation.item.input_audio_transcription.completed", + "transcript": "hi", + "item_id": "item_1", + } + ).encode() + + backend_ws = MagicMock() + backend_ws.recv = AsyncMock(side_effect=[completed, ConnectionClosed(None, None)]) + backend_ws.send = AsyncMock() + + logging_obj = MagicMock() + logging_obj.async_success_handler = AsyncMock() + logging_obj.success_handler = MagicMock() + + streaming = RealTimeStreaming(client_ws, backend_ws, logging_obj) + await streaming.backend_to_client_send_messages() + + assert streaming._is_transcription_session is False + sent_to_backend = [ + json.loads(c.args[0]) for c in backend_ws.send.call_args_list if c.args + ] + assert any(e.get("type") == "response.create" for e in sent_to_backend) + + +def test_client_session_update_marks_transcription_session(): + """A client session.update with type=transcription flags the session.""" + streaming = RealTimeStreaming(MagicMock(), MagicMock(), MagicMock()) + assert streaming._is_transcription_session is False + streaming._collect_user_input_from_client_event( + json.dumps({"type": "session.update", "session": {"type": "transcription"}}) + ) + assert streaming._is_transcription_session is True + + @pytest.mark.asyncio async def test_client_ack_caches_setup_to_prevent_duplicate_session_update_setup(): websocket = MagicMock() diff --git a/tests/test_litellm/llms/azure/realtime/test_azure_realtime_handler.py b/tests/test_litellm/llms/azure/realtime/test_azure_realtime_handler.py index 41d301c5d5fd..33e960fa1fd8 100644 --- a/tests/test_litellm/llms/azure/realtime/test_azure_realtime_handler.py +++ b/tests/test_litellm/llms/azure/realtime/test_azure_realtime_handler.py @@ -147,6 +147,78 @@ async def test_construct_url_ga_protocol(): assert "deployment" not in url +@pytest.mark.asyncio +async def test_construct_url_forwards_transcription_intent_ga(): + """ + Transcription sessions connect with intent=transcription. The Azure handler + must forward that query param so gpt-realtime-whisper opens a transcription + session instead of a normal realtime session. + """ + from litellm.llms.azure.realtime.handler import AzureOpenAIRealtime + + handler = AzureOpenAIRealtime() + url = handler._construct_url( + api_base="https://my-endpoint.openai.azure.com", + model="gpt-realtime-whisper", + api_version="2025-04-01-preview", + realtime_protocol="GA", + query_params={"model": "gpt-realtime-whisper", "intent": "transcription"}, + ) + + assert "/openai/v1/realtime?" in url + assert "intent=transcription" in url + + +@pytest.mark.asyncio +async def test_construct_url_forwards_transcription_intent_beta(): + from litellm.llms.azure.realtime.handler import AzureOpenAIRealtime + + handler = AzureOpenAIRealtime() + url = handler._construct_url( + api_base="https://my-endpoint.openai.azure.com", + model="whisper-deploy", + api_version="2024-10-01-preview", + query_params={"intent": "transcription"}, + ) + + assert "/openai/realtime?" in url + assert "deployment=whisper-deploy" in url + assert "intent=transcription" in url + + +@pytest.mark.asyncio +async def test_construct_url_encodes_intent_value(): + """A crafted intent value must be URL-encoded, not injected as raw query params.""" + from litellm.llms.azure.realtime.handler import AzureOpenAIRealtime + + handler = AzureOpenAIRealtime() + url = handler._construct_url( + api_base="https://my-endpoint.openai.azure.com", + model="gpt-realtime-whisper", + api_version="2025-04-01-preview", + realtime_protocol="GA", + query_params={"intent": "transcription&foo=bar"}, + ) + assert "intent=transcription%26foo%3Dbar" in url + assert "&foo=bar" not in url + + +@pytest.mark.asyncio +async def test_construct_url_no_intent_when_absent(): + """No intent param leaks into the URL when not provided.""" + from litellm.llms.azure.realtime.handler import AzureOpenAIRealtime + + handler = AzureOpenAIRealtime() + url = handler._construct_url( + api_base="https://my-endpoint.openai.azure.com", + model="gpt-4o-realtime-preview", + api_version="2024-10-01-preview", + realtime_protocol="GA", + query_params={"model": "gpt-4o-realtime-preview"}, + ) + assert "intent=" not in url + + @pytest.mark.asyncio async def test_construct_url_v1_protocol(): """ diff --git a/tests/test_litellm/llms/openai/realtime/test_transcription_sessions.py b/tests/test_litellm/llms/openai/realtime/test_transcription_sessions.py new file mode 100644 index 000000000000..afcfcc753362 --- /dev/null +++ b/tests/test_litellm/llms/openai/realtime/test_transcription_sessions.py @@ -0,0 +1,174 @@ +""" +Tests for the Realtime transcription_sessions surface used by gpt-realtime-whisper: + - OpenAI / Azure URL construction (POST /v1/realtime/transcription_sessions) + - RealtimeTranscriptionSessionRequest model-resolution + passthrough + - BaseLLMHTTPHandler.async_realtime_transcription_session_handler targeting +""" + +import os +import sys +from unittest.mock import AsyncMock, MagicMock + +import httpx +import pytest + +sys.path.insert(0, os.path.abspath("../../../../..")) + +from litellm.llms.azure.realtime.http_transformation import AzureRealtimeHTTPConfig +from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler +from litellm.llms.custom_httpx.llm_http_handler import BaseLLMHTTPHandler +from litellm.llms.openai.realtime.http_transformation import OpenAIRealtimeHTTPConfig +from litellm.types.realtime import RealtimeTranscriptionSessionRequest + + +def test_openai_transcription_session_url(): + cfg = OpenAIRealtimeHTTPConfig() + assert ( + cfg.get_transcription_session_url( + api_base="https://api.openai.com", model="gpt-realtime-whisper" + ) + == "https://api.openai.com/v1/realtime/transcription_sessions" + ) + + +def test_openai_transcription_session_url_strips_trailing_v1(): + """A /v1 suffix must not be duplicated in the path.""" + cfg = OpenAIRealtimeHTTPConfig() + assert ( + cfg.get_transcription_session_url( + api_base="https://api.openai.com/v1", model="gpt-realtime-whisper" + ) + == "https://api.openai.com/v1/realtime/transcription_sessions" + ) + + +def test_azure_transcription_session_url_uses_deployment_and_api_version(): + cfg = AzureRealtimeHTTPConfig() + url = cfg.get_transcription_session_url( + api_base="https://my.openai.azure.com", + model="whisper-deploy", + api_version="2025-04-01-preview", + ) + assert ( + url + == "https://my.openai.azure.com/openai/realtime/transcription_sessions?api-version=2025-04-01-preview" + ) + + +def test_request_resolves_model_from_top_level_hint(): + req = RealtimeTranscriptionSessionRequest( + model="openai/gpt-realtime-whisper", + input_audio_transcription={"model": "gpt-realtime-whisper"}, + ) + assert req.resolved_model() == "openai/gpt-realtime-whisper" + + +def test_request_resolves_model_from_input_audio_transcription(): + req = RealtimeTranscriptionSessionRequest( + input_audio_transcription={"model": "gpt-realtime-whisper", "language": "en"}, + ) + assert req.resolved_model() == "gpt-realtime-whisper" + + +def test_request_passthrough_excludes_routing_hint(): + """Unknown fields pass through; the litellm-only `model` hint is not forwarded.""" + req = RealtimeTranscriptionSessionRequest( + model="openai/gpt-realtime-whisper", + input_audio_format="pcm16", + input_audio_transcription={"model": "gpt-realtime-whisper"}, + turn_detection=None, + ) + forwarded = req.model_dump(exclude_none=True, exclude={"model"}) + assert "model" not in forwarded + assert forwarded["input_audio_format"] == "pcm16" + assert forwarded["input_audio_transcription"] == {"model": "gpt-realtime-whisper"} + + +@pytest.mark.asyncio +async def test_handler_posts_to_transcription_sessions_url(): + handler = BaseLLMHTTPHandler() + + mock_response = MagicMock(spec=httpx.Response) + mock_client = MagicMock(spec=AsyncHTTPHandler) + mock_client.post = AsyncMock(return_value=mock_response) + + logging_obj = MagicMock() + logging_obj.pre_call = MagicMock() + + request_body = {"input_audio_transcription": {"model": "gpt-realtime-whisper"}} + result = await handler.async_realtime_transcription_session_handler( + api_base="https://api.openai.com", + api_key="sk-test", + request_data=request_body, + logging_obj=logging_obj, + timeout=10.0, + provider_config=OpenAIRealtimeHTTPConfig(), + model="gpt-realtime-whisper", + client=mock_client, + ) + + assert result is mock_response + _, kwargs = mock_client.post.call_args + assert kwargs["url"] == "https://api.openai.com/v1/realtime/transcription_sessions" + assert kwargs["json"] == request_body + assert kwargs["headers"]["Authorization"] == "Bearer sk-test" + + +@pytest.mark.asyncio +async def test_client_secret_handler_still_targets_client_secrets_url(): + """Refactor regression: the client_secrets handler must keep its own URL.""" + handler = BaseLLMHTTPHandler() + + mock_response = MagicMock(spec=httpx.Response) + mock_client = MagicMock(spec=AsyncHTTPHandler) + mock_client.post = AsyncMock(return_value=mock_response) + + logging_obj = MagicMock() + logging_obj.pre_call = MagicMock() + + await handler.async_realtime_client_secret_handler( + api_base="https://api.openai.com", + api_key="sk-test", + request_data={"session": {"type": "realtime"}}, + logging_obj=logging_obj, + timeout=10.0, + provider_config=OpenAIRealtimeHTTPConfig(), + model="gpt-4o-realtime-preview", + client=mock_client, + ) + + _, kwargs = mock_client.post.call_args + assert kwargs["url"] == "https://api.openai.com/v1/realtime/client_secrets" + + +@pytest.mark.asyncio +async def test_sdk_fn_routes_openai_transcription_session(monkeypatch): + """ + litellm.acreate_realtime_transcription_session resolves the OpenAI provider + from the transcription model and POSTs to the OpenAI transcription_sessions URL. + """ + import litellm + + monkeypatch.setenv("OPENAI_API_KEY", "sk-unit-test") + + mock_response = MagicMock(spec=httpx.Response) + mock_client = MagicMock(spec=AsyncHTTPHandler) + mock_client.post = AsyncMock(return_value=mock_response) + + result = await litellm.acreate_realtime_transcription_session( + model="openai/gpt-realtime-whisper", + transcription_session={ + "input_audio_format": "pcm16", + "input_audio_transcription": {"model": "gpt-realtime-whisper"}, + }, + client=mock_client, + ) + + assert result is mock_response + _, kwargs = mock_client.post.call_args + assert kwargs["url"].endswith("/v1/realtime/transcription_sessions") + # The litellm-only routing hint must not be forwarded upstream. + assert "model" not in kwargs["json"] + assert kwargs["json"]["input_audio_transcription"] == { + "model": "gpt-realtime-whisper" + } diff --git a/tests/test_litellm/proxy/realtime_endpoints/test_realtime_webrtc_endpoints.py b/tests/test_litellm/proxy/realtime_endpoints/test_realtime_webrtc_endpoints.py index 367c89a05f31..ec541903dd0c 100644 --- a/tests/test_litellm/proxy/realtime_endpoints/test_realtime_webrtc_endpoints.py +++ b/tests/test_litellm/proxy/realtime_endpoints/test_realtime_webrtc_endpoints.py @@ -311,3 +311,191 @@ async def test_realtime_calls_success_with_valid_encrypted_token( assert response.status_code == 201 assert response.content.startswith(b"v=0") assert b"application/sdp" in response.headers.get("content-type", "").encode() + + +def test_token_payload_carries_session_type(): + """The encrypted token records the session kind so /realtime/calls can replay it.""" + payload = _encode_realtime_token_payload( + ephemeral_key="epk", + model_id="gpt-realtime-whisper", + user_id=None, + team_id=None, + expires_at=None, + session_type="transcription", + ) + decoded = _decode_realtime_token_payload(payload) + assert decoded is not None + assert decoded["session_type"] == "transcription" + + +@pytest.mark.asyncio +async def test_realtime_calls_replays_transcription_session_type( + proxy_app, + mock_add_litellm_data, + mock_pre_call_hook, +): + """ + A token minted for a transcription session must drive /realtime/calls to send + session.type == "transcription" upstream, not the default "realtime". + """ + captured = {} + + async def _capturing_route(*args, **kwargs): + captured["session"] = kwargs.get("data", {}).get("session") + + async def _inner(): + resp = MagicMock(spec=httpx.Response) + resp.status_code = 201 + resp.content = b"v=0\r\n" + resp.headers = {"content-type": "application/sdp"} + return resp + + return _inner() + + token_payload = _encode_realtime_token_payload( + ephemeral_key="epk", + model_id="gpt-realtime-whisper", + user_id=None, + team_id=None, + expires_at=int(time.time()) + 3600, + session_type="transcription", + ) + encrypted_token = encrypt_value_helper(token_payload) + + client = TestClient(proxy_app) + with ( + patch( + "litellm.proxy.proxy_server.route_request", + side_effect=_capturing_route, + ), + patch( + "litellm.proxy.proxy_server.add_litellm_data_to_request", + side_effect=mock_add_litellm_data, + ), + patch("litellm.proxy.proxy_server.proxy_logging_obj") as mock_logging, + ): + mock_logging.pre_call_hook = AsyncMock(side_effect=mock_pre_call_hook) + mock_logging.post_call_failure_hook = AsyncMock() + + client.post( + "/v1/realtime/calls", + headers={"Authorization": f"Bearer {encrypted_token}"}, + content=b"v=0\r\n", + ) + + assert captured["session"]["type"] == "transcription" + + +# --- transcription_sessions endpoint --- + + +@pytest.fixture +def mock_route_request_transcription_sessions(): + """Mock route_request to return a fake transcription_sessions upstream response.""" + future_expires_at = int(time.time()) + 3600 + body = { + "id": "sess_abc", + "object": "realtime.transcription_session", + "client_secret": { + "value": "upstream_ephemeral_key", + "expires_at": future_expires_at, + }, + } + mock_resp = MagicMock(spec=httpx.Response) + mock_resp.status_code = 200 + mock_resp.text = json.dumps(body) + mock_resp.content = json.dumps(body).encode() + mock_resp.headers = {} + mock_resp.json.return_value = body + + async def _mock_route(*args, **kwargs): + async def _inner(): + return mock_resp + + return _inner() + + return _mock_route + + +def test_transcription_sessions_requires_auth(proxy_app): + """POST /v1/realtime/transcription_sessions returns 401 without Authorization.""" + from fastapi import HTTPException + + def _raise_401(): + raise HTTPException(status_code=401, detail="Unauthorized") + + proxy_app.dependency_overrides[user_api_key_auth] = _raise_401 + try: + client = TestClient(proxy_app, raise_server_exceptions=False) + response = client.post( + "/v1/realtime/transcription_sessions", + json={"input_audio_transcription": {"model": "gpt-realtime-whisper"}}, + ) + assert response.status_code == 401 + finally: + proxy_app.dependency_overrides.pop(user_api_key_auth, None) + + +@pytest.mark.asyncio +async def test_transcription_sessions_encrypts_client_secret( + proxy_app, + mock_route_request_transcription_sessions, + mock_add_litellm_data, + mock_pre_call_hook, +): + """ + POST /v1/realtime/transcription_sessions returns 200 and the ephemeral key + under client_secret.value must be encrypted (never the raw upstream key). + """ + proxy_app.dependency_overrides[user_api_key_auth] = lambda: UserAPIKeyAuth( + user_id="test-user", team_id="test-team" + ) + captured_route_type = {} + + async def _capturing_route(*args, **kwargs): + captured_route_type["route_type"] = kwargs.get("route_type") + return await mock_route_request_transcription_sessions(*args, **kwargs) + + try: + client = TestClient(proxy_app) + with ( + patch( + "litellm.proxy.proxy_server.route_request", + side_effect=_capturing_route, + ), + patch( + "litellm.proxy.proxy_server.add_litellm_data_to_request", + side_effect=mock_add_litellm_data, + ), + patch("litellm.proxy.proxy_server.proxy_logging_obj") as mock_logging, + ): + mock_logging.pre_call_hook = AsyncMock(side_effect=mock_pre_call_hook) + mock_logging.post_call_failure_hook = AsyncMock() + + response = client.post( + "/v1/realtime/transcription_sessions", + headers={"Authorization": "Bearer sk-test-master-key"}, + json={ + "input_audio_format": "pcm16", + "input_audio_transcription": {"model": "gpt-realtime-whisper"}, + }, + ) + + assert response.status_code == 200 + data = response.json() + assert data["client_secret"]["value"] != "upstream_ephemeral_key" + # The encrypted value must decrypt back to a payload carrying the raw key. + decrypted = decrypt_value_helper( + data["client_secret"]["value"], + key="client_secret.value", + exception_type="debug", + ) + assert decrypted is not None + assert "upstream_ephemeral_key" in decrypted + # Routed through the dedicated transcription_sessions route type. + assert ( + captured_route_type["route_type"] + == "acreate_realtime_transcription_session" + ) + finally: + proxy_app.dependency_overrides.pop(user_api_key_auth, None) diff --git a/tests/test_litellm/test_cost_calculator.py b/tests/test_litellm/test_cost_calculator.py index 3d45a3409d80..a9cdaf271978 100644 --- a/tests/test_litellm/test_cost_calculator.py +++ b/tests/test_litellm/test_cost_calculator.py @@ -418,9 +418,130 @@ def test_realtime_logging_object_allows_null_transcript_in_conversation_item_add usage=usage, results=results, ) - assert logging_result.usage.total_tokens == 18 assert logging_result.results[0]["item"]["content"][0]["transcript"] is None + assert logging_result.results[0]["item"]["content"][0]["transcript"] is None +def test_realtime_transcription_duration_cost(monkeypatch): + """ + gpt-realtime-whisper transcription sessions are billed by input audio duration + ($0.017/min). The .completed events carry usage {type: duration, seconds: N}; + cost must equal total_seconds * input_cost_per_second. + """ + monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True") + monkeypatch.setattr(litellm, "model_cost", litellm.get_model_cost_map(url="")) + + from litellm.cost_calculator import RealtimeAPITokenUsageProcessor + + results: OpenAIRealtimeStreamList = [ + { + "type": "session.created", + "session": { + "type": "transcription", + "audio": { + "input": {"transcription": {"model": "gpt-realtime-whisper"}} + }, + }, + }, + { + "type": "conversation.item.input_audio_transcription.completed", + "transcript": "hello", + "usage": {"type": "duration", "seconds": 60.0}, + }, + { + "type": "conversation.item.input_audio_transcription.completed", + "transcript": "world", + "usage": {"type": "duration", "seconds": 30.0}, + }, + ] + + combined = RealtimeAPITokenUsageProcessor.collect_and_combine_usage_from_realtime_stream_results( + results=results + ) + cost = handle_realtime_stream_cost_calculation( + results=results, + combined_usage_object=combined, + custom_llm_provider="openai", + litellm_model_name="gpt-realtime-whisper", + ) + + # 90 seconds at $0.017/minute. + expected = 90.0 * (0.017 / 60) + assert abs(cost - expected) < 1e-9 + assert cost > 0 # guards against the duration branch being dropped + + +def test_realtime_transcription_duration_cost_resolves_model_from_litellm_name( + monkeypatch, +): + """When no session event carries the ASR model, the litellm_model_name is used.""" + monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True") + monkeypatch.setattr(litellm, "model_cost", litellm.get_model_cost_map(url="")) + + results: OpenAIRealtimeStreamList = [ + { + "type": "conversation.item.input_audio_transcription.completed", + "usage": {"type": "duration", "seconds": 120.0}, + }, + ] + cost = handle_realtime_stream_cost_calculation( + results=results, + combined_usage_object=Usage(), + custom_llm_provider="azure", + litellm_model_name="azure/gpt-realtime-whisper", + ) + assert abs(cost - 120.0 * (0.017 / 60)) < 1e-9 + + +def test_realtime_transcription_no_completed_events_is_zero(monkeypatch): + """A realtime stream without transcription completed events adds no extra cost.""" + monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True") + monkeypatch.setattr(litellm, "model_cost", litellm.get_model_cost_map(url="")) + + from litellm.cost_calculator import handle_realtime_transcription_cost_calculation + + results: OpenAIRealtimeStreamList = [ + {"type": "session.created", "session": {"model": "gpt-realtime-whisper"}}, + {"type": "response.done", "response": {"usage": {}}}, + ] + assert ( + handle_realtime_transcription_cost_calculation( + results=results, + custom_llm_provider="openai", + litellm_model_name="gpt-realtime-whisper", + ) + == 0.0 + ) + + +def test_realtime_transcription_token_billed_fallback(monkeypatch): + """ + Token-billed transcription models price by audio/text tokens. Verify the + fallback path multiplies audio tokens by the model's audio token cost. + """ + monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True") + monkeypatch.setattr(litellm, "model_cost", litellm.get_model_cost_map(url="")) + + from litellm.cost_calculator import _transcription_usage_cost + + # gpt-4o-transcribe: input_cost_per_audio_token = 2.5e-06, input_cost_per_token = 2.5e-06, + # output_cost_per_token = 1e-05 + model_info = litellm.get_model_info( + model="gpt-4o-transcribe", custom_llm_provider="openai" + ) + usage = { + "type": "tokens", + "input_tokens": 40, + "output_tokens": 10, + "total_tokens": 50, + "input_token_details": {"audio_tokens": 30, "text_tokens": 10}, + } + cost = _transcription_usage_cost(usage, model_info) + expected = ( + 30 * 2.5e-06 # audio tokens + + 10 * 2.5e-06 # text tokens + + 10 * 1e-05 # output tokens + ) + assert abs(cost - expected) < 1e-12 def test_custom_pricing_with_router_model_id(): From 312f51a10bf4b858e6e83f4c3798f1b09b599c97 Mon Sep 17 00:00:00 2001 From: Emerson Gomes Date: Fri, 5 Jun 2026 07:34:59 -0500 Subject: [PATCH 2/9] Address PR review: URL-encode all Azure WS query params; forward query_params through provider_config branch --- litellm/llms/azure/realtime/handler.py | 12 +++-- litellm/llms/custom_httpx/llm_http_handler.py | 20 +++++++- litellm/realtime_api/main.py | 2 + .../realtime/test_transcription_sessions.py | 49 +++++++++++++++++++ 4 files changed, 77 insertions(+), 6 deletions(-) diff --git a/litellm/llms/azure/realtime/handler.py b/litellm/llms/azure/realtime/handler.py index 8f0037570b07..a95f65be4179 100644 --- a/litellm/llms/azure/realtime/handler.py +++ b/litellm/llms/azure/realtime/handler.py @@ -5,7 +5,6 @@ """ from typing import Any, Optional, cast -from urllib.parse import quote from litellm._logging import _redact_string, verbose_proxy_logger from litellm.constants import REALTIME_WEBSOCKET_MAX_MESSAGE_SIZE_BYTES @@ -57,6 +56,8 @@ def _construct_url( beta/default: "wss://.../openai/realtime?api-version=2024-10-01-preview&deployment=gpt-4o-realtime-preview" GA/v1: "wss://.../openai/v1/realtime?model=gpt-realtime-deployment" """ + from urllib.parse import urlencode + api_base = api_base.replace("https://", "wss://") # Determine path based on realtime_protocol (case-insensitive) @@ -66,16 +67,17 @@ def _construct_url( ) if _is_ga: path = "/openai/v1/realtime" - url = f"{api_base}{path}?model={model}" + qs = urlencode({"model": model}) else: # Default to beta path for backwards compatibility path = "/openai/realtime" - url = f"{api_base}{path}?api-version={api_version}&deployment={model}" + qs = urlencode({"api-version": api_version, "deployment": model}) intent = (query_params or {}).get("intent") if intent: - url = f"{url}&intent={quote(str(intent), safe='')}" - return url + qs = f"{qs}&{urlencode({'intent': intent})}" + + return f"{api_base}{path}?{qs}" async def async_realtime( self, diff --git a/litellm/llms/custom_httpx/llm_http_handler.py b/litellm/llms/custom_httpx/llm_http_handler.py index 377faf1400c8..321bab822084 100644 --- a/litellm/llms/custom_httpx/llm_http_handler.py +++ b/litellm/llms/custom_httpx/llm_http_handler.py @@ -5260,6 +5260,21 @@ def _handle_error( headers=error_headers, ) + @staticmethod + def _append_query_params(url: str, query_params: Optional[Dict[str, Any]]) -> str: + """Append query_params to url, skipping keys already present in the URL.""" + if not query_params: + return url + from urllib.parse import parse_qsl, urlencode, urlparse, urlunparse + + parsed = urlparse(url) + existing = dict(parse_qsl(parsed.query)) + extras = {k: v for k, v in query_params.items() if k not in existing} + if not extras: + return url + new_query = parsed.query + ("&" if parsed.query else "") + urlencode(extras) + return urlunparse(parsed._replace(query=new_query)) + async def async_realtime( self, model: str, @@ -5273,11 +5288,14 @@ async def async_realtime( timeout: Optional[float] = None, user_api_key_dict: Optional[Any] = None, litellm_metadata: Optional[Dict[str, Any]] = None, + query_params: Optional[Dict[str, Any]] = None, ): import websockets from websockets.asyncio.client import ClientConnection - url = provider_config.get_complete_url(api_base, model, api_key) + url = self._append_query_params( + provider_config.get_complete_url(api_base, model, api_key), query_params + ) headers = provider_config.validate_environment( headers=headers, model=model, diff --git a/litellm/realtime_api/main.py b/litellm/realtime_api/main.py index 4ec00e1f65bc..60f6fabd5617 100644 --- a/litellm/realtime_api/main.py +++ b/litellm/realtime_api/main.py @@ -345,6 +345,7 @@ async def _arealtime( # noqa: PLR0915 headers=headers, user_api_key_dict=kwargs.get("user_api_key_dict"), litellm_metadata=_build_litellm_metadata(kwargs), + query_params=query_params, ) elif _custom_llm_provider == "azure": api_base = ( @@ -518,6 +519,7 @@ async def _arealtime( # noqa: PLR0915 headers=headers, user_api_key_dict=kwargs.get("user_api_key_dict"), litellm_metadata=_build_litellm_metadata(kwargs), + query_params=query_params, ) else: raise ValueError(f"Unsupported model: {model}") diff --git a/tests/test_litellm/llms/openai/realtime/test_transcription_sessions.py b/tests/test_litellm/llms/openai/realtime/test_transcription_sessions.py index afcfcc753362..08ef94856775 100644 --- a/tests/test_litellm/llms/openai/realtime/test_transcription_sessions.py +++ b/tests/test_litellm/llms/openai/realtime/test_transcription_sessions.py @@ -172,3 +172,52 @@ async def test_sdk_fn_routes_openai_transcription_session(monkeypatch): assert kwargs["json"]["input_audio_transcription"] == { "model": "gpt-realtime-whisper" } + + +def test_append_query_params_skips_existing_keys(): + from litellm.llms.custom_httpx.llm_http_handler import BaseLLMHTTPHandler + + url = "wss://example.com/v1/realtime?model=gpt-4o" + result = BaseLLMHTTPHandler._append_query_params( + url, {"model": "ignored", "intent": "transcription"} + ) + assert "model=ignored" not in result + assert "intent=transcription" in result + + +def test_append_query_params_no_params_returns_unchanged(): + from litellm.llms.custom_httpx.llm_http_handler import BaseLLMHTTPHandler + + url = "wss://example.com/v1/realtime?model=gpt-4o" + assert BaseLLMHTTPHandler._append_query_params(url, None) == url + assert BaseLLMHTTPHandler._append_query_params(url, {}) == url + + +def test_append_query_params_encodes_special_chars(): + from litellm.llms.custom_httpx.llm_http_handler import BaseLLMHTTPHandler + + url = "wss://example.com/v1/realtime" + result = BaseLLMHTTPHandler._append_query_params(url, {"intent": "a&b=c"}) + assert "intent=a%26b%3Dc" in result + assert "&b=c" not in result + + +def test_azure_construct_url_encodes_model_and_api_version(): + """model and api-version must be URL-encoded to prevent query-string injection.""" + from litellm.llms.azure.realtime.handler import AzureOpenAIRealtime + + h = AzureOpenAIRealtime() + url = h._construct_url( + "https://x.openai.azure.com", + "deploy&evil=1", + "2024-10-01-preview", + ) + assert "evil=1" not in url.split("?", 1)[1] + + url_ga = h._construct_url( + "https://x.openai.azure.com", + "deploy&evil=1", + None, + realtime_protocol="GA", + ) + assert "evil=1" not in url_ga.split("?", 1)[1] From 1d4a72a38563c284595a775908aaa60ced94c37c Mon Sep 17 00:00:00 2001 From: Emerson Gomes Date: Fri, 5 Jun 2026 08:29:55 -0500 Subject: [PATCH 3/9] Address PR review: session_type validation, model auth fix, cost perf, billing fallback, detail/docs cleanup --- litellm/cost_calculator.py | 13 ++-- litellm/proxy/realtime_endpoints/endpoints.py | 4 +- litellm/realtime_api/README.md | 62 ++----------------- litellm/realtime_api/main.py | 6 ++ 4 files changed, 21 insertions(+), 64 deletions(-) diff --git a/litellm/cost_calculator.py b/litellm/cost_calculator.py index aea1ba3f2201..69309c6b4c3a 100644 --- a/litellm/cost_calculator.py +++ b/litellm/cost_calculator.py @@ -2534,11 +2534,12 @@ def handle_realtime_stream_cost_calculation( break # exit if we find a valid model total_cost = input_cost_per_token + output_cost_per_token - total_cost += handle_realtime_transcription_cost_calculation( - results=results, - custom_llm_provider=custom_llm_provider, - litellm_model_name=litellm_model_name, - ) + if any(r.get("type") == _TRANSCRIPTION_COMPLETED_EVENT_TYPE for r in results): + total_cost += handle_realtime_transcription_cost_calculation( + results=results, + custom_llm_provider=custom_llm_provider, + litellm_model_name=litellm_model_name, + ) return total_cost @@ -2602,7 +2603,7 @@ def _get_transcription_model_name_from_results( transcription = ( (session.get("audio", {}) or {}).get("input", {}) or {} ).get("transcription", {}) or session.get("input_audio_transcription", {}) - model = (transcription or {}).get("model") + model = (transcription or {}).get("model") or session.get("model") if model: return model return None diff --git a/litellm/proxy/realtime_endpoints/endpoints.py b/litellm/proxy/realtime_endpoints/endpoints.py index a2afd868ec40..94b1384b712c 100644 --- a/litellm/proxy/realtime_endpoints/endpoints.py +++ b/litellm/proxy/realtime_endpoints/endpoints.py @@ -291,6 +291,8 @@ async def proxy_realtime_calls( user_id = decoded_payload.get("user_id") or None team_id = decoded_payload.get("team_id") or None session_type = decoded_payload.get("session_type") or "realtime" + if session_type not in ("realtime", "transcription"): + session_type = "realtime" else: # Backward compatibility: older tokens contained only encrypted upstream key. openai_ephemeral_key = decrypted_token_value @@ -464,7 +466,7 @@ async def create_realtime_transcription_session( ) if isinstance(e, HTTPException): raise ProxyException( - message=getattr(e, "message", str(e)), + message=getattr(e, "detail", getattr(e, "message", str(e))), type=getattr(e, "type", "None"), param=getattr(e, "param", "None"), code=getattr(e, "status_code", http_status.HTTP_400_BAD_REQUEST), diff --git a/litellm/realtime_api/README.md b/litellm/realtime_api/README.md index cf3d2973cfc6..d810de2f24fc 100644 --- a/litellm/realtime_api/README.md +++ b/litellm/realtime_api/README.md @@ -1,61 +1,9 @@ Abstraction / Routing logic for OpenAI's `/v1/realtime` endpoints. -## Realtime transcription (`gpt-realtime-whisper`) +Supported endpoints: +- WebSocket: `/v1/realtime` (with `intent=transcription` for transcription-only sessions) +- HTTP: `/v1/realtime/client_secrets`, `/v1/realtime/transcription_sessions` -`gpt-realtime-whisper` is the low-latency streaming speech-to-text model. It is a -Realtime transcription session, not the file-based `/audio/transcriptions` path. Use -the standard `gpt-4o-transcribe` / `whisper-1` models for request/response or file -transcription; use `gpt-realtime-whisper` for live streaming transcript deltas. +Supported providers: OpenAI, Azure OpenAI, Bedrock, Vertex AI, xAI. -Both OpenAI and Azure OpenAI (Microsoft Foundry) are supported. Cost is tracked by input -audio duration (OpenAI: $0.017/minute), derived from the -`conversation.item.input_audio_transcription.completed` usage events. - -### WebSocket - -Connect to the proxy realtime WebSocket with `intent=transcription`, then send a -`session.update` configuring a transcription session: - -``` -wss:///v1/realtime?model=gpt-realtime-whisper&intent=transcription -``` - -```json -{ - "type": "session.update", - "session": { - "type": "transcription", - "audio": { - "input": { - "format": { "type": "audio/pcm", "rate": 24000 }, - "transcription": { "model": "gpt-realtime-whisper", "language": "en" } - } - } - } -} -``` - -Append audio with `input_audio_buffer.append`, then `input_audio_buffer.commit` (when not -using server VAD). Listen for `conversation.item.input_audio_transcription.delta` and -`.completed` events. The proxy does not auto-trigger `response.create` for transcription -sessions. - -### Ephemeral transcription session (WebRTC) - -`POST /v1/realtime/transcription_sessions` mints an ephemeral session for browser/WebRTC -clients. The returned `client_secret.value` is encrypted by the proxy and exchanged via -`POST /v1/realtime/calls`. - -```bash -curl https:///v1/realtime/transcription_sessions \ - -H "Authorization: Bearer $LITELLM_KEY" \ - -H "Content-Type: application/json" \ - -d '{ - "input_audio_format": "pcm16", - "input_audio_transcription": { "model": "gpt-realtime-whisper", "language": "en" } - }' -``` - -For Azure, route to an `azure/gpt-realtime-whisper` deployment; the proxy targets -`/openai/realtime/transcription_sessions?api-version=...` and forwards -`intent=transcription` on the WebSocket. \ No newline at end of file +For user-facing documentation and usage examples, see the litellm-docs repo. \ No newline at end of file diff --git a/litellm/realtime_api/main.py b/litellm/realtime_api/main.py index 60f6fabd5617..1883d1ea19f5 100644 --- a/litellm/realtime_api/main.py +++ b/litellm/realtime_api/main.py @@ -212,6 +212,12 @@ async def acreate_realtime_transcription_session( custom_llm_provider=custom_llm_provider, ) request_data = req.model_dump(exclude_none=True, exclude={"model"}) + # Ensure the upstream body's input_audio_transcription.model matches the + # authorized routing model. This prevents a caller from supplying an allowed + # top-level model for auth while sneaking a different model into the nested + # transcription config that gets forwarded to the provider. + if isinstance(request_data.get("input_audio_transcription"), dict): + request_data["input_audio_transcription"]["model"] = model_name return await base_llm_http_handler.async_realtime_transcription_session_handler( api_base=resolved_api_base, api_key=resolved_api_key, From 897816243edf89b5c0242466922d106950bbee96 Mon Sep 17 00:00:00 2001 From: Emerson Gomes Date: Fri, 5 Jun 2026 08:50:43 -0500 Subject: [PATCH 4/9] Improve test coverage: detection from backend, error paths, unknown usage type, resolved_model None --- .../test_realtime_streaming.py | 56 +++++++++ .../realtime/test_transcription_sessions.py | 5 +- .../test_realtime_webrtc_endpoints.py | 110 ++++++++++++++++++ tests/test_litellm/test_cost_calculator.py | 23 +++- 4 files changed, 192 insertions(+), 2 deletions(-) diff --git a/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py b/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py index 870b0051f786..10bfd473b419 100644 --- a/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py +++ b/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py @@ -632,6 +632,62 @@ def test_client_session_update_marks_transcription_session(): assert streaming._is_transcription_session is True +def test_detect_transcription_session_from_backend_transcription_session_events(): + """Backend transcription_session.created/updated events flag the session.""" + streaming = RealTimeStreaming(MagicMock(), MagicMock(), MagicMock()) + assert streaming._is_transcription_session is False + streaming._detect_transcription_session_from_backend( + {"type": "transcription_session.created"} + ) + assert streaming._is_transcription_session is True + + streaming2 = RealTimeStreaming(MagicMock(), MagicMock(), MagicMock()) + streaming2._detect_transcription_session_from_backend( + {"type": "transcription_session.updated"} + ) + assert streaming2._is_transcription_session is True + + +def test_detect_transcription_session_from_backend_session_created_with_type(): + """Backend session.created with type=transcription flags the session.""" + streaming = RealTimeStreaming(MagicMock(), MagicMock(), MagicMock()) + streaming._detect_transcription_session_from_backend( + {"type": "session.created", "session": {"type": "transcription"}} + ) + assert streaming._is_transcription_session is True + + +def test_detect_transcription_session_from_backend_ignores_non_transcription(): + """Backend session.created without type=transcription does not flag the session.""" + streaming = RealTimeStreaming(MagicMock(), MagicMock(), MagicMock()) + streaming._detect_transcription_session_from_backend( + {"type": "session.created", "session": {"model": "gpt-4o-realtime-preview"}} + ) + assert streaming._is_transcription_session is False + + +def test_capture_transcription_usage_deduplicates_when_already_stored(): + """ + When the event is already in messages (logged via store_message), it must not + be appended a second time by _capture_transcription_usage. + """ + import litellm + + streaming = RealTimeStreaming(MagicMock(), MagicMock(), MagicMock()) + # Add the event type to the default logged list so _should_store_message returns True. + streaming.logged_real_time_event_types = [ + "conversation.item.input_audio_transcription.completed" + ] + event = { + "type": "conversation.item.input_audio_transcription.completed", + "usage": {"type": "duration", "seconds": 5.0}, + } + streaming.store_message(json.dumps(event)) + initial_count = len(streaming.messages) + streaming._capture_transcription_usage(event) + assert len(streaming.messages) == initial_count # no duplicate + + @pytest.mark.asyncio async def test_client_ack_caches_setup_to_prevent_duplicate_session_update_setup(): websocket = MagicMock() diff --git a/tests/test_litellm/llms/openai/realtime/test_transcription_sessions.py b/tests/test_litellm/llms/openai/realtime/test_transcription_sessions.py index 08ef94856775..62fc3a8d0aa9 100644 --- a/tests/test_litellm/llms/openai/realtime/test_transcription_sessions.py +++ b/tests/test_litellm/llms/openai/realtime/test_transcription_sessions.py @@ -55,7 +55,10 @@ def test_azure_transcription_session_url_uses_deployment_and_api_version(): ) -def test_request_resolves_model_from_top_level_hint(): +def test_request_resolves_model_returns_none_when_both_absent(): + req = RealtimeTranscriptionSessionRequest(input_audio_format="pcm16") + assert req.resolved_model() is None + req = RealtimeTranscriptionSessionRequest( model="openai/gpt-realtime-whisper", input_audio_transcription={"model": "gpt-realtime-whisper"}, diff --git a/tests/test_litellm/proxy/realtime_endpoints/test_realtime_webrtc_endpoints.py b/tests/test_litellm/proxy/realtime_endpoints/test_realtime_webrtc_endpoints.py index ec541903dd0c..362307252ee5 100644 --- a/tests/test_litellm/proxy/realtime_endpoints/test_realtime_webrtc_endpoints.py +++ b/tests/test_litellm/proxy/realtime_endpoints/test_realtime_webrtc_endpoints.py @@ -499,3 +499,113 @@ async def _capturing_route(*args, **kwargs): ) finally: proxy_app.dependency_overrides.pop(user_api_key_auth, None) + + +def test_session_type_coerced_for_unknown_value(): + """An unrecognized session_type in the token falls back to 'realtime'.""" + payload = _encode_realtime_token_payload( + ephemeral_key="epk", + model_id="gpt-4o", + user_id=None, + team_id=None, + expires_at=None, + session_type="INJECTED_TYPE", + ) + # Force-deserialize and check the coercion that happens in proxy_realtime_calls. + decoded = json.loads(payload) + session_type = decoded.get("session_type") or "realtime" + if session_type not in ("realtime", "transcription"): + session_type = "realtime" + assert session_type == "realtime" + + +@pytest.mark.asyncio +async def test_transcription_sessions_returns_upstream_error_verbatim( + proxy_app, + mock_add_litellm_data, + mock_pre_call_hook, +): + """Non-200 upstream response is forwarded unchanged (no encryption attempted).""" + mock_resp = MagicMock(spec=httpx.Response) + mock_resp.status_code = 400 + mock_resp.content = b'{"error":"bad_request"}' + mock_resp.headers = {} + mock_resp.json.return_value = {"error": "bad_request"} + mock_resp.text = '{"error":"bad_request"}' + + async def _mock_route(*args, **kwargs): + async def _inner(): + return mock_resp + + return _inner() + + proxy_app.dependency_overrides[user_api_key_auth] = lambda: UserAPIKeyAuth( + user_id="test-user", team_id="test-team" + ) + try: + client = TestClient(proxy_app) + with ( + patch( + "litellm.proxy.proxy_server.route_request", + side_effect=_mock_route, + ), + patch( + "litellm.proxy.proxy_server.add_litellm_data_to_request", + side_effect=mock_add_litellm_data, + ), + patch("litellm.proxy.proxy_server.proxy_logging_obj") as mock_logging, + ): + mock_logging.pre_call_hook = AsyncMock(side_effect=mock_pre_call_hook) + mock_logging.post_call_failure_hook = AsyncMock() + + response = client.post( + "/v1/realtime/transcription_sessions", + headers={"Authorization": "Bearer sk-test-master-key"}, + json={"input_audio_transcription": {"model": "gpt-realtime-whisper"}}, + ) + assert response.status_code == 400 + assert response.content == b'{"error":"bad_request"}' + finally: + proxy_app.dependency_overrides.pop(user_api_key_auth, None) + + +@pytest.mark.asyncio +async def test_transcription_sessions_wraps_route_exception( + proxy_app, + mock_add_litellm_data, + mock_pre_call_hook, +): + """A route exception is wrapped in a ProxyException with a human-readable message.""" + from fastapi import HTTPException + + async def _raise_http(*args, **kwargs): + raise HTTPException(status_code=403, detail="Model not allowed") + + proxy_app.dependency_overrides[user_api_key_auth] = lambda: UserAPIKeyAuth( + user_id="test-user" + ) + try: + client = TestClient(proxy_app, raise_server_exceptions=False) + with ( + patch( + "litellm.proxy.proxy_server.route_request", + side_effect=_raise_http, + ), + patch( + "litellm.proxy.proxy_server.add_litellm_data_to_request", + side_effect=mock_add_litellm_data, + ), + patch("litellm.proxy.proxy_server.proxy_logging_obj") as mock_logging, + ): + mock_logging.pre_call_hook = AsyncMock(side_effect=mock_pre_call_hook) + mock_logging.post_call_failure_hook = AsyncMock() + + response = client.post( + "/v1/realtime/transcription_sessions", + headers={"Authorization": "Bearer sk-test-master-key"}, + json={"input_audio_transcription": {"model": "gpt-realtime-whisper"}}, + ) + assert response.status_code == 403 + assert "Model not allowed" in response.text + finally: + proxy_app.dependency_overrides.pop(user_api_key_auth, None) diff --git a/tests/test_litellm/test_cost_calculator.py b/tests/test_litellm/test_cost_calculator.py index a9cdaf271978..1590a3f59357 100644 --- a/tests/test_litellm/test_cost_calculator.py +++ b/tests/test_litellm/test_cost_calculator.py @@ -421,6 +421,8 @@ def test_realtime_logging_object_allows_null_transcript_in_conversation_item_add assert logging_result.usage.total_tokens == 18 assert logging_result.results[0]["item"]["content"][0]["transcript"] is None assert logging_result.results[0]["item"]["content"][0]["transcript"] is None + + def test_realtime_transcription_duration_cost(monkeypatch): """ gpt-realtime-whisper transcription sessions are billed by input audio duration @@ -544,7 +546,26 @@ def test_realtime_transcription_token_billed_fallback(monkeypatch): assert abs(cost - expected) < 1e-12 -def test_custom_pricing_with_router_model_id(): +def test_transcription_usage_cost_returns_zero_for_unknown_type(): + """An unrecognized usage type yields 0 (safe fallback, no exception).""" + from litellm.cost_calculator import _transcription_usage_cost + + assert _transcription_usage_cost({"type": "future_billing_type"}, {}) == 0.0 + assert _transcription_usage_cost({}, {}) == 0.0 + + +def test_get_transcription_model_falls_back_to_session_model(monkeypatch): + """session.model is used when transcription-specific model fields are absent.""" + monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True") + monkeypatch.setattr(litellm, "model_cost", litellm.get_model_cost_map(url="")) + + from litellm.cost_calculator import _get_transcription_model_name_from_results + + results: OpenAIRealtimeStreamList = [ + {"type": "session.created", "session": {"model": "gpt-realtime-whisper"}}, + ] + assert _get_transcription_model_name_from_results(results) == "gpt-realtime-whisper" + from litellm import Router router = Router( From 65b87340c4520ccfbc74631f77fe3694af0dcd5f Mon Sep 17 00:00:00 2001 From: Emerson Gomes Date: Sat, 6 Jun 2026 09:40:07 -0500 Subject: [PATCH 5/9] Backport realtime transcription websocket fixes --- litellm/llms/azure/realtime/handler.py | 16 +++-- litellm/proxy/proxy_server.py | 29 +++++++-- litellm/realtime_api/main.py | 15 ++++- .../realtime/test_openai_realtime.py | 35 ++++++++++ tests/proxy_unit_tests/test_realtime_cache.py | 4 ++ .../realtime/test_azure_realtime_handler.py | 64 +++++++++++++++++++ 6 files changed, 148 insertions(+), 15 deletions(-) diff --git a/litellm/llms/azure/realtime/handler.py b/litellm/llms/azure/realtime/handler.py index a95f65be4179..5340eb4916bd 100644 --- a/litellm/llms/azure/realtime/handler.py +++ b/litellm/llms/azure/realtime/handler.py @@ -65,19 +65,25 @@ def _construct_url( "GA", "V1", ) + intent = (query_params or {}).get("intent") + if _is_ga: path = "/openai/v1/realtime" - qs = urlencode({"model": model}) + query_parts = [] + if intent != "transcription" and ( + query_params is None or "model" in query_params + ): + query_parts.append(urlencode({"model": model})) else: # Default to beta path for backwards compatibility path = "/openai/realtime" - qs = urlencode({"api-version": api_version, "deployment": model}) + query_parts = [urlencode({"api-version": api_version, "deployment": model})] - intent = (query_params or {}).get("intent") if intent: - qs = f"{qs}&{urlencode({'intent': intent})}" + query_parts.append(urlencode({"intent": intent})) - return f"{api_base}{path}?{qs}" + qs = "&".join(query_parts) + return f"{api_base}{path}?{qs}" if qs else f"{api_base}{path}" async def async_realtime( self, diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index 72423b2a7968..387d4b6f6b3e 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -9426,13 +9426,15 @@ async def vertex_ai_live_passthrough_endpoint( @lru_cache(maxsize=_REALTIME_BODY_CACHE_SIZE) def _realtime_query_params_template( - model: str, intent: Optional[str] + model: Optional[str], intent: Optional[str] ) -> Tuple[Tuple[str, str], ...]: """ Build a hashable representation of the realtime query params so we can cache the repetitive model/intent combinations. """ - params: List[Tuple[str, str]] = [("model", model)] + params: List[Tuple[str, str]] = [] + if model is not None: + params.append(("model", model)) if intent is not None: params.append(("intent", intent)) return tuple(params) @@ -9443,8 +9445,10 @@ def _realtime_query_params_template( @app.websocket("/realtime") async def realtime_websocket_endpoint( websocket: WebSocket, - model: str, - intent: str = fastapi.Query( + model: Optional[str] = fastapi.Query( + None, description="The model to use for the websocket connection." + ), + intent: Optional[str] = fastapi.Query( None, description="The intent of the websocket connection." ), guardrails: Optional[str] = fastapi.Query( @@ -9461,6 +9465,17 @@ async def realtime_websocket_endpoint( accept_kwargs: dict = {} if requested_protocols: accept_kwargs["subprotocol"] = requested_protocols[0] + + route_model = model + if route_model is None: + if intent == "transcription": + route_model = "gpt-realtime-whisper" + else: + await websocket.close( + code=1008, reason="model query parameter is required" + ) + return + assert route_model is not None await websocket.accept(**accept_kwargs) # Only use explicit parameters, not all query params @@ -9469,7 +9484,7 @@ async def realtime_websocket_endpoint( ) data: Dict[str, Any] = { - "model": model, + "model": route_model, "websocket": websocket, "query_params": query_params, # Only explicit params } @@ -9489,7 +9504,7 @@ async def realtime_websocket_endpoint( request._url = websocket.url async def return_body(): - return _realtime_request_body(model) + return _realtime_request_body(route_model) request.body = return_body # type: ignore @@ -9515,7 +9530,7 @@ async def return_body(): user_request_timeout=user_request_timeout, user_max_tokens=user_max_tokens, user_api_base=user_api_base, - model=model, + model=route_model, route_type="_arealtime", ) except Exception as e: diff --git a/litellm/realtime_api/main.py b/litellm/realtime_api/main.py index 1883d1ea19f5..7031ecaa1a0b 100644 --- a/litellm/realtime_api/main.py +++ b/litellm/realtime_api/main.py @@ -319,9 +319,13 @@ async def _arealtime( # noqa: PLR0915 api_key=api_key, ) - # Ensure query params use the normalized provider model (no proxy aliases). + # If the client supplied `model` in the URL, ensure it uses the normalized + # provider model (no proxy aliases). If they omitted it, preserve that shape + # for transcription-only sessions like OpenAI's `?intent=transcription`. if query_params is not None: - query_params = {**query_params, "model": model} + query_params = {**query_params} + if "model" in query_params: + query_params["model"] = model litellm_logging_obj.update_from_kwargs( kwargs=kwargs, @@ -374,8 +378,13 @@ async def _arealtime( # noqa: PLR0915 kwargs.get("realtime_protocol") or litellm_params.get("realtime_protocol") or os.environ.get("LITELLM_AZURE_REALTIME_PROTOCOL") - or "beta" ) + if ( + realtime_protocol is None + and (query_params or {}).get("intent") == "transcription" + ): + realtime_protocol = "GA" + realtime_protocol = realtime_protocol or "beta" await azure_realtime.async_realtime( model=model, websocket=websocket, diff --git a/tests/llm_translation/realtime/test_openai_realtime.py b/tests/llm_translation/realtime/test_openai_realtime.py index fc9f938b4cde..0e50e2792d6d 100644 --- a/tests/llm_translation/realtime/test_openai_realtime.py +++ b/tests/llm_translation/realtime/test_openai_realtime.py @@ -393,3 +393,38 @@ def fake_get_llm_provider(model, api_base=None, api_key=None): called_kwargs = mock_async_realtime.call_args.kwargs assert called_kwargs["query_params"]["model"] == "gpt-4o-realtime-preview" assert called_kwargs["query_params"]["intent"] == "chat" + + +@pytest.mark.asyncio +async def test_realtime_query_params_preserve_missing_model(monkeypatch): + """ + OpenAI-compatible transcription clients can connect with only + ?intent=transcription and send the model in session.update. Do not add + model= back into the upstream query params when the client omitted it. + """ + from litellm.realtime_api import main as realtime_main + + mock_async_realtime = AsyncMock() + monkeypatch.setattr( + realtime_main, + "openai_realtime", + MagicMock(async_realtime=mock_async_realtime), + ) + + def fake_get_llm_provider(model, api_base=None, api_key=None): + return ("gpt-realtime-whisper", "openai", None, None) + + monkeypatch.setattr(realtime_main, "get_llm_provider", fake_get_llm_provider) + + query_params: RealtimeQueryParams = {"intent": "transcription"} + + await realtime_main._arealtime( + model="gpt-realtime-whisper", + websocket=MagicMock(), + api_key="sk-test", + query_params=query_params, + litellm_logging_obj=MagicMock(), + ) + + called_kwargs = mock_async_realtime.call_args.kwargs + assert called_kwargs["query_params"] == {"intent": "transcription"} diff --git a/tests/proxy_unit_tests/test_realtime_cache.py b/tests/proxy_unit_tests/test_realtime_cache.py index c4cb4ea8e02b..8316ed1d29ae 100644 --- a/tests/proxy_unit_tests/test_realtime_cache.py +++ b/tests/proxy_unit_tests/test_realtime_cache.py @@ -44,10 +44,14 @@ def test_realtime_query_params_template_caches_each_pair_separately(): params_with_intent_first = _realtime_query_params_template("gpt-4o", "intent-a") params_with_intent_second = _realtime_query_params_template("gpt-4o", "intent-a") params_without_intent = _realtime_query_params_template("gpt-4o", None) + params_transcription_without_model = _realtime_query_params_template( + None, "transcription" + ) assert params_with_intent_first is params_with_intent_second assert params_with_intent_first == (("model", "gpt-4o"), ("intent", "intent-a")) assert params_without_intent == (("model", "gpt-4o"),) + assert params_transcription_without_model == (("intent", "transcription"),) assert params_with_intent_first is not params_without_intent diff --git a/tests/test_litellm/llms/azure/realtime/test_azure_realtime_handler.py b/tests/test_litellm/llms/azure/realtime/test_azure_realtime_handler.py index 33e960fa1fd8..4638bc4df0ff 100644 --- a/tests/test_litellm/llms/azure/realtime/test_azure_realtime_handler.py +++ b/tests/test_litellm/llms/azure/realtime/test_azure_realtime_handler.py @@ -167,6 +167,31 @@ async def test_construct_url_forwards_transcription_intent_ga(): assert "/openai/v1/realtime?" in url assert "intent=transcription" in url + assert "model=" not in url + + +@pytest.mark.asyncio +async def test_construct_url_forwards_transcription_intent_ga_without_model_query(): + """ + OpenAI-compatible transcription clients may connect with only + intent=transcription and send the transcription model in session.update. + Preserve that query shape instead of forcing model= into the upstream URL. + """ + from litellm.llms.azure.realtime.handler import AzureOpenAIRealtime + + handler = AzureOpenAIRealtime() + url = handler._construct_url( + api_base="https://my-endpoint.openai.azure.com", + model="gpt-realtime-whisper", + api_version="2025-04-01-preview", + realtime_protocol="GA", + query_params={"intent": "transcription"}, + ) + + assert url == ( + "wss://my-endpoint.openai.azure.com/openai/v1/realtime" + "?intent=transcription" + ) @pytest.mark.asyncio @@ -440,6 +465,45 @@ async def test_realtime_protocol_from_litellm_params(): assert litellm_params.get("realtime_protocol") == "GA" +@pytest.mark.asyncio +async def test_arealtime_transcription_intent_defaults_to_ga(monkeypatch): + """ + Azure gpt-realtime-whisper transcription connects on the GA /openai/v1/realtime + path. If the DB model lacks realtime_protocol, infer GA from intent=transcription. + """ + from litellm.realtime_api import main as realtime_main + + mock_async_realtime = AsyncMock() + monkeypatch.setattr( + realtime_main, + "azure_realtime", + MagicMock(async_realtime=mock_async_realtime), + ) + + def fake_get_llm_provider(model, api_base=None, api_key=None): + return ( + "gpt-realtime-whisper", + "azure", + "test-key", + "https://my-endpoint.openai.azure.com", + ) + + monkeypatch.setattr(realtime_main, "get_llm_provider", fake_get_llm_provider) + + await realtime_main._arealtime( + model="azure/gpt-realtime-whisper", + websocket=MagicMock(), + api_key="test-key", + api_version="2025-04-01-preview", + query_params={"intent": "transcription"}, + litellm_logging_obj=MagicMock(), + ) + + called_kwargs = mock_async_realtime.call_args.kwargs + assert called_kwargs["realtime_protocol"] == "GA" + assert called_kwargs["query_params"] == {"intent": "transcription"} + + @pytest.mark.asyncio async def test_async_realtime_default_maintains_backwards_compatibility(): """ From 3c5c83f4e49300c7e6dd001b03aea67663d570de Mon Sep 17 00:00:00 2001 From: Emerson Gomes Date: Sat, 6 Jun 2026 10:09:10 -0500 Subject: [PATCH 6/9] Enforce authorized realtime transcription model --- .../litellm_core_utils/realtime_streaming.py | 79 +++++++++++- litellm/llms/azure/realtime/handler.py | 6 + litellm/llms/custom_httpx/llm_http_handler.py | 5 + litellm/llms/openai/realtime/handler.py | 6 + .../test_realtime_streaming.py | 117 ++++++++++++++++++ 5 files changed, 212 insertions(+), 1 deletion(-) diff --git a/litellm/litellm_core_utils/realtime_streaming.py b/litellm/litellm_core_utils/realtime_streaming.py index cfabb6e5e1a5..1deaddb444c7 100644 --- a/litellm/litellm_core_utils/realtime_streaming.py +++ b/litellm/litellm_core_utils/realtime_streaming.py @@ -47,6 +47,7 @@ def __init__( user_api_key_dict: Optional[Any] = None, request_data: Optional[Dict] = None, backend_uses_beta_protocol: Optional[bool] = None, + force_transcription_model: Optional[str] = None, ): self.websocket = websocket self.backend_ws = backend_ws @@ -103,7 +104,8 @@ def __init__( # Whether this is a transcription-only session (session.type == "transcription", # e.g. gpt-realtime-whisper). Such sessions must not be sent response.create and # their input_audio_transcription.completed usage drives duration-based cost. - self._is_transcription_session: bool = False + self._force_transcription_model = force_transcription_model + self._is_transcription_session: bool = force_transcription_model is not None # Per-connection caps for pre-setup audio frames (message count + total bytes). _MAX_BUFFERED_MESSAGES: int = 200 @@ -340,6 +342,7 @@ async def _send_to_backend(self, message: str) -> bool: backend, False if the provider transformation produced no output and the message was effectively dropped. """ + message = self._enforce_transcription_session_model(message) if self.provider_config: transformed = self.provider_config.transform_realtime_request( message, self.model, self.session_configuration_request @@ -359,6 +362,80 @@ async def _send_to_backend(self, message: str) -> bool: await self.backend_ws.send(message) # type: ignore[union-attr, attr-defined] return True + def _enforce_transcription_session_model(self, message: str) -> str: + """Force client transcription session updates to the authorized model. + + `/v1/realtime?intent=transcription` may intentionally omit `model` from + the upstream URL for Azure compatibility, but the proxy still authorizes + a resolved LiteLLM model before opening the backend websocket. If a + client later sends a transcription `session.update`, any model embedded + in that update must be rewritten to the same authorized model instead of + allowing a post-auth model/deployment switch. + + Normal realtime sessions keep their independent nested transcription + model behavior because `_force_transcription_model` is only set for + transcription-intent websocket routes. + """ + if self._force_transcription_model is None: + return message + + try: + message_obj = json.loads(message) + except (json.JSONDecodeError, TypeError): + return message + + if message_obj.get("type") not in ( + "session.update", + "transcription_session.update", + ): + return message + + session = message_obj.get("session") + if not isinstance(session, dict): + return message + + if session.get("type") == "transcription": + self._is_transcription_session = True + + authorized_model = self._force_transcription_model + changed = False + + transcription = session.get("input_audio_transcription") + if ( + isinstance(transcription, dict) + and transcription.get("model") != authorized_model + ): + session["input_audio_transcription"] = { + **transcription, + "model": authorized_model, + } + changed = True + + audio = session.get("audio") + if isinstance(audio, dict): + audio_input = audio.get("input") + if isinstance(audio_input, dict): + nested_transcription = audio_input.get("transcription") + if ( + isinstance(nested_transcription, dict) + and nested_transcription.get("model") != authorized_model + ): + session["audio"] = { + **audio, + "input": { + **audio_input, + "transcription": { + **nested_transcription, + "model": authorized_model, + }, + }, + } + changed = True + + if not changed: + return message + return json.dumps(message_obj) + def _uses_deferred_backend_setup(self) -> bool: """True when setup is deferred until the client's first session.update.""" if self.provider_config is None: diff --git a/litellm/llms/azure/realtime/handler.py b/litellm/llms/azure/realtime/handler.py index 5340eb4916bd..a2efc00271b1 100644 --- a/litellm/llms/azure/realtime/handler.py +++ b/litellm/llms/azure/realtime/handler.py @@ -134,9 +134,15 @@ async def async_realtime( websocket, cast(ClientConnection, backend_ws), logging_obj, + model=model, user_api_key_dict=user_api_key_dict, request_data={"litellm_metadata": litellm_metadata or {}}, backend_uses_beta_protocol=backend_uses_beta_protocol, + force_transcription_model=( + model + if (query_params or {}).get("intent") == "transcription" + else None + ), ) await realtime_streaming.bidirectional_forward() diff --git a/litellm/llms/custom_httpx/llm_http_handler.py b/litellm/llms/custom_httpx/llm_http_handler.py index 321bab822084..1a968707f1a9 100644 --- a/litellm/llms/custom_httpx/llm_http_handler.py +++ b/litellm/llms/custom_httpx/llm_http_handler.py @@ -5336,6 +5336,11 @@ async def async_realtime( model, user_api_key_dict=user_api_key_dict, request_data=_request_data, + force_transcription_model=( + model + if (query_params or {}).get("intent") == "transcription" + else None + ), ) if _session_config: realtime_streaming.session_configuration_request = _session_config diff --git a/litellm/llms/openai/realtime/handler.py b/litellm/llms/openai/realtime/handler.py index f34dae2df091..6751004f1b1a 100644 --- a/litellm/llms/openai/realtime/handler.py +++ b/litellm/llms/openai/realtime/handler.py @@ -157,8 +157,14 @@ async def async_realtime( websocket, cast(ClientConnection, backend_ws), logging_obj, + model=model, user_api_key_dict=user_api_key_dict, request_data={"litellm_metadata": litellm_metadata or {}}, + force_transcription_model=( + model + if (query_params or {}).get("intent") == "transcription" + else None + ), ) await realtime_streaming.bidirectional_forward() diff --git a/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py b/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py index 10bfd473b419..9015d2299cd9 100644 --- a/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py +++ b/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py @@ -632,6 +632,123 @@ def test_client_session_update_marks_transcription_session(): assert streaming._is_transcription_session is True +@pytest.mark.asyncio +async def test_transcription_session_update_enforces_authorized_flat_model(): + backend_ws = MagicMock() + backend_ws.send = AsyncMock() + streaming = RealTimeStreaming( + MagicMock(), + backend_ws, + MagicMock(), + model="gpt-realtime-whisper", + force_transcription_model="gpt-realtime-whisper", + ) + + await streaming._send_to_backend( + json.dumps( + { + "type": "session.update", + "session": { + "type": "transcription", + "input_audio_transcription": { + "model": "restricted-transcription-model", + "language": "en", + }, + }, + } + ) + ) + + sent = json.loads(backend_ws.send.await_args.args[0]) + assert sent["session"]["input_audio_transcription"] == { + "model": "gpt-realtime-whisper", + "language": "en", + } + assert streaming._is_transcription_session is True + + +@pytest.mark.asyncio +async def test_transcription_session_update_enforces_authorized_nested_model(): + backend_ws = MagicMock() + backend_ws.send = AsyncMock() + streaming = RealTimeStreaming( + MagicMock(), + backend_ws, + MagicMock(), + model="gpt-realtime-whisper", + force_transcription_model="gpt-realtime-whisper", + ) + + await streaming._send_to_backend( + json.dumps( + { + "type": "session.update", + "session": { + "type": "transcription", + "audio": { + "input": { + "transcription": { + "model": "restricted-transcription-model", + "prompt": "domain words", + }, + "format": {"type": "audio/pcm", "rate": 24000}, + } + }, + }, + } + ) + ) + + sent = json.loads(backend_ws.send.await_args.args[0]) + assert sent["session"]["audio"]["input"]["transcription"] == { + "model": "gpt-realtime-whisper", + "prompt": "domain words", + } + assert sent["session"]["audio"]["input"]["format"] == { + "type": "audio/pcm", + "rate": 24000, + } + assert streaming._is_transcription_session is True + + +@pytest.mark.asyncio +async def test_normal_realtime_session_keeps_nested_transcription_model(): + backend_ws = MagicMock() + backend_ws.send = AsyncMock() + streaming = RealTimeStreaming( + MagicMock(), + backend_ws, + MagicMock(), + model="gpt-4o-realtime-preview", + ) + + await streaming._send_to_backend( + json.dumps( + { + "type": "session.update", + "session": { + "type": "realtime", + "audio": { + "input": { + "transcription": { + "model": "whisper-1", + "language": "en", + } + } + }, + }, + } + ) + ) + + sent = json.loads(backend_ws.send.await_args.args[0]) + assert sent["session"]["audio"]["input"]["transcription"] == { + "model": "whisper-1", + "language": "en", + } + assert streaming._is_transcription_session is False + + def test_detect_transcription_session_from_backend_transcription_session_events(): """Backend transcription_session.created/updated events flag the session.""" streaming = RealTimeStreaming(MagicMock(), MagicMock(), MagicMock()) From b94a3061dd9f42e3069fc598bcea9f2edf6b8df6 Mon Sep 17 00:00:00 2001 From: Emerson Gomes Date: Sat, 6 Jun 2026 14:24:14 -0500 Subject: [PATCH 7/9] Enforce realtime transcription model access --- litellm/proxy/auth/auth_checks.py | 21 ++++++++ litellm/proxy/proxy_server.py | 11 ++++ litellm/proxy/realtime_endpoints/endpoints.py | 10 ++++ .../test_realtime_webrtc_endpoints.py | 54 +++++++++++++++++++ 4 files changed, 96 insertions(+) diff --git a/litellm/proxy/auth/auth_checks.py b/litellm/proxy/auth/auth_checks.py index 93a64889458f..6e4a2dde6850 100644 --- a/litellm/proxy/auth/auth_checks.py +++ b/litellm/proxy/auth/auth_checks.py @@ -3120,6 +3120,27 @@ async def can_key_call_model( raise +async def can_key_call_resolved_model( + model: str, + llm_model_list: Optional[list], + valid_token: UserAPIKeyAuth, + llm_router: Optional[litellm.Router], +) -> None: + if valid_token.config: + return + if ( + isinstance(valid_token.models, list) + and SpecialModelNames.all_team_models.value in valid_token.models + ): + return + await can_key_call_model( + model=model, + llm_model_list=llm_model_list, + valid_token=valid_token, + llm_router=llm_router, + ) + + def can_org_access_model( model: str, org_object: Optional[LiteLLM_OrganizationTable], diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index 387d4b6f6b3e..e5fd6b655c98 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -254,6 +254,7 @@ def generate_feedback_box(): ) from litellm.proxy.auth.auth_checks import ( ExperimentalUIJWTToken, + can_key_call_resolved_model, get_team_object, log_db_metrics, ) @@ -9476,6 +9477,16 @@ async def realtime_websocket_endpoint( ) return assert route_model is not None + try: + await can_key_call_resolved_model( + model=route_model, + llm_model_list=llm_model_list, + valid_token=user_api_key_dict, + llm_router=llm_router, + ) + except ProxyException as e: + await websocket.close(code=1008, reason=e.message[:120]) + return await websocket.accept(**accept_kwargs) # Only use explicit parameters, not all query params diff --git a/litellm/proxy/realtime_endpoints/endpoints.py b/litellm/proxy/realtime_endpoints/endpoints.py index 94b1384b712c..f457ef4efe33 100644 --- a/litellm/proxy/realtime_endpoints/endpoints.py +++ b/litellm/proxy/realtime_endpoints/endpoints.py @@ -10,6 +10,7 @@ from litellm._logging import verbose_proxy_logger from litellm.proxy._types import ProxyException, UserAPIKeyAuth +from litellm.proxy.auth.auth_checks import can_key_call_resolved_model from litellm.proxy.auth.user_api_key_auth import user_api_key_auth from litellm.proxy.common_utils.encrypt_decrypt_utils import ( decrypt_value_helper, @@ -410,6 +411,7 @@ async def create_realtime_transcription_session( add_litellm_data_to_request, general_settings, llm_router, + llm_model_list, proxy_config, proxy_logging_obj, route_request, @@ -423,6 +425,12 @@ async def create_realtime_transcription_session( req = RealtimeTranscriptionSessionRequest(**body) model: str = req.resolved_model() or "gpt-realtime-whisper" + await can_key_call_resolved_model( + model=model, + valid_token=user_api_key_dict, + llm_model_list=llm_model_list, + llm_router=llm_router, + ) transcription_session = {k: v for k, v in body.items() if k != "model"} data = {"model": model, "transcription_session": transcription_session} @@ -464,6 +472,8 @@ async def create_realtime_transcription_session( "litellm.proxy.realtime_endpoints.create_realtime_transcription_session(): Exception - %s", str(e), ) + if isinstance(e, ProxyException): + raise e if isinstance(e, HTTPException): raise ProxyException( message=getattr(e, "detail", getattr(e, "message", str(e))), diff --git a/tests/test_litellm/proxy/realtime_endpoints/test_realtime_webrtc_endpoints.py b/tests/test_litellm/proxy/realtime_endpoints/test_realtime_webrtc_endpoints.py index 362307252ee5..e443d9ce1369 100644 --- a/tests/test_litellm/proxy/realtime_endpoints/test_realtime_webrtc_endpoints.py +++ b/tests/test_litellm/proxy/realtime_endpoints/test_realtime_webrtc_endpoints.py @@ -436,6 +436,60 @@ def _raise_401(): proxy_app.dependency_overrides.pop(user_api_key_auth, None) +@pytest.mark.asyncio +async def test_transcription_sessions_rejects_disallowed_resolved_model( + proxy_app, +): + proxy_app.dependency_overrides[user_api_key_auth] = lambda: UserAPIKeyAuth( + user_id="test-user", + models=["gpt-4o-realtime-preview"], + ) + try: + client = TestClient(proxy_app, raise_server_exceptions=False) + with ( + patch("litellm.proxy.proxy_server.route_request") as mock_route_request, + patch("litellm.proxy.proxy_server.proxy_logging_obj") as mock_logging, + ): + mock_logging.post_call_failure_hook = AsyncMock() + + response = client.post( + "/v1/realtime/transcription_sessions", + headers={"Authorization": "Bearer sk-test-master-key"}, + json={ + "input_audio_transcription": {"model": "gpt-realtime-whisper"} + }, + ) + + assert response.status_code == 403 + assert "Tried to access gpt-realtime-whisper" in response.text + mock_route_request.assert_not_called() + finally: + proxy_app.dependency_overrides.pop(user_api_key_auth, None) + + +@pytest.mark.asyncio +async def test_realtime_transcription_websocket_default_model_checks_key_scope(): + from litellm.proxy import proxy_server + + websocket = MagicMock() + websocket.headers = {} + websocket.close = AsyncMock() + websocket.accept = AsyncMock() + + await proxy_server.realtime_websocket_endpoint( + websocket=websocket, + model=None, + intent="transcription", + user_api_key_dict=UserAPIKeyAuth(models=["gpt-4o-realtime-preview"]), + ) + + websocket.accept.assert_not_awaited() + websocket.close.assert_awaited_once() + _, close_kwargs = websocket.close.call_args + assert close_kwargs["code"] == 1008 + assert "not allowed to access model" in close_kwargs["reason"] + + @pytest.mark.asyncio async def test_transcription_sessions_encrypts_client_secret( proxy_app, From 2961910849613fe2133cd2e31461e1d261ea4500 Mon Sep 17 00:00:00 2001 From: Emerson Gomes Date: Sat, 6 Jun 2026 16:34:52 -0500 Subject: [PATCH 8/9] Enforce realtime resolved model scopes --- litellm/proxy/auth/auth_checks.py | 91 ++++++++- .../test_realtime_webrtc_endpoints.py | 188 ++++++++++++++++++ 2 files changed, 269 insertions(+), 10 deletions(-) diff --git a/litellm/proxy/auth/auth_checks.py b/litellm/proxy/auth/auth_checks.py index 6e4a2dde6850..6877aea63cf1 100644 --- a/litellm/proxy/auth/auth_checks.py +++ b/litellm/proxy/auth/auth_checks.py @@ -3126,19 +3126,90 @@ async def can_key_call_resolved_model( valid_token: UserAPIKeyAuth, llm_router: Optional[litellm.Router], ) -> None: - if valid_token.config: - return - if ( + from litellm.proxy.proxy_server import ( + prisma_client, + proxy_logging_obj, + user_api_key_cache, + ) + + skip_key_model_check = valid_token.config or ( isinstance(valid_token.models, list) and SpecialModelNames.all_team_models.value in valid_token.models - ): - return - await can_key_call_model( - model=model, - llm_model_list=llm_model_list, - valid_token=valid_token, - llm_router=llm_router, ) + if not skip_key_model_check: + await can_key_call_model( + model=model, + llm_model_list=llm_model_list, + valid_token=valid_token, + llm_router=llm_router, + ) + + team_object: Optional[LiteLLM_TeamTableCachedObj] = None + team_object_from_lookup = False + if valid_token.team_id is not None: + try: + team_object = await get_team_object( + team_id=valid_token.team_id, + prisma_client=prisma_client, + user_api_key_cache=user_api_key_cache, + parent_otel_span=valid_token.parent_otel_span, + proxy_logging_obj=proxy_logging_obj, + ) + team_object_from_lookup = True + except Exception: + team_object = LiteLLM_TeamTableCachedObj( + team_id=valid_token.team_id, + models=valid_token.team_models, + blocked=valid_token.team_blocked, + team_alias=valid_token.team_alias, + metadata=valid_token.team_metadata, + object_permission_id=valid_token.team_object_permission_id, + object_permission=valid_token.team_object_permission, + ) + + if team_object is not None: + try: + await can_team_access_model( + model=model, + team_object=team_object, + llm_router=llm_router, + team_model_aliases=valid_token.team_model_aliases, + ) + except ProxyException as team_denial: + if team_denial.type != ProxyErrorTypes.team_model_access_denied: + raise + if not await _key_access_group_grants_model( + model=model, + valid_token=valid_token, + team_object=team_object, + llm_router=llm_router, + ): + raise + + if valid_token.user_id is not None and team_object_from_lookup: + await _check_team_member_model_access( + model=model, + team_object=team_object, + valid_token=valid_token, + llm_router=llm_router, + prisma_client=prisma_client, + user_api_key_cache=user_api_key_cache, + proxy_logging_obj=proxy_logging_obj, + ) + + if valid_token.project_id is not None: + project_object = await get_project_object( + project_id=valid_token.project_id, + prisma_client=prisma_client, + user_api_key_cache=user_api_key_cache, + proxy_logging_obj=proxy_logging_obj, + ) + if project_object is not None and len(project_object.models) > 0: + can_project_access_model( + model=model, + project_object=project_object, + llm_router=llm_router, + ) def can_org_access_model( diff --git a/tests/test_litellm/proxy/realtime_endpoints/test_realtime_webrtc_endpoints.py b/tests/test_litellm/proxy/realtime_endpoints/test_realtime_webrtc_endpoints.py index e443d9ce1369..ffd438bd015e 100644 --- a/tests/test_litellm/proxy/realtime_endpoints/test_realtime_webrtc_endpoints.py +++ b/tests/test_litellm/proxy/realtime_endpoints/test_realtime_webrtc_endpoints.py @@ -467,6 +467,152 @@ async def test_transcription_sessions_rejects_disallowed_resolved_model( proxy_app.dependency_overrides.pop(user_api_key_auth, None) +@pytest.mark.asyncio +async def test_transcription_sessions_rejects_disallowed_team_model_scope( + proxy_app, +): + from litellm.proxy._types import LiteLLM_TeamTableCachedObj + + team = LiteLLM_TeamTableCachedObj( + team_id="team-a", + models=["gpt-4o-realtime-preview"], + ) + proxy_app.dependency_overrides[user_api_key_auth] = lambda: UserAPIKeyAuth( + user_id="test-user", + team_id="team-a", + models=["*"], + ) + try: + client = TestClient(proxy_app, raise_server_exceptions=False) + with ( + patch("litellm.proxy.proxy_server.route_request") as mock_route_request, + patch("litellm.proxy.proxy_server.proxy_logging_obj") as mock_logging, + patch( + "litellm.proxy.auth.auth_checks.get_team_object", + new=AsyncMock(return_value=team), + ), + patch( + "litellm.proxy.auth.auth_checks.get_team_membership", + new=AsyncMock(return_value=None), + ), + ): + mock_logging.post_call_failure_hook = AsyncMock() + + response = client.post( + "/v1/realtime/transcription_sessions", + headers={"Authorization": "Bearer sk-test-master-key"}, + json={ + "input_audio_transcription": {"model": "gpt-realtime-whisper"} + }, + ) + + assert response.status_code == 403 + assert "team" in response.text.lower() + assert "Tried to access gpt-realtime-whisper" in response.text + mock_route_request.assert_not_called() + finally: + proxy_app.dependency_overrides.pop(user_api_key_auth, None) + + +@pytest.mark.asyncio +async def test_transcription_sessions_rejects_disallowed_project_model_scope( + proxy_app, +): + from litellm.proxy._types import LiteLLM_ProjectTableCachedObj + + project = LiteLLM_ProjectTableCachedObj( + project_id="project-a", + models=["gpt-4o-realtime-preview"], + created_by="test-user", + updated_by="test-user", + ) + proxy_app.dependency_overrides[user_api_key_auth] = lambda: UserAPIKeyAuth( + user_id="test-user", + project_id="project-a", + models=["*"], + ) + try: + client = TestClient(proxy_app, raise_server_exceptions=False) + with ( + patch("litellm.proxy.proxy_server.route_request") as mock_route_request, + patch("litellm.proxy.proxy_server.proxy_logging_obj") as mock_logging, + patch( + "litellm.proxy.auth.auth_checks.get_project_object", + new=AsyncMock(return_value=project), + ), + ): + mock_logging.post_call_failure_hook = AsyncMock() + + response = client.post( + "/v1/realtime/transcription_sessions", + headers={"Authorization": "Bearer sk-test-master-key"}, + json={ + "input_audio_transcription": {"model": "gpt-realtime-whisper"} + }, + ) + + assert response.status_code == 403 + assert "project" in response.text.lower() + assert "Tried to access gpt-realtime-whisper" in response.text + mock_route_request.assert_not_called() + finally: + proxy_app.dependency_overrides.pop(user_api_key_auth, None) + + +@pytest.mark.asyncio +async def test_transcription_sessions_rejects_disallowed_team_member_model_scope( + proxy_app, +): + from litellm.proxy._types import ( + LiteLLM_BudgetTable, + LiteLLM_TeamMembership, + LiteLLM_TeamTableCachedObj, + ) + + team = LiteLLM_TeamTableCachedObj(team_id="team-a", models=["*"]) + membership = LiteLLM_TeamMembership( + user_id="test-user", + team_id="team-a", + litellm_budget_table=LiteLLM_BudgetTable( + allowed_models=["gpt-4o-realtime-preview"], + ), + ) + proxy_app.dependency_overrides[user_api_key_auth] = lambda: UserAPIKeyAuth( + user_id="test-user", + team_id="team-a", + models=["*"], + ) + try: + client = TestClient(proxy_app, raise_server_exceptions=False) + with ( + patch("litellm.proxy.proxy_server.route_request") as mock_route_request, + patch("litellm.proxy.proxy_server.proxy_logging_obj") as mock_logging, + patch( + "litellm.proxy.auth.auth_checks.get_team_object", + new=AsyncMock(return_value=team), + ), + patch( + "litellm.proxy.auth.auth_checks.get_team_membership", + new=AsyncMock(return_value=membership), + ), + ): + mock_logging.post_call_failure_hook = AsyncMock() + + response = client.post( + "/v1/realtime/transcription_sessions", + headers={"Authorization": "Bearer sk-test-master-key"}, + json={ + "input_audio_transcription": {"model": "gpt-realtime-whisper"} + }, + ) + + assert response.status_code == 403 + assert "Team member not allowed to access model" in response.text + mock_route_request.assert_not_called() + finally: + proxy_app.dependency_overrides.pop(user_api_key_auth, None) + + @pytest.mark.asyncio async def test_realtime_transcription_websocket_default_model_checks_key_scope(): from litellm.proxy import proxy_server @@ -490,6 +636,48 @@ async def test_realtime_transcription_websocket_default_model_checks_key_scope() assert "not allowed to access model" in close_kwargs["reason"] +@pytest.mark.asyncio +async def test_realtime_transcription_websocket_default_model_checks_team_scope(): + from litellm.proxy import proxy_server + from litellm.proxy._types import LiteLLM_TeamTableCachedObj + + team = LiteLLM_TeamTableCachedObj( + team_id="team-a", + models=["gpt-4o-realtime-preview"], + ) + websocket = MagicMock() + websocket.headers = {} + websocket.close = AsyncMock() + websocket.accept = AsyncMock() + + with ( + patch( + "litellm.proxy.auth.auth_checks.get_team_object", + new=AsyncMock(return_value=team), + ), + patch( + "litellm.proxy.auth.auth_checks.get_team_membership", + new=AsyncMock(return_value=None), + ), + ): + await proxy_server.realtime_websocket_endpoint( + websocket=websocket, + model=None, + intent="transcription", + user_api_key_dict=UserAPIKeyAuth( + user_id="test-user", + team_id="team-a", + models=["*"], + ), + ) + + websocket.accept.assert_not_awaited() + websocket.close.assert_awaited_once() + _, close_kwargs = websocket.close.call_args + assert close_kwargs["code"] == 1008 + assert "not allowed to access model" in close_kwargs["reason"] + + @pytest.mark.asyncio async def test_transcription_sessions_encrypts_client_secret( proxy_app, From 9b0602f019b0ecac48808a127e14ae0f3a463e53 Mon Sep 17 00:00:00 2001 From: Emerson Gomes Date: Sat, 6 Jun 2026 17:14:26 -0500 Subject: [PATCH 9/9] Enforce WebRTC transcription model scope --- litellm/proxy/realtime_endpoints/endpoints.py | 170 ++++++++++++++++-- .../test_realtime_webrtc_endpoints.py | 140 +++++++++++++++ 2 files changed, 294 insertions(+), 16 deletions(-) diff --git a/litellm/proxy/realtime_endpoints/endpoints.py b/litellm/proxy/realtime_endpoints/endpoints.py index f457ef4efe33..a953dbec6b70 100644 --- a/litellm/proxy/realtime_endpoints/endpoints.py +++ b/litellm/proxy/realtime_endpoints/endpoints.py @@ -27,6 +27,136 @@ router = APIRouter() _REALTIME_TOKEN_VERSION = "realtime_v1" +_DEFAULT_REALTIME_MODEL = "gpt-4o-realtime-preview" +_DEFAULT_TRANSCRIPTION_MODEL = "gpt-realtime-whisper" +_ALLOWED_SESSION_TYPES = ("realtime", "transcription") + + +def _coerce_realtime_session_type(session_type: Optional[str]) -> str: + if session_type in _ALLOWED_SESSION_TYPES: + return session_type + return "realtime" + + +def _append_model_candidate(candidates: list[str], model: Any) -> None: + if isinstance(model, str) and model and model not in candidates: + candidates.append(model) + + +def _transcription_model_candidates_from_session(session: dict) -> list[str]: + candidates: list[str] = [] + + audio = session.get("audio") + if isinstance(audio, dict): + audio_input = audio.get("input") + if isinstance(audio_input, dict): + nested_transcription = audio_input.get("transcription") + if isinstance(nested_transcription, dict): + _append_model_candidate( + candidates, + nested_transcription.get("model"), + ) + + flat_transcription = session.get("input_audio_transcription") + if isinstance(flat_transcription, dict): + _append_model_candidate(candidates, flat_transcription.get("model")) + + return candidates + + +def _set_transcription_model_on_session( + session: dict, + model: str, + create_if_missing: bool = False, +) -> None: + updated_existing_config = False + + flat_transcription = session.get("input_audio_transcription") + if isinstance(flat_transcription, dict): + session["input_audio_transcription"] = { + **flat_transcription, + "model": model, + } + updated_existing_config = True + + audio = session.get("audio") + if isinstance(audio, dict): + audio_input = audio.get("input") + if isinstance(audio_input, dict): + nested_transcription = audio_input.get("transcription") + if isinstance(nested_transcription, dict): + session["audio"] = { + **audio, + "input": { + **audio_input, + "transcription": { + **nested_transcription, + "model": model, + }, + }, + } + updated_existing_config = True + + if updated_existing_config or not create_if_missing: + return + + audio = audio if isinstance(audio, dict) else {} + audio_input = audio.get("input") + audio_input = audio_input if isinstance(audio_input, dict) else {} + session["audio"] = { + **audio, + "input": { + **audio_input, + "transcription": {"model": model}, + }, + } + + +async def _prepare_client_secret_session( + req: RealtimeClientSecretRequest, + user_api_key_dict: UserAPIKeyAuth, + llm_model_list: Optional[list], + llm_router: Any, +) -> tuple[str, Optional[dict], str]: + session_type = _coerce_realtime_session_type( + req.session.type if req.session else None + ) + session_data: Optional[dict] = ( + req.session.model_dump(exclude_none=True) if req.session else None + ) + if session_data is not None: + session_data["type"] = session_type + + session_model = req.session.model if req.session else None + model: str = session_model or req.model or _DEFAULT_REALTIME_MODEL + if session_type != "transcription": + return model, session_data, session_type + + transcription_model_candidates = _transcription_model_candidates_from_session( + session_data or {} + ) + if not transcription_model_candidates: + _append_model_candidate(transcription_model_candidates, session_model) + _append_model_candidate(transcription_model_candidates, req.model) + if not transcription_model_candidates: + transcription_model_candidates.append(_DEFAULT_TRANSCRIPTION_MODEL) + + model = transcription_model_candidates[0] + for transcription_model in transcription_model_candidates: + await can_key_call_resolved_model( + model=transcription_model, + valid_token=user_api_key_dict, + llm_model_list=llm_model_list, + llm_router=llm_router, + ) + if session_data is not None: + _set_transcription_model_on_session( + session=session_data, + model=model, + create_if_missing=True, + ) + session_data.pop("model", None) + return model, session_data, session_type def _encode_realtime_token_payload( @@ -99,6 +229,7 @@ async def create_realtime_client_secret( add_litellm_data_to_request, general_settings, llm_router, + llm_model_list, proxy_config, proxy_logging_obj, route_request, @@ -111,17 +242,18 @@ async def create_realtime_client_secret( body = await _read_request_body(request=request) req = RealtimeClientSecretRequest(**body) - model: str = ( - (req.session.model if req.session else None) - or req.model - or "gpt-4o-realtime-preview" + model, session_data, session_type = await _prepare_client_secret_session( + req=req, + user_api_key_dict=user_api_key_dict, + llm_model_list=llm_model_list, + llm_router=llm_router, ) data = {"model": model} # If session is provided, use it; otherwise create one from model - if req.session: - data["session"] = req.session.model_dump(exclude_none=True) + if session_data is not None: + data["session"] = session_data elif req.model: # User provided model at root level, convert to session format data["session"] = {"type": "realtime", "model": model} @@ -166,6 +298,8 @@ async def create_realtime_client_secret( "litellm.proxy.realtime_endpoints.webrtc.create_realtime_client_secret(): Exception - %s", str(e), ) + if isinstance(e, ProxyException): + raise e if isinstance(e, HTTPException): raise ProxyException( message=getattr(e, "message", str(e)), @@ -204,9 +338,7 @@ async def create_realtime_client_secret( user_id=getattr(user_api_key_dict, "user_id", None), team_id=getattr(user_api_key_dict, "team_id", None), expires_at=expires_at if isinstance(expires_at, int) else None, - session_type=( - req.session.type if req.session and req.session.type else "realtime" - ), + session_type=session_type, ) encrypted_token: str = encrypt_value_helper(token_payload) upstream_json["value"] = encrypted_token @@ -287,17 +419,17 @@ async def proxy_realtime_calls( model = ( decoded_payload.get("model_id") or request.query_params.get("model") - or "gpt-4o-realtime-preview" + or _DEFAULT_REALTIME_MODEL ) user_id = decoded_payload.get("user_id") or None team_id = decoded_payload.get("team_id") or None - session_type = decoded_payload.get("session_type") or "realtime" - if session_type not in ("realtime", "transcription"): - session_type = "realtime" + session_type = _coerce_realtime_session_type( + decoded_payload.get("session_type") + ) else: # Backward compatibility: older tokens contained only encrypted upstream key. openai_ephemeral_key = decrypted_token_value - model = request.query_params.get("model", "gpt-4o-realtime-preview") + model = request.query_params.get("model", _DEFAULT_REALTIME_MODEL) user_id = None team_id = None session_type = "realtime" @@ -311,11 +443,17 @@ async def proxy_realtime_calls( data: dict = {} try: - # Build session config for the multipart form data session_config = { "type": session_type, - "model": model, } + if session_type == "transcription": + _set_transcription_model_on_session( + session=session_config, + model=model, + create_if_missing=True, + ) + else: + session_config["model"] = model data = { "model": model, diff --git a/tests/test_litellm/proxy/realtime_endpoints/test_realtime_webrtc_endpoints.py b/tests/test_litellm/proxy/realtime_endpoints/test_realtime_webrtc_endpoints.py index ffd438bd015e..65853df392f4 100644 --- a/tests/test_litellm/proxy/realtime_endpoints/test_realtime_webrtc_endpoints.py +++ b/tests/test_litellm/proxy/realtime_endpoints/test_realtime_webrtc_endpoints.py @@ -241,6 +241,142 @@ async def test_client_secrets_success_with_mock( proxy_app.dependency_overrides.pop(user_api_key_auth, None) +@pytest.mark.asyncio +async def test_client_secrets_transcription_rejects_disallowed_nested_model( + proxy_app, +): + proxy_app.dependency_overrides[user_api_key_auth] = lambda: UserAPIKeyAuth( + user_id="test-user", + models=["gpt-4o-realtime-preview"], + ) + try: + client = TestClient(proxy_app, raise_server_exceptions=False) + with ( + patch("litellm.proxy.proxy_server.route_request") as mock_route_request, + patch("litellm.proxy.proxy_server.proxy_logging_obj") as mock_logging, + ): + mock_logging.post_call_failure_hook = AsyncMock() + + response = client.post( + "/v1/realtime/client_secrets", + headers={"Authorization": "Bearer sk-test-master-key"}, + json={ + "model": "gpt-4o-realtime-preview", + "session": { + "type": "transcription", + "model": "gpt-4o-realtime-preview", + "audio": { + "input": { + "transcription": { + "model": "gpt-realtime-whisper" + } + } + }, + }, + }, + ) + + assert response.status_code == 403 + assert "Tried to access gpt-realtime-whisper" in response.text + mock_route_request.assert_not_called() + finally: + proxy_app.dependency_overrides.pop(user_api_key_auth, None) + + +@pytest.mark.asyncio +async def test_client_secrets_transcription_routes_on_nested_model( + proxy_app, + mock_add_litellm_data, + mock_pre_call_hook, +): + proxy_app.dependency_overrides[user_api_key_auth] = lambda: UserAPIKeyAuth( + user_id="test-user", + models=["gpt-4o-realtime-preview", "gpt-realtime-whisper"], + ) + captured = {} + future_expires_at = int(time.time()) + 3600 + + async def _capturing_route(*args, **kwargs): + captured["data"] = kwargs.get("data") + + async def _inner(): + resp = MagicMock(spec=httpx.Response) + resp.status_code = 200 + resp.text = ( + f'{{"value":"upstream_ephemeral_key","expires_at":{future_expires_at}}}' + ) + resp.content = ( + f'{{"value":"upstream_ephemeral_key","expires_at":{future_expires_at}}}' + ).encode() + resp.headers = {} + resp.json.return_value = { + "value": "upstream_ephemeral_key", + "expires_at": future_expires_at, + } + return resp + + return _inner() + + try: + client = TestClient(proxy_app) + with ( + patch( + "litellm.proxy.proxy_server.route_request", + side_effect=_capturing_route, + ), + patch( + "litellm.proxy.proxy_server.add_litellm_data_to_request", + side_effect=mock_add_litellm_data, + ), + patch("litellm.proxy.proxy_server.proxy_logging_obj") as mock_logging, + ): + mock_logging.pre_call_hook = AsyncMock(side_effect=mock_pre_call_hook) + mock_logging.post_call_failure_hook = AsyncMock() + + response = client.post( + "/v1/realtime/client_secrets", + headers={"Authorization": "Bearer sk-test-master-key"}, + json={ + "model": "gpt-4o-realtime-preview", + "session": { + "type": "transcription", + "model": "gpt-4o-realtime-preview", + "audio": { + "input": { + "transcription": { + "model": "gpt-realtime-whisper" + } + } + }, + }, + }, + ) + + assert response.status_code == 200 + assert captured["data"]["model"] == "gpt-realtime-whisper" + session = captured["data"]["session"] + assert session["type"] == "transcription" + assert "model" not in session + assert ( + session["audio"]["input"]["transcription"]["model"] + == "gpt-realtime-whisper" + ) + encrypted_value = response.json()["value"] + decoded = _decode_realtime_token_payload( + decrypt_value_helper( + encrypted_value, + key="client_secret.value", + exception_type="debug", + ) + or "" + ) + assert decoded is not None + assert decoded["model_id"] == "gpt-realtime-whisper" + assert decoded["session_type"] == "transcription" + finally: + proxy_app.dependency_overrides.pop(user_api_key_auth, None) + + def test_realtime_calls_requires_auth(proxy_app): """POST /v1/realtime/calls returns 401 without Authorization. @@ -384,6 +520,10 @@ async def _inner(): ) assert captured["session"]["type"] == "transcription" + assert ( + captured["session"]["audio"]["input"]["transcription"]["model"] + == "gpt-realtime-whisper" + ) # --- transcription_sessions endpoint ---