This commit is contained in:
A Sai Charan 2026-10-19 21:11:30 +00:00 • committed by GitHub
commit f7b9ae3788
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
7 changed files with 287 additions and 8 deletions

View file

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

View file

@ -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,
)

View file

@ -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:

View file

@ -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,
)

View file

View file

@ -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,
)

View file

@ -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(),