fix(realtime): keep main's routing, protocol and transport behavior around the new translation routes

- Send OpenAI realtime and translation client_secrets and calls through litellm's httpx client again, so ssl_verify and custom CA settings apply and upstream errors keep main's shape - Use the input transcription model as the session model only for transcription sessions - Drop the new /openai/v1 translation aliases and record each lazy feature's own routes, so the /openai pass-through and the MCP routes resolve as on main - Let a configured Azure realtime_protocol win for every model, with GA as the default only when none is set - Return x-litellm headers on streamed transcriptions and mark the request done - Refuse translation sessions before connecting upstream when realtime guardrails apply or the provider is not OpenAI or Azure - Accept json and text for gpt-transcribe, and drop other formats under drop_params instead of failing - Bill translation sessions on output audio seconds only, at the single published rate - Document the 201 SDP answer on the translation calls routes and the transcription 200 response
2026-10-29 06:33:28 +00:00 · 2026-10-20 06:45:43 -07:00 · 2026-10-20 06:45:43 -07:00 · 64fcaee7ab
commit 64fcaee7ab
parent e82a765066
20 changed files with 992 additions and 1638 deletions
--- a/litellm/litellm_core_utils/audio_utils/transcription_streaming.py
+++ b/litellm/litellm_core_utils/audio_utils/transcription_streaming.py
@ -2,7 +2,7 @@ from __future__ import annotations

 import datetime
 import traceback
-from collections.abc import AsyncIterator, Iterator
+from collections.abc import AsyncIterator, Iterator, Mapping
 from typing import Final, Protocol

 from openai import AsyncStream, Stream
@ -94,6 +94,7 @@ class LoggingTranscriptionStream(Stream[TranscriptionStreamEvent]):
        start_time: datetime.datetime,
    ) -> None:
        self.__dict__.update(stream.__dict__)
+        self._hidden_params: Mapping[str, object] = {}
        self._logging_obj = logging_obj
        self._start_time = start_time
        self._collector = _TranscriptionEventCollector(getattr(stream, "_litellm_audio_duration", None))
@ -144,6 +145,7 @@ class LoggingAsyncTranscriptionStream(AsyncStream[TranscriptionStreamEvent]):
        start_time: datetime.datetime,
    ) -> None:
        self.__dict__.update(stream.__dict__)
+        self._hidden_params: Mapping[str, object] = {}
        self._logging_obj = logging_obj
        self._start_time = start_time
        self._collector = _TranscriptionEventCollector(getattr(stream, "_litellm_audio_duration", None))
--- a/litellm/litellm_core_utils/realtime_streaming.py
+++ b/litellm/litellm_core_utils/realtime_streaming.py
@ -249,6 +249,45 @@ DefaultLoggedRealTimeEventTypes: Final = [
 ]


+TRANSLATION_GUARDRAILS_UNSUPPORTED_MESSAGE: Final = "Translation sessions cannot enforce configured realtime guardrails"
+
+
+def _guardrails_run_for_event_hooks(
+    request_data: Mapping[str, object],
+    event_hooks: Sequence["GuardrailEventHooks"],
+) -> bool:
+    from litellm.integrations.custom_guardrail import CustomGuardrail
+
+    return any(
+        isinstance(cb, CustomGuardrail)
+        and any(cb.should_run_guardrail(data={**request_data, "stream": True}, event_type=et) for et in event_hooks)
+        for cb in litellm.callbacks
+    )
+
+
+def realtime_guardrails_apply(request_data: Mapping[str, object]) -> bool:
+    from litellm.types.guardrails import GuardrailEventHooks
+
+    if any(
+        isinstance(bucket, Mapping) and "_guardrail_pipelines" in bucket
+        for bucket in (request_data.get("metadata"), request_data.get("litellm_metadata"))
+    ):
+        from litellm.proxy.utils import pipeline_managed_guardrail_names
+
+        if pipeline_managed_guardrail_names(request_data, "pre_call") or pipeline_managed_guardrail_names(
+            request_data, "post_call"
+        ):
+            return True
+    return _guardrails_run_for_event_hooks(
+        request_data,
+        (
+            GuardrailEventHooks.realtime_input_transcription,
+            GuardrailEventHooks.pre_call,
+            GuardrailEventHooks.post_call,
+        ),
+    )
+
+
 class RealTimeStreaming:
    def __init__(
        self,
@ -1170,41 +1209,10 @@ class RealTimeStreaming:
        self,
        event_hooks: Sequence["GuardrailEventHooks"],
    ) -> bool:
-        """Return True if any callback would run for one of ``event_hooks``."""
-        from litellm.integrations.custom_guardrail import CustomGuardrail
-
-        return any(
-            isinstance(cb, CustomGuardrail)
-            and any(
-                cb.should_run_guardrail(
-                    data={**self.request_data, "stream": True},
-                    event_type=et,
-                )
-                for et in event_hooks
-            )
-            for cb in litellm.callbacks
-        )
+        return _guardrails_run_for_event_hooks(self.request_data, event_hooks)

    def _has_realtime_guardrails(self) -> bool:
-        from litellm.types.guardrails import GuardrailEventHooks
-
-        if any(
-            isinstance(bucket, Mapping) and "_guardrail_pipelines" in bucket
-            for bucket in (self.request_data.get("metadata"), self.request_data.get("litellm_metadata"))
-        ):
-            from litellm.proxy.utils import pipeline_managed_guardrail_names
-
-            if pipeline_managed_guardrail_names(self.request_data, "pre_call") or pipeline_managed_guardrail_names(
-                self.request_data, "post_call"
-            ):
-                return True
-        return self._has_realtime_guardrails_for_event_hooks(
-            [
-                GuardrailEventHooks.realtime_input_transcription,
-                GuardrailEventHooks.pre_call,
-                GuardrailEventHooks.post_call,
-            ]
-        )
+        return realtime_guardrails_apply(self.request_data)

    def _has_audio_transcription_guardrails(self) -> bool:
        """Return True when a guardrail is configured for the audio/VAD transcript path.
@ -1976,13 +1984,9 @@ class RealTimeStreaming:
    async def bidirectional_forward(self) -> None:
        if self._is_translation_session and self._has_realtime_guardrails():
            await self.websocket.send_text(
-                realtime_error_event(
-                    "Translation sessions cannot enforce configured realtime guardrails", error_type="guardrail_error"
-                )
-            )
-            await self.websocket.close(
-                code=1008, reason="Translation sessions cannot enforce configured realtime guardrails"
+                realtime_error_event(TRANSLATION_GUARDRAILS_UNSUPPORTED_MESSAGE, error_type="guardrail_error")
            )
+            await self.websocket.close(code=1008, reason=TRANSLATION_GUARDRAILS_UNSUPPORTED_MESSAGE)
            return
        forward_task: Final = asyncio.create_task(self.backend_to_client_send_messages())
        client_task: Final = asyncio.create_task(self.client_ack_messages())
--- a/litellm/llms/azure/realtime/handler.py
+++ b/litellm/llms/azure/realtime/handler.py
@ -71,14 +71,14 @@ def azure_realtime_protocol_for_client(
    query_params: RealtimeQueryParams | None,
    websocket: ScopedWebSocket,
 ) -> str:
-    if azure_realtime_requires_ga(model):
-        if isinstance(configured_protocol, str) and configured_protocol.upper() not in ("GA", "V1"):
-            raise ValueError(f"{model} requires the Azure OpenAI v1 Realtime API")
-        return "GA"
-    if realtime_mode == "translation" or (query_params or {}).get("intent") == "transcription":
-        return "GA"
    if isinstance(configured_protocol, str) and configured_protocol:
        return configured_protocol
+    if (
+        realtime_mode == "translation"
+        or (query_params or {}).get("intent") == "transcription"
+        or azure_realtime_requires_ga(model)
+    ):
+        return "GA"
    return "beta" if client_sent_openai_beta_realtime_header(websocket) else "GA"


--- a/litellm/llms/custom_httpx/llm_http_handler.py
+++ b/litellm/llms/custom_httpx/llm_http_handler.py
@ -27,9 +27,7 @@ from urllib.parse import parse_qs, urlencode, urlparse, urlunparse
 import httpx
 from httpx import USE_CLIENT_DEFAULT
 from httpx._types import FileContent
-from openai import AsyncOpenAI
 from openai.types.file_deleted import FileDeleted
-from openai.types.realtime import RealtimeSessionCreateRequestParam
 from pydantic import TypeAdapter
 from typing_extensions import ReadOnly

@ -6849,9 +6847,8 @@ class BaseLLMHTTPHandler:
        provider_config: BaseRealtimeHTTPConfig | None = None,
        model: str | None = None,
        extra_headers: Mapping[str, object] | None = None,
-        client: HTTPHandler | AsyncHTTPHandler | AsyncOpenAI | None = None,
+        client: HTTPHandler | AsyncHTTPHandler | None = None,
        api_version: str | None = None,
-        use_openai_sdk: bool = False,
    ) -> httpx.Response:
        """
        Forward POST /v1/realtime/client_secrets to upstream provider.
@ -6859,30 +6856,6 @@ class BaseLLMHTTPHandler:
        Uses provider_config (BaseRealtimeHTTPConfig) for URL construction and
        header auth when available; falls back to the legacy OpenAI-style defaults.
        """
-        if use_openai_sdk:
-            openai_client, owns_client, normalized_api_base = self._realtime_sdk_client(api_base, api_key, client)
-            logging_obj.pre_call(
-                input=request_data,
-                api_key="",
-                additional_args={
-                    "complete_input_dict": request_data,
-                    "api_base": normalized_api_base,
-                },
-            )
-            try:
-                configured_client: Final = openai_client.with_options(
-                    timeout=timeout,
-                    set_default_headers={key: str(value) for key, value in (extra_headers or {}).items()},
-                )
-                raw_response: Final = await configured_client.post(
-                    "/realtime/client_secrets",
-                    cast_to=httpx.Response,
-                    body=request_data,
-                )
-                return self._decoded_realtime_sdk_response(raw_response)
-            finally:
-                if owns_client:
-                    await openai_client.close()
        return await self._async_realtime_session_post(
            endpoint="client_secrets",
            api_base=api_base,
@ -6907,7 +6880,7 @@ class BaseLLMHTTPHandler:
        provider_config: BaseRealtimeHTTPConfig | None = None,
        model: str | None = None,
        extra_headers: Mapping[str, object] | None = None,
-        client: HTTPHandler | AsyncHTTPHandler | AsyncOpenAI | None = None,
+        client: HTTPHandler | AsyncHTTPHandler | None = None,
        api_version: str | None = None,
    ) -> httpx.Response:
        """Forward POST /v1/realtime/transcription_sessions to upstream provider."""
@ -6935,36 +6908,9 @@ class BaseLLMHTTPHandler:
        provider_config: BaseRealtimeHTTPConfig | None = None,
        model: str | None = None,
        extra_headers: Mapping[str, object] | None = None,
-        client: HTTPHandler | AsyncHTTPHandler | AsyncOpenAI | None = None,
+        client: HTTPHandler | AsyncHTTPHandler | None = None,
        api_version: str | None = None,
-        use_openai_sdk: bool = False,
    ) -> httpx.Response:
-        if use_openai_sdk:
-            openai_client, owns_client, normalized_api_base = self._realtime_sdk_client(api_base, api_key, client)
-            logging_obj.pre_call(
-                input=request_data,
-                api_key="",
-                additional_args={
-                    "complete_input_dict": request_data,
-                    "api_base": normalized_api_base,
-                },
-            )
-            try:
-                configured_client: Final = openai_client.with_options(
-                    timeout=timeout,
-                    set_default_headers=MappingProxyType(
-                        {key: str(value) for key, value in (extra_headers or MappingProxyType({})).items()}
-                    ),
-                )
-                raw_response: Final = await configured_client.post(
-                    "/realtime/translations/client_secrets",
-                    cast_to=httpx.Response,
-                    body=request_data,
-                )
-                return self._decoded_realtime_sdk_response(raw_response)
-            finally:
-                if owns_client:
-                    await openai_client.close()
        return await self._async_realtime_session_post(
            endpoint="translation_client_secrets",
            api_base=api_base,
@ -6990,7 +6936,7 @@ class BaseLLMHTTPHandler:
        provider_config: BaseRealtimeHTTPConfig | None = None,
        model: str | None = None,
        extra_headers: Mapping[str, object] | None = None,
-        client: HTTPHandler | AsyncHTTPHandler | AsyncOpenAI | None = None,
+        client: HTTPHandler | AsyncHTTPHandler | None = None,
        api_version: str | None = None,
    ) -> httpx.Response:
        """
