scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-29 06:33:28 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 267

fix(realtime): reconcile audio protocols billing and current CI

Browse source
This commit is contained in:
Emerson Gomes 2026-10-06 09:24:20 -05:00
parent 913c5dd250
commit 515f6da598
No known key found for this signature in database
GPG key ID: D3DF28AB5D1B5E17
18 changed files with 193 additions and 118 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
.github/workflows/test-unit.yml vendored
View file

@ -422,6 +422,7 @@ jobs:
tests/unit/ocr
tests/unit/passthrough
tests/unit/rag
tests/unit/cookbook
tests/unit/realtime_api
tests/unit/rerank_api
tests/unit/search

8
litellm/litellm_core_utils/audio_utils/transcription_streaming.py
View file

@ -72,13 +72,7 @@ class _TranscriptionEventCollector:
response: Final = TranscriptionResponse(
text=done_event.text if done_event is not None else "".join(self.text_deltas),
usage=done_event.usage.model_dump() if done_event is not None and done_event.usage is not None else None,
languages=(
[ # mutable-ok: the response model requires a concrete serialized language list
language.model_dump() for language in done_languages
]
if done_languages is not None
else None
),
languages=([language.model_dump() for language in done_languages] if done_languages is not None else None),
)
if self.duration is not None:
response.set_audio_transcription_duration(self.duration)

67
litellm/litellm_core_utils/realtime_streaming.py
View file

