diff --git a/litellm/proxy/common_request_processing.py b/litellm/proxy/common_request_processing.py index ce39ecf52d..9a0f371d1c 100644 --- a/litellm/proxy/common_request_processing.py +++ b/litellm/proxy/common_request_processing.py @@ -744,6 +744,8 @@ class ProxyBaseLLMRequestProcessing: "aembedding", "aresponses", "_arealtime", + "acreate_realtime_client_secret", + "arealtime_calls", "aget_responses", "adelete_responses", "acancel_responses", diff --git a/litellm/proxy/realtime_endpoints/endpoints.py b/litellm/proxy/realtime_endpoints/endpoints.py new file mode 100644 index 0000000000..70fb897c14 --- /dev/null +++ b/litellm/proxy/realtime_endpoints/endpoints.py @@ -0,0 +1,352 @@ +#### Realtime WebRTC Endpoints ##### + +import json +from typing import Any, Dict, Optional + +from fastapi import APIRouter, Depends, HTTPException, Request, Response +from fastapi import status as http_status + +from litellm._logging import verbose_proxy_logger +from litellm.proxy._types import ProxyException, UserAPIKeyAuth +from litellm.proxy.auth.user_api_key_auth import user_api_key_auth +from litellm.proxy.common_request_processing import ProxyBaseLLMRequestProcessing +from litellm.proxy.common_utils.encrypt_decrypt_utils import ( + decrypt_value_helper, + encrypt_value_helper, +) +from litellm.proxy.common_utils.http_parsing_utils import _read_request_body +from litellm.types.realtime import ( + RealtimeClientSecretRequest, + RealtimeClientSecretResponse, +) + +router = APIRouter() + +_REALTIME_TOKEN_VERSION = "realtime_v1" + + +def _encode_realtime_token_payload( + ephemeral_key: str, + model_id: str, + user_id: Optional[str], + team_id: Optional[str], + expires_at: Optional[int], +) -> str: + """ + Encode metadata with the upstream ephemeral key so /realtime/calls can + route without requiring model as a query param. + """ + payload: Dict[str, Any] = { + "v": _REALTIME_TOKEN_VERSION, + "ephemeral_key": ephemeral_key, + "model_id": model_id, + "user_id": user_id or "", + "team_id": team_id or "", + "expires_at": expires_at, + } + return json.dumps(payload, separators=(",", ":")) + + +def _decode_realtime_token_payload( + decrypted_value: str, +) -> Optional[Dict[str, Any]]: + """ + Decode realtime token payload; returns None for legacy/raw ephemeral tokens. + """ + try: + decoded = json.loads(decrypted_value) + except Exception: + return None + + if not isinstance(decoded, dict): + return None + if decoded.get("v") != _REALTIME_TOKEN_VERSION: + return None + if not isinstance(decoded.get("ephemeral_key"), str): + return None + if not isinstance(decoded.get("model_id"), str): + return None + return decoded + + +@router.post( + "/v1/realtime/client_secrets", + dependencies=[Depends(user_api_key_auth)], + tags=["realtime"], +) +@router.post( + "/realtime/client_secrets", + dependencies=[Depends(user_api_key_auth)], + tags=["realtime"], +) +@router.post( + "/openai/v1/realtime/client_secrets", + dependencies=[Depends(user_api_key_auth)], + tags=["realtime"], +) +async def create_realtime_client_secret( + request: Request, + fastapi_response: Response, + user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth), +) -> RealtimeClientSecretResponse: + 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 = RealtimeClientSecretRequest(**body) + + model: str = ( + (req.session.model if req.session else None) + or req.model + or "gpt-4o-realtime-preview" + ) + + 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) + elif req.model: + # User provided model at root level, convert to session format + data["session"] = {"type": "realtime", "model": model} + + if req.expires_after: + data["expires_after"] = req.expires_after.model_dump(exclude_none=True) + + 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_client_secret", + ) + + verbose_proxy_logger.debug( + "WebRTC: /v1/realtime/client_secrets (model=%s)", model + ) + + llm_call = await route_request( + data=data, + route_type="acreate_realtime_client_secret", + llm_router=llm_router, + user_model=user_model, + ) + upstream_resp = await llm_call + + 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.webrtc.create_realtime_client_secret(): 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( + "WebRTC client_secrets upstream error %s: %s", + upstream_resp.status_code, + upstream_resp.text, + ) + return Response( + content=upstream_resp.content, + status_code=upstream_resp.status_code, + media_type="application/json", + ) + + upstream_json: dict = upstream_resp.json() + + # Encrypt upstream ephemeral key with routing metadata so /realtime/calls + # can recover model without requiring query params. + raw_value: str = upstream_json.get("value", "") + expires_at = upstream_json.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, + ) + encrypted_token: str = encrypt_value_helper(token_payload) + upstream_json["value"] = encrypted_token + + session_obj: Optional[dict] = upstream_json.get("session") + if isinstance(session_obj, dict): + cs = session_obj.get("client_secret") + if isinstance(cs, dict) and "value" in cs: + cs["value"] = encrypted_token + upstream_json["session"] = session_obj + + return RealtimeClientSecretResponse(**upstream_json) + + +@router.post( + "/v1/realtime/calls", + tags=["realtime"], +) +@router.post( + "/realtime/calls", + tags=["realtime"], +) +@router.post( + "/openai/v1/realtime/calls", + tags=["realtime"], +) +async def proxy_realtime_calls( + request: Request, + fastapi_response: Response, +) -> Response: + 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, + ) + + # Auth: the Bearer token is the encrypted ephemeral key issued by + # /realtime/client_secrets, not a standard proxy API key. + auth_header: Optional[str] = request.headers.get("Authorization") + if not auth_header or not auth_header.startswith("Bearer "): + return Response( + content=json.dumps({"error": "Missing or invalid Authorization header"}), + status_code=http_status.HTTP_401_UNAUTHORIZED, + media_type="application/json", + ) + + encrypted_token = auth_header.removeprefix("Bearer ").strip() + decrypted_token_value = decrypt_value_helper( + value=encrypted_token, + key="realtime_calls_auth", + ) + if not decrypted_token_value: + return Response( + content=json.dumps({"error": "Invalid or expired token"}), + status_code=http_status.HTTP_401_UNAUTHORIZED, + media_type="application/json", + ) + + sdp_body: bytes = await request.body() + decoded_payload = _decode_realtime_token_payload(decrypted_token_value) + if decoded_payload is not None: + openai_ephemeral_key = decoded_payload.get("ephemeral_key", "") + model = ( + decoded_payload.get("model_id") + or request.query_params.get("model") + or "gpt-4o-realtime-preview" + ) + 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") + + # Build a minimal UserAPIKeyAuth so we can pass through the logging pipeline + # even though this endpoint uses the provider ephemeral key for auth. + minimal_auth = UserAPIKeyAuth() + + data: dict = {} + try: + # Build session config for the multipart form data + session_config = { + "type": "realtime", + "model": model, + } + + data = { + "model": model, + "openai_ephemeral_key": openai_ephemeral_key, + "sdp_body": sdp_body, + "session": session_config, + } + + data = await add_litellm_data_to_request( + data=data, + request=request, + general_settings=general_settings, + user_api_key_dict=minimal_auth, + version=version, + proxy_config=proxy_config, + ) + + data = await proxy_logging_obj.pre_call_hook( + user_api_key_dict=minimal_auth, + data=data, + call_type="arealtime_calls", + ) + + verbose_proxy_logger.debug( + "WebRTC: /v1/realtime/calls (model=%s)", model + ) + + llm_call = await route_request( + data=data, + route_type="arealtime_calls", + llm_router=llm_router, + user_model=user_model, + ) + upstream_resp = await llm_call + + except Exception as e: + await proxy_logging_obj.post_call_failure_hook( + user_api_key_dict=minimal_auth, + original_exception=e, + request_data=data, + ) + verbose_proxy_logger.error( + "litellm.proxy.realtime_endpoints.webrtc.proxy_realtime_calls(): 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), + ) + + return Response( + content=upstream_resp.content, + status_code=upstream_resp.status_code, + media_type=upstream_resp.headers.get("content-type", "application/sdp"), + ) diff --git a/litellm/types/utils.py b/litellm/types/utils.py index 341e5117fd..74afb5fe2e 100644 --- a/litellm/types/utils.py +++ b/litellm/types/utils.py @@ -492,6 +492,8 @@ CallTypesLiteral = Literal[ "aresponses", "responses", "acreate_skill", + "acreate_realtime_client_secret", + "arealtime_calls", ] # Mapping of API routes to their corresponding call types