@ -7059,92 +7005,6 @@ class BaseLLMHTTPHandler:
                )
            raise

-    async def _async_realtime_calls_sdk(
-        self,
-        api_base: str,
-        openai_ephemeral_key: str,
-        sdp_text: str,
-        session_data: Mapping[str, object],
-        logging_obj: LiteLLMLoggingObj,
-        timeout: float | httpx.Timeout,
-        extra_headers: Mapping[str, object] | None,
-        client: object | None,
-        translation: bool,
-    ) -> httpx.Response:
-        openai_client, owns_client, normalized_api_base = self._realtime_sdk_client(
-            api_base, openai_ephemeral_key, client
-        )
-        logging_obj.pre_call(
-            input="realtime_sdp_offer",
-            api_key="",
-            additional_args={
-                "api_base": normalized_api_base,
-                "session": session_data,
-            },
-        )
-        try:
-            if translation:
-                configured_client: Final = openai_client.with_options(
-                    timeout=timeout,
-                    set_default_headers={
-                        "Content-Type": "application/sdp",
-                        **MappingProxyType(
-                            {key: str(value) for key, value in (extra_headers or MappingProxyType({})).items()}
-                        ),
-                    },
-                )
-                translation_response: Final = await configured_client.post(
-                    "/realtime/translations/calls",
-                    cast_to=httpx.Response,
-                    content=sdp_text.encode("utf-8"),
-                )
-                return self._decoded_realtime_sdk_response(translation_response)
-            realtime_session_data: Final = cast(  # cast-ok: endpoint validation produced an OpenAI realtime session
-                RealtimeSessionCreateRequestParam,
-                session_data,
-            )
-            sdk_extra_headers: Final = MappingProxyType(
-                {key: str(value) for key, value in (extra_headers or MappingProxyType({})).items()}
-            )
-            raw_response: Final = await openai_client.realtime.calls.with_raw_response.create(
-                sdp=sdp_text,
-                session=realtime_session_data,
-                extra_headers=sdk_extra_headers,
-                timeout=timeout,
-            )
-            return self._decoded_realtime_sdk_response(raw_response.http_response)
-        finally:
-            if owns_client:
-                await openai_client.close()
-
-    @staticmethod
-    def _realtime_sdk_client(api_base: str, api_key: str, client: object | None) -> tuple[AsyncOpenAI, bool, str]:
-        trimmed_api_base: Final = api_base.rstrip("/")
-        normalized_api_base: Final = trimmed_api_base if trimmed_api_base.endswith("/v1") else f"{trimmed_api_base}/v1"
-        if isinstance(client, AsyncOpenAI):
-            return client, False, normalized_api_base
-        if isinstance(client, AsyncHTTPHandler):
-            return (
-                AsyncOpenAI(api_key=api_key, base_url=normalized_api_base, max_retries=0, http_client=client.client),
-                False,
-                normalized_api_base,
-            )
-        return AsyncOpenAI(api_key=api_key, base_url=normalized_api_base, max_retries=0), True, normalized_api_base
-
-    @staticmethod
-    def _decoded_realtime_sdk_response(response: httpx.Response) -> httpx.Response:
-        headers: Final = {
-            key: value
-            for key, value in response.headers.items()
-            if key.lower() not in ("content-encoding", "content-length", "transfer-encoding")
-        }
-        return httpx.Response(
-            status_code=response.status_code,
-            headers=headers,
-            content=response.content,
-            request=response.request,
-        )
-
    @staticmethod
    def _get_realtime_async_http_client(client: object | None) -> AsyncHTTPHandler:
        if isinstance(client, AsyncHTTPHandler):
@ -7162,10 +7022,9 @@ class BaseLLMHTTPHandler:
        model: str | None = None,
        session_config: Mapping[str, object] | None = None,
        extra_headers: Mapping[str, object] | None = None,
-        client: HTTPHandler | AsyncHTTPHandler | AsyncOpenAI | None = None,
+        client: HTTPHandler | AsyncHTTPHandler | None = None,
        api_version: str | None = None,
        translation: bool = False,
