feat(realtime): support Azure GPT-Live WebSockets

litellm/llms/azure/live/__init__.py

@@ -0,0 +1,3 @@
+from .handler import AzureOpenAILiveSessions
+
+__all__ = ("AzureOpenAILiveSessions",)

litellm/llms/azure/live/handler.py

@@ -0,0 +1,66 @@
+from collections.abc import Mapping
+from types import MappingProxyType
+from typing import Final, Protocol
+from urllib.parse import urlsplit, urlunsplit
+
+from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLogging
+from litellm.llms.openai.live.handler import LiveClientWebSocket, OpenAILiveSessions
+
+from ..realtime.handler import AzureOpenAIRealtime
+
+_LIVE_WEBSOCKET_SCHEMES: Final = MappingProxyType({"https": "wss", "http": "ws"})
+
+
+class LiveSessionsRelay(Protocol):
+    async def async_live_session(
+        self,
+        *,
+        model: str,
+        websocket: LiveClientWebSocket,
+        logging_obj: LiteLLMLogging,
+        session_start: Mapping[str, object],
+        api_base: str | None,
+        api_key: str | None,
+        upstream_url: str | None = None,
+        additional_headers: Mapping[str, str] | None = None,
+    ) -> None: ...
+
+
+class AzureOpenAILiveSessions:
+    def __init__(self, live_sessions: LiveSessionsRelay | None = None) -> None:
+        self._live_sessions = live_sessions or OpenAILiveSessions()
+
+    @staticmethod
+    def construct_live_url(api_base: str) -> str:
+        parsed_url: Final = urlsplit(api_base)
+        websocket_scheme: Final = _LIVE_WEBSOCKET_SCHEMES.get(parsed_url.scheme, parsed_url.scheme)
+        return urlunsplit((websocket_scheme, parsed_url.netloc, "/openai/v1/live/sessions", "", ""))
+
+    async def async_live_session(
+        self,
+        *,
+        model: str,
+        websocket: LiveClientWebSocket,
+        logging_obj: LiteLLMLogging,
+        session_start: Mapping[str, object],
+        api_base: str | None,
+        api_key: str | None,
+        azure_ad_token: str | None,
+    ) -> None:
+        if api_base is None:
+            raise ValueError("api_base is required for Azure OpenAI Live session calls")
+        auth_headers: Final = AzureOpenAIRealtime.get_auth_headers(
+            api_key=api_key,
+            azure_ad_token=azure_ad_token,
+        )
+        credential: Final = api_key or azure_ad_token
+        await self._live_sessions.async_live_session(
+            model=model,
+            websocket=websocket,
+            logging_obj=logging_obj,
+            session_start=session_start,
+            api_base=api_base,
+            api_key=credential,
+            upstream_url=self.construct_live_url(api_base),
+            additional_headers=auth_headers,
+        )

litellm/llms/openai/live/handler.py

@@ -411,14 +411,16 @@ class OpenAILiveSessions(OpenAIRealtime):
         session_start: Mapping[str, object],
         api_base: str | None,
         api_key: str | None,
+        upstream_url: str | None = None,
+        additional_headers: Mapping[str, str] | None = None,
     ) -> None:
         if api_key is None:
-            raise ValueError("api_key is required for OpenAI Live session calls")
+            raise ValueError("api_key is required for Live session calls")
         from websockets.exceptions import InvalidStatus
 
         resolved_api_base: Final = api_base or self._get_default_api_base()
-        url: Final = self._construct_live_url(resolved_api_base)
-        headers: Final = self._get_additional_headers(api_key)
+        url: Final = upstream_url or self._construct_live_url(resolved_api_base)
+        headers: Final = additional_headers or self._get_additional_headers(api_key)
         ssl_config: Final = self._get_ssl_config(url)
         session: Final = as_str_mapping(session_start.get("session"))
         if session is None:

litellm/realtime_api/main.py

