Add webrtc transformations and http handler

litellm/llms/azure/realtime/http_transformation.py

@@ -0,0 +1,47 @@
+"""Azure OpenAI realtime HTTP transformation config (client_secrets + realtime_calls)."""
+
+from typing import Optional
+
+import litellm
+from litellm.llms.base_llm.realtime.http_transformation import BaseRealtimeHTTPConfig
+from litellm.secret_managers.main import get_secret_str
+
+
+class AzureRealtimeHTTPConfig(BaseRealtimeHTTPConfig):
+    def get_api_base(self, api_base: Optional[str], **kwargs) -> str:
+        return (
+            api_base
+            or litellm.api_base
+            or get_secret_str("AZURE_API_BASE")
+            or ""
+        )
+
+    def get_api_key(self, api_key: Optional[str], **kwargs) -> str:
+        return (
+            api_key
+            or litellm.api_key
+            or get_secret_str("AZURE_API_KEY")
+            or ""
+        )
+
+    def get_complete_url(self, api_base: Optional[str], model: str) -> str:
+        base = self.get_api_base(api_base).rstrip("/")
+        return f"{base}/v1/realtime/client_secrets"
+
+    def validate_environment(
+        self,
+        headers: dict,
+        model: str,
+        api_key: Optional[str] = None,
+    ) -> dict:
+        return {
+            **headers,
+            "api-key": api_key or "",
+            "Content-Type": "application/json",
+        }
+
+    def get_realtime_calls_headers(self, ephemeral_key: str) -> dict:
+        return {
+            "api-key": ephemeral_key,
+            "Content-Type": "application/sdp",
+        }

litellm/llms/base_llm/realtime/http_transformation.py

@@ -0,0 +1,115 @@
+"""
+Base transformation class for realtime HTTP endpoints (client_secrets, realtime_calls).
+
+These are HTTP (not WebSocket) endpoints used by the WebRTC flow:
+  POST /v1/realtime/client_secrets  — obtains a short-lived ephemeral key
+  POST /v1/realtime/calls           — exchanges an SDP offer using that key
+"""
+
+from abc import ABC, abstractmethod
+from typing import Optional, Union
+
+import httpx
+
+
+class BaseRealtimeHTTPConfig(ABC):
+    """
+    Abstract base for provider-specific realtime HTTP credential / URL logic.
+
+    Implement one subclass per provider (OpenAI, Azure, …).
+    """
+
+    # ------------------------------------------------------------------ #
+    # Credential resolution                                                #
+    # ------------------------------------------------------------------ #
+
+    @abstractmethod
+    def get_api_base(
+        self,
+        api_base: Optional[str],
+        **kwargs,
+    ) -> str:
+        """
+        Resolve the provider API base URL.
+
+        Resolution order (provider-specific):
+          explicit api_base → litellm.api_base → env var → hard-coded default
+        """
+
+    @abstractmethod
+    def get_api_key(
+        self,
+        api_key: Optional[str],
+        **kwargs,
+    ) -> str:
+        """
+        Resolve the provider API key.
+
+        Resolution order (provider-specific):
+          explicit api_key → litellm.api_key → env var → ""
+        """
+
+    # ------------------------------------------------------------------ #
+    # client_secrets endpoint                                              #
+    # ------------------------------------------------------------------ #
+
+    @abstractmethod
+    def get_complete_url(self, api_base: Optional[str], model: str) -> str:
+        """Return the full URL for POST /realtime/client_secrets."""
+
+    @abstractmethod
+    def validate_environment(
+        self,
+        headers: dict,
+        model: str,
+        api_key: Optional[str] = None,
+    ) -> dict:
+        """
+        Build and return the request headers for the client_secrets call.
+
+        Merge `headers` (caller-supplied extras) with auth / content-type
+        headers required by this provider.
+        """
+
+    # ------------------------------------------------------------------ #
+    # realtime_calls endpoint                                              #
+    # ------------------------------------------------------------------ #
+
+    def get_realtime_calls_url(
+        self, api_base: Optional[str], model: str
+    ) -> str:
+        """Return the full URL for POST /realtime/calls (SDP exchange)."""
+        base = (api_base or "").rstrip("/")
+        return f"{base}/v1/realtime/calls"
+
+    def get_realtime_calls_headers(self, ephemeral_key: str) -> dict:
+        """
+        Build headers for the realtime_calls POST.
+
+        The Bearer token here is the ephemeral key obtained from
+        client_secrets, not the long-lived provider key.
+        """
+        return {
+            "Authorization": f"Bearer {ephemeral_key}",
+        }
+
+    # ------------------------------------------------------------------ #
+    # Error handling                                                      #
+    # ------------------------------------------------------------------ #
+
+    def get_error_class(
+        self, error_message: str, status_code: int, headers: Union[dict, httpx.Headers]
+    ):
+        """
+        Map HTTP errors to LiteLLM exception types.
+
+        Default: generic exception. Override in subclasses for provider-specific
+        error mapping (e.g., Azure uses different error codes).
+        """
+        from litellm.llms.base_llm.chat.transformation import BaseLLMException
+
+        return BaseLLMException(
+            status_code=status_code,
+            message=error_message,
+            headers=headers,
+        )

