scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-13 03:59:12 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 66

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 b2cb45bf1e
commit a9adf50656
No known key found for this signature in database
GPG key ID: D3DF28AB5D1B5E17
19 changed files with 194 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

@ -471,6 +471,7 @@ jobs:
test-path: >-
tests/unit/anthropic_interface
tests/unit/compression
tests/unit/cookbook
tests/unit/enterprise/enterprise_callbacks/test_callback_controls.py
tests/unit/enterprise/enterprise_callbacks/test_llm_guard.py
tests/unit/enterprise/enterprise_callbacks/test_secret_detection.py

2
litellm/cost_calculator.py
View file

@ -3056,7 +3056,7 @@ def handle_realtime_stream_cost_calculation(
)
total_cost: Final = input_cost_per_token + output_cost_per_token + transcription_cost + translation_cost
additional_costs: Final = { # mutable-ok: logging stores a mutable per-request cost breakdown
additional_costs: Final = {
key: value
for key, value in (
("transcription_cost", transcription_cost),

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)

66
litellm/litellm_core_utils/realtime_streaming.py
View file

@ -7,6 +7,7 @@ from dataclasses import dataclass
from enum import Enum, auto
from typing import TYPE_CHECKING, Any, Final, NoReturn, Protocol, TypedDict, cast
from pydantic import TypeAdapter, ValidationError
from typing_extensions import ReadOnly
import litellm
@ -112,6 +113,25 @@ def _decode_json_object(payload: str) -> Mapping[str, object]:
return json.loads(payload)
_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: ...
@ -155,13 +175,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
)
@ -226,8 +247,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] = {
@ -492,7 +513,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",
@ -515,35 +536,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

@ -49,7 +49,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

@ -6421,7 +6421,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,
},
@ -6429,10 +6429,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",
@ -6504,7 +6501,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,
},
@ -6637,7 +6634,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,
},
@ -6646,7 +6643,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()}
@ -6687,7 +6684,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")
@ -6732,9 +6729,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

@ -200,7 +200,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({})),
@ -241,7 +241,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

@ -35,7 +35,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

@ -7922,7 +7922,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(
@ -8273,7 +8275,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

@ -43,7 +43,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_MODEL_ROUTES: Final = frozenset(
f"{prefix}/{operation}"
for prefix, operation in product(

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

@ -479,10 +479,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

@ -13122,9 +13122,7 @@ async def realtime_websocket_endpoint(
)
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

@ -282,18 +282,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,
@ -446,15 +446,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

@ -73,24 +73,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,
},
@ -228,7 +222,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,
}
@ -257,7 +251,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,
@ -293,18 +287,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)
@ -482,7 +474,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,
@ -492,13 +484,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

@ -604,9 +604,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(
@ -614,6 +615,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"}},
},
}
@ -637,6 +640,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():
@ -2935,8 +2943,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
@ -2949,8 +2971,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,8 +1,12 @@
import asyncio
import json
from collections.abc import Mapping
from typing import Final
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from openai import AsyncOpenAI, omit
from pydantic import TypeAdapter
class DummySDKConnectionManager:
@ -407,24 +411,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",
@ -434,13 +449,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,
@ -453,7 +462,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
@ -574,6 +588,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