@ -8,7 +8,7 @@ from enum import Enum, auto
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final, NoReturn, Protocol, TypeAlias, TypedDict, cast, runtime_checkable
from pydantic import TypeAdapter
from pydantic import TypeAdapter, ValidationError
from typing_extensions import ReadOnly
import litellm
@ -207,6 +207,25 @@ def _gate_session_update(
return {**client_session, "turn_detection": gate(None)}
_TRANSLATION_AUDIO_FORMAT: Final = TypeAdapter(str | Mapping[str, object] | None)
def _translation_audio_bytes_per_second(audio_format: str | Mapping[str, object] | None) -> float | None:
if isinstance(audio_format, str):
return 8000.0 if audio_format in ("g711_ulaw", "g711_alaw") else None
if audio_format is None:
return None
format_type: Final = audio_format.get("type")
rate: Final = audio_format.get("rate")
if format_type in ("audio/pcmu", "audio/pcma") and rate is None:
return 8000.0
if not isinstance(rate, (int, float)) or rate <= 0:
return None
if format_type == "audio/pcm":
return float(rate) * 2
return float(rate) if format_type in ("audio/pcmu", "audio/pcma") else None
class RealtimeEventNormalizer(Protocol):
def should_drop(self, event: object) -> bool: ...
def normalize(self, event: dict) -> dict: ...
@ -261,13 +280,14 @@ class RealTimeStreaming:
self.tool_calls: list[dict] = []
self._is_translation_session = translation_session
self._translation_input_seconds = 0.0
self._translation_input_bytes_per_second = 48000.0
self._translation_output_audio_bytes = 0
self._translation_output_bytes_per_second = 48000.0
self._translation_usage_finalized = False
# Detect whether the client is explicitly opting into the beta protocol.
self._client_wants_beta = self.detect_beta_header(websocket)
self._backend_uses_beta_protocol = (
self._backend_uses_beta_protocol = not translation_session and (
self._client_wants_beta if backend_uses_beta_protocol is None else backend_uses_beta_protocol
)
@ -349,8 +369,8 @@ class RealTimeStreaming:
_CLIENT_AUDIO_BUFFER_COMMIT_TYPES = frozenset(["input_audio_buffer.commit", "input_audio_buffer.end"])
_AUDIO_FORMAT_MAP: dict[str, dict[str, str | int]] = {
"pcm16": {"type": "audio/pcm", "rate": 24000},
"g711_ulaw": {"type": "audio/G711-ulaw", "rate": 8000},
"g711_alaw": {"type": "audio/G711-alaw", "rate": 8000},
"g711_ulaw": {"type": "audio/pcmu"},
"g711_alaw": {"type": "audio/pcma"},
}
# GA name → beta name (when client WebSocket includes OpenAI-Beta: realtime=v1)
_GA_TO_BETA_EVENT_TYPES: dict[str, str] = {
@ -616,7 +636,7 @@ class RealTimeStreaming:
)
self._translation_usage_finalized = True
return
self._capture_translation_output_format(event_obj)
self._capture_translation_audio_formats(event_obj)
if event_obj.get("type") not in (
"session.output_audio.delta",
"response.output_audio.delta",
@ -639,35 +659,38 @@ class RealTimeStreaming:
event: Final = _decode_json_object(message)
except (json.JSONDecodeError, TypeError):
return
self._capture_translation_audio_formats(event)
if event.get("type") != "session.input_audio_buffer.append" or not isinstance(audio := event.get("audio"), str):
return
try:
decoded: Final = base64.b64decode(audio, validate=True)
except (ValueError, TypeError):
return
self._translation_input_seconds += len(decoded) / 48000.0
self._translation_input_seconds += len(decoded) / self._translation_input_bytes_per_second
def _capture_translation_output_format(self, event_obj: Mapping[str, object]) -> None:
def _capture_translation_audio_formats(self, event_obj: Mapping[str, object]) -> None:
session: Final = event_obj.get("session")
if not isinstance(session, dict):
return
audio: Final = session.get("audio")
output: Final = audio.get("output") if isinstance(audio, dict) else None
audio_format: Final = output.get("format") if isinstance(output, dict) else None
if isinstance(audio_format, str):
if audio_format in ("g711_ulaw", "g711_alaw"):
self._translation_output_bytes_per_second = 8000.0
if not isinstance(audio, dict):
return
if not isinstance(audio_format, dict):
return
format_type: Final = audio_format.get("type")
rate: Final = audio_format.get("rate")
if not isinstance(rate, (int, float)) or rate <= 0:
return
if format_type == "audio/pcm":
self._translation_output_bytes_per_second = float(rate) * 2
elif format_type in ("audio/pcmu", "audio/pcma"):
self._translation_output_bytes_per_second = float(rate)
for direction, configuration in (("input", audio.get("input")), ("output", audio.get("output"))):
if not isinstance(configuration, dict):
continue
try:
if (
bytes_per_second := _translation_audio_bytes_per_second(
_TRANSLATION_AUDIO_FORMAT.validate_python(configuration.get("format"))
)
) is None:
continue
except ValidationError:
continue
if direction == "input":
self._translation_input_bytes_per_second = bytes_per_second
else:
self._translation_output_bytes_per_second = bytes_per_second
def _finalize_translation_usage(self) -> None:
if self._translation_usage_finalized:

2
litellm/llms/azure/realtime/http_transformation.py
View file

@ -52,7 +52,7 @@ class AzureRealtimeHTTPConfig(BaseRealtimeHTTPConfig):
model: str,
api_key: str | None = None,
) -> dict:
validated_headers: Final = { # mutable-ok: provider authentication headers are extended before dispatch
validated_headers: Final = {
**headers,
"Content-Type": "application/json",
}

19
litellm/llms/custom_httpx/llm_http_handler.py
View file

@ -6865,7 +6865,7 @@ class BaseLLMHTTPHandler:
logging_obj.pre_call(
input=request_data,
api_key="",
additional_args={ # mutable-ok: logging owns a mutable request metadata payload
additional_args={
"complete_input_dict": request_data,
"api_base": normalized_api_base,
},
@ -6873,10 +6873,7 @@ class BaseLLMHTTPHandler:
try:
configured_client: Final = openai_client.with_options(
timeout=timeout,
set_default_headers={ # mutable-ok: OpenAI SDK accepts a mutable custom-header mapping
key: str(value)
for key, value in (extra_headers or {}).items() # mutable-ok: SDK requires concrete headers
},
set_default_headers={key: str(value) for key, value in (extra_headers or {}).items()},
)
raw_response: Final = await configured_client.post(
"/realtime/client_secrets",
@ -6948,7 +6945,7 @@ class BaseLLMHTTPHandler:
logging_obj.pre_call(
input=request_data,
api_key="",
additional_args={ # mutable-ok: logging owns a mutable request metadata payload
additional_args={
"complete_input_dict": request_data,
"api_base": normalized_api_base,
},
@ -7081,7 +7078,7 @@ class BaseLLMHTTPHandler:
logging_obj.pre_call(
input="realtime_sdp_offer",
api_key="",
additional_args={ # mutable-ok: logging owns a mutable request metadata payload
additional_args={
"api_base": normalized_api_base,
"session": session_data,
},
@ -7090,7 +7087,7 @@ class BaseLLMHTTPHandler:
if translation:
configured_client: Final = openai_client.with_options(
timeout=timeout,
set_default_headers={ # mutable-ok: OpenAI SDK accepts a mutable custom-header mapping
set_default_headers={
"Content-Type": "application/sdp",
**MappingProxyType(
{key: str(value) for key, value in (extra_headers or MappingProxyType({})).items()}
@ -7131,7 +7128,7 @@ class BaseLLMHTTPHandler:
@staticmethod
def _decoded_realtime_sdk_response(response: httpx.Response) -> httpx.Response:
headers: Final = { # mutable-ok: httpx accepts a concrete response header mapping
headers: Final = {
key: value
for key, value in response.headers.items()
if key.lower() not in ("content-encoding", "content-length", "transfer-encoding")
@ -7176,9 +7173,7 @@ class BaseLLMHTTPHandler:
- session: JSON string with {"type": "realtime", "model": "...", ...}
"""
session_data: Final[dict[str, object]] = { # mutable-ok: model and session type are resolved locally
**(
session_config or {} # mutable-ok: absent session configuration starts from an empty provider payload
)
**(session_config or {})
}
if "type" not in session_data:
session_data["type"] = "translation" if translation else "realtime"

6
litellm/llms/openai/realtime/handler.py
View file

@ -212,7 +212,7 @@ class OpenAIRealtime(OpenAIChatCompletion):
model=sdk_model,
extra_query=extra_query,
extra_headers=headers,
websocket_connection_options={ # mutable-ok: OpenAI SDK forwards a mutable options mapping
websocket_connection_options={
"max_size": REALTIME_WEBSOCKET_MAX_MESSAGE_SIZE_BYTES,
**(MappingProxyType({"ssl": ssl_config}) if url.startswith("wss://") else MappingProxyType({})),
**(MappingProxyType({"open_timeout": timeout}) if timeout is not None else MappingProxyType({})),
@ -253,7 +253,9 @@ class OpenAIRealtime(OpenAIChatCompletion):
# Get provider-specific SSL configuration
ssl_config: Final = self._get_ssl_config(url)
openai_beta_realtime: Final = client_sent_openai_beta_realtime_header(websocket)
openai_beta_realtime: Final = realtime_mode != "translation" and client_sent_openai_beta_realtime_header(
websocket
)
if not openai_beta_realtime:
verbose_logger.debug(
"OpenAI Realtime: connecting with GA protocol (no OpenAI-Beta header). "

4
litellm/llms/openai/transcriptions/gpt_transformation.py
View file

@ -14,7 +14,7 @@ class OpenAIGPTAudioTranscriptionConfig(OpenAIWhisperAudioTranscriptionConfig):
"""
Get the supported OpenAI params for the `gpt-4o-transcribe` models
"""
return [ # mutable-ok: base transcription interface requires a mutable supported-parameter list
return [
"language",
"prompt",
"response_format",
@ -56,7 +56,7 @@ class OpenAIGPTTranscribeAudioTranscriptionConfig(OpenAIGPTAudioTranscriptionCon
optional_params: dict, # mutable-ok: base transformation interface supplies a mutable request payload
litellm_params: dict, # mutable-ok: base transformation interface supplies mutable provider parameters
) -> AudioTranscriptionRequestData:
data: Final = { # mutable-ok: OpenAI SDK consumes this multipart request mapping
data: Final = {
"model": model,
"file": audio_file,
"response_format": "json",

2
litellm/llms/openai/transcriptions/handler.py
View file

@ -36,7 +36,7 @@ class OpenAIAudioTranscription(OpenAIChatCompletion):
sdk_data: Final = sdk_compatible_transcription_request_data(data)
if data.get("stream") is True:
stream_response: Final = await openai_aclient.audio.transcriptions.create(**sdk_data, timeout=timeout)
return {}, stream_response # mutable-ok: response headers use the existing mutable mapping contract
return {}, stream_response
raw_response: Final = await openai_aclient.audio.transcriptions.with_raw_response.create(
**sdk_data, timeout=timeout
) # pyright: ignore[reportArgumentType] # SDK TypedDict lags accepted transcription options

8
litellm/main.py
View file

@ -8218,7 +8218,9 @@ async def atranscription(
if file is not None:
calculated_duration = calculate_request_duration(file)
if calculated_duration is not None:
response._litellm_audio_duration = calculated_duration # pyright: ignore[reportAttributeAccess] # SDK streams permit runtime metadata but do not type this field
response._litellm_audio_duration = ( # pyright: ignore[reportAttributeAccessIssue] # SDK omits it
calculated_duration
)
return response
if not isinstance(response, TranscriptionResponse):
raise ValueError(
@ -8567,7 +8569,9 @@ def transcription(
calculated_duration: Final = calculate_request_duration(file)
if isinstance(response, (Stream, AsyncStream)):
if calculated_duration is not None:
response._litellm_audio_duration = calculated_duration # pyright: ignore[reportAttributeAccess] # SDK streams permit runtime metadata but do not type this field
response._litellm_audio_duration = ( # pyright: ignore[reportAttributeAccessIssue] # SDK omits it
calculated_duration
)
elif getattr(response, "duration", None) is None and calculated_duration is not None:
response.set_audio_transcription_duration(calculated_duration)

2
litellm/proxy/_lazy_features.py
View file

@ -47,7 +47,7 @@ def _include_router_prepend(attr_name: str = "router") -> Callable[["FastAPI", o
route_count: Final = len(router.routes)
app.include_router(getattr(module, attr_name))
new_routes: Final = router.routes[route_count:]
router.routes[:] = [ # mutable-ok: FastAPI route registration requires in-place list replacement
router.routes[:] = [
*new_routes,
*router.routes[:route_count],
]

11
litellm/proxy/agent_endpoints/auth/managed_authorization.py
View file

@ -11,7 +11,16 @@ from litellm.proxy.agent_endpoints.managed_identity import raise_identity_failur
from litellm.types.agents import AgentResponse
from litellm.types.proxy.agent_identity import AgentIdentityFailure, ManagedAgentContext
_MANAGED_REALTIME_ROUTES: Final = frozenset(("/realtime", "/v1/realtime", "/openai/v1/realtime"))
_MANAGED_REALTIME_ROUTES: Final = frozenset(
(
"/realtime",
"/v1/realtime",
"/openai/v1/realtime",
"/realtime/translations",
"/v1/realtime/translations",
"/openai/v1/realtime/translations",
)
)
_MANAGED_LIVE_ROUTES: Final = frozenset(("/live/sessions", "/v1/live/sessions", "/openai/v1/live/sessions"))
_MANAGED_MODEL_ROUTES: Final = frozenset(
f"{prefix}/{operation}"

6
litellm/proxy/common_utils/http_parsing_utils.py
View file

@ -527,10 +527,8 @@ async def get_form_data(request: Request) -> dict[str, Any]:
array_keys: Final = frozenset(key[:-2] for key, _ in form_items if key.endswith("[]"))
normalized_items: Final = tuple((key.removesuffix("[]"), value) for key, value in form_items)
normalized_keys: Final = frozenset(key for key, _ in normalized_items)
return { # mutable-ok: request parsers expose a mutable form-data mapping to endpoint handlers
key: [ # mutable-ok: repeated multipart values follow the established mutable list contract
value for item_key, value in normalized_items if item_key == key
]
return {
key: [value for item_key, value in normalized_items if item_key == key]
if key in array_keys
else next(value for item_key, value in reversed(normalized_items) if item_key == key)
for key in normalized_keys

4
litellm/proxy/proxy_server.py
View file

@ -13754,9 +13754,7 @@ async def _route_realtime_websocket_session(
)
query_params: Final = cast( # cast-ok: cached tuples contain only the declared realtime query keys
RealtimeQueryParams,
dict( # mutable-ok: downstream realtime routing normalizes this request-scoped query mapping
_realtime_query_params_template(query_model, intent)
),
dict(_realtime_query_params_template(query_model, intent)),
)
data: dict[str, object] = {

18
litellm/proxy/realtime_endpoints/endpoints.py
View file

@ -289,18 +289,18 @@ def _decode_realtime_token_payload(
)
@router.post(
"/v1/realtime/translations/client_secrets",
dependencies=[Depends(user_api_key_auth)], # mutable-ok: FastAPI decorator requires dependency lists
tags=["realtime"], # mutable-ok: FastAPI decorator requires tag lists
dependencies=[Depends(user_api_key_auth)],
tags=["realtime"],
)
@router.post(
"/realtime/translations/client_secrets",
dependencies=[Depends(user_api_key_auth)], # mutable-ok: FastAPI decorator requires dependency lists
tags=["realtime"], # mutable-ok: FastAPI decorator requires tag lists
dependencies=[Depends(user_api_key_auth)],
tags=["realtime"],
)
@router.post(
"/openai/v1/realtime/translations/client_secrets",
dependencies=[Depends(user_api_key_auth)], # mutable-ok: FastAPI decorator requires dependency lists
tags=["realtime"], # mutable-ok: FastAPI decorator requires tag lists
dependencies=[Depends(user_api_key_auth)],
tags=["realtime"],
)
async def create_realtime_client_secret(
request: Request,
@ -453,15 +453,15 @@ async def create_realtime_client_secret(
)
@router.post(
"/v1/realtime/translations/calls",
tags=["realtime"], # mutable-ok: FastAPI decorator requires tag lists
tags=["realtime"],
)
@router.post(
"/realtime/translations/calls",
tags=["realtime"], # mutable-ok: FastAPI decorator requires tag lists
tags=["realtime"],
)
@router.post(
"/openai/v1/realtime/translations/calls",
tags=["realtime"], # mutable-ok: FastAPI decorator requires tag lists
tags=["realtime"],
)
async def proxy_realtime_calls(
request: Request,

40
litellm/realtime_api/main.py
View file

@ -81,24 +81,18 @@ def _model_params_with_stored_credentials(model_params: Mapping[str, object]) ->
def _with_resolved_session_model(session: dict[str, object], model_name: str) -> dict[str, object]:
if session.get("type") == "transcription":
audio = session.get("audio")
audio = audio if isinstance(audio, dict) else {} # mutable-ok: nested session model is rebuilt locally
audio = audio if isinstance(audio, dict) else {}
audio_input = audio.get("input")
audio_input = (
audio_input if isinstance(audio_input, dict) else {} # mutable-ok: nested session model is rebuilt locally
)
audio_input = audio_input if isinstance(audio_input, dict) else {}
transcription = audio_input.get("transcription")
transcription = (
transcription
if isinstance(transcription, dict)
else {} # mutable-ok: nested session model is rebuilt locally
)
return { # mutable-ok: provider routing requires an independently mutable session payload
transcription = transcription if isinstance(transcription, dict) else {}
return {
**session,
"audio": { # mutable-ok: provider routing rebuilds nested audio configuration
"audio": {
**audio,
"input": { # mutable-ok: provider routing rebuilds nested input configuration
"input": {
**audio_input,
"transcription": { # mutable-ok: resolved deployment replaces only the transcription model
"transcription": {
**transcription,
"model": model_name,
},
@ -248,7 +242,7 @@ async def acreate_realtime_client_secret(
if isinstance(request_data.get("session"), dict):
request_data["session"] = _with_resolved_session_model(request_data["session"], model_name)
elif req.model is not None:
request_data["session"] = { # mutable-ok: OpenAI SDK consumes this request-scoped session payload
request_data["session"] = {
"type": "realtime",
"model": model_name,
}
@ -279,7 +273,7 @@ async def acreate_realtime_translation_client_secret(
) -> httpx.Response:
requested_model_name: Final = model or (session or {}).get("model") or "gpt-realtime-translate"
session_config: Final = RealtimeSessionConfig.model_validate(
{ # mutable-ok: Pydantic validates this request-scoped translation session payload
{
**(session or {}),
"type": "translation",
"model": requested_model_name,
@ -315,18 +309,16 @@ async def acreate_realtime_translation_client_secret(
litellm_logging_obj.update_from_kwargs(
kwargs=kwargs,
model=model_name,
optional_params={ # mutable-ok: logging owns a mutable request metadata payload
optional_params={
"expires_after": expires_after,
"session": session,
},
litellm_params={ # mutable-ok: logging owns a mutable provider metadata payload
"api_base": resolved_api_base
},
litellm_params={"api_base": resolved_api_base},
custom_llm_provider=custom_llm_provider,
)
request_data: Final = req.model_dump(
exclude_none=True,
exclude={"model"}, # mutable-ok: Pydantic requires a mutable field-exclusion set
exclude={"model"},
)
request_data["session"] = _with_resolved_session_model(request_data["session"], model_name)
request_data["session"].pop("type", None)
@ -508,7 +500,7 @@ async def arealtime_translation_calls(
litellm_params=litellm_params,
)
session_config: Final = _with_resolved_session_model(
{ # mutable-ok: provider routing requires an independently mutable session payload
{
**(session or {}),
"type": "translation",
"model": model_name,
@ -518,13 +510,11 @@ async def arealtime_translation_calls(
litellm_logging_obj.update_from_kwargs(
kwargs=kwargs,
model=model_name,
optional_params={ # mutable-ok: logging owns a mutable request metadata payload
optional_params={
"realtime_translation_calls": True,
"session": session_config,
},
litellm_params={ # mutable-ok: logging owns a mutable provider metadata payload
"api_base": resolved_api_base
},
litellm_params={"api_base": resolved_api_base},
custom_llm_provider=custom_llm_provider,
)
return await base_llm_http_handler.async_realtime_calls_handler(

43
tests/unit/litellm_core_utils/test_realtime_streaming.py
View file

@ -648,9 +648,10 @@ async def test_client_ack_messages_keeps_beta_session_shape_for_beta_backend():
@pytest.mark.asyncio
async def test_translation_session_update_omits_session_type():
@pytest.mark.parametrize("beta_client", (False, True))
async def test_translation_session_update_omits_session_type(beta_client: bool) -> None:
client_ws = MagicMock()
client_ws.scope = {"headers": []}
client_ws.scope = {"headers": [(b"openai-beta", b"realtime=v1")] if beta_client else []}
client_ws.receive_text = AsyncMock(
side_effect=[
json.dumps(
@ -658,6 +659,8 @@ async def test_translation_session_update_omits_session_type():
"type": "session.update",
"session": {
"type": "translation",
"input_audio_format": "g711_ulaw",
"output_audio_format": "pcm16",
"audio": {"output": {"language": "fr"}},
},
}
@ -681,6 +684,11 @@ async def test_translation_session_update_omits_session_type():
sent_to_backend = json.loads(backend_ws.send.call_args_list[0].args[0])
assert "type" not in sent_to_backend["session"]
assert sent_to_backend["session"]["audio"]["output"]["language"] == "fr"
assert sent_to_backend["session"]["audio"]["input"]["format"]["type"] == "audio/pcmu", (
"OpenAI Realtime audio formats, checked 2026-10-06: "
"https://github.com/openai/openai-python/blob/v2.33.0/src/openai/types/realtime/realtime_audio_formats.py"
)
assert sent_to_backend["session"]["audio"]["output"]["format"]["type"] == "audio/pcm"
def test_translate_event_to_beta_renames_delta_types():
@ -3068,8 +3076,22 @@ def test_translation_audio_duration_is_finalized_once(event_type: str):
@pytest.mark.asyncio
@pytest.mark.parametrize("output_bytes", (0, 48000))
@pytest.mark.parametrize("format_event_type", ("session.created", "session.update"))
@pytest.mark.parametrize(
"audio_format,bytes_per_second",
(
("pcm16", 48000),
("g711_ulaw", 8000),
("g711_alaw", 8000),
({"type": "audio/pcm", "rate": 16000}, 32000),
({"type": "audio/pcmu", "rate": 8000}, 8000),
({"type": "audio/pcma", "rate": 8000}, 8000),
({"type": "audio/pcmu"}, 8000),
({"type": "audio/pcma"}, 8000),
),
)
async def test_translation_disconnect_bills_sent_input_audio(
output_bytes: int,
output_bytes: int, format_event_type: str, audio_format: str | Mapping[str, object], bytes_per_second: int
) -> None:
import base64
@ -3082,8 +3104,21 @@ async def test_translation_disconnect_bills_sent_input_audio(
model="gpt-realtime-translate",
translation_session=True,
)
format_event: Final = {
"type": format_event_type,
"session": {"audio": {"input": {"format": audio_format}}},
}
if format_event_type == "session.update":
await streaming._send_to_backend(json.dumps(format_event))
else:
streaming._capture_translation_output_audio(format_event)
await streaming._send_to_backend(
json.dumps({"type": "session.input_audio_buffer.append", "audio": base64.b64encode(bytes(96000)).decode()})
json.dumps(
{
"type": "session.input_audio_buffer.append",
"audio": base64.b64encode(bytes(bytes_per_second * 2)).decode(),
}
)
)
streaming._capture_translation_output_audio(
{"type": "session.output_audio.delta", "delta": base64.b64encode(bytes(output_bytes)).decode()}

53
tests/unit/llms/openai/realtime/test_openai_realtime_handler.py
View file

@ -1,4 +1,7 @@
import asyncio
import json
from collections.abc import Mapping
from typing import Final
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
@ -9,6 +12,7 @@ from litellm.types.realtime import RealtimeQueryParams
from openai import AsyncOpenAI, omit
from pydantic import TypeAdapter
class DummySDKConnectionManager:
@ -487,24 +491,35 @@ def test_translation_url_uses_dedicated_path():
@pytest.mark.asyncio
async def test_translation_websocket_uses_direct_transport():
@pytest.mark.parametrize("beta_client", (False, True))
async def test_translation_websocket_uses_direct_transport(beta_client: bool) -> None:
from litellm.llms.openai.realtime.handler import OpenAIRealtime
backend = AsyncMock()
backend: Final = AsyncMock()
pending: Final = asyncio.Event()
class TranslationConnectionManager:
async def __aenter__(self):
return backend
async def receive_backend() -> str:
await pending.wait()
return ""
async def __aexit__(self, exc_type, exc, tb):
return None
websocket = MagicMock()
websocket.scope = {"headers": []}
backend.recv.side_effect = receive_backend
websocket: Final = MagicMock()
websocket.scope = {"headers": [(b"openai-beta", b"realtime=v1")] if beta_client else []}
websocket.receive_text = AsyncMock(
side_effect=(
json.dumps(
{
"type": "session.update",
"session": {"type": "translation", "audio": {"output": {"language": "fr"}}},
}
),
RuntimeError("connection closed"),
)
)
websocket.close = AsyncMock()
logging_obj = MagicMock()
handler = OpenAIRealtime()
expected_url = "wss://api.openai.com/v1/realtime/translations?model=gpt-realtime-translate"
logging_obj: Final = MagicMock()
handler: Final = OpenAIRealtime()
expected_url: Final = "wss://api.openai.com/v1/realtime/translations?model=gpt-realtime-translate"
assert (
handler._construct_url(
api_base="https://api.openai.com/v1",
@ -514,13 +529,7 @@ async def test_translation_websocket_uses_direct_transport():
== expected_url
)
with (
patch("websockets.connect", return_value=TranslationConnectionManager()) as connect,
patch( # test-quality-ok: transport test replaces the unbounded streaming loop
"litellm.llms.openai.realtime.handler.RealTimeStreaming"
) as streaming,
):
streaming.return_value.bidirectional_forward = AsyncMock()
with patch("websockets.connect", return_value=DummySDKConnectionManager(backend)) as connect:
await handler.async_realtime(
model="gpt-realtime-translate",
websocket=websocket,
@ -533,7 +542,9 @@ async def test_translation_websocket_uses_direct_transport():
connect.assert_called_once()
assert connect.call_args.args[0] == expected_url
assert streaming.call_args.kwargs["translation_session"] is True
assert "OpenAI-Beta" not in connect.call_args.kwargs["additional_headers"]
sent: Final = TypeAdapter(Mapping[str, object]).validate_json(backend.send.call_args_list[0].args[0])
assert sent["session"] == {"audio": {"output": {"language": "fr"}}}
@pytest.mark.parametrize("sdk_client", [True, False])

17
tests/unit/proxy/agent_endpoints/auth/test_managed_authorization.py
View file

@ -364,6 +364,10 @@ async def test_unknown_invocation_target_leaves_billing_unset(monkeypatch: pytes
("/v1beta/models/gemini-model:generateContent", "POST", True),
("/v1/realtime", "GET", True),
("/v1/realtime", "POST", False),
("/v1/realtime/translations", "GET", True),
("/v1/realtime/translations", "POST", False),
("/v1/realtime/translations/client_secrets", "POST", False),
("/v1/realtime/translations/calls", "POST", False),
("/v1/realtime/client_secrets", "POST", False),
("/mcp/tools/call", "POST", True),
("/a2a/target/message/send", "POST", True),
@ -437,7 +441,17 @@ def test_managed_inference_ignores_unsupported_query_model():
)
@pytest.mark.parametrize("route", ["/realtime", "/v1/realtime", "/openai/v1/realtime"])
@pytest.mark.parametrize(
"route",
(
"/realtime",
"/v1/realtime",
"/openai/v1/realtime",
"/realtime/translations",
"/v1/realtime/translations",
"/openai/v1/realtime/translations",
),
)
def test_managed_realtime_requires_a_model_and_ignores_completion_defaults(route: str) -> None:
from litellm.proxy.agent_endpoints.auth.managed_authorization import managed_inference_request
@ -588,6 +602,7 @@ def test_registered_inference_routes_have_an_explicit_managed_access_decision(ro
"/vector_stores", "/vector_store/", "/search", "/containers", "/skills", "/claude-code/",
"/interactions", "/agents", "/responses/{", "/responses/input_tokens",
"/realtime/client_secrets", "/realtime/calls", "/realtime/transcription_sessions",
"/realtime/translations/client_secrets", "/realtime/translations/calls",
)) or normalized in ("/models", "/cursor/models", "/cursor/v1/models")
concrete: Final = route.split("?")[0].replace("{model}", "model").replace("{model_name:path}", "model")
assert managed_agent_route_allowed(concrete, None) is not unsupported, route