@@ -39,6 +39,7 @@ 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 ..llms.azure.common_utils import get_azure_ad_token
+from ..llms.azure.live.handler import AzureOpenAILiveSessions
 from ..llms.azure.realtime.handler import AzureOpenAIRealtime, azure_realtime_protocol_for_client
 from ..llms.bedrock.realtime.handler import BedrockRealtime, raise_if_session_committed
 from ..llms.custom_httpx.http_handler import realtime_ssl_for_url
@@ -59,6 +60,7 @@ if TYPE_CHECKING:
 azure_realtime: Final = AzureOpenAIRealtime()
 openai_realtime: Final = OpenAIRealtime()
 openai_live_sessions: Final = OpenAILiveSessions()
+azure_live_sessions: Final = AzureOpenAILiveSessions()
 bedrock_realtime: Final = BedrockRealtime()
 xai_realtime: Final = XAIRealtime()
 vertex_llm_base: Final = VertexBase()
@@ -409,7 +411,7 @@ _LiveSessionRejection = _LiveProviderUnsupported | _LiveGuardrailsUnsupported
 def _live_session_rejection(
     custom_llm_provider: str, litellm_metadata: Mapping[str, object]
 ) -> _LiveSessionRejection | None:
-    if custom_llm_provider != "openai":
+    if custom_llm_provider != "openai" and custom_llm_provider not in AZURE_OPENAI_AUDIO_PROVIDERS:
         return _LiveProviderUnsupported(provider=custom_llm_provider)
     if _realtime_guardrail_configured(litellm_metadata):
         return _LiveGuardrailsUnsupported()