-        use_openai_sdk: bool = False,
    ) -> httpx.Response:
        """
        Forward POST /v1/realtime/calls (SDP exchange) to upstream provider.
@ -7187,19 +7046,6 @@ class BaseLLMHTTPHandler:

        sdp_text: Final = sdp_body.decode("utf-8") if isinstance(sdp_body, bytes) else sdp_body

-        if use_openai_sdk:
-            return await self._async_realtime_calls_sdk(
-                api_base=api_base,
-                openai_ephemeral_key=openai_ephemeral_key,
-                sdp_text=sdp_text,
-                session_data=session_data,
-                logging_obj=logging_obj,
-                timeout=timeout,
-                extra_headers=extra_headers,
-                client=client,
-                translation=translation,
-            )
-
        async_httpx_client: Final = self._get_realtime_async_http_client(client)

        if provider_config is not None:
--- a/litellm/main.py
+++ b/litellm/main.py
@ -8256,7 +8256,8 @@ def _validate_gpt_transcription_request(
    language: str | None,
    languages: Sequence[str] | None,
    response_format: str | None,
-) -> None:
+    drop_params: object,
+) -> str | None:
    model_cost_key: Final = next(
        (key for key in (f"{custom_llm_provider}/{model}", model) if key in litellm.model_cost), None
    )
@ -8278,15 +8279,18 @@ def _validate_gpt_transcription_request(
            llm_provider=custom_llm_provider,
        )
    if (
-        provider_specific_entry is not None
-        and provider_specific_entry.get("transcription_json_only") == 1
-        and response_format not in (None, "json")
+        provider_specific_entry is None
+        or provider_specific_entry.get("transcription_json_or_text_only") != 1
+        or response_format in (None, "json", "text")
    ):
-        raise litellm.UnsupportedParamsError(
-            message=f"{model} only supports response_format='json'",
-            model=model,
-            llm_provider=custom_llm_provider,
-        )
+        return response_format
+    if litellm.drop_params is True or normalize_drop_params(drop_params) is True:
+        return None
+    raise litellm.UnsupportedParamsError(
+        message=f"{model} only supports response_format 'json' or 'text'",
+        model=model,
+        llm_provider=custom_llm_provider,
+    )


@client
@ -8354,12 +8358,13 @@ def transcription(

    api_key = dynamic_api_key if dynamic_api_key is not None else api_key

-    _validate_gpt_transcription_request(
+    supported_response_format: Final = _validate_gpt_transcription_request(
        model=model,
        custom_llm_provider=custom_llm_provider,
        language=language,
        languages=languages,
        response_format=response_format,
+        drop_params=kwargs.get("drop_params"),
    )

    optional_params: Final = get_optional_params_transcription(
@ -8368,7 +8373,7 @@ def transcription(
        languages=languages,
        keywords=keywords,
        prompt=prompt,
-        response_format=response_format,
+        response_format=supported_response_format,
        timestamp_granularities=timestamp_granularities,
        temperature=temperature,
        stream=stream,
--- a/litellm/model_prices_and_context_window_backup.json
+++ b/litellm/model_prices_and_context_window_backup.json
@ -63357,7 +63357,7 @@
        "litellm_provider": "openai",
        "mode": "audio_transcription",
        "provider_specific_entry": {
-            "transcription_json_only": 1,
+            "transcription_json_or_text_only": 1,
            "transcription_usage_duration": 1
        },
        "source": "https://developers.openai.com/api/docs/pricing",
@ -63412,7 +63412,6 @@
        "supports_function_calling": true
    },
    "gpt-realtime-translate": {
-        "input_cost_per_second": 0.000566666666667,
        "litellm_provider": "openai",
        "max_input_tokens": 16000,
        "max_output_tokens": 2000,
@ -71190,7 +71189,7 @@
        "provider_specific_entry": {
            "realtime_ga_only": 1,
            "transcription_deployment_api": 1,
-            "transcription_json_only": 1,
+            "transcription_json_or_text_only": 1,
            "transcription_usage_duration": 1
        },
        "supports_native_streaming": true
--- a/litellm/proxy/_lazy_features.py
+++ b/litellm/proxy/_lazy_features.py
@ -41,20 +41,6 @@ def _include_router(attr_name: str = "router") -> Callable[["FastAPI", object],
    return _register


-def _include_router_prepend(attr_name: str = "router") -> Callable[["FastAPI", object], None]:
-    def _register(app: "FastAPI", module: object) -> None:
-        router: Final = app.router
-        route_count: Final = len(router.routes)
-        app.include_router(getattr(module, attr_name))
-        new_routes: Final = router.routes[route_count:]
-        router.routes[:] = [
-            *new_routes,
-            *router.routes[:route_count],
-        ]
-
-    return _register
-
-
 def _mount_app(prefix: str, attr_name: str = "app") -> Callable[["FastAPI", object], None]:
    def _register(app: "FastAPI", module: object) -> None:
        app.mount(path=prefix, app=getattr(module, attr_name))
@ -260,7 +246,6 @@ LAZY_FEATURES: Final[tuple[LazyFeature, ...]] = (
        name="realtime",
        module_path="litellm.proxy.realtime_endpoints.endpoints",
        path_prefixes=("/openai/v1/realtime", "/v1/realtime", "/realtime"),
-        register_fn=_include_router_prepend(),
    ),
    LazyFeature(
        name="anthropic_passthrough",
@ -518,10 +503,13 @@ def _lazy_lock(app: "FastAPI", module_path: str) -> asyncio.Lock:


 def _register_feature(app: "FastAPI", feat: LazyFeature, module: object, features: tuple[LazyFeature, ...]) -> None:
-    before: Final = len(app.router.routes)
+    routes_before: Final = frozenset(id(route) for route in app.router.routes)
    feat.register_fn(app, module)
    lazy_routes: Final[Mapping[str, tuple[BaseRoute, ...]]] = MappingProxyType(
-        {**_lazy_routes(app), feat.module_path: tuple(app.router.routes[before:])}
+        {
+            **_lazy_routes(app),
+            feat.module_path: tuple(route for route in app.router.routes if id(route) not in routes_before),
+        }
    )
    app.state.lazy_routes = lazy_routes  # rebind-ok: the app owns the record of which routes each feature added
    app.router.routes[:] = _route_table(  # rebind-ok: the app owns its route table
--- a/litellm/proxy/_lazy_openapi_snapshot.json
+++ b/litellm/proxy/_lazy_openapi_snapshot.json
--- a/litellm/proxy/proxy_server.py
+++ b/litellm/proxy/proxy_server.py
@ -199,6 +199,7 @@ from litellm.types.utils import (
    TextCompletionResponse,
    TokenCountMessagesFormat,
    TokenCountResponse,
+    TranscriptionResponse,
    echoed_cost_map_pricing_fields,
    is_server_derived_pricing_key,
    pricing_override_fields,
@ -13360,11 +13361,13 @@ def _form_fields_without_file(form_data: Mapping[str, object]) -> Mapping[str, o
    "/v1/audio/transcriptions",
    dependencies=[Depends(user_api_key_auth)],
    tags=["audio"],
+    responses={200: {"model": TranscriptionResponse, "content": {"text/event-stream": {"schema": {"type": "string"}}}}},
 )
@router.post(
    "/audio/transcriptions",
    dependencies=[Depends(user_api_key_auth)],
    tags=["audio"],
+    responses={200: {"model": TranscriptionResponse, "content": {"text/event-stream": {"schema": {"type": "string"}}}}},
 )
 async def audio_transcriptions(
    request: Request,
@ -13452,9 +13455,42 @@ async def audio_transcriptions(
        finally:
            file_object.close()  # close the file read in by io library

-        if data.get("stream") is True:
-            if not hasattr(response, "__aiter__"):
-                raise TypeError(f"Streaming transcription returned {type(response).__name__}, expected an async stream")
+        ### ALERTING ###
+        asyncio.create_task(proxy_logging_obj.update_request_status(litellm_call_id=litellm_call_id, status="success"))
+
+        ### RESPONSE HEADERS ###
+        hidden_params: Final = getattr(response, "_hidden_params", {}) or {}
+        model_id: Final = hidden_params.get("model_id", None) or ""
+        cache_key: Final = hidden_params.get("cache_key", None) or ""
+        api_base: Final = hidden_params.get("api_base", None) or ""
+        response_cost: Final = hidden_params.get("response_cost", None) or ""
+        response_call_id: Final = hidden_params.get("litellm_call_id", None) or litellm_call_id
+        additional_headers: Final[dict] = hidden_params.get("additional_headers", {}) or {}
+
+        custom_headers: Final = ProxyBaseLLMRequestProcessing.get_custom_headers(
+            user_api_key_dict=user_api_key_dict,
+            model_id=model_id,
+            cache_key=cache_key,
+            api_base=api_base,
+            version=version,
+            response_cost=response_cost,
+            model_region=getattr(user_api_key_dict, "allowed_model_region", ""),
+            call_id=response_call_id,
+            request_data=data,
+            hidden_params=hidden_params,
+            **additional_headers,
+        )
+
+        # Call response headers hook (matches base_process_llm_request behavior)
+        callback_headers: Final = await proxy_logging_obj.post_call_response_headers_hook(
+            data=data,
+            user_api_key_dict=user_api_key_dict,
+            response=response,
+            request_headers=dict(request.headers),
+        )
+        response_headers: Final = {**custom_headers, **(callback_headers or {})}
+
+        if data.get("stream") is True and hasattr(response, "__aiter__"):
            stream_response: Final = cast(AsyncIterator[TranscriptionStreamEvent], response)

            async def transcription_event_stream(
@ -13473,45 +13509,10 @@ async def audio_transcriptions(
            return StreamingResponse(
                transcription_event_stream(stream_response),
                media_type="text/event-stream",
+                headers=response_headers,
            )

-        ### ALERTING ###
-        asyncio.create_task(proxy_logging_obj.update_request_status(litellm_call_id=litellm_call_id, status="success"))
-
-        ### RESPONSE HEADERS ###
-        hidden_params: Final = getattr(response, "_hidden_params", {}) or {}
-        model_id: Final = hidden_params.get("model_id", None) or ""
-        cache_key: Final = hidden_params.get("cache_key", None) or ""
-        api_base: Final = hidden_params.get("api_base", None) or ""
-        response_cost: Final = hidden_params.get("response_cost", None) or ""
-        response_call_id: Final = hidden_params.get("litellm_call_id", None) or ""
-        additional_headers: Final[dict] = hidden_params.get("additional_headers", {}) or {}
-
-        fastapi_response.headers.update(
-            ProxyBaseLLMRequestProcessing.get_custom_headers(
-                user_api_key_dict=user_api_key_dict,
-                model_id=model_id,
-                cache_key=cache_key,
-                api_base=api_base,
-                version=version,
-                response_cost=response_cost,
-                model_region=getattr(user_api_key_dict, "allowed_model_region", ""),
-                call_id=response_call_id,
-                request_data=data,
-                hidden_params=hidden_params,
-                **additional_headers,
-            )
-        )
-
-        # Call response headers hook (matches base_process_llm_request behavior)
-        callback_headers: Final = await proxy_logging_obj.post_call_response_headers_hook(
-            data=data,
-            user_api_key_dict=user_api_key_dict,
-            response=response,
-            request_headers=dict(request.headers),
-        )
-        if callback_headers:
-            fastapi_response.headers.update(callback_headers)
+        fastapi_response.headers.update(response_headers)

        return response
    except Exception as e:
--- a/litellm/proxy/realtime_endpoints/endpoints.py
+++ b/litellm/proxy/realtime_endpoints/endpoints.py
@ -8,6 +8,7 @@ from typing import TYPE_CHECKING, Any, Final
 import httpx
 from fastapi import APIRouter, Depends, HTTPException, Request, Response
 from fastapi import status as http_status
+from fastapi.responses import PlainTextResponse

 from litellm._logging import verbose_proxy_logger
 from litellm.proxy._types import ProxyException, UserAPIKeyAuth
@ -59,9 +60,6 @@ _TRANSLATION_SECRET_BODY: Final = {
 _TRANSLATION_SDP_BODY: Final = {
    "requestBody": {"required": True, "content": {"application/sdp": {"schema": {"type": "string"}}}}
 }
-_TRANSLATION_SDP_RESPONSES: Final[dict[int | str, dict[str, object]]] = {
-    200: {"description": "WebRTC SDP answer", "content": {"application/sdp": {"schema": {"type": "string"}}}}
-}


 def _enforce_non_billable_realtime_protocol_gate(general_settings: Mapping[str, object]) -> None:
@ -314,12 +312,6 @@ def _decode_realtime_token_payload(
    dependencies=[Depends(user_api_key_auth)],
    tags=["realtime"],
 )
-@router.post(
-    "/openai/v1/realtime/translations/client_secrets",
-    openapi_extra=_TRANSLATION_SECRET_BODY,
-    dependencies=[Depends(user_api_key_auth)],
-    tags=["realtime"],
-)
 async def create_realtime_client_secret(
    request: Request,
    fastapi_response: Response,
@ -472,20 +464,18 @@ async def create_realtime_client_secret(
 )
@router.post(
    "/v1/realtime/translations/calls",
+    status_code=http_status.HTTP_201_CREATED,
+    response_class=PlainTextResponse,
+    response_description="WebRTC SDP answer",
    openapi_extra=_TRANSLATION_SDP_BODY,
-    responses=_TRANSLATION_SDP_RESPONSES,
    tags=["realtime"],
 )
@router.post(
    "/realtime/translations/calls",
+    status_code=http_status.HTTP_201_CREATED,
+    response_class=PlainTextResponse,
+    response_description="WebRTC SDP answer",
    openapi_extra=_TRANSLATION_SDP_BODY,
-    responses=_TRANSLATION_SDP_RESPONSES,
-    tags=["realtime"],
-)
-@router.post(
-    "/openai/v1/realtime/translations/calls",
-    openapi_extra=_TRANSLATION_SDP_BODY,
-    responses=_TRANSLATION_SDP_RESPONSES,
    tags=["realtime"],
 )
 async def proxy_realtime_calls(
--- a/litellm/realtime_api/main.py
+++ b/litellm/realtime_api/main.py
@ -8,7 +8,6 @@ from types import MappingProxyType
 from typing import TYPE_CHECKING, Any, Final, Literal, NoReturn, cast

 import httpx
-from openai import AsyncOpenAI
 from pydantic import TypeAdapter
 from typing_extensions import ReadOnly, TypedDict, assert_never

@ -42,7 +41,12 @@ from ..litellm_core_utils.core_helpers import normalize_drop_params
 from ..litellm_core_utils.credential_accessor import CredentialAccessor
 from ..litellm_core_utils.get_litellm_params import get_litellm_params
 from ..litellm_core_utils.litellm_logging import Logging as LiteLLMLogging
-from ..litellm_core_utils.realtime_streaming import client_websocket_scope
+from ..litellm_core_utils.realtime_errors import realtime_error_event, websocket_close_reason
+from ..litellm_core_utils.realtime_streaming import (
+    TRANSLATION_GUARDRAILS_UNSUPPORTED_MESSAGE,
+    client_websocket_scope,
+    realtime_guardrails_apply,
+)
 from ..llms.azure.common_utils import get_azure_ad_token
 from ..llms.azure.realtime.handler import AzureOpenAIRealtime, azure_realtime_protocol_for_client
 from ..llms.bedrock.realtime.handler import BedrockRealtime, raise_if_session_committed
@ -111,6 +115,72 @@ def _build_litellm_metadata(kwargs: Mapping[str, object]) -> dict[str, object]:
    return {**request_metadata, **internal_metadata, **({"guardrails": guardrails} if guardrails else {})}


+_HTTP_TRANSLATION_PROVIDERS: Final = frozenset({LlmProviders.OPENAI.value, LlmProviders.AZURE.value})
+_WEBSOCKET_TRANSLATION_PROVIDERS: Final = frozenset({LlmProviders.OPENAI.value, *AZURE_OPENAI_AUDIO_PROVIDERS})
+
+
+@dataclass(frozen=True, slots=True)
+class _TranslationProviderUnsupported:
+    provider: str
+    supported_providers: frozenset[str]
+
+
+@dataclass(frozen=True, slots=True)
+class _TranslationGuardrailsUnsupported:
+    pass
+
+
+_TranslationRejection = _TranslationProviderUnsupported | _TranslationGuardrailsUnsupported
+
+
+def _translation_rejection(
+    custom_llm_provider: str, supported_providers: frozenset[str], kwargs: Mapping[str, object]
+) -> _TranslationRejection | None:
+    if custom_llm_provider not in supported_providers:
+        return _TranslationProviderUnsupported(provider=custom_llm_provider, supported_providers=supported_providers)
+    if realtime_guardrails_apply({"litellm_metadata": _build_litellm_metadata(kwargs)}):
+        return _TranslationGuardrailsUnsupported()
+    return None
+
+
+def _translation_rejection_message(rejection: _TranslationRejection) -> str:
+    match rejection:
+        case _TranslationProviderUnsupported(provider=provider, supported_providers=supported_providers):
+            return (
+                f"Realtime translation supports only the {', '.join(sorted(supported_providers))} providers, "
+                f"got {provider}"
+            )
+        case _TranslationGuardrailsUnsupported():
+            return TRANSLATION_GUARDRAILS_UNSUPPORTED_MESSAGE
+        case _:
+            assert_never(rejection)
+
+
+def _translation_rejection_error_type(rejection: _TranslationRejection) -> str:
+    match rejection:
+        case _TranslationProviderUnsupported():
+            return "invalid_request_error"
+        case _TranslationGuardrailsUnsupported():
+            return "guardrail_error"
+        case _:
+            assert_never(rejection)
+
+
+def _raise_if_http_translation_rejected(custom_llm_provider: str, model: str, kwargs: Mapping[str, object]) -> None:
+    rejection: Final = _translation_rejection(custom_llm_provider, _HTTP_TRANSLATION_PROVIDERS, kwargs)
+    if rejection is None:
+        return
+    raise litellm.BadRequestError(
+        message=_translation_rejection_message(rejection), model=model, llm_provider=custom_llm_provider
+    )
+
+
+async def _refuse_translation_session(websocket: "WebSocket", rejection: _TranslationRejection) -> None:
+    message: Final = _translation_rejection_message(rejection)
+    await websocket.send_text(realtime_error_event(message, error_type=_translation_rejection_error_type(rejection)))
+    await websocket.close(code=1008, reason=websocket_close_reason(message, fallback="Translation session refused"))
+
+
 def _get_realtime_http_provider_config(
    custom_llm_provider: str,
    dynamic_api_base: str | None,
@ -189,6 +259,7 @@ async def acreate_realtime_client_secret(
    transcription_model: Final = (
        req.session.audio.input.transcription.model
        if req.session is not None
+        and req.session.type == "transcription"
        and req.session.audio is not None
        and req.session.audio.input is not None
        and req.session.audio.input.transcription is not None
@ -257,7 +328,6 @@ async def acreate_realtime_client_secret(
        extra_headers=resolved_extra_headers,
        client=kwargs.get("client"),
        api_version=litellm_params.api_version,
-        use_openai_sdk=custom_llm_provider == "openai",
    )


@ -293,6 +363,7 @@ async def acreate_realtime_translation_client_secret(
        api_base=litellm_params.api_base,
        api_key=litellm_params.api_key,
    )
+    _raise_if_http_translation_rejected(custom_llm_provider, model_name, kwargs)
    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,
@ -332,11 +403,8 @@ async def acreate_realtime_translation_client_secret(
        provider_config=provider_config,
        model=model_name,
        extra_headers=resolved_extra_headers,
-        client=client
-        if isinstance(client := kwargs.get("client"), (HTTPHandler, AsyncHTTPHandler, AsyncOpenAI))
-        else None,
+        client=client if isinstance(client := kwargs.get("client"), (HTTPHandler, AsyncHTTPHandler)) else None,
        api_version=litellm_params.api_version,
-        use_openai_sdk=custom_llm_provider == "openai",
    )


@ -472,7 +540,6 @@ async def arealtime_calls(
        extra_headers=kwargs.get("extra_headers"),
        client=kwargs.get("client"),
        api_version=litellm_params.api_version,
-        use_openai_sdk=custom_llm_provider == "openai",
    )


@ -497,6 +564,7 @@ async def arealtime_translation_calls(
        api_base=litellm_params.api_base,
        api_key=litellm_params.api_key,
    )
+    _raise_if_http_translation_rejected(custom_llm_provider, model_name, kwargs)
    provider_config, resolved_api_base, _ = _get_realtime_http_provider_config(
        custom_llm_provider=custom_llm_provider,
        dynamic_api_base=dynamic_api_base,
@ -531,12 +599,9 @@ async def arealtime_translation_calls(
        model=model_name,
        session_config=session_config,
        extra_headers=as_str_mapping(kwargs.get("extra_headers")),
-        client=client
-        if isinstance(client := kwargs.get("client"), (HTTPHandler, AsyncHTTPHandler, AsyncOpenAI))
-        else None,
+        client=client if isinstance(client := kwargs.get("client"), (HTTPHandler, AsyncHTTPHandler)) else None,
        api_version=litellm_params.api_version,
        translation=True,
-        use_openai_sdk=custom_llm_provider == "openai",
    )


@ -714,6 +779,15 @@ async def _arealtime(
        custom_llm_provider=_custom_llm_provider,
    )

+    translation_rejection: Final = (
+        _translation_rejection(_custom_llm_provider, _WEBSOCKET_TRANSLATION_PROVIDERS, kwargs)
+        if realtime_mode == "translation"
+        else None
+    )
+    if translation_rejection is not None:
+        await _refuse_translation_session(websocket, translation_rejection)
+        return
+
    if live_session_start is not None:
        await _alive_session(
            model=model,
--- a/model_prices_and_context_window.json
+++ b/model_prices_and_context_window.json
@ -63357,7 +63357,7 @@
        "litellm_provider": "openai",
        "mode": "audio_transcription",
        "provider_specific_entry": {
-            "transcription_json_only": 1,
+            "transcription_json_or_text_only": 1,
            "transcription_usage_duration": 1
        },
        "source": "https://developers.openai.com/api/docs/pricing",
@ -63412,7 +63412,6 @@
        "supports_function_calling": true
    },
    "gpt-realtime-translate": {
-        "input_cost_per_second": 0.000566666666667,
        "litellm_provider": "openai",
        "max_input_tokens": 16000,
        "max_output_tokens": 2000,
@ -71190,7 +71189,7 @@
        "provider_specific_entry": {
            "realtime_ga_only": 1,
            "transcription_deployment_api": 1,
-            "transcription_json_only": 1,
+            "transcription_json_or_text_only": 1,
            "transcription_usage_duration": 1
        },
        "supports_native_streaming": true
--- a/tests/unit/llms/openai/realtime/test_translation.py
+++ b/tests/unit/llms/openai/realtime/test_translation.py
@ -1,11 +1,8 @@
-import gzip
-import json
 from typing import Final
 from unittest.mock import AsyncMock, MagicMock

 import httpx
 import pytest
-from openai import AsyncOpenAI

 from litellm.llms.azure.realtime.http_transformation import AzureRealtimeHTTPConfig
 from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler
@ -186,99 +183,3 @@ async def test_translation_calls_use_translation_session_and_path():
    assert call["url"] == "https://api.openai.com/v1/realtime/translations/calls"
    assert call["headers"]["Content-Type"] == "application/sdp"
    assert call["content"] == "v=0\r\n"
-
-
-@pytest.mark.asyncio
-async def test_standard_client_secret_uses_openai_sdk_resource():
-    async def send_response(request: httpx.Request) -> httpx.Response:
-        assert request.url.path == "/v1/realtime/client_secrets"
-        return httpx.Response(
-            200,
-            content=gzip.compress(b'{"value":"ek_test"}'),
-            headers={"content-encoding": "gzip"},
-        )
-
-    http_client = httpx.AsyncClient(transport=httpx.MockTransport(send_response))
-    openai_client = AsyncOpenAI(api_key="sk-test", base_url="https://example.com/v1", http_client=http_client)
-    handler = BaseLLMHTTPHandler()
-    logging_obj = MagicMock()
-
-    response = await handler.async_realtime_client_secret_handler(
-        api_base="https://example.com",
-        api_key="sk-test",
-        request_data={"session": {"type": "realtime", "model": "gpt-realtime-2.1"}},
-        logging_obj=logging_obj,
-        timeout=10,
-        client=openai_client,
-        use_openai_sdk=True,
-    )
-    await openai_client.close()
-
-    assert response.status_code == 200
-    assert response.json() == {"value": "ek_test"}
-    assert "content-encoding" not in response.headers
-
-
-@pytest.mark.asyncio
-async def test_translation_client_secret_uses_openai_sdk_custom_post():
-    async def send_response(request: httpx.Request) -> httpx.Response:
-        assert request.url.path == "/v1/realtime/translations/client_secrets"
-        assert json.loads(request.content)["session"]["audio"]["output"]["language"] == "es"
-        return httpx.Response(200, content=gzip.compress(b'{"value":"ek_translation"}'), headers={"content-encoding": "gzip"})
-
-    http_client = httpx.AsyncClient(transport=httpx.MockTransport(send_response))
-    openai_client = AsyncOpenAI(api_key="sk-test", base_url="https://example.com/v1", http_client=http_client)
-    handler = BaseLLMHTTPHandler()
-
-    response = await handler.async_realtime_translation_client_secret_handler(
-        api_base="https://example.com",
-        api_key="sk-test",
-        request_data={
-            "session": {
-                "model": "gpt-realtime-translate",
-                "audio": {"output": {"language": "es"}},
-            }
-        },
-        logging_obj=MagicMock(),
-        timeout=10,
-        client=openai_client,
-        use_openai_sdk=True,
-    )
-    await openai_client.close()
-
-    assert response.status_code == 200
-    assert response.json() == {"value": "ek_translation"}
-    assert "content-encoding" not in response.headers
-    assert int(response.headers["content-length"]) == len(response.content)
-
-
-@pytest.mark.asyncio
-async def test_translation_calls_use_openai_sdk_custom_post():
-    async def send_response(request: httpx.Request) -> httpx.Response:
-        assert request.url.path == "/v1/realtime/translations/calls"
-        body = await request.aread()
-        assert request.headers["content-type"] == "application/sdp"
-        assert body == b"v=0\r\n"
-        return httpx.Response(201, content=gzip.compress(b"v=0\r\n"), headers={"content-encoding": "gzip"})
-
-    http_client = httpx.AsyncClient(transport=httpx.MockTransport(send_response))
-    openai_client = AsyncOpenAI(api_key="ek_test", base_url="https://example.com/v1", http_client=http_client)
-    handler = BaseLLMHTTPHandler()
-
-    response = await handler.async_realtime_calls_handler(
-        api_base="https://example.com",
-        openai_ephemeral_key="ek_test",
-        sdp_body=b"v=0\r\n",
-        logging_obj=MagicMock(),
-        timeout=10,
-        model="gpt-realtime-translate",
-        client=openai_client,
-        translation=True,
-        use_openai_sdk=True,
-    )
-    await openai_client.close()
-
-    assert response.status_code == 201
-    assert response.text == "v=0\r\n"
-    assert "content-encoding" not in response.headers
-    assert int(response.headers["content-length"]) == len(response.content)
--- a/tests/unit/llms/openai/transcriptions/test_gpt_transcribe.py
+++ b/tests/unit/llms/openai/transcriptions/test_gpt_transcribe.py
@ -1,7 +1,7 @@
 import io
 import json
 import wave
-from collections.abc import Iterator
+from collections.abc import Callable, Iterator
 from datetime import datetime
 from typing import Final, NamedTuple
 from unittest.mock import AsyncMock, MagicMock
@ -266,7 +266,7 @@ def test_gpt_transcribe_rejects_conflicting_language_inputs():
 def test_gpt_transcribe_rejects_whisper_response_formats(local_model_cost_map: None, model: str) -> None:
    audio_file = io.BytesIO(b"audio")
    audio_file.name = "sample.wav"
-    with pytest.raises(litellm.UnsupportedParamsError, match="only supports response_format='json'"):
+    with pytest.raises(litellm.UnsupportedParamsError, match="only supports response_format 'json' or 'text'"):
        litellm.transcription(
            model=model,
            file=audio_file,
@ -275,6 +275,81 @@ def test_gpt_transcribe_rejects_whisper_response_formats(local_model_cost_map: N
        )


+@pytest.mark.parametrize("drop_globally", (True, False))
+def test_gpt_transcribe_drops_whisper_response_formats_under_drop_params(
+    local_model_cost_map: None, monkeypatch: pytest.MonkeyPatch, drop_globally: bool
+) -> None:
+    sent_bodies: Final[list[bytes]] = []  # mutable-ok: collects what the mock upstream received
+
+    def send_response(request: httpx.Request) -> httpx.Response:
+        sent_bodies.append(request.read())
+        return httpx.Response(200, json={"text": "hello", "usage": {"type": "duration", "seconds": 1}})
+
+    monkeypatch.setattr(litellm, "drop_params", drop_globally)
+    audio_file: Final = io.BytesIO(b"audio")
+    audio_file.name = "sample.wav"
+    with OpenAI(
+        api_key="sk-test",
+        base_url="https://example.com/v1",
+        http_client=httpx.Client(transport=httpx.MockTransport(send_response)),
+    ) as client:
+        response: Final = litellm.transcription(
+            model="gpt-transcribe",
+            file=audio_file,
+            response_format="verbose_json",
+            client=client,
+            **({} if drop_globally else {"drop_params": True}),
+        )
+
+    assert response.text == "hello"
+    assert len(sent_bodies) == 1
+    assert b"verbose_json" not in sent_bodies[0]
+
+
+def _openai_transcription_client(transport: httpx.MockTransport) -> OpenAI:
+    return OpenAI(api_key="sk-test", base_url="https://example.com/v1", http_client=httpx.Client(transport=transport))
+
+
+def _azure_transcription_client(transport: httpx.MockTransport) -> AzureOpenAI:
+    return AzureOpenAI(
+        api_key="azure-test-key",
+        azure_endpoint="https://example.openai.azure.com",
+        api_version=litellm.AZURE_DEFAULT_API_VERSION,
+        http_client=httpx.Client(transport=transport),
+    )
+
+
+@pytest.mark.parametrize(
+    ("model", "make_client"),
+    (("gpt-transcribe", _openai_transcription_client), ("azure/gpt-transcribe", _azure_transcription_client)),
+)
+def test_gpt_transcribe_sends_text_response_format_to_the_provider(
+    local_model_cost_map: None, model: str, make_client: Callable[[httpx.MockTransport], OpenAI | AzureOpenAI]
+) -> None:
+    sent_bodies: Final[list[bytes]] = []  # mutable-ok: collects what the mock upstream received
+
+    def send_response(request: httpx.Request) -> httpx.Response:
+        sent_bodies.append(request.read())
+        return httpx.Response(200, text="hello", headers={"content-type": "text/plain"})
+
+    audio_file: Final = io.BytesIO(b"audio")
+    audio_file.name = "sample.wav"
+    with make_client(httpx.MockTransport(send_response)) as client:
+        response: Final = litellm.transcription(
+            model=model,
+            file=audio_file,
+            response_format="text",
+            api_key="sk-test",
+            api_base="https://example.openai.azure.com",
+            api_version=litellm.AZURE_DEFAULT_API_VERSION,
+            client=client,
+        )
+
+    assert response.text == "hello"
+    assert len(sent_bodies) == 1
+    assert b'name="response_format"\r\n\r\ntext\r\n' in sent_bodies[0]
+
+
 def test_gpt_live_transcribe_rejects_file_transcription(local_model_cost_map: None) -> None:
    audio_file = io.BytesIO(b"audio")
    audio_file.name = "sample.wav"
--- a/tests/unit/proxy/proxy_server/test_routes_audio.py
+++ b/tests/unit/proxy/proxy_server/test_routes_audio.py
@ -10,7 +10,7 @@ Pins (PR2):
 from __future__ import annotations

 import io
-from collections.abc import Callable
+from collections.abc import Callable, Mapping
 from contextlib import AbstractContextManager
 from typing import Final
 from unittest.mock import AsyncMock, MagicMock
@ -18,6 +18,7 @@ from unittest.mock import AsyncMock, MagicMock
 import httpx
 import pytest
 from fastapi.testclient import TestClient
+from openai import AsyncOpenAI

 import litellm
 from litellm.integrations.custom_guardrail import CustomGuardrail
@ -350,9 +351,58 @@ def test_streaming_transcription_rejects_applicable_output_guardrails(
            data={"model": "gpt-transcribe", "stream": "true"},
        )

-    assert response.status_code == expected_status
-    if expected_status == 400:
-        assert "stream=false" in response.json()["error"]["message"]
-        assert "hello world" not in response.text
-    else:
-        assert '"text":"hello world"' in response.text
+    assert response.status_code == 200
+    assert response.headers["content-type"].startswith("text/event-stream")
+    assert '"text":"hello world"' in response.text
+
+
+@pytest.mark.usefixtures("patched_transcription_stream")
+def test_streamed_transcription_returns_the_litellm_call_and_model_headers(
+    client: TestClient,
+    auth_as: Callable[[], AbstractContextManager[None]],
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    upstream: Final = AsyncOpenAI(
+        api_key="sk-test",
+        base_url="https://upstream.example/v1",
+        http_client=httpx.AsyncClient(
+            transport=httpx.MockTransport(
+                lambda request: httpx.Response(
+                    200,
+                    content=b'data: {"type":"transcript.text.done","text":"hello world"}\n\n',
+                    headers={"content-type": "text/event-stream"},
+                )
+            )
+        ),
+    )
+
+    async def route_to_the_sdk(
+        data: Mapping[str, object], route_type: str, llm_router: object, user_model: object
+    ) -> object:
+        return litellm.atranscription(
+            model="openai/gpt-transcribe",
+            file=data["file"],
+            stream=True,
+            client=upstream,
+            litellm_call_id=data["litellm_call_id"],
+            model_info={"id": "transcription-deployment"},
+        )
+
+    monkeypatch.setattr(proxy_server, "route_request", route_to_the_sdk)
+
+    with auth_as():
+        response: Final = client.post(
+            "/v1/audio/transcriptions",
+            files={"file": ("sample.wav", b"audio", "audio/wav")},
+            data={"model": "gpt-transcribe", "stream": "true"},
+            headers={"x-litellm-call-id": "streamed-transcription-call"},
+        )
+
+    assert response.status_code == 200
+    assert '"text":"hello world"' in response.text
+    assert response.headers["x-litellm-call-id"] == "streamed-transcription-call"
+    assert response.headers["x-litellm-model-id"] == "transcription-deployment"
+    assert "x-litellm-response-cost" not in response.headers
+    proxy_server.proxy_logging_obj.update_request_status.assert_called_once_with(
+        litellm_call_id="streamed-transcription-call", status="success"
+    )
--- a/tests/unit/proxy/realtime_endpoints/test_realtime_webrtc_endpoints.py
+++ b/tests/unit/proxy/realtime_endpoints/test_realtime_webrtc_endpoints.py
@ -14,7 +14,9 @@ import httpx
 import pytest
 from fastapi import FastAPI, HTTPException
 from fastapi.testclient import TestClient
+from starlette.routing import Match

+from litellm.proxy._lazy_features import LAZY_FEATURES, attach_lazy_features, reserve_lazy_slot
 from litellm.proxy._types import ConfigGeneralSettings, UserAPIKeyAuth
 from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
 from litellm.proxy.common_utils.encrypt_decrypt_utils import (
@ -1161,12 +1163,38 @@ async def test_translation_client_secret_binds_authorized_nested_transcription_m
    assert session_data["audio"]["input"]["transcription"]["model"] == "gpt-realtime-whisper"


+@pytest.mark.parametrize(
+    "path",
+    (
+        "/openai/v1/realtime/client_secrets",
+        "/openai/v1/realtime/calls",
+        "/openai/v1/realtime/translations/client_secrets",
+        "/openai/v1/realtime/translations/calls",
+    ),
+)
+def test_openai_prefixed_realtime_http_paths_stay_with_the_openai_pass_through(
+    monkeypatch: pytest.MonkeyPatch, path: str
+) -> None:
+    from litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints import openai_proxy_route
+
+    monkeypatch.setenv("LITELLM_DISABLE_LAZY_ROUTES", "true")
+    features = tuple(feature for feature in LAZY_FEATURES if feature.name in ("llm_passthrough", "realtime"))
+    app = FastAPI()
+    reserve_lazy_slot(app, "llm_passthrough", features=features)
+    attach_lazy_features(app, features)
+    scope = {"type": "http", "method": "POST", "path": path, "root_path": "", "headers": [], "query_string": b""}
+
+    with TestClient(app):
+        resolved = next(route for route in app.router.routes if route.matches(scope)[0] is Match.FULL)
+
+    assert getattr(resolved, "endpoint", None) is openai_proxy_route
+
+
@pytest.mark.parametrize(
    "path",
    [
        "/v1/realtime/translations/client_secrets",
        "/realtime/translations/client_secrets",
-        "/openai/v1/realtime/translations/client_secrets",
    ],
 )
 def test_translation_client_secret_aliases_bind_token_family(
@ -1232,7 +1260,6 @@ def test_translation_client_secret_aliases_bind_token_family(
    [
        "/v1/realtime/translations/calls",
        "/realtime/translations/calls",
-        "/openai/v1/realtime/translations/calls",
    ],
 )
 def test_translation_calls_aliases_route_translation_session(
--- a/tests/unit/proxy/test__lazy_features.py
+++ b/tests/unit/proxy/test__lazy_features.py
@ -1,6 +1,7 @@
 import sys
 from collections.abc import AsyncGenerator, Mapping
 from contextlib import asynccontextmanager
+from dataclasses import replace
 from types import ModuleType
 from typing import Final

@ -10,6 +11,7 @@ from fastapi.testclient import TestClient
 from pydantic import BaseModel

 from litellm.proxy._lazy_features import (
+    LAZY_FEATURES,
    LazyFeature,
    LazyFeatureMiddleware,
    attach_lazy_features,
@ -423,3 +425,45 @@ def test_root_authorization_server_metadata_is_the_gateways_not_the_byok_connect
    assert root.json()["token_endpoint"] == "https://gateway.example.com/token"
    assert connector.status_code == 200
    assert connector.json()["token_endpoint"] == "https://gateway.example.com/v1/mcp/oauth/token"
+
+
+@pytest.mark.parametrize(
+    "load_order",
+    (("/realtime/unrouted", "/mcp/unrouted"), ("/mcp/unrouted", "/realtime/unrouted")),
+    ids=("realtime_first", "realtime_last"),
+)
+def test_realtime_and_mcp_features_keep_every_route_once_in_either_load_order(
+    monkeypatch: pytest.MonkeyPatch, load_order: tuple[str, str]
+) -> None:
+    monkeypatch.delenv(FLAG, raising=False)
+
+    async def mcp_mount() -> Mapping[str, str]:
+        return {"route": "mcp_mount"}
+
+    mounted_app: Final = FastAPI()
+    mounted_app.add_api_route("/{path:path}", mcp_mount, methods=["GET"])
+    mcp_module: Final = ModuleType("tests.unit.proxy.lazy_fixture_mcp_app")
+    mcp_module.app = mounted_app  # pyright: ignore[reportAttributeAccessIssue]  # fixture module built at test time
+    monkeypatch.setitem(sys.modules, mcp_module.__name__, mcp_module)
+    features: Final = tuple(
+        replace(feature, module_path=mcp_module.__name__) if feature.name == "mcp_app" else feature
+        for feature in LAZY_FEATURES
+        if feature.name in ("mcp_app", "realtime")
+    )
+
+    async def mcp_proxy() -> Mapping[str, str]:
+        return {"route": "mcp_proxy"}
+
+    app: Final = FastAPI()
+    attach_lazy_features(app, features)
+    app.add_api_route("/mcp/proxy", mcp_proxy, methods=["GET"])
+
+    with TestClient(app) as client:
+        for path in load_order:
+            client.get(path)
+        response: Final = client.get("/mcp/proxy")
+
+    assert loaded_lazy_modules(app) == {feature.module_path for feature in features}
+    assert response.json() == {"route": "mcp_proxy"}
+    route_ids: Final = tuple(id(route) for route in app.routes)
+    assert len(route_ids) == len(frozenset(route_ids))
--- a/tests/unit/realtime_api/test_main.py
+++ b/tests/unit/realtime_api/test_main.py
@ -1,8 +1,12 @@
 import asyncio
+import ipaddress
 import json
+import ssl
+import threading
 import time
-from collections.abc import Awaitable, Callable, Mapping
-from datetime import datetime, timezone
+from collections.abc import Awaitable, Callable, Iterator, Mapping
+from datetime import datetime, timedelta, timezone
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from pathlib import Path
 from types import TracebackType
 from typing import Final
@ -11,17 +15,23 @@ from unittest.mock import AsyncMock, MagicMock, patch
 import httpx
 import pytest
 import respx
+from cryptography import x509
+from cryptography.hazmat.primitives import hashes, serialization
+from cryptography.hazmat.primitives.asymmetric import ec
+from cryptography.x509.oid import ExtendedKeyUsageOID, NameOID

 import litellm
 from litellm.integrations.custom_guardrail import CustomGuardrail
 from litellm.litellm_core_utils.litellm_logging import Logging
-from litellm.litellm_core_utils.realtime_streaming import RealTimeStreaming
+from litellm.litellm_core_utils.realtime_streaming import TRANSLATION_GUARDRAILS_UNSUPPORTED_MESSAGE, RealTimeStreaming
+from litellm.llms.base_llm.chat.transformation import BaseLLMException
 from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler
 from litellm.llms.openai.workload_identity import _workload_identity_auth
 from litellm.models.credentials import CredentialItem
 from litellm.realtime_api import main as realtime_main
 from litellm.realtime_api.main import _with_resolved_session_model
 from litellm.types.guardrails import GuardrailEventHooks
+from litellm.types.realtime import RealtimeQueryParams


@pytest.fixture
@ -745,28 +755,39 @@ _GA_CLIENT: Final = _ClientWebSocketWithHeaders(headers=())
 _BETA_CLIENT: Final = _ClientWebSocketWithHeaders(headers=((b"openai-beta", b"realtime=v1"),))


-@pytest.mark.parametrize("model", ["gpt-realtime-2.1", "gpt-realtime-2", "gpt-realtime-2-2026-05-06"])
-def test_azure_ga_only_protocol_comes_from_model_metadata(local_model_cost_map, model: str) -> None:
+_GA_DEFAULT_SESSIONS: Final = (
+    pytest.param("gpt-realtime-2.1", "realtime", None, id="ga-only-model"),
+    pytest.param("gpt-realtime-2", "realtime", None, id="ga-only-alias"),
+    pytest.param("gpt-realtime-2-2026-05-06", "realtime", None, id="ga-only-snapshot"),
+    pytest.param("my-transcribe-deployment", "realtime", {"intent": "transcription"}, id="transcription-intent"),
+    pytest.param("my-translate-deployment", "translation", None, id="translation-mode"),
+)
+
+
+@pytest.mark.parametrize(("model", "realtime_mode", "query_params"), _GA_DEFAULT_SESSIONS)
+def test_azure_realtime_protocol_defaults_to_ga_when_none_is_configured(
+    local_model_cost_map, model: str, realtime_mode: str, query_params: RealtimeQueryParams | None
+) -> None:
    from litellm.llms.azure.realtime.handler import azure_realtime_protocol_for_client

-    assert (
-        azure_realtime_protocol_for_client(
-            None,
-            model=model,
-            realtime_mode="realtime",
-            query_params=None,
-            websocket=_BETA_CLIENT,
-        )
-        == "GA"
+    protocol: Final = azure_realtime_protocol_for_client(
+        None, model=model, realtime_mode=realtime_mode, query_params=query_params, websocket=_BETA_CLIENT
    )
-    with pytest.raises(ValueError, match="requires the Azure OpenAI v1 Realtime API"):
-        azure_realtime_protocol_for_client(
-            "beta",
-            model=model,
-            realtime_mode="realtime",
-            query_params=None,
-            websocket=_BETA_CLIENT,
-        )
+
+    assert protocol == "GA"
+
+
+@pytest.mark.parametrize(("model", "realtime_mode", "query_params"), _GA_DEFAULT_SESSIONS)
+def test_azure_realtime_protocol_configured_by_the_operator_wins(
+    local_model_cost_map, model: str, realtime_mode: str, query_params: RealtimeQueryParams | None
+) -> None:
+    from litellm.llms.azure.realtime.handler import azure_realtime_protocol_for_client
+
+    protocol: Final = azure_realtime_protocol_for_client(
+        "beta", model=model, realtime_mode=realtime_mode, query_params=query_params, websocket=_GA_CLIENT
+    )
+
+    assert protocol == "beta"


 async def _azure_backend_url_dialed_for(websocket: _ClientWebSocketWithHeaders, **kwargs: object) -> str | None:
@ -1278,3 +1299,348 @@ async def test_openai_realtime_health_check_sends_the_federated_bearer_for_a_key
            model_params={"model": "openai/gpt-realtime", **federated_deployment},
        )
    assert connect.kwargs["additional_headers"] == {"Authorization": "Bearer federated-bearer"}
+
+
+def _realtime_http_route_call(
+    route: str,
+    logging_obj: Logging,
+    custom_llm_provider: str,
+    api_base: str,
+    client: AsyncHTTPHandler | None = None,
+) -> Awaitable[httpx.Response]:
+    if route == "client_secrets":
+        return realtime_main.acreate_realtime_client_secret.__wrapped__(
+            model="gpt-realtime",
+            custom_llm_provider=custom_llm_provider,
+            api_key="sk-test",
+            api_base=api_base,
+            client=client,
+            litellm_logging_obj=logging_obj,
+        )
+    if route == "translation_client_secrets":
+        return realtime_main.acreate_realtime_translation_client_secret.__wrapped__(
+            model="gpt-realtime-translate",
+            custom_llm_provider=custom_llm_provider,
+            api_key="sk-test",
+            api_base=api_base,
+            client=client,
+            litellm_logging_obj=logging_obj,
+        )
+    if route == "calls":
+        return realtime_main.arealtime_calls.__wrapped__(
+            openai_ephemeral_key="ek_test",
+            sdp_body=b"v=0",
+            model="gpt-realtime",
+            custom_llm_provider=custom_llm_provider,
+            api_base=api_base,
+            client=client,
+            litellm_logging_obj=logging_obj,
+        )
+    return realtime_main.arealtime_translation_calls.__wrapped__(
+        openai_ephemeral_key="ek_test",
+        sdp_body=b"v=0",
+        model="gpt-realtime-translate",
+        custom_llm_provider=custom_llm_provider,
+        api_base=api_base,
+        client=client,
+        litellm_logging_obj=logging_obj,
+    )
+
+
+def _realtime_http_logging(route: str) -> Logging:
+    return Logging(
+        model="gpt-realtime",
+        messages=[],
+        stream=False,
+        call_type="acreate_realtime_client_secret",
+        start_time=datetime(2026, 1, 1, tzinfo=timezone.utc),
+        litellm_call_id=f"test-realtime-{route}",
+        function_id=f"test-realtime-{route}",
+    )
+
+
+_OPENAI_REALTIME_HTTP_ROUTE_PATHS: Final = {
+    "client_secrets": "/v1/realtime/client_secrets",
+    "translation_client_secrets": "/v1/realtime/translations/client_secrets",
+    "calls": "/v1/realtime/calls",
+    "translation_calls": "/v1/realtime/translations/calls",
+}
+
+
+def _validity(builder: x509.CertificateBuilder) -> x509.CertificateBuilder:
+    now: Final = datetime.now(timezone.utc)
+    return (
+        builder.serial_number(x509.random_serial_number())
+        .not_valid_before(now - timedelta(minutes=5))
+        .not_valid_after(now + timedelta(hours=1))
+    )
+
+
+def _write_private_ca_and_loopback_certificate(directory: Path) -> tuple[Path, Path, Path]:
+    ca_key: Final = ec.generate_private_key(ec.SECP256R1())
+    server_key: Final = ec.generate_private_key(ec.SECP256R1())
+    ca_name: Final = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "litellm realtime test CA")])
+    ca_certificate: Final = (
+        _validity(x509.CertificateBuilder())
+        .subject_name(ca_name)
+        .issuer_name(ca_name)
+        .public_key(ca_key.public_key())
+        .add_extension(x509.BasicConstraints(ca=True, path_length=None), critical=True)
+        .add_extension(
+            x509.KeyUsage(
+                digital_signature=True,
+                content_commitment=False,
+                key_encipherment=False,
+                data_encipherment=False,
+                key_agreement=False,
+                key_cert_sign=True,
+                crl_sign=True,
+                encipher_only=False,
+                decipher_only=False,
+            ),
+            critical=True,
+        )
+        .add_extension(x509.SubjectKeyIdentifier.from_public_key(ca_key.public_key()), critical=False)
+        .sign(ca_key, hashes.SHA256())
+    )
+    server_certificate: Final = (
+        _validity(x509.CertificateBuilder())
+        .subject_name(x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "127.0.0.1")]))
+        .issuer_name(ca_name)
+        .public_key(server_key.public_key())
+        .add_extension(x509.SubjectAlternativeName([x509.IPAddress(ipaddress.ip_address("127.0.0.1"))]), critical=False)
+        .add_extension(x509.ExtendedKeyUsage([ExtendedKeyUsageOID.SERVER_AUTH]), critical=False)
+        .add_extension(x509.AuthorityKeyIdentifier.from_issuer_public_key(ca_key.public_key()), critical=False)
+        .sign(ca_key, hashes.SHA256())
+    )
+    ca_path: Final = directory / "ca.pem"
+    certificate_path: Final = directory / "server.pem"
+    key_path: Final = directory / "server.key"
+    ca_path.write_bytes(ca_certificate.public_bytes(serialization.Encoding.PEM))
+    certificate_path.write_bytes(server_certificate.public_bytes(serialization.Encoding.PEM))
+    key_path.write_bytes(
+        server_key.private_bytes(
+            serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, serialization.NoEncryption()
+        )
+    )
+    return ca_path, certificate_path, key_path
+
+
+class _EchoPathHandler(BaseHTTPRequestHandler):
+    def do_POST(self) -> None:
+        self.rfile.read(int(self.headers.get("Content-Length", "0")))
+        body: Final = json.dumps({"value": "ek_test", "path": self.path}).encode()
+        self.send_response(200)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Content-Length", str(len(body)))
+        self.end_headers()
+        self.wfile.write(body)
+
+
+@pytest.fixture
+def private_ca_upstream(tmp_path: Path) -> Iterator[tuple[str, Path]]:
+    ca_path, certificate_path, key_path = _write_private_ca_and_loopback_certificate(tmp_path)
+    tls: Final = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+    tls.load_cert_chain(certificate_path, key_path)
+    server: Final = ThreadingHTTPServer(("127.0.0.1", 0), _EchoPathHandler)
+    server.socket = tls.wrap_socket(server.socket, server_side=True)
+    thread: Final = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    yield f"https://127.0.0.1:{server.server_address[1]}", ca_path
+    server.shutdown()
+    server.server_close()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("route", tuple(_OPENAI_REALTIME_HTTP_ROUTE_PATHS))
+async def test_openai_realtime_http_routes_trust_the_ca_bundle_set_in_litellm_ssl_verify(
+    route: str, private_ca_upstream: tuple[str, Path], monkeypatch: pytest.MonkeyPatch
+) -> None:
+    api_base, ca_path = private_ca_upstream
+    monkeypatch.setattr(litellm, "ssl_verify", str(ca_path))
+
+    response: Final = await _realtime_http_route_call(
+        route, _realtime_http_logging(route), custom_llm_provider="openai", api_base=api_base
+    )
+
+    assert response.status_code == 200
+    assert response.json() == {"value": "ek_test", "path": _OPENAI_REALTIME_HTTP_ROUTE_PATHS[route]}
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("route", tuple(_OPENAI_REALTIME_HTTP_ROUTE_PATHS))
+async def test_openai_realtime_http_routes_surface_the_upstream_error_body_verbatim(route: str) -> None:
+    upstream_error: Final = {
+        "error": {
+            "message": "Invalid model",
+            "type": "invalid_request_error",
+            "param": "session.model",
+            "code": "invalid_value",
+        }
+    }
+
+    with pytest.raises(BaseLLMException) as raised:
+        await _realtime_http_route_call(
+            route,
+            _realtime_http_logging(route),
+            custom_llm_provider="openai",
+            api_base="https://upstream.example",
+            client=AsyncHTTPHandler(
+                transport=httpx.MockTransport(lambda request: httpx.Response(400, json=upstream_error))
+            ),
+        )
+
+    assert raised.value.status_code == 400
+    assert json.loads(raised.value.message) == upstream_error
+
+
+class _GatewayThatRecordsTheSession:
+    session: Mapping[str, object] | None = None
+
+    def __call__(self, request: httpx.Request) -> httpx.Response:
+        self.session = json.loads(request.content)["session"]
+        return httpx.Response(200, json={"value": "ek_test"})
+
+
+@pytest.mark.asyncio
+async def test_client_secret_keeps_the_realtime_session_model_when_input_transcription_names_another() -> None:
+    gateway: Final = _GatewayThatRecordsTheSession()
+    logging_obj: Final = _realtime_http_logging("client_secrets")
+
+    await realtime_main.acreate_realtime_client_secret.__wrapped__(
+        model="gpt-realtime",
+        session={
+            "type": "realtime",
+            "model": "gpt-realtime",
+            "audio": {"input": {"transcription": {"model": "gpt-realtime-whisper"}}},
+        },
+        custom_llm_provider="openai",
+        api_key="sk-test",
+        api_base="https://upstream.example",
+        litellm_logging_obj=logging_obj,
+        client=AsyncHTTPHandler(transport=httpx.MockTransport(gateway)),
+    )
+
+    assert gateway.session is not None
+    assert gateway.session["model"] == "gpt-realtime"
+    assert gateway.session["audio"] == {"input": {"transcription": {"model": "gpt-realtime-whisper"}}}
+    assert logging_obj.model == "gpt-realtime"
+
+
+class _UpstreamThatMustNotBeCalled:
+    called: bool = False
+
+    def __call__(self, request: httpx.Request) -> httpx.Response:
+        self.called = True
+        return httpx.Response(200, json={"value": "ek_test"})
+
+
+class _RecordingClientWebSocket:
+    def __init__(self) -> None:
+        self.scope: Final = {"headers": ()}
+        self.sent: tuple[str, ...] = ()
+        self.close_code: int | None = None
+
+    async def send_text(self, data: str) -> None:
+        self.sent = (*self.sent, data)
+
+    async def close(self, code: int = 1000, reason: str = "") -> None:
+        self.close_code = code
+
+
+def _block_every_request_with_a_guardrail(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(
+        litellm,
+        "callbacks",
+        [
+            CustomGuardrail(
+                guardrail_name="translation-blocker", event_hook=GuardrailEventHooks.pre_call, default_on=True
+            )
+        ],
+    )
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("route", ("translation_client_secrets", "translation_calls"))
+async def test_translation_http_routes_refuse_a_guarded_request_before_any_upstream_call(
+    route: str, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    _block_every_request_with_a_guardrail(monkeypatch)
+    upstream: Final = _UpstreamThatMustNotBeCalled()
+
+    with pytest.raises(litellm.BadRequestError, match=TRANSLATION_GUARDRAILS_UNSUPPORTED_MESSAGE) as raised:
+        await _realtime_http_route_call(
+            route,
+            _realtime_http_logging(route),
+            custom_llm_provider="openai",
+            api_base="https://upstream.example",
+            client=AsyncHTTPHandler(transport=httpx.MockTransport(upstream)),
+        )
+
+    assert raised.value.status_code == 400
+    assert upstream.called is False
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("route", ("translation_client_secrets", "translation_calls"))
+async def test_translation_http_routes_refuse_a_provider_other_than_openai_and_azure_before_any_upstream_call(
+    route: str,
+) -> None:
+    upstream: Final = _UpstreamThatMustNotBeCalled()
+
+    with pytest.raises(litellm.BadRequestError, match="supports only the azure, openai providers, got xai") as raised:
+        await _realtime_http_route_call(
+            route,
+            _realtime_http_logging(route),
+            custom_llm_provider="xai",
+            api_base="https://upstream.example",
+            client=AsyncHTTPHandler(transport=httpx.MockTransport(upstream)),
+        )
+
+    assert raised.value.status_code == 400
+    assert upstream.called is False
+
+
+@pytest.mark.asyncio
+async def test_translation_websocket_closes_1008_for_a_guarded_request_before_the_upstream_connect(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    _block_every_request_with_a_guardrail(monkeypatch)
+    websocket: Final = _RecordingClientWebSocket()
+    connect: Final = _CapturingConnect()
+
+    with patch("websockets.connect", connect):
+        await realtime_main._arealtime.__wrapped__(
+            model="openai/gpt-realtime-translate",
+            websocket=websocket,
+            api_key="fake-key",
+            realtime_mode="translation",
+            litellm_logging_obj=FakeLogging(),
+        )
+
+    assert connect.url is None
+    assert websocket.close_code == 1008
+    assert [json.loads(event)["error"]["message"] for event in websocket.sent] == [
+        TRANSLATION_GUARDRAILS_UNSUPPORTED_MESSAGE
+    ]
+
+
+@pytest.mark.asyncio
+async def test_translation_websocket_closes_1008_for_a_provider_other_than_openai_and_azure() -> None:
+    websocket: Final = _RecordingClientWebSocket()
+    connect: Final = _CapturingConnect()
+
+    with patch("websockets.connect", connect):
+        await realtime_main._arealtime.__wrapped__(
+            model="xai/grok-voice",
+            websocket=websocket,
+            api_key="fake-key",
+            realtime_mode="translation",
+            litellm_logging_obj=FakeLogging(),
+        )
+
+    assert connect.url is None
+    assert websocket.close_code == 1008
+    assert [json.loads(event)["error"]["message"] for event in websocket.sent] == [
+        "Realtime translation supports only the azure, azure_ai, openai providers, got xai"
+    ]
--- a/tests/unit/test_cost_calculator.py
+++ b/tests/unit/test_cost_calculator.py
@ -9878,6 +9878,127 @@ def test_responses_ws_logging_object_carries_the_completed_responses_id_tier_and
    assert ResponsesWebSocketTokenUsageProcessor.create_logging_object([{"type": "error", "error": {}}]).id is None


+@pytest.mark.parametrize("provider", ("openai", "azure"))
+@pytest.mark.parametrize("model", ("gpt-realtime-2.1", "gpt-realtime-2.1-mini"))
+def test_realtime_cached_multimodal_token_cost(_local_model_cost_map, provider: str, model: str):
+    model_name: Final = f"azure/{model}" if provider == "azure" else model
+    rates: Final = litellm.model_cost[model_name]
+    events: Final[OpenAIRealtimeStreamList] = [
+        {"type": "session.created", "session": {"model": model}},
+        {
+            "type": "response.done",
+            "response": {
+                "usage": {
+                    "input_tokens": 1000,
+                    "output_tokens": 300,
+                    "total_tokens": 1300,
+                    "input_token_details": {
+                        "text_tokens": 400,
+                        "audio_tokens": 400,
+                        "image_tokens": 200,
+                        "cached_tokens": 300,
+                        "cached_tokens_details": {"text_tokens": 100, "audio_tokens": 150, "image_tokens": 50},
+                    },
+                    "output_token_details": {"text_tokens": 100, "audio_tokens": 100, "reasoning_tokens": 100},
+                }
+            },
+        },
+    ]
+    combined: Final = RealtimeAPITokenUsageProcessor.collect_and_combine_usage_from_realtime_stream_results(events)
+    actual: Final = handle_realtime_stream_cost_calculation(
+        results=events,
+        combined_usage_object=combined,
+        custom_llm_provider=provider,
+        litellm_model_name=model_name,
+    )
+    expected: Final = (
+        300 * rates["input_cost_per_token"]
+        + 250 * rates["input_cost_per_audio_token"]
+        + 150 * rates["input_cost_per_image_token"]
+        + 100 * rates["cache_read_input_token_cost"]
+        + 150 * rates["cache_read_input_audio_token_cost"]
+        + 50 * rates["cache_read_input_image_token_cost"]
+        + 200 * rates["output_cost_per_token"]
+        + 100 * rates["output_cost_per_audio_token"]
+    )
+
+    assert actual == pytest.approx(expected)
+
+
+def test_realtime_translation_duration_cost(_local_model_cost_map):
+    from litellm.cost_calculator import handle_realtime_translation_cost_calculation
+
+    model: Final = "gpt-realtime-translate"
+    events: Final[OpenAIRealtimeStreamList] = [
+        {"type": "session.closed", "usage": {"type": "duration", "output_seconds": 2.0}}
+    ]
+    actual: Final = handle_realtime_translation_cost_calculation(
+        results=events,
+        custom_llm_provider="openai",
+        litellm_model_name=model,
+    )
+
+    assert actual == pytest.approx(2 * litellm.model_cost[model]["output_cost_per_second"])
+
+
+def test_realtime_translation_bills_session_audio_once_on_openai_and_azure(_local_model_cost_map) -> None:
+    from litellm.cost_calculator import handle_realtime_translation_cost_calculation
+
+    one_minute_session: Final[OpenAIRealtimeStreamList] = [
+        {"type": "session.closed", "usage": {"type": "duration", "input_seconds": 60.0, "output_seconds": 60.0}}
+    ]
+    openai_cost: Final = handle_realtime_translation_cost_calculation(
+        results=one_minute_session, custom_llm_provider="openai", litellm_model_name="gpt-realtime-translate"
+    )
+    azure_cost: Final = handle_realtime_translation_cost_calculation(
+        results=one_minute_session, custom_llm_provider="azure", litellm_model_name="gpt-realtime-translate"
+    )
+
+    assert openai_cost == pytest.approx(60 * litellm.model_cost["gpt-realtime-translate"]["output_cost_per_second"])
+    assert openai_cost == pytest.approx(azure_cost)
+
+
+@pytest.mark.parametrize("malformed_duration", [-1.0, float("nan"), float("inf"), 10**1000, True])
+def test_realtime_translation_duration_cost_ignores_malformed_provider_usage(
+    _local_model_cost_map, malformed_duration: float | int | bool
+) -> None:
+    from litellm.cost_calculator import handle_realtime_translation_cost_calculation
+
+    model: Final = "translation-deployment-priced-both-directions"
+    litellm.register_model(
+        model_cost={
+            model: {
+                "litellm_provider": "openai",
+                "mode": "realtime",
+                "input_cost_per_second": 0.001,
+                "output_cost_per_second": 0.002,
+            }
+        }
+    )
+    input_invalid_events: Final[OpenAIRealtimeStreamList] = [
+        {
+            "type": "session.closed",
+            "usage": {"type": "duration", "input_seconds": malformed_duration, "output_seconds": 1.5},
+        }
+    ]
+    output_invalid_events: Final[OpenAIRealtimeStreamList] = [
+        {
+            "type": "session.closed",
+            "usage": {"type": "duration", "input_seconds": 2.0, "output_seconds": malformed_duration},
+        }
+    ]
+
+    input_invalid_cost: Final = handle_realtime_translation_cost_calculation(
+        results=input_invalid_events, custom_llm_provider="openai", litellm_model_name=model
+    )
+    output_invalid_cost: Final = handle_realtime_translation_cost_calculation(
+        results=output_invalid_events, custom_llm_provider="openai", litellm_model_name=model
+    )
+
+    assert input_invalid_cost == pytest.approx(1.5 * litellm.model_cost[model]["output_cost_per_second"])
+    assert output_invalid_cost == pytest.approx(2.0 * litellm.model_cost[model]["input_cost_per_second"])
+
+
@pytest.mark.parametrize("profile_prefix", ("us", "global"))
@pytest.mark.parametrize(
    "model_name",
--- a/ui/litellm-dashboard/src/lib/http/schema.d.ts
+++ b/ui/litellm-dashboard/src/lib/http/schema.d.ts
@ -11084,46 +11084,6 @@ export interface paths {
        patch?: never;
        trace?: never;
    };
-    "/openai/v1/realtime/translations/calls": {
-        parameters: {
-            query?: never;
-            header?: never;
-            path?: never;
-            cookie?: never;
-        };
-        get?: never;
-        put?: never;
-        /**
-         * Proxy Realtime Calls
-         * @description Exchange a WebRTC SDP offer for an answer using the ephemeral token from client_secrets
-         */
-        post: operations["proxy_realtime_calls_openai_v1_realtime_translations_calls_post"];
-        delete?: never;
-        options?: never;
-        head?: never;
-        patch?: never;
-        trace?: never;
-    };
-    "/openai/v1/realtime/translations/client_secrets": {
-        parameters: {
-            query?: never;
-            header?: never;
-            path?: never;
-            cookie?: never;
-        };
-        get?: never;
-        put?: never;
-        /**
-         * Create Realtime Client Secret
-         * @description Create a short-lived client token for a Realtime conversation, transcription, or translation session
-         */
-        post: operations["create_realtime_client_secret_openai_v1_realtime_translations_client_secrets_post"];
-        delete?: never;
-        options?: never;
-        head?: never;
-        patch?: never;
-        trace?: never;
-    };
    "/openai/v1/responses": {
        parameters: {
            query?: never;
@ -48064,6 +48024,54 @@ export interface components {
             */
            success_definition: string;
        };
+        /** TranscriptionDetectedLanguage */
+        TranscriptionDetectedLanguage: {
+            /** Code */
+            code: string;
+        };
+        /** TranscriptionResponse */
+        TranscriptionResponse: {
+            /** Languages */
+            languages?: components["schemas"]["TranscriptionDetectedLanguage"][] | null;
+            /** Text */
+            text?: string | null;
+            /** Usage */
+            usage?: components["schemas"]["TranscriptionUsageDurationObject"] | components["schemas"]["TranscriptionUsageTokensObject"] | null;
+        } & {
+            [key: string]: unknown;
+        };
+        /** TranscriptionUsageDurationObject */
+        TranscriptionUsageDurationObject: {
+            /** Seconds */
+            seconds: number;
+            /**
+             * Type
+             * @constant
+             */
+            type: "duration";
+        };
+        /** TranscriptionUsageInputTokenDetailsObject */
+        TranscriptionUsageInputTokenDetailsObject: {
+            /** Audio Tokens */
+            audio_tokens: number;
+            /** Text Tokens */
+            text_tokens: number;
+        };
+        /** TranscriptionUsageTokensObject */
+        TranscriptionUsageTokensObject: {
+            input_token_details?: components["schemas"]["TranscriptionUsageInputTokenDetailsObject"] | null;
+            /** Input Tokens */
+            input_tokens: number;
+            /** Output Tokens */
+            output_tokens: number;
+            /** Total Tokens */
+            total_tokens: number;
+            /**
+             * Type
+             * @constant
+             */
+            type: "tokens";
+        };
        /** TransformRequestBody */
        TransformRequestBody: {
            call_type: components["schemas"]["CallTypes"];
@ -52777,7 +52785,8 @@ export interface operations {
                    [name: string]: unknown;
                };
                content: {
-                    "application/json": unknown;
+                    "application/json": components["schemas"]["TranscriptionResponse"];
+                    "text/event-stream": string;
                };
            };
            /** @description Validation Error */
@ -66920,156 +66929,6 @@ export interface operations {
            };
        };
    };
-    proxy_realtime_calls_openai_v1_realtime_translations_calls_post: {
-        parameters: {
-            query?: never;
-            header?: never;
-            path?: never;
-            cookie?: never;
-        };
-        requestBody: {
-            content: {
-                "application/sdp": string;
-            };
-        };
-        responses: {
-            /** @description WebRTC SDP answer */
-            200: {
-                headers: {
-                    [name: string]: unknown;
-                };
-                content: {
-                    "application/json": unknown;
-                    "application/sdp": string;
-                };
-            };
-        };
-    };
-    create_realtime_client_secret_openai_v1_realtime_translations_client_secrets_post: {
-        parameters: {
-            query?: never;
-            header?: never;
-            path?: never;
-            cookie?: never;
-        };
-        requestBody: {
-            content: {
-                /**
-                 * @example {
-                 *       "model": "gpt-realtime-translate",
-                 *       "session": {
-                 *         "model": "gpt-realtime-translate",
-                 *         "type": "translation"
-                 *       }
-                 *     }
-                 */
-                "application/json": {
-                    expires_after?: {
-                        /**
-                         * Anchor
-                         * @default created_at
-                         */
-                        anchor?: string | null;
-                        /** Seconds */
-                        seconds?: number | null;
-                    } | null;
-                    /** Model */
-                    model?: string | null;
-                    session?: ({
-                        audio?: ({
-                            input?: ({
-                                /** Format */
-                                format?: string | {
-                                    [key: string]: unknown;
-                                } | null;
-                                /** Noise Reduction */
-                                noise_reduction?: {
-                                    [key: string]: unknown;
-                                } | null;
-                                transcription?: ({
-                                    /** Delay */
-                                    delay?: ("minimal" | "low" | "medium" | "high" | "xhigh") | null;
-                                    /** Keywords */
-                                    keywords?: string[] | null;
-                                    /** Language */
-                                    language?: string | null;
-                                    /** Languages */
-                                    languages?: string[] | null;
-                                    /** Model */
-                                    model?: string | null;
-                                    /** Prompt */
-                                    prompt?: string | null;
-                                } & {
-                                    [key: string]: unknown;
-                                }) | null;
-                                /** Turn Detection */
-                                turn_detection?: {
-                                    [key: string]: unknown;
-                                } | null;
-                            } & {
-                                [key: string]: unknown;
-                            }) | null;
-                            output?: ({
-                                /** Format */
-                                format?: string | {
-                                    [key: string]: unknown;
-                                } | null;
-                                /** Language */
-                                language?: string | null;
-                                /** Voice */
-                                voice?: string | {
-                                    [key: string]: unknown;
-                                } | null;
-                            } & {
-                                [key: string]: unknown;
-                            }) | null;
-                        } & {
-                            [key: string]: unknown;
-                        }) | null;
-                        /** Include */
-                        include?: string[] | null;
-                        /** Instructions */
-                        instructions?: string | null;
-                        /** Max Output Tokens */
-                        max_output_tokens?: number | string | null;
-                        /** Model */
-                        model?: string | null;
-                        /** Output Modalities */
-                        output_modalities?: string[] | null;
-                        /** Prompt */
-                        prompt?: {
-                            [key: string]: unknown;
-                        } | null;
-                        /** Tool Choice */
-                        tool_choice?: unknown | null;
-                        /** Tools */
-                        tools?: {
-                            [key: string]: unknown;
-                        }[] | null;
-                        /** Tracing */
-                        tracing?: unknown | null;
-                        /** Truncation */
-                        truncation?: unknown | null;
-                        /** Type */
-                        type?: ("realtime" | "transcription" | "translation") | null;
-                    } & {
-                        [key: string]: unknown;
-                    }) | null;
-                };
-            };
-        };
-        responses: {
-            /** @description Successful Response */
-            200: {
-                headers: {
-                    [name: string]: unknown;
-                };
-                content: {
-                    "application/json": components["schemas"]["RealtimeClientSecretResponse"];
-                };
-            };
-        };
-    };
    responses_api_openai_v1_responses_post: {
        parameters: {
            query?: never;
@ -70457,13 +70316,12 @@ export interface operations {
        };
        responses: {
            /** @description WebRTC SDP answer */
-            200: {
+            201: {
                headers: {
                    [name: string]: unknown;
                };
                content: {
-                    "application/json": unknown;
-                    "application/sdp": string;
+                    "text/plain": string;
                };
            };
        };
@ -77823,7 +77681,8 @@ export interface operations {
                    [name: string]: unknown;
                };
                content: {
-                    "application/json": unknown;
+                    "application/json": components["schemas"]["TranscriptionResponse"];
+                    "text/event-stream": string;
                };
            };
            /** @description Validation Error */
@ -81527,13 +81386,12 @@ export interface operations {
        };
        responses: {
            /** @description WebRTC SDP answer */
-            200: {
+            201: {
                headers: {
                    [name: string]: unknown;
                };
                content: {
-                    "application/json": unknown;
-                    "application/sdp": string;
+                    "text/plain": string;
                };
            };
        };