Merge dcd949350d into 82ca10cc2d

litellm/llms/azure/realtime/handler.py

@@ -22,7 +22,7 @@ from ....litellm_core_utils.realtime_streaming import (
     ScopedWebSocket,
     client_sent_openai_beta_realtime_header,
 )
-from ....llms.custom_httpx.http_handler import get_shared_realtime_ssl_context
+from ....llms.custom_httpx.http_handler import realtime_ssl_for_url
 from ..azure import AzureChatCompletion
 
 # BACKEND_WS_URL = "ws://localhost:8080/v1/realtime?model=gpt-4o-realtime-preview-2024-10-01"
@@ -96,7 +96,7 @@ class AzureOpenAIRealtime(AzureChatCompletion):
         Construct Azure realtime WebSocket URL.
 
         Args:
-            api_base: Azure API base URL (will be converted from https:// to wss://)
+            api_base: Azure API base URL (https:// becomes wss://, http:// becomes ws://)
             model: Model deployment name
             api_version: Azure API version
             realtime_protocol: Protocol version to use:
@@ -113,7 +113,7 @@ class AzureOpenAIRealtime(AzureChatCompletion):
         """
         from urllib.parse import urlencode
 
-        api_base = api_base.replace("https://", "wss://")
+        ws_base: Final = api_base.replace("https://", "wss://").replace("http://", "ws://")
 
         # Determine path based on realtime_protocol (case-insensitive)
         _is_ga: Final = realtime_protocol is not None and realtime_protocol.upper() in (
@@ -136,7 +136,7 @@ class AzureOpenAIRealtime(AzureChatCompletion):
             query_parts.append(urlencode({"intent": intent}))
 
         qs: Final = "&".join(query_parts)
-        return f"{api_base}{path}?{qs}" if qs else f"{api_base}{path}"
+        return f"{ws_base}{path}?{qs}" if qs else f"{ws_base}{path}"
 
     def construct_url(
         self,
@@ -184,12 +184,11 @@ class AzureOpenAIRealtime(AzureChatCompletion):
         auth_headers: Final = self.get_auth_headers(api_key=api_key, azure_ad_token=azure_ad_token)
 
         try:
-            ssl_context: Final = get_shared_realtime_ssl_context()
             async with websockets.connect(
                 url,
                 additional_headers=auth_headers,
                 max_size=REALTIME_WEBSOCKET_MAX_MESSAGE_SIZE_BYTES,
-                ssl=ssl_context,
+                ssl=realtime_ssl_for_url(url),
             ) as backend_ws:
                 realtime_streaming: Final = RealTimeStreaming(
                     websocket,

litellm/llms/custom_httpx/llm_http_handler.py

@@ -201,7 +201,7 @@ def _rust_responses_websocket_enabled(
     return decision(context) is not Decision.PYTHON
 
 
-from .http_handler import get_shared_realtime_ssl_context, realtime_ssl_for_url
+from .http_handler import realtime_ssl_for_url
 
 if TYPE_CHECKING:
     from aiohttp import ClientSession
@@ -6935,11 +6935,7 @@ class BaseLLMHTTPHandler:
                 ws_url = urlunparse(_parsed._replace(query=urlencode({k: v[0] for k, v in _qs.items()})))
 
         try:
-            ssl_context = get_shared_realtime_ssl_context()
-            if ws_url.startswith("wss://") and ssl_context is False:
-                ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
-                ssl_context.check_hostname = False
-                ssl_context.verify_mode = ssl.CERT_NONE
+            ssl_context: Final = realtime_ssl_for_url(ws_url)
 
             logging_obj.pre_call(
                 input=None,

litellm/llms/openai/realtime/handler.py

@@ -19,7 +19,7 @@ from ....litellm_core_utils.realtime_streaming import (
     RealtimeUsageMeter,
     client_sent_openai_beta_realtime_header,
 )
-from ....llms.custom_httpx.http_handler import get_shared_realtime_ssl_context
+from ....llms.custom_httpx.http_handler import realtime_ssl_for_url
 from ..common_utils import is_openai_backed_api_base
 from ..openai import OpenAIChatCompletion
 
@@ -69,20 +69,9 @@ class OpenAIRealtime(OpenAIChatCompletion):
             url: WebSocket URL (ws:// or wss://)
 
         Returns:
-            SSL configuration (None, True, or SSLContext)
+            None for ws://, otherwise the shared SSL context, or an unverified one when ssl_verify is off
         """
-        if url.startswith("ws://"):
-            return None
-
-        # Use the shared SSL context which respects custom CA certs and SSL settings
-        ssl_config: Final = get_shared_realtime_ssl_context()
-
-        # If ssl_config is False (ssl_verify=False), websockets library needs True instead
-        # to establish connection without verification (False would fail)
-        if ssl_config is False:
-            return True
-
-        return ssl_config
+        return realtime_ssl_for_url(url)
 
     def _construct_url(self, api_base: str, query_params: RealtimeQueryParams) -> str:
         """

tests/integration/_support/upstream.py

@@ -414,10 +414,9 @@ class Provider:
         self.interactions[interaction_id] = cancelled
         return JSONResponse(_interaction_body(interaction_id, cancelled))
 
-    async def realtime(self, websocket: WebSocket) -> None:
+    def _observed_upgrade(self, websocket: WebSocket) -> tuple[str, str, str]:
         authorization: Final = websocket.headers.get("authorization", "")
         api_key: Final = websocket.headers.get("api-key", "")
-        scenario_id: Final = authorization.removeprefix("Bearer ") if authorization else api_key
         self.observations.put(
             Observation(
                 websocket.url.path,
@@ -427,6 +426,10 @@ class Provider:
                 api_key,
             )
         )
+        return authorization.removeprefix("Bearer ") if authorization else api_key, authorization, api_key
+
+    async def realtime(self, websocket: WebSocket) -> None:
+        scenario_id, authorization, api_key = self._observed_upgrade(websocket)
         response: Final = self.scenario_store.get(scenario_id)
         if not isinstance(response, RealtimeResponse):
             await websocket.close(code=4404)
@@ -483,6 +486,21 @@ class Provider:
         if frame_type == "input_audio_buffer.commit" and connection.auto_answers():
             await _reply_while_open(websocket, _rendered_realtime_event(dict(_AUTO_ANSWER), connection.scenario_id))
 
+    async def responses_websocket(self, websocket: WebSocket) -> None:
+        scenario_id, authorization, api_key = self._observed_upgrade(websocket)
+        response: Final = self.scenario_store.get(scenario_id)
+        if not isinstance(response, RealtimeResponse):
+            await websocket.close(code=4404)
+            return
+        await websocket.accept()
+        async for message in websocket.iter_json():
+            payload: Final = JSON_OBJECT.validate_python(message)
+            self.observations.put(Observation(websocket.url.path, authorization, payload, "WEBSOCKET_FRAME", api_key))
+            if payload.get("type") != "response.create":
+                continue
+            for event in response.events:
+                await websocket.send_json(_rendered_realtime_event(event, scenario_id))
+
     async def muse_realtime(self, websocket: WebSocket) -> None:
         await websocket.accept()
         handshake: Final = JSON_OBJECT.validate_python(await websocket.receive_json())
@@ -653,6 +671,8 @@ class Provider:
                 WebSocketRoute("/v1/realtime", self.realtime),
                 WebSocketRoute("/openai/v1/realtime", self.realtime),
                 WebSocketRoute("/openai/realtime", self.realtime),
+                WebSocketRoute("/v1/responses", self.responses_websocket),
+                WebSocketRoute("/openai/v1/responses", self.responses_websocket),
                 WebSocketRoute("/v1/asr/realtime", self.muse_realtime),
                 WebSocketRoute(GEMINI_LIVE_PATH, self.gemini_live),
                 WebSocketRoute(GEMINI_AI_STUDIO_LIVE_PATH, self.gemini_ai_studio_live),

tests/integration/providers/test_realtime_websocket_dial_schemes.py

@@ -0,0 +1,340 @@
+"""Backend websocket dials carry a TLS context only when the resolved URL is ``wss://``.
+
+An Azure or OpenAI realtime session and a Responses websocket session each open one backend socket. A
+plaintext ``api_base`` (``http://`` or ``ws://``) resolves to ``ws://``, where the websockets client refuses
+any ``ssl`` argument, and ``https://`` resolves to ``wss://``, where ``litellm_settings.ssl_verify: false``
+has to become a context that skips verification rather than ``False``, which the client rejects. Every row
+runs against the scripted upstream on 127.0.0.1 and asserts the upgrade it saw, so a dial that never left
+the proxy fails on the observation, not only on the client's close.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import uuid
+from collections.abc import AsyncIterator, Iterator
+from pathlib import Path
+from typing import Final
+
+import httpx
+import pytest
+import websockets
+import yaml
+from pydantic import JsonValue
+from websockets.asyncio.client import ClientConnection
+from websockets.exceptions import ConnectionClosed
+
+from tests.integration._support.client import (
+    JSON_OBJECT,
+    Gateway,
+    Scenario,
+    eventually,
+    gateway_from_environment,
+    object_value,
+    string_value,
+)
+from tests.integration._support.process import UpstreamSlot, owned_proxy, owned_upstream
+from tests.integration._support.upstream import ScenarioHandle, delete_scenario, register_scenario
+from tests.integration.cost_calculation.cost_tracking_case import RealtimeResponse
+
+pytestmark: Final = pytest.mark.timeout(240)
+
+REALTIME_MODEL: Final = "gpt-realtime"
+RESPONSES_MODEL: Final = "gpt-5.6"
+API_VERSION: Final = "2025-04-01-preview"
+PLAINTEXT_SCHEMES: Final = ("http", "ws")
+AZURE_REALTIME_PATHS: Final = frozenset({"/openai/realtime", "/openai/v1/realtime"})
+OPENAI_REALTIME_PATH: Final = "/v1/realtime"
+RESPONSES_DEPLOYMENTS: Final = (("openai", "/v1", "/v1/responses"), ("azure", "", "/openai/v1/responses"))
+RESPONSES_TERMINAL: Final = frozenset({"response.completed", "response.failed", "error"})
+COMPLETED_REALTIME_TURN: Final = ("session.created", "response.done")
+COMPLETED_RESPONSES_TURN: Final = ("response.created", "response.output_text.delta", "response.completed")
+CONVERGENCE_SECONDS: Final = 30
+REALTIME_USAGE: Final[dict[str, JsonValue]] = {
+    "total_tokens": 12,
+    "input_tokens": 7,
+    "output_tokens": 5,
+    "input_token_details": {"text_tokens": 7, "audio_tokens": 0, "cached_tokens": 0},
+    "output_token_details": {"text_tokens": 5, "audio_tokens": 0},
+}
+RESPONSES_USAGE: Final[dict[str, JsonValue]] = {"input_tokens": 5, "output_tokens": 2, "total_tokens": 7}
+
+
+def _realtime_turn() -> RealtimeResponse:
+    return RealtimeResponse(
+        content_type="application/x-realtime",
+        session_model=REALTIME_MODEL,
+        events=(
+            {
+                "type": "response.done",
+                "event_id": "evt_$UNIQUE_ID",
+                "response": {
+                    "id": "resp_$REQUEST_ID",
+                    "object": "realtime.response",
+                    "status": "completed",
+                    "output": [],
+                    "usage": REALTIME_USAGE,
+                },
+            },
+        ),
+    )
+
+
+def _responses_turn() -> RealtimeResponse:
+    response: Final[dict[str, JsonValue]] = {
+        "id": "resp_$REQUEST_ID",
+        "object": "response",
+        "created_at": 1700000000,
+        "model": RESPONSES_MODEL,
+    }
+    message: Final[dict[str, JsonValue]] = {
+        "type": "message",
+        "id": "msg_$REQUEST_ID",
+        "status": "completed",
+        "role": "assistant",
+        "content": [{"type": "output_text", "text": "scripted $REQUEST_ID", "annotations": []}],
+    }
+    return RealtimeResponse(
+        content_type="application/x-realtime",
+        events=(
+            {"type": "response.created", "response": {**response, "status": "in_progress", "output": []}},
+            {
+                "type": "response.output_text.delta",
+                "item_id": "msg_$REQUEST_ID",
+                "output_index": 0,
+                "content_index": 0,
+                "delta": "scripted $REQUEST_ID",
+            },
+            {
+                "type": "response.completed",
+                "response": {**response, "status": "completed", "output": [message], "usage": RESPONSES_USAGE},
+            },
+        ),
+    )
+
+
+def _scripted(scenario: Scenario, response: RealtimeResponse, *, control_url: str) -> ScenarioHandle:
+    handle: Final = register_scenario(f"ws-dial-{uuid.uuid4().hex[:12]}", response, control_url=control_url)
+    scenario.cleanups.callback(delete_scenario, handle)
+    return handle
+
+
+def _ws_base(http_url: str) -> str:
+    return http_url.replace("https://", "wss://").replace("http://", "ws://")
+
+
+def _with_scheme(http_url: str, scheme: str) -> str:
+    return f"{scheme}://{http_url.removeprefix('http://')}"
+
+
+def _proxy_ws(gateway: Gateway) -> str:
+    return _ws_base(str(gateway.client.base_url).rstrip("/"))
+
+
+def _upstream_url(gateway: Gateway) -> str:
+    return gateway.upstream_url.rstrip("/")
+
+
+def _realtime_deployment(scenario: Scenario, provider: str, scenario_id: str, api_base: str) -> str:
+    return scenario.model(
+        model=f"{provider}/{REALTIME_MODEL}",
+        api_key=scenario_id,
+        api_base=api_base,
+        api_version=API_VERSION,
+        model_info={"mode": "realtime"},
+    )
+
+
+def _responses_deployment(scenario: Scenario, provider: str, scenario_id: str, api_base: str) -> str:
+    return scenario.model(
+        model=f"{provider}/{RESPONSES_MODEL}",
+        api_key=scenario_id,
+        api_base=api_base,
+        **({"api_version": API_VERSION} if provider == "azure" else {}),
+    )
+
+
+def _close_code(closed: ConnectionClosed) -> int:
+    return 1006 if closed.rcvd is None else closed.rcvd.code
+
+
+async def _realtime_frames(socket: ClientConnection) -> AsyncIterator[dict[str, JsonValue]]:
+    try:
+        first: Final = JSON_OBJECT.validate_json(await socket.recv())
+        yield first
+        if first.get("type") != "session.created":
+            async for message in socket:
+                yield JSON_OBJECT.validate_json(message)
+            return
+        await socket.send(json.dumps({"type": "response.create"}))
+        async for message in socket:
+            event: Final = JSON_OBJECT.validate_json(message)
+            yield event
+            if event.get("type") == "response.done":
+                break
+    except ConnectionClosed as closed:
+        yield {"type": "closed", "code": _close_code(closed)}
+
+
+def _create_frame(model: str) -> str:
+    return json.dumps(
+        {
+            "type": "response.create",
+            "model": model,
+            "input": [{"type": "message", "role": "user", "content": [{"type": "input_text", "text": "say it"}]}],
+        }
+    )
+
+
+async def _responses_frames(socket: ClientConnection, model: str) -> AsyncIterator[dict[str, JsonValue]]:
+    try:
+        await socket.send(_create_frame(model))
+        async for message in socket:
+            event: Final = JSON_OBJECT.validate_json(message)
+            yield event
+            if event.get("type") in RESPONSES_TERMINAL:
+                break
+    except ConnectionClosed as closed:
+        yield {"type": "closed", "code": _close_code(closed)}
+
+
+async def _collect(frames: AsyncIterator[dict[str, JsonValue]]) -> tuple[dict[str, JsonValue], ...]:
+    return tuple([frame async for frame in frames])
+
+
+async def _realtime_session(ws_base: str, model: str, key: str) -> tuple[dict[str, JsonValue], ...]:
+    headers: Final = {"Authorization": f"Bearer {key}"}
+    async with websockets.connect(f"{ws_base}/v1/realtime?model={model}", additional_headers=headers) as socket:
+        return await asyncio.wait_for(_collect(_realtime_frames(socket)), 60)
+
+
+async def _responses_session(ws_base: str, model: str, key: str) -> tuple[dict[str, JsonValue], ...]:
+    headers: Final = {"Authorization": f"Bearer {key}"}
+    async with websockets.connect(f"{ws_base}/v1/responses?model={model}", additional_headers=headers) as socket:
+        return await asyncio.wait_for(_collect(_responses_frames(socket, model)), 60)
+
+
+def _types(events: tuple[dict[str, JsonValue], ...]) -> tuple[str, ...]:
+    return tuple(string_value(event["type"]) for event in events)
+
+
+def _observations(control_url: str) -> tuple[dict[str, JsonValue], ...]:
+    verify: Final = not control_url.startswith("https://")
+    with httpx.Client(base_url=control_url, timeout=5, trust_env=False, verify=verify) as upstream:
+        return tuple(map(object_value, upstream.get("/__observations").json()["requests"]))
+
+
+def _dialed_by(request: dict[str, JsonValue], scenario_id: str) -> bool:
+    bearer: Final = string_value(request["authorization"]).removeprefix("Bearer ")
+    return request["method"] == "WEBSOCKET" and scenario_id in (bearer, request["api_key"])
+
+
+def _upgrade_paths(control_url: str, scenario_id: str) -> tuple[str, ...]:
+    observed: Final = _observations(control_url)
+    return tuple(string_value(request["path"]) for request in observed if _dialed_by(request, scenario_id))
+
+
+def _assert_azure_realtime_upgrade(control_url: str, scenario_id: str) -> None:
+    paths: Final = _upgrade_paths(control_url, scenario_id)
+    assert len(paths) == 1 and paths[0] in AZURE_REALTIME_PATHS, paths
+
+
+def _health(gateway: Gateway, model: str) -> dict[str, JsonValue]:
+    response: Final = gateway.request("GET", "/health", params={"model": model})
+    return JSON_OBJECT.validate_json(response.content)
+
+
+def _probed(body: dict[str, JsonValue]) -> int:
+    counts: Final = (body.get("healthy_count", 0), body.get("unhealthy_count", 0))
+    return sum(count for count in counts if isinstance(count, int))
+
+
+def _converged_health(gateway: Gateway, model: str) -> dict[str, JsonValue]:
+    return eventually(lambda: _health(gateway, model), lambda body: _probed(body) == 1, seconds=CONVERGENCE_SECONDS)
+
+
+@pytest.mark.parametrize("scheme", PLAINTEXT_SCHEMES)
+def test_azure_realtime_session_over_a_plaintext_api_base(gateway: Gateway, scheme: str) -> None:
+    with gateway.scenario() as scenario:
+        handle: Final = _scripted(scenario, _realtime_turn(), control_url=_upstream_url(gateway))
+        api_base: Final = _with_scheme(_upstream_url(gateway), scheme)
+        model: Final = _realtime_deployment(scenario, "azure", handle.scenario_id, api_base)
+        key: Final = scenario.key(models=[model])
+        events: Final = asyncio.run(_realtime_session(_proxy_ws(gateway), model, key))
+        assert _types(events) == COMPLETED_REALTIME_TURN, events
+        _assert_azure_realtime_upgrade(_upstream_url(gateway), handle.scenario_id)
+
+
+@pytest.mark.parametrize("scheme", PLAINTEXT_SCHEMES)
+def test_azure_realtime_health_check_over_a_plaintext_api_base(gateway: Gateway, scheme: str) -> None:
+    with gateway.scenario() as scenario:
+        handle: Final = _scripted(scenario, _realtime_turn(), control_url=_upstream_url(gateway))
+        api_base: Final = _with_scheme(_upstream_url(gateway), scheme)
+        model: Final = _realtime_deployment(scenario, "azure", handle.scenario_id, api_base)
+        health: Final = _converged_health(gateway, model)
+        assert (health["healthy_count"], health["unhealthy_count"]) == (1, 0), health
+        _assert_azure_realtime_upgrade(_upstream_url(gateway), handle.scenario_id)
+
+
+@pytest.mark.parametrize(("provider", "suffix", "path"), RESPONSES_DEPLOYMENTS, ids=["openai", "azure"])
+def test_responses_websocket_session_over_a_plaintext_api_base(
+    gateway: Gateway, provider: str, suffix: str, path: str
+) -> None:
+    with gateway.scenario() as scenario:
+        handle: Final = _scripted(scenario, _responses_turn(), control_url=_upstream_url(gateway))
+        model: Final = _responses_deployment(scenario, provider, handle.scenario_id, _upstream_url(gateway) + suffix)
+        key: Final = scenario.key(models=[model])
+        events: Final = asyncio.run(_responses_session(_proxy_ws(gateway), model, key))
+        assert _types(events) == COMPLETED_RESPONSES_TURN, events
+        assert _upgrade_paths(_upstream_url(gateway), handle.scenario_id) == (path,)
+
+
+def _unverified_config(directory: Path) -> Path:
+    config: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text())
+    unverified: Final = {**config, "litellm_settings": {**config["litellm_settings"], "ssl_verify": False}}
+    path: Final = directory / "ssl-verify-off.yaml"
+    path.write_text(yaml.safe_dump(unverified))
+    return path
+
+
+@pytest.fixture(scope="module")
+def tls_upstream(tmp_path_factory: pytest.TempPathFactory) -> Iterator[UpstreamSlot]:
+    with owned_upstream(tmp_path_factory.mktemp("ws-dial-tls-upstream"), tls=True) as slot:
+        yield slot
+
+
+@pytest.fixture(scope="module")
+def unverified_proxy(tmp_path_factory: pytest.TempPathFactory) -> Iterator[Gateway]:
+    directory: Final = tmp_path_factory.mktemp("ws-dial-unverified-proxy")
+    with (
+        gateway_from_environment() as base,
+        owned_proxy(base, directory, {}, config=_unverified_config(directory)) as proxy,
+    ):
+        yield proxy
+
+
+@pytest.mark.parametrize("provider", ["azure", "openai"])
+def test_realtime_session_with_verification_off_over_a_self_signed_upstream(
+    unverified_proxy: Gateway, tls_upstream: UpstreamSlot, provider: str
+) -> None:
+    with unverified_proxy.scenario() as scenario:
+        handle: Final = _scripted(scenario, _realtime_turn(), control_url=tls_upstream.url)
+        model: Final = _realtime_deployment(scenario, provider, handle.scenario_id, tls_upstream.url)
+        key: Final = scenario.key(models=[model])
+        events: Final = asyncio.run(_realtime_session(_proxy_ws(unverified_proxy), model, key))
+        assert _types(events) == COMPLETED_REALTIME_TURN, events
+        paths: Final = _upgrade_paths(tls_upstream.url, handle.scenario_id)
+        assert len(paths) == 1 and paths[0] in (AZURE_REALTIME_PATHS | {OPENAI_REALTIME_PATH}), paths
+
+
+def test_responses_websocket_session_with_verification_off_over_a_self_signed_upstream(
+    unverified_proxy: Gateway, tls_upstream: UpstreamSlot
+) -> None:
+    with unverified_proxy.scenario() as scenario:
+        handle: Final = _scripted(scenario, _responses_turn(), control_url=tls_upstream.url)
+        model: Final = _responses_deployment(scenario, "openai", handle.scenario_id, f"{tls_upstream.url}/v1")
+        key: Final = scenario.key(models=[model])
+        events: Final = asyncio.run(_responses_session(_proxy_ws(unverified_proxy), model, key))
+        assert _types(events) == COMPLETED_RESPONSES_TURN, events
+        assert _upgrade_paths(tls_upstream.url, handle.scenario_id) == ("/v1/responses",)

tests/unit/llms/azure/realtime/test_azure_realtime_handler.py

@@ -740,3 +740,55 @@ async def test_arealtime_forwards_deployment_azure_ad_token(monkeypatch, no_ambi
     )
 
     assert mock_async_realtime.call_args.kwargs["azure_ad_token"] == "deployment-entra-token"
+
+
+@pytest.mark.asyncio
+async def test_construct_url_maps_http_to_ws():
+    from litellm.llms.azure.realtime.handler import AzureOpenAIRealtime
+
+    url = AzureOpenAIRealtime()._construct_url("http://127.0.0.1:8113", "gpt-realtime", "2024-10-01-preview")
+
+    assert url.startswith("ws://127.0.0.1:8113/openai/realtime?")
+
+
+async def _dial_through_mocked_connect(api_base: str) -> tuple[str, dict]:
+    from litellm.llms.azure.realtime.handler import AzureOpenAIRealtime
+
+    with (
+        patch("websockets.connect", return_value=_DummyAsyncContextManager(AsyncMock())) as mock_ws_connect,
+        patch("litellm.llms.azure.realtime.handler.RealTimeStreaming") as mock_realtime_streaming,
+    ):
+        mock_realtime_streaming.return_value.bidirectional_forward = AsyncMock()
+        await AzureOpenAIRealtime().async_realtime(
+            model="gpt-realtime",
+            websocket=AsyncMock(),
+            logging_obj=MagicMock(),
+            api_base=api_base,
+            api_key="test-key",
+            api_version="2024-10-01-preview",
+        )
+    mock_ws_connect.assert_called_once()
+    return mock_ws_connect.call_args[0][0], mock_ws_connect.call_args[1]
+
+
+@pytest.mark.parametrize("api_base", ["http://127.0.0.1:8113", "ws://127.0.0.1:8113"])
+@pytest.mark.asyncio
+async def test_async_realtime_plaintext_api_base_dials_ws_without_ssl(api_base):
+    url, kwargs = await _dial_through_mocked_connect(api_base)
+
+    assert url.startswith("ws://127.0.0.1:8113/openai/realtime?")
+    assert kwargs["ssl"] is None
+
+
+@pytest.mark.asyncio
+async def test_async_realtime_wss_with_ssl_verify_off_dials_with_an_unverified_context(monkeypatch):
+    import ssl
+
+    monkeypatch.setattr("litellm.llms.custom_httpx.http_handler._shared_realtime_ssl_context", False)
+
+    url, kwargs = await _dial_through_mocked_connect("https://127.0.0.1:8113")
+
+    assert url.startswith("wss://127.0.0.1:8113/openai/realtime?")
+    assert isinstance(kwargs["ssl"], ssl.SSLContext)
+    assert kwargs["ssl"].verify_mode is ssl.CERT_NONE
+    assert kwargs["ssl"].check_hostname is False

tests/unit/llms/openai/realtime/test_openai_realtime_handler.py

@@ -526,3 +526,43 @@ async def test_realtime_query_params_preserve_missing_model(monkeypatch):
 
     called_kwargs = mock_async_realtime.call_args.kwargs
     assert called_kwargs["query_params"] == {"intent": "transcription"}
+
+
+@pytest.mark.asyncio
+async def test_async_realtime_wss_with_ssl_verify_off_dials_with_an_unverified_context(monkeypatch):
+    import ssl
+
+    from litellm.llms.openai.realtime.handler import OpenAIRealtime
+
+    monkeypatch.setattr("litellm.llms.custom_httpx.http_handler._shared_realtime_ssl_context", False)
+
+    class DummyAsyncContextManager:
+        def __init__(self, value):
+            self.value = value
+
+        async def __aenter__(self):
+            return self.value
+
+        async def __aexit__(self, exc_type, exc, tb):
+            return None
+
+    with (
+        patch("websockets.connect", return_value=DummyAsyncContextManager(AsyncMock())) as mock_ws_connect,
+        patch("litellm.llms.openai.realtime.handler.RealTimeStreaming") as mock_realtime_streaming,
+    ):
+        mock_realtime_streaming.return_value.bidirectional_forward = AsyncMock()
+        await OpenAIRealtime().async_realtime(
+            model="gpt-realtime",
+            websocket=AsyncMock(),
+            logging_obj=MagicMock(),
+            api_base="https://127.0.0.1:8113",
+            api_key="test-key",
+            query_params={"model": "gpt-realtime"},
+        )
+
+    mock_ws_connect.assert_called_once()
+    assert mock_ws_connect.call_args[0][0].startswith("wss://127.0.0.1:8113/v1/realtime?")
+    dialed_ssl = mock_ws_connect.call_args[1]["ssl"]
+    assert isinstance(dialed_ssl, ssl.SSLContext)
+    assert dialed_ssl.verify_mode is ssl.CERT_NONE
+    assert dialed_ssl.check_hostname is False

tests/unit/responses/test_responses_websocket_all_providers.py

@@ -3159,6 +3159,75 @@ def _wrapped_reasoning_item():
     }
 
 
+class TestNativeWebSocketDialSsl:
+    """The backend dial's ssl argument follows the resolved URL scheme: none for ws://, the shared
+    context for wss://, and an unverified context for wss:// when ssl_verify is off."""
+
+    @staticmethod
+    async def _dial(ws_url: str) -> dict:
+        from unittest.mock import AsyncMock, MagicMock, patch
+
+        from litellm.llms.custom_httpx.llm_http_handler import BaseLLMHTTPHandler
+
+        captured = {}
+
+        class FakeConnect:
+            def __init__(self, url, **kwargs):
+                captured.update(kwargs)
+
+            async def __aenter__(self):
+                raise Exception("stop")
+
+            async def __aexit__(self, *args):
+                pass
+
+        mock_config = MagicMock(spec=OpenAIResponsesAPIConfig)
+        mock_config.supports_native_websocket.return_value = True
+        mock_config.get_websocket_url.return_value = ws_url
+        mock_config.validate_environment.return_value = {}
+        mock_logging = MagicMock()
+        mock_logging.pre_call = MagicMock()
+        mock_ws = MagicMock()
+        mock_ws.close = AsyncMock()
+
+        with patch("websockets.connect", FakeConnect):
+            await BaseLLMHTTPHandler().async_responses_websocket(
+                model="gpt-5.6",
+                websocket=mock_ws,
+                logging_obj=mock_logging,
+                responses_api_provider_config=mock_config,
+                api_key="sk-test",
+            )
+        assert "ssl" in captured, "the backend dial never happened"
+        return captured
+
+    @pytest.mark.asyncio
+    async def test_ws_url_dials_without_ssl(self):
+        captured = await self._dial("ws://127.0.0.1:8113/v1/responses")
+
+        assert captured["ssl"] is None
+
+    @pytest.mark.asyncio
+    async def test_wss_url_dials_with_the_shared_context(self):
+        from litellm.llms.custom_httpx.http_handler import get_shared_realtime_ssl_context
+
+        captured = await self._dial("wss://127.0.0.1:8113/v1/responses")
+
+        assert captured["ssl"] is get_shared_realtime_ssl_context()
+
+    @pytest.mark.asyncio
+    async def test_wss_url_with_ssl_verify_off_dials_with_an_unverified_context(self, monkeypatch):
+        import ssl
+
+        monkeypatch.setattr("litellm.llms.custom_httpx.http_handler._shared_realtime_ssl_context", False)
+
+        captured = await self._dial("wss://127.0.0.1:8113/v1/responses")
+
+        assert isinstance(captured["ssl"], ssl.SSLContext)
+        assert captured["ssl"].verify_mode is ssl.CERT_NONE
+        assert captured["ssl"].check_hostname is False
+
+
 class TestNativeWebSocketEncryptedContentAffinity:
 
     @pytest.mark.asyncio