@@ -419,9 +421,9 @@ def _live_session_rejection(
 def _raise_live_session_rejection(rejection: _LiveSessionRejection) -> NoReturn:
     match rejection:
         case _LiveProviderUnsupported(provider=provider):
-            raise ValueError(f"OpenAI Live sessions require the openai provider, got {provider}")
+            raise ValueError(f"Live sessions require the openai or azure provider, got {provider}")
         case _LiveGuardrailsUnsupported():
-            raise ValueError("Guardrails are not supported on OpenAI Live sessions")
+            raise ValueError("Guardrails are not supported on Live sessions")
         case _:
             assert_never(rejection)
 
@@ -434,12 +436,30 @@ async def _alive_session(
     session_start: Mapping[str, object],
     api_base: str | None,
     api_key: str | None,
+    azure_ad_token: str | None,
     litellm_metadata: Mapping[str, object],
     litellm_params: Mapping[str, object],
 ) -> None:
     rejection: Final = _live_session_rejection(custom_llm_provider, litellm_metadata)
     if rejection is not None:
         _raise_live_session_rejection(rejection)
+    if custom_llm_provider in AZURE_OPENAI_AUDIO_PROVIDERS:
+        azure_api_base: Final = api_base or litellm.api_base or get_secret_str("AZURE_API_BASE")
+        resolved_api_key: Final = api_key or litellm.api_key or get_secret_str("AZURE_API_KEY")
+        await azure_live_sessions.async_live_session(
+            model=model,
+            websocket=websocket,
+            logging_obj=litellm_logging_obj,
+            session_start=session_start,
+            api_base=azure_api_base,
+            api_key=resolved_api_key,
+            azure_ad_token=(
+                None
+                if resolved_api_key
+                else get_azure_ad_token(GenericLiteLLMParams(**litellm_params, azure_ad_token=azure_ad_token))
+            ),
+        )
+        return
     resolved_api_base: Final = api_base or litellm.api_base or "https://api.openai.com/"
     await openai_live_sessions.async_live_session(
         model=model,
@@ -518,6 +538,7 @@ async def _arealtime(
             session_start=live_session_start,
             api_base=dynamic_api_base or litellm_params.api_base,
             api_key=dynamic_api_key,
+            azure_ad_token=azure_ad_token,
             litellm_metadata=_build_litellm_metadata(kwargs),
             litellm_params=kwargs,
         )

tests/unit/llms/azure/live/test_handler.py

@@ -0,0 +1,141 @@
+from collections.abc import Mapping
+from types import MappingProxyType
+from typing import Final
+from unittest.mock import MagicMock
+
+import pytest
+
+from litellm.litellm_core_utils.litellm_logging import Logging
+from litellm.llms.azure.live.handler import AzureOpenAILiveSessions
+from litellm.llms.openai.live.handler import LiveClientWebSocket
+
+
+class _RecordingLiveSessions:
+    def __init__(self) -> None:
+        self.kwargs: Mapping[str, object] | None = None
+
+    async def async_live_session(
+        self,
+        *,
+        model: str,
+        websocket: LiveClientWebSocket,
+        logging_obj: Logging,
+        session_start: Mapping[str, object],
+        api_base: str | None,
+        api_key: str | None,
+        upstream_url: str | None = None,
+        additional_headers: Mapping[str, str] | None = None,
+    ) -> None:
+        self.kwargs = MappingProxyType(
+            {
+                "model": model,
+                "websocket": websocket,
+                "logging_obj": logging_obj,
+                "session_start": session_start,
+                "api_base": api_base,
+                "api_key": api_key,
+                "upstream_url": upstream_url,
+                "additional_headers": additional_headers,
+            }
+        )
+
+
+@pytest.mark.parametrize(
+    ("api_base", "expected_url"),
+    [
+        (
+            "https://voice-resource.openai.azure.com",
+            "wss://voice-resource.openai.azure.com/openai/v1/live/sessions",
+        ),
+        (
+            "https://voice-resource.services.ai.azure.com/openai/deployments/ignored?api-version=preview",
+            "wss://voice-resource.services.ai.azure.com/openai/v1/live/sessions",
+        ),
+        (
+            "http://localhost:8080/azure",
+            "ws://localhost:8080/openai/v1/live/sessions",
+        ),
+    ],
+)
+def test_construct_live_url_uses_azure_ga_path(api_base: str, expected_url: str) -> None:
+    assert AzureOpenAILiveSessions.construct_live_url(api_base) == expected_url
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("api_key", "azure_ad_token", "expected_headers", "expected_credential"),
+    [
+        ("azure-key", None, {"api-key": "azure-key"}, "azure-key"),
+        (None, "entra-token", {"Authorization": "Bearer entra-token"}, "entra-token"),
+    ],
+)
+async def test_async_live_session_uses_azure_endpoint_and_authentication(
+    api_key: str | None,
+    azure_ad_token: str | None,
+    expected_headers: Mapping[str, str],
+    expected_credential: str,
+) -> None:
+    relay: Final = _RecordingLiveSessions()
+    handler: Final = AzureOpenAILiveSessions(relay)
+    websocket: Final = MagicMock(spec=LiveClientWebSocket)
+    logging_obj: Final = MagicMock(spec=Logging)
+    session_start: Final[Mapping[str, object]] = MappingProxyType(
+        {
+            "type": "session.start",
+            "session": {"model": "azure-live-alias"},
+        }
+    )
+
+    await handler.async_live_session(
+        model="gpt-live-1-deployment",
+        websocket=websocket,
+        logging_obj=logging_obj,
+        session_start=session_start,
+        api_base="https://voice-resource.openai.azure.com",
+        api_key=api_key,
+        azure_ad_token=azure_ad_token,
+    )
+
+    assert relay.kwargs == {
+        "model": "gpt-live-1-deployment",
+        "websocket": websocket,
+        "logging_obj": logging_obj,
+        "session_start": session_start,
+        "api_base": "https://voice-resource.openai.azure.com",
+        "api_key": expected_credential,
+        "upstream_url": "wss://voice-resource.openai.azure.com/openai/v1/live/sessions",
+        "additional_headers": expected_headers,
+    }
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("api_base", "api_key", "azure_ad_token", "expected_message"),
+    [
+        (None, "azure-key", None, "api_base is required for Azure OpenAI Live session calls"),
+        (
+            "https://voice-resource.openai.azure.com",
+            None,
+            None,
+            "Missing Azure credentials for the realtime endpoint",
+        ),
+    ],
+)
+async def test_async_live_session_rejects_missing_configuration(
+    api_base: str | None,
+    api_key: str | None,
+    azure_ad_token: str | None,
+    expected_message: str,
+) -> None:
+    handler: Final = AzureOpenAILiveSessions(_RecordingLiveSessions())
+
+    with pytest.raises(ValueError, match=expected_message):
+        await handler.async_live_session(
+            model="gpt-live-1-deployment",
+            websocket=MagicMock(spec=LiveClientWebSocket),
+            logging_obj=MagicMock(spec=Logging),
+            session_start=MappingProxyType({"type": "session.start"}),
+            api_base=api_base,
+            api_key=api_key,
+            azure_ad_token=azure_ad_token,
+        )