litellm/llms/custom_httpx/llm_http_handler.py

@@ -4735,6 +4735,151 @@ class BaseLLMHTTPHandler:
                         f"Unexpected error while closing WebSocket: {close_error}"
                     )
 
+    async def async_realtime_client_secret_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,
+    ) -> httpx.Response:
+        """
+        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.
+        """
+        if client is None or not isinstance(client, AsyncHTTPHandler):
+            async_httpx_client = get_async_httpx_client(
+                llm_provider=litellm.LlmProviders.OPENAI,
+            )
+        else:
+            async_httpx_client = client
+
+        if provider_config is not None:
+            url = provider_config.get_complete_url(api_base=api_base, model=model or "")
+            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"
+            headers = {
+                "Authorization": f"Bearer {api_key}",
+                "Content-Type": "application/json",
+                "OpenAI-Beta": "realtime=v1",
+            }
+
+        if extra_headers:
+            headers.update(extra_headers)
+
+        logging_obj.pre_call(
+            input=request_data,
+            api_key="",
+            additional_args={
+                "complete_input_dict": request_data,
+                "api_base": url,
+                "headers": headers,
+            },
+        )
+
+        try:
+            return await async_httpx_client.post(
+                url=url,
+                headers=headers,
+                json=request_data,
+                timeout=timeout,
+            )
+        except Exception as e:
+            raise self._handle_error(
+                e=e,
+                provider_config=provider_config,
+            )
+
+    async def async_realtime_calls_handler(
+        self,
+        api_base: str,
+        openai_ephemeral_key: str,
+        sdp_body: bytes,
+        logging_obj: LiteLLMLoggingObj,
+        timeout: Union[float, httpx.Timeout],
+        provider_config: Optional[Any] = None,
+        model: Optional[str] = None,
+        session_config: Optional[Dict[str, Any]] = None,
+        extra_headers: Optional[Dict[str, Any]] = None,
+        client: Optional[Union[HTTPHandler, AsyncHTTPHandler]] = None,
+    ) -> httpx.Response:
+        """
+        Forward POST /v1/realtime/calls (SDP exchange) to upstream provider.
+
+        Uses provider_config (BaseRealtimeHTTPConfig) for URL construction and
+        header auth when available; falls back to the legacy OpenAI-style defaults.
+
+        OpenAI's GA realtime API expects multipart/form-data with:
+          - sdp: the SDP offer (text)
+          - session: JSON string with {"type": "realtime", "model": "...", ...}
+        """
+        if client is None or not isinstance(client, AsyncHTTPHandler):
+            async_httpx_client = get_async_httpx_client(
+                llm_provider=litellm.LlmProviders.OPENAI,
+            )
+        else:
+            async_httpx_client = client
+
+        if provider_config is not None:
+            url = provider_config.get_realtime_calls_url(api_base=api_base, model=model or "")
+            headers: Dict[str, Any] = provider_config.get_realtime_calls_headers(
+                ephemeral_key=openai_ephemeral_key
+            )
+        else:
+            url = f"{api_base.rstrip('/')}/v1/realtime/calls"
+            headers = {
+                "Authorization": f"Bearer {openai_ephemeral_key}",
+            }
+
+        if extra_headers:
+            headers.update(extra_headers)
+
+        # Build multipart form data: sdp + session JSON
+        session_data = session_config or {}
+        if "type" not in session_data:
+            session_data["type"] = "realtime"
+        if "model" not in session_data and model:
+            session_data["model"] = model
+
+        sdp_text = sdp_body.decode("utf-8") if isinstance(sdp_body, bytes) else sdp_body
+
+        files = {
+            "sdp": (None, sdp_text, "text/plain"),
+            "session": (None, json.dumps(session_data), "application/json"),
+        }
+
+        logging_obj.pre_call(
+            input="realtime_sdp_offer",
+            api_key="",
+            additional_args={
+                "api_base": url,
+                "headers": headers,
+                "session": session_data,
+            },
+        )
+
+        try:
+            return await async_httpx_client.post(
+                url=url,
+                headers=headers,
+                files=files,
+                timeout=timeout,
+            )
+        except Exception as e:
+            raise self._handle_error(
+                e=e,
+                provider_config=provider_config,
+            )
+
     async def async_responses_websocket(
         self,
         model: str,

litellm/llms/openai/realtime/http_transformation.py

@@ -0,0 +1,50 @@
+"""OpenAI realtime HTTP transformation config (client_secrets + realtime_calls)."""
+
+from typing import Optional
+
+import litellm
+from litellm.llms.base_llm.realtime.http_transformation import BaseRealtimeHTTPConfig
+from litellm.secret_managers.main import get_secret_str
+
+
+class OpenAIRealtimeHTTPConfig(BaseRealtimeHTTPConfig):
+    def get_api_base(self, api_base: Optional[str], **kwargs) -> str:
+        return (
+            api_base
+            or litellm.api_base
+            or get_secret_str("OPENAI_API_BASE")
+            or "https://api.openai.com"
+        )
+
+    def get_api_key(self, api_key: Optional[str], **kwargs) -> str:
+        return (
+            api_key
+            or litellm.api_key
+            or litellm.openai_key
+            or get_secret_str("OPENAI_API_KEY")
+            or ""
+        )
+
+    def get_complete_url(self, api_base: Optional[str], model: str) -> str:
+        base = self.get_api_base(api_base).rstrip("/")
+        if base.endswith("/v1"):
+            base = base[:-3]
+        return f"{base}/v1/realtime/client_secrets"
+
+    def get_realtime_calls_url(self, api_base: Optional[str], model: str) -> str:
+        base = self.get_api_base(api_base).rstrip("/")
+        if base.endswith("/v1"):
+            base = base[:-3]
+        return f"{base}/v1/realtime/calls"
+
+    def validate_environment(
+        self,
+        headers: dict,
+        model: str,
+        api_key: Optional[str] = None,
+    ) -> dict:
+        return {
+            **headers,
+            "Authorization": f"Bearer {api_key or ''}",
+            "Content-Type": "application/json",
+        }

litellm/proxy/proxy_server.py

@@ -260,6 +260,7 @@ from litellm.proxy.anthropic_endpoints.claude_code_endpoints import (
     claude_code_marketplace_router,
 )
 from litellm.proxy.anthropic_endpoints.endpoints import router as anthropic_router
+from litellm.proxy.realtime_endpoints.endpoints import router as webrtc_router
 from litellm.proxy.anthropic_endpoints.skills_endpoints import (
     router as anthropic_skills_router,
 )
@@ -13169,6 +13170,7 @@ app.include_router(vector_store_management_router)
 app.include_router(vector_store_files_router)
 app.include_router(credential_router)
 app.include_router(llm_passthrough_router)
+app.include_router(webrtc_router)
 app.include_router(mcp_management_router)
 app.include_router(mcp_byok_oauth_router)
 app.include_router(anthropic_router)