tests/unit/realtime_api/test_main.py

@@ -290,6 +290,52 @@ async def test_arealtime_live_session_dispatches_to_openai_live_handler(monkeypa
     assert captured.values["api_key"] == "resolved-key"
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("provider", ["azure", "azure_ai"])
+async def test_arealtime_live_session_dispatches_to_azure_live_handler(
+    provider: str, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    captured: Final = _CallCapture()
+
+    def mock_get_llm_provider(
+        model: str,
+        api_base: str | None,
+        api_key: str | None,
+        custom_llm_provider: str | None = None,
+    ) -> tuple[str, str, str | None, str | None]:
+        return "gpt-live-deployment", provider, "azure-key", "https://voice-resource.openai.azure.com"
+
+    class FakeAzureLiveHandler:
+        async def async_live_session(self, **kwargs: object) -> None:
+            captured.values = kwargs
+
+    monkeypatch.setattr(realtime_main, "get_llm_provider", mock_get_llm_provider)
+    monkeypatch.setattr(realtime_main, "azure_live_sessions", FakeAzureLiveHandler())
+    start_frame: Final[dict[str, object]] = {
+        "type": "session.start",
+        "session": {"model": "azure-live-alias", "instructions": "Stay brief"},
+    }
+    websocket: Final = MagicMock()
+    logging_obj: Final = FakeLogging()
+
+    await realtime_main._arealtime.__wrapped__(
+        model="azure-live-alias",
+        websocket=websocket,
+        litellm_logging_obj=logging_obj,
+        live_session_start=start_frame,
+    )
+
+    assert captured.values == {
+        "model": "gpt-live-deployment",
+        "websocket": websocket,
+        "logging_obj": logging_obj,
+        "session_start": start_frame,
+        "api_base": "https://voice-resource.openai.azure.com",
+        "api_key": "azure-key",
+        "azure_ad_token": None,
+    }
+
+
 @pytest.mark.asyncio
 async def test_arealtime_live_session_rejects_matching_guardrail_before_dispatch(
     monkeypatch: pytest.MonkeyPatch,
@@ -329,7 +375,7 @@ async def test_arealtime_live_session_rejects_matching_guardrail_before_dispatch
     monkeypatch.setattr(realtime_main, "openai_live_sessions", live_handler)
     monkeypatch.setattr(litellm, "callbacks", [guardrail])
 
-    with pytest.raises(ValueError, match="Guardrails are not supported on OpenAI Live sessions"):
+    with pytest.raises(ValueError, match="Guardrails are not supported on Live sessions"):
         await realtime_main._arealtime.__wrapped__(
             model="live-alias",
             websocket=MagicMock(),
@@ -373,7 +419,7 @@ async def test_arealtime_live_session_rejects_streaming_only_pre_call_guardrail(
     monkeypatch.setattr(realtime_main, "openai_live_sessions", live_handler)
     monkeypatch.setattr(litellm, "callbacks", [guardrail])
 
-    with pytest.raises(ValueError, match="Guardrails are not supported on OpenAI Live sessions"):
+    with pytest.raises(ValueError, match="Guardrails are not supported on Live sessions"):
         await realtime_main._arealtime.__wrapped__(
             model="live-alias",
             websocket=MagicMock(),