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

fix(realtime): keep main's transcription, client secret and route behavior where the PR changed it

Browse source 
Transcription reaches every provider as on main again, sends stream only when asked, forwards language and
languages together, keeps form key order, bills provider seconds only when above zero, honors base_model for the
new flags, keeps main's Azure api_version on the proxy path, and streams carry chat's headers, an error event on a
mid-stream failure, and a spend row when the stream ends early. Translation cost reads both per-second rates from
one source and spend rows carry the billed seconds.

Realtime transcription WebSockets keep model in the upstream URL outside OpenAI's hosts, client secrets forward
explicit nulls, accept conversation and delay, and answer 400 on a malformed body, the /openai translation paths
are back to main, translation refuses only for guardrails realtime sessions run, nested aliases resolve and later
session.update models are authorized, translation client secrets run per-model auth, beta clients bridged to GA
get beta-shaped session events, and transcription sessions send the deployment model in both shapes.
This commit is contained in:
mateo-berri 2026-10-20 14:12:28 -07:00 • committed by Emerson Gomes
parent c57d273aa6
commit 34319dff7d
No known key found for this signature in database
GPG key ID: D3DF28AB5D1B5E17
34 changed files with 1427 additions and 539 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

71
litellm/cost_calculator.py
View file

@ -352,22 +352,24 @@ def _transcription_uses_token_pricing(
return (prompt_tokens_val > 0) or (completion_tokens_val > 0)
def _transcription_duration_for_cost(
completion_response: object, model: str | None, custom_llm_provider: str | None
) -> float:
provider_duration: Final = _get_transcription_usage_duration(completion_response)
transcription_info: Final = (
_get_model_info_or_none(model, custom_llm_provider)
if model is not None and custom_llm_provider is not None
else None
)
def _bills_transcription_usage_duration(pricing_model: str | None, custom_llm_provider: str | None) -> bool:
if pricing_model is None or custom_llm_provider is None:
return False
transcription_info: Final = _get_model_info_or_none(pricing_model, custom_llm_provider)
transcription_metadata: Final = (
as_str_mapping(transcription_info.get("provider_specific_entry")) if transcription_info is not None else None
)
return transcription_metadata is not None and transcription_metadata.get("transcription_usage_duration") == 1
def _transcription_duration_for_cost(
completion_response: object, pricing_model: str | None, model: str | None, custom_llm_provider: str | None
) -> float:
provider_duration: Final = _get_transcription_usage_duration(completion_response)
if (
provider_duration is not None
and transcription_metadata is not None
and transcription_metadata.get("transcription_usage_duration") == 1
and provider_duration > 0
and any(_bills_transcription_usage_duration(name, custom_llm_provider) for name in (pricing_model, model))
):
return provider_duration
hidden_params: Final = as_str_mapping(attribute_or_none(completion_response, "_hidden_params")) or {}
@ -1907,7 +1909,7 @@ def completion_cost(
prompt_characters = litellm.utils.count_characters(text=prompt)
elif call_type in _TRANSCRIPTION_CALL_TYPES:
audio_transcription_file_duration = _transcription_duration_for_cost(
completion_response, requested_model, custom_llm_provider
completion_response, model, requested_model, custom_llm_provider
)
elif call_type in _RERANK_CALL_TYPES:
if completion_response is not None and isinstance(completion_response, RerankResponse):
@ -3213,8 +3215,11 @@ class RealtimeAPITokenUsageProcessor(BaseTokenUsageProcessor):
def create_logging_realtime_object(
usage: Usage, results: OpenAIRealtimeStreamList
) -> LiteLLMRealtimeStreamLoggingObject:
_, translated_output_seconds = _translation_session_seconds(results)
return LiteLLMRealtimeStreamLoggingObject(
usage=usage,
usage=usage.model_copy(update={"seconds": translated_output_seconds})
if translated_output_seconds > 0
else usage,
results=results,
)
@ -3661,12 +3666,7 @@ def handle_realtime_stream_cost_calculation(
return total_cost
def handle_realtime_translation_cost_calculation(
results: OpenAIRealtimeStreamList,
custom_llm_provider: str,
litellm_model_name: str,
potential_model_names: Sequence[str | None] = (),
) -> float:
def _translation_session_seconds(results: OpenAIRealtimeStreamList) -> tuple[float, float]:
usage_events: Final = tuple(
usage
for result in results
@ -3684,6 +3684,19 @@ def handle_realtime_translation_cost_calculation(
for usage in usage_events
if (output_duration := normalized_audio_duration_seconds(usage.get("output_seconds"))) is not None
)
return float(input_seconds), float(output_seconds)
_TRANSLATION_RATE_KEYS: Final = ("input_cost_per_second", "output_cost_per_second")
def handle_realtime_translation_cost_calculation(
results: OpenAIRealtimeStreamList,
custom_llm_provider: str,
litellm_model_name: str,
potential_model_names: Sequence[str | None] = (),
) -> float:
input_seconds, output_seconds = _translation_session_seconds(results)
if input_seconds <= 0 and output_seconds <= 0:
return 0.0
model_infos: Final = tuple(
@ -3691,22 +3704,12 @@ def handle_realtime_translation_cost_calculation(
for model in (*potential_model_names, litellm_model_name)
if model is not None
)
input_cost_per_second: Final = next(
(
rate
for info in model_infos
if (rate := _declared_transcription_rate(info, ("input_cost_per_second",))) is not None
),
0.0,
)
output_cost_per_second: Final = next(
(
output_rate
for info in model_infos
if (output_rate := _declared_transcription_rate(info, ("output_cost_per_second",))) is not None
),
0.0,
priced_info: Final = next(
(info for info in model_infos if _declared_transcription_rate(info, _TRANSLATION_RATE_KEYS) is not None),
None,
)
input_cost_per_second: Final = _declared_transcription_rate(priced_info, ("input_cost_per_second",)) or 0.0
output_cost_per_second: Final = _declared_transcription_rate(priced_info, ("output_cost_per_second",)) or 0.0
return input_seconds * input_cost_per_second + output_seconds * output_cost_per_second

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

@ -2,7 +2,7 @@ from __future__ import annotations
import datetime
import traceback
from collections.abc import AsyncIterator, Iterator, Mapping
from collections.abc import AsyncIterator, Iterator
from typing import Final, Protocol
from openai import AsyncStream, Stream
@ -11,11 +11,9 @@ from openai.types.audio import (
TranscriptionTextDeltaEvent,
TranscriptionTextDoneEvent,
)
from pydantic import TypeAdapter
from litellm.types.utils import TranscriptionDetectedLanguage, TranscriptionResponse
_DETECTED_LANGUAGES: Final = TypeAdapter(list[TranscriptionDetectedLanguage])
from litellm.litellm_core_utils.core_helpers import process_response_headers
from litellm.types.utils import TranscriptionResponse
class TranscriptionStreamLogging(Protocol):
@ -71,15 +69,10 @@ class _TranscriptionEventCollector:
def response(self) -> TranscriptionResponse:
done_event: Final = self.done_event
done_languages: Final = getattr(done_event, "languages", None) if done_event is not None else None
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=(
_DETECTED_LANGUAGES.validate_python(done_languages, from_attributes=True)
if done_languages is not None
else None
),
languages=getattr(done_event, "languages", None) if done_event is not None else None,
)
if self.duration is not None:
response.set_audio_transcription_duration(self.duration)
@ -94,7 +87,9 @@ class LoggingTranscriptionStream(Stream[TranscriptionStreamEvent]):
start_time: datetime.datetime,
) -> None:
self.__dict__.update(stream.__dict__)
self._hidden_params: Mapping[str, object] = {}
self._hidden_params: dict[str, object] = { # mutable-ok: ResponseMetadata fills in the call's metadata
"additional_headers": process_response_headers(stream.response.headers)
}
self._logging_obj = logging_obj
self._start_time = start_time
self._collector = _TranscriptionEventCollector(getattr(stream, "_litellm_audio_duration", None))
@ -122,8 +117,6 @@ class LoggingTranscriptionStream(Stream[TranscriptionStreamEvent]):
if self._finalized or self._failed:
return
self._finalized = True
if self._collector.done_event is None and self._collector.duration is None:
return
self._logging_obj.success_handler(
self._collector.response(),
self._start_time,
@ -136,6 +129,10 @@ class LoggingTranscriptionStream(Stream[TranscriptionStreamEvent]):
finally:
self._finalize()
@property
def logging_obj(self) -> TranscriptionStreamLogging:
return self._logging_obj
class LoggingAsyncTranscriptionStream(AsyncStream[TranscriptionStreamEvent]):
def __init__(
@ -145,7 +142,9 @@ class LoggingAsyncTranscriptionStream(AsyncStream[TranscriptionStreamEvent]):
start_time: datetime.datetime,
) -> None:
self.__dict__.update(stream.__dict__)
self._hidden_params: Mapping[str, object] = {}
self._hidden_params: dict[str, object] = { # mutable-ok: ResponseMetadata fills in the call's metadata
"additional_headers": process_response_headers(stream.response.headers)
}
self._logging_obj = logging_obj
self._start_time = start_time
self._collector = _TranscriptionEventCollector(getattr(stream, "_litellm_audio_duration", None))
@ -179,8 +178,6 @@ class LoggingAsyncTranscriptionStream(AsyncStream[TranscriptionStreamEvent]):
if self._finalized or self._failed:
return
self._finalized = True
if self._collector.done_event is None and self._collector.duration is None:
return
response: Final = self._collector.response()
end_time: Final = datetime.datetime.now() # noqa: DTZ005 # callback timestamps use the legacy naive contract
self._logging_obj.handle_sync_success_callbacks_for_async_calls(
@ -200,6 +197,10 @@ class LoggingAsyncTranscriptionStream(AsyncStream[TranscriptionStreamEvent]):
finally:
await self._finalize()
@property
def logging_obj(self) -> TranscriptionStreamLogging:
return self._logging_obj
def wrap_transcription_stream(
stream: Stream[TranscriptionStreamEvent] | AsyncStream[TranscriptionStreamEvent],

9
litellm/litellm_core_utils/llm_response_utils/convert_dict_to_response.py
View file

@ -44,7 +44,6 @@ from litellm.types.utils import (
StreamingChoices,
TextChoices,
TextCompletionResponse,
TranscriptionDetectedLanguage,
TranscriptionResponse,
TranscriptionUsageDurationObject,
TranscriptionUsageTokensObject,
@ -947,11 +946,9 @@ def convert_to_model_response_object(
if key in response_object:
setattr(model_response_object, key, response_object[key])
if "languages" in response_object and response_object["languages"] is not None:
transcription_response: Final = model_response_object
transcription_response.languages = TypeAdapter(
tuple[TranscriptionDetectedLanguage, ...]
).validate_python(response_object["languages"])
detected_languages: Final = TranscriptionResponse.detected_languages(response_object.get("languages"))
if detected_languages is not None:
model_response_object.languages = detected_languages
if "usage" in response_object and response_object["usage"] is not None:
tr_usage_object: TranscriptionUsageDurationObject | TranscriptionUsageTokensObject | None = None

154
litellm/litellm_core_utils/realtime_streaming.py
View file

@ -43,6 +43,7 @@ if TYPE_CHECKING:
from websockets.asyncio.client import ClientConnection
from websockets.exceptions import ConnectionClosed
from litellm.integrations.custom_guardrail import CustomGuardrail
from litellm.types.guardrails import GuardrailEventHooks
CLIENT_CONNECTION_CLASS = ClientConnection
@ -151,6 +152,52 @@ def _mapping_field(container: Mapping[str, object], key: str) -> Mapping[str, ob
return _json_object(container.get(key))
_BETA_AUDIO_FORMAT_BY_GA_TYPE: Final[Mapping[str, str]] = MappingProxyType(
{"audio/pcm": "pcm16", "audio/pcmu": "g711_ulaw", "audio/pcma": "g711_alaw"}
)
_GA_TO_BETA_TOP_LEVEL_SESSION_FIELDS: Final[Mapping[str, str]] = MappingProxyType(
{"output_modalities": "modalities", "max_output_tokens": "max_response_output_tokens"}
)
_GA_TO_BETA_AUDIO_INPUT_FIELDS: Final[Mapping[str, str]] = MappingProxyType(
{
"format": "input_audio_format",
"transcription": "input_audio_transcription",
"noise_reduction": "input_audio_noise_reduction",
"turn_detection": "turn_detection",
}
)
_GA_TO_BETA_AUDIO_OUTPUT_FIELDS: Final[Mapping[str, str]] = MappingProxyType(
{"format": "output_audio_format", "voice": "voice", "speed": "speed"}
)
_GA_ONLY_SESSION_FIELDS: Final = frozenset(("type", "audio", *_GA_TO_BETA_TOP_LEVEL_SESSION_FIELDS))
def _beta_session_value(ga_field: str, value: object) -> object:
if ga_field == "output_modalities":
return ["audio", "text"] if value == ["audio"] else value
if ga_field != "format":
return value
ga_format: Final = _json_object(value)
format_type: Final = None if ga_format is None else ga_format.get("type")
return _BETA_AUDIO_FORMAT_BY_GA_TYPE.get(format_type, value) if isinstance(format_type, str) else value
def _beta_session_fields(ga_fields: Mapping[str, object], beta_names: Mapping[str, str]) -> Mapping[str, object]:
return {
beta_names[name]: _beta_session_value(name, value) for name, value in ga_fields.items() if name in beta_names
}
def ga_session_to_beta(session: Mapping[str, object]) -> Mapping[str, object]:
audio: Final = _mapping_field(session, "audio") or _EMPTY_SESSION
return {
**{key: value for key, value in session.items() if key not in _GA_ONLY_SESSION_FIELDS},
**_beta_session_fields(session, _GA_TO_BETA_TOP_LEVEL_SESSION_FIELDS),
**_beta_session_fields(_mapping_field(audio, "input") or _EMPTY_SESSION, _GA_TO_BETA_AUDIO_INPUT_FIELDS),
**_beta_session_fields(_mapping_field(audio, "output") or _EMPTY_SESSION, _GA_TO_BETA_AUDIO_OUTPUT_FIELDS),
}
def _guardrail_event_id() -> str:
return f"{_GUARDRAIL_EVENT_ID_PREFIX}{uuid.uuid4().hex}"
@ -265,26 +312,29 @@ def _guardrails_run_for_event_hooks(
)
def _realtime_runner_guardrail(
callback: object,
check_data: Mapping[str, object],
event_hooks: Sequence["GuardrailEventHooks"],
) -> "CustomGuardrail | None":
from litellm.integrations.custom_guardrail import CustomGuardrail
if not isinstance(callback, CustomGuardrail) or callback.use_native_lifecycle_hooks:
return None
runs: Final = any(
callback.should_run_guardrail(data=check_data, event_type=event_hook) for event_hook in event_hooks
)
return callback if runs else None
def realtime_guardrails_apply(request_data: Mapping[str, object]) -> bool:
from litellm.types.guardrails import GuardrailEventHooks
if any(
isinstance(bucket, Mapping) and "_guardrail_pipelines" in bucket
for bucket in (request_data.get("metadata"), request_data.get("litellm_metadata"))
):
from litellm.proxy.utils import pipeline_managed_guardrail_names
if pipeline_managed_guardrail_names(request_data, "pre_call") or pipeline_managed_guardrail_names(
request_data, "post_call"
):
return True
return _guardrails_run_for_event_hooks(
request_data,
(
GuardrailEventHooks.realtime_input_transcription,
GuardrailEventHooks.pre_call,
GuardrailEventHooks.post_call,
),
check_data: Final = {**request_data, "stream": True}
runner_event_hooks: Final = (GuardrailEventHooks.realtime_input_transcription, GuardrailEventHooks.pre_call)
return any(
_realtime_runner_guardrail(callback, check_data, runner_event_hooks) is not None
for callback in litellm.callbacks
)
@ -933,40 +983,43 @@ class RealTimeStreaming:
return self._rewrite_session_update_transcription_model(message, self._force_transcription_model)
async def _enforce_translation_nested_transcription_model(self, message: str) -> str:
if self._bound_nested_transcription_model is not None:
return self._rewrite_session_update_transcription_model(message, self._bound_nested_transcription_model)
message_obj: Final = self._session_update_message_obj(message)
if message_obj is None:
return message
session: Final = as_str_mapping(message_obj.get("session"))
if session is None:
return message
nested_models: Final = self._nested_transcription_models_from_session(session)
if not nested_models:
return message
session: Final = as_str_mapping(message_obj.get("session")) if message_obj is not None else None
bound_model: Final = self._bound_nested_transcription_model
requested_models: Final = (
tuple(model for model in self._nested_transcription_models_from_session(session) if model != bound_model)
if session is not None
else ()
)
if not requested_models:
return (
message
if bound_model is None
else self._rewrite_session_update_transcription_model(message, bound_model)
)
valid_token: Final = self.user_api_key_dict
if valid_token is None:
return message
from litellm.proxy._types import UserAPIKeyAuth
from litellm.proxy.auth.auth_checks import can_key_call_resolved_model
from litellm.proxy.proxy_server import llm_model_list, llm_router
from litellm.proxy.realtime_endpoints.endpoints import nested_transcription_upstream_model
if not isinstance(valid_token, UserAPIKeyAuth):
return message
for nested_model in nested_models:
for requested_model in requested_models:
await can_key_call_resolved_model(
model=nested_model,
model=requested_model,
valid_token=valid_token,
llm_model_list=llm_model_list,
llm_router=llm_router,
)
bound_model: Final = nested_models[0]
self._bound_nested_transcription_model = bound_model
return self._rewrite_session_update_transcription_model(message, bound_model)
upstream_model: Final = nested_transcription_upstream_model(
requested_models[0], valid_token.team_id, llm_router
)
self._bound_nested_transcription_model = upstream_model
return self._rewrite_session_update_transcription_model(message, upstream_model)
def _uses_deferred_backend_setup(self) -> bool:
"""True when setup is deferred until the client's first session.update."""
@ -1087,7 +1140,18 @@ class RealTimeStreaming:
return event
def _event_to_client_json(self, event: dict) -> str:
return json.dumps(self._normalize_event_for_ga_client(event))
return json.dumps(self._session_event_for_client(self._normalize_event_for_ga_client(event)))
def _session_event_for_client(self, event: Mapping[str, object]) -> Mapping[str, object]:
if (
not self._client_wants_beta
or self._backend_uses_beta_protocol
or self._is_translation_session
or event.get("type") not in self._SESSION_EVENT_TYPES
):
return event
session: Final = _mapping_field(event, "session")
return event if session is None else {**event, "session": ga_session_to_beta(session)}
async def _send_event_to_client(self, event: object, event_str: str) -> bool:
if self._should_drop_event_from_client(event):
@ -1100,7 +1164,7 @@ class RealTimeStreaming:
translated: Final = self._translate_event_to_beta(event)
if translated is None:
return False
await self.websocket.send_text(json.dumps(translated))
await self.websocket.send_text(json.dumps(self._session_event_for_client(translated)))
return True
except Exception as e:
verbose_logger.warning(
@ -1253,7 +1317,6 @@ class RealTimeStreaming:
transcript completion uses ``realtime_input_transcription`` only;
typed user messages and tool outputs use ``pre_call``.
"""
from litellm.integrations.custom_guardrail import CustomGuardrail
from litellm.types.guardrails import GuardrailEventHooks
if event_hooks is None:
@ -1262,14 +1325,11 @@ class RealTimeStreaming:
_check_data: Final = {**self.request_data, "transcript": transcript, "stream": True}
_already_run: Final[set] = set()
for callback in litellm.callbacks:
if not isinstance(callback, CustomGuardrail):
continue
if callback.use_native_lifecycle_hooks:
continue
if id(callback) in _already_run:
continue
if not any(callback.should_run_guardrail(data=_check_data, event_type=et) for et in _realtime_event_types):
runner_guardrails: Final = (
_realtime_runner_guardrail(callback, _check_data, _realtime_event_types) for callback in litellm.callbacks
)
for callback in runner_guardrails:
if callback is None or id(callback) in _already_run:
continue
_already_run.add(id(callback))
try:
@ -1547,7 +1607,7 @@ class RealTimeStreaming:
translated = self._translate_event_to_beta(event)
if translated is None:
continue
await self.websocket.send_text(json.dumps(translated))
await self.websocket.send_text(json.dumps(self._session_event_for_client(translated)))
async def backend_to_client_send_messages(self) -> BackendClose:
import websockets

13
litellm/llms/azure/audio_transcriptions.py
View file

@ -1,7 +1,7 @@
from collections.abc import Coroutine, Mapping
from typing import TYPE_CHECKING, Final, cast
from openai import AsyncAzureOpenAI, AsyncOpenAI, AsyncStream, AzureOpenAI, OpenAI, Stream
from openai import AsyncAzureOpenAI, AsyncStream, AzureOpenAI, Stream
from openai.types.audio import TranscriptionStreamEvent
from openai.types.audio.transcription_create_params import (
TranscriptionCreateParamsBase,
@ -9,7 +9,6 @@ from openai.types.audio.transcription_create_params import (
)
from pydantic import BaseModel
import litellm
from litellm._uuid import uuid
from litellm.litellm_core_utils.audio_utils.utils import get_audio_file_name
from litellm.llms.base_llm.audio_transcription.transformation import (
@ -55,9 +54,6 @@ class AzureAudioTranscription(AzureChatCompletion):
):
data: Final = {"model": model, "file": audio_file, **optional_params}
sdk_data: Final = sdk_compatible_transcription_request_data(data)
resolved_api_version: Final = (
litellm.AZURE_DEFAULT_API_VERSION if api_version in ("v1", "latest", "preview") else api_version
)
if atranscription is True:
return self.async_audio_transcriptions(
@ -67,7 +63,6 @@ class AzureAudioTranscription(AzureChatCompletion):
timeout=timeout,
api_key=api_key,
api_base=api_base,
api_version=resolved_api_version,
client=client,
max_retries=max_retries,
logging_obj=logging_obj,
@ -77,7 +72,7 @@ class AzureAudioTranscription(AzureChatCompletion):
)
azure_client: Final = self.get_azure_openai_client(
api_version=resolved_api_version,
api_version=None if self._is_azure_v1_api_version(api_version) else api_version,
api_base=api_base,
api_key=api_key,
model=model,
@ -85,7 +80,7 @@ class AzureAudioTranscription(AzureChatCompletion):
client=client,
litellm_params=litellm_params,
)
if not isinstance(azure_client, (AzureOpenAI, OpenAI)):
if not isinstance(azure_client, AzureOpenAI):
raise AzureOpenAIError(
status_code=500,
message="azure_client is not an instance of AzureOpenAI",
@ -169,7 +164,7 @@ class AzureAudioTranscription(AzureChatCompletion):
client=client,
litellm_params=litellm_params,
)
if not isinstance(async_azure_client, (AsyncAzureOpenAI, AsyncOpenAI)):
if not isinstance(async_azure_client, AsyncAzureOpenAI):
raise AzureOpenAIError(
status_code=500,
message="async_azure_client is not an instance of AsyncAzureOpenAI",

58
litellm/llms/base_llm/audio_transcription/transformation.py
View file

@ -50,6 +50,64 @@ def sdk_compatible_transcription_response(response: object) -> BaseModel | str:
return _TRANSCRIPTION_TEXT.validate_python(response, strict=True)
def _content_type(http_response: httpx.Response) -> str:
return next(iter(http_response.headers.get_list("content-type")), "")
def _is_event_stream(http_response: httpx.Response) -> bool:
return _content_type(http_response).startswith("text/event-stream")
def _unstreamed_transcription_response(http_response: httpx.Response) -> BaseModel | str:
if _content_type(http_response).startswith("application/json"):
return Transcription.model_validate_json(http_response.content)
return http_response.text
def _non_streaming_params(data: Mapping[str, object]) -> TranscriptionCreateParamsBase:
return cast( # cast-ok: SDK file handles have no runtime validation schema
TranscriptionCreateParamsBase, {key: value for key, value in data.items() if key != "stream"}
)
def _streaming_params(data: Mapping[str, object]) -> TranscriptionCreateParamsStreaming:
return cast( # cast-ok: SDK file handles have no runtime validation schema
TranscriptionCreateParamsStreaming, data
)
def create_sdk_transcription(
client: OpenAI, data: Mapping[str, object], timeout: float | httpx.Timeout
) -> tuple[dict[str, str], Stream[TranscriptionStreamEvent] | BaseModel | str]:
if data.get("stream") is not True:
raw_response: Final = client.audio.transcriptions.with_raw_response.create(
**_non_streaming_params(data), timeout=timeout
)
return dict(raw_response.headers), sdk_compatible_transcription_response(raw_response.parse())
raw_stream: Final = client.audio.transcriptions.with_raw_response.create(**_streaming_params(data), timeout=timeout)
if _is_event_stream(raw_stream.http_response):
return dict(raw_stream.headers), raw_stream.parse(to=Stream[TranscriptionStreamEvent])
raw_stream.http_response.read()
return dict(raw_stream.headers), _unstreamed_transcription_response(raw_stream.http_response)
async def acreate_sdk_transcription(
client: AsyncOpenAI, data: Mapping[str, object], timeout: float | httpx.Timeout
) -> tuple[dict[str, str], AsyncStream[TranscriptionStreamEvent] | BaseModel | str]:
if data.get("stream") is not True:
raw_response: Final = await client.audio.transcriptions.with_raw_response.create(
**_non_streaming_params(data), timeout=timeout
)
return dict(raw_response.headers), sdk_compatible_transcription_response(raw_response.parse())
raw_stream: Final = await client.audio.transcriptions.with_raw_response.create(
**_streaming_params(data), timeout=timeout
)
if _is_event_stream(raw_stream.http_response):
return dict(raw_stream.headers), raw_stream.parse(to=AsyncStream[TranscriptionStreamEvent])
await raw_stream.http_response.aread()
return dict(raw_stream.headers), _unstreamed_transcription_response(raw_stream.http_response)
@dataclass
class AudioTranscriptionRequestData:
"""

36
litellm/main.py
View file

@ -8252,37 +8252,25 @@ async def atranscription(
def _validate_gpt_transcription_request(
model: str,
pricing_model: str,
custom_llm_provider: str,
language: str | None,
languages: Sequence[str] | None,
response_format: str | None,
drop_params: object,
) -> str | None:
model_cost_key: Final = next(
(
key
for key in (f"{custom_llm_provider}/{model}", model)
for key in (f"{custom_llm_provider}/{pricing_model}", pricing_model)
if (as_str_mapping(litellm.model_cost.get(key)) or {}).get("litellm_provider") == custom_llm_provider
),
None,
)
model_info: Final = (
get_model_info(model=model, custom_llm_provider=custom_llm_provider) if model_cost_key is not None else None
get_model_info(model=pricing_model, custom_llm_provider=custom_llm_provider)
if model_cost_key is not None
else None
)
supported_endpoints: Final = model_info.get("supported_endpoints") if model_info is not None else None
provider_specific_entry: Final = model_info.get("provider_specific_entry") if model_info is not None else None
if language is not None and languages is not None:
raise litellm.UnsupportedParamsError(
message="language and languages cannot be used together",
model=model,
llm_provider=custom_llm_provider,
)
if supported_endpoints is not None and "/v1/audio/transcriptions" not in supported_endpoints:
raise litellm.UnsupportedParamsError(
message=f"{model} is available through the Realtime API, not file transcription",
model=model,
llm_provider=custom_llm_provider,
)
if (
provider_specific_entry is None
or provider_specific_entry.get("transcription_json_or_text_only") != 1
@ -8363,11 +8351,18 @@ def transcription(
api_key = dynamic_api_key if dynamic_api_key is not None else api_key
litellm_params_dict: Final = get_litellm_params(**kwargs)
configured_base_model: Final = litellm_params_dict.get("base_model")
pricing_model: Final = (
configured_base_model.removeprefix(f"{custom_llm_provider}/")
if isinstance(configured_base_model, str)
else model
)
supported_response_format: Final = _validate_gpt_transcription_request(
model=model,
pricing_model=pricing_model,
custom_llm_provider=custom_llm_provider,
language=language,
languages=languages,
response_format=response_format,
drop_params=kwargs.get("drop_params"),
)
@ -8386,8 +8381,6 @@ def transcription(
**non_default_params,
)
litellm_params_dict: Final = get_litellm_params(**kwargs)
litellm_logging_obj.update_environment_variables(
model=model,
user=user,
@ -8415,6 +8408,7 @@ def transcription(
provider_config: Final = ProviderConfigManager.get_provider_audio_transcription_config(
model=model,
provider=LlmProviders(custom_llm_provider),
base_model=pricing_model,
)
uses_openai_transport: Final = custom_llm_provider in OPENAI_AUDIO_TRANSCRIPTION_PROVIDERS and not (

24
litellm/proxy/_lazy_openapi_snapshot.json
View file

@ -53586,13 +53586,6 @@
"delay": {
"anyOf": [
{
"enum": [
"minimal",
"low",
"medium",
"high",
"xhigh"
],
"type": "string"
},
{
@ -53877,11 +53870,6 @@
"type": {
"anyOf": [
{
"enum": [
"realtime",
"transcription",
"translation"
],
"type": "string"
},
{
@ -54152,13 +54140,6 @@
"delay": {
"anyOf": [
{
"enum": [
"minimal",
"low",
"medium",
"high",
"xhigh"
],
"type": "string"
},
{
@ -54443,11 +54424,6 @@
"type": {
"anyOf": [
{
"enum": [
"realtime",
"transcription",
"translation"
],
"type": "string"
},
{

3
litellm/proxy/_types.py
View file

@ -329,7 +329,6 @@ class LiteLLMRoutes(enum.Enum):
"/openai/v1/realtime?{model}",
"/realtime/translations",
"/v1/realtime/translations",
"/openai/v1/realtime/translations",
# realtime (GA WebRTC HTTP routes)
"/realtime/client_secrets",
"/v1/realtime/client_secrets",
@ -342,10 +341,8 @@ class LiteLLMRoutes(enum.Enum):
"/openai/v1/realtime/transcription_sessions",
"/realtime/translations/client_secrets",
"/v1/realtime/translations/client_secrets",
"/openai/v1/realtime/translations/client_secrets",
"/realtime/translations/calls",
"/v1/realtime/translations/calls",
"/openai/v1/realtime/translations/calls",
# responses API
"/responses",
"/v1/responses",

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

@ -18,7 +18,6 @@ _MANAGED_REALTIME_ROUTES: Final = frozenset(
"/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"))

1
litellm/proxy/auth/auth_utils.py
View file

@ -1958,6 +1958,7 @@ _MODEL_ROUTING_COMPLETION_MODEL_ROUTE_MARKERS: Final = ("/evals",)
# can_key_call_model() validates the real target model.
_MODEL_ROUTING_SESSION_MODEL_ROUTE_MARKERS: Final = (
"/realtime/client_secrets",
"/realtime/translations/client_secrets",
"/realtime/calls",
)
_MODEL_ROUTING_ID_FIELDS: Final = (

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

@ -536,7 +536,7 @@ async def get_form_data(request: Request) -> dict[str, Any]:
form_items: Final = tuple(form.multi_items() if hasattr(form, "multi_items") else form.items())
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)
normalized_keys: Final = tuple(dict.fromkeys(key for key, _ in normalized_items))
return {
key: [value for item_key, value in normalized_items if item_key == key]
if key in array_keys

62
litellm/proxy/proxy_server.py
View file

@ -10382,6 +10382,23 @@ def _make_keepalive_resolver(request_data: Mapping[str, object]) -> Callable[[ob
return _resolve
def _sse_error_frame(client_error: Exception) -> str:
error_returned: Final = json.dumps(
{
"error": sse_error_payload(client_error)[1]
if isinstance(client_error, HTTPException)
else ProxyException(
message=getattr(client_error, "message", str(client_error)),
type=getattr(client_error, "type", "None"),
param=getattr(client_error, "param", "None"),
code=getattr(client_error, "status_code", 500),
).to_dict()
},
default=str,
)
return f"data: {error_returned}\n\n"
async def async_data_generator(
response,
user_api_key_dict: UserAPIKeyAuth,
@ -10623,21 +10640,8 @@ async def async_data_generator(
return
if isinstance(e, HTTPException) and (not stream_started or not ends_with_openai_done):
raise client_error
error_returned: Final = json.dumps(
{
"error": sse_error_payload(client_error)[1]
if isinstance(client_error, HTTPException)
else ProxyException(
message=getattr(client_error, "message", str(client_error)),
type=getattr(client_error, "type", "None"),
param=getattr(client_error, "param", "None"),
code=getattr(client_error, "status_code", 500),
).to_dict()
},
default=str,
)
stream_completed = True
yield f"data: {error_returned}\n\n"
yield _sse_error_frame(client_error)
if ends_with_openai_done:
yield _OPENAI_STREAM_DONE_FRAME
finally:
@ -13466,6 +13470,7 @@ async def audio_transcriptions(
response_cost: Final = hidden_params.get("response_cost", None) or ""
response_call_id: Final = hidden_params.get("litellm_call_id", None) or litellm_call_id
additional_headers: Final[dict] = hidden_params.get("additional_headers", {}) or {}
stream_logging_obj: Final = getattr(response, "logging_obj", None)
custom_headers: Final = ProxyBaseLLMRequestProcessing.get_custom_headers(
user_api_key_dict=user_api_key_dict,
@ -13478,6 +13483,7 @@ async def audio_transcriptions(
call_id=response_call_id,
request_data=data,
hidden_params=hidden_params,
litellm_logging_obj=stream_logging_obj if isinstance(stream_logging_obj, LiteLLMLoggingObj) else None,
**additional_headers,
)
@ -13499,6 +13505,13 @@ async def audio_transcriptions(
try:
async for event in stream:
yield f"data: {event.model_dump_json()}\n\n"
except Exception as stream_error: # noqa: BLE001 # stream boundary: a mid-stream failure ends the SSE body as an error event
replacement: Final = await proxy_logging_obj.post_call_failure_hook(
user_api_key_dict=user_api_key_dict,
original_exception=stream_error,
request_data=data,
)
yield _sse_error_frame(stream_error if replacement is None else replacement)
finally:
close: Final = getattr(stream, "aclose", None) or getattr(stream, "close", None)
if callable(close):
@ -13659,23 +13672,9 @@ async def _reject_realtime_session(
await _release_realtime_max_parallel_slot(user_api_key_dict)
def _resolve_realtime_upstream_query_model(
model: str | None,
intent: str | None,
is_translation: bool,
route_model: str,
) -> str | None:
if intent == "transcription":
return None
if is_translation:
return route_model
return model
@app.websocket("/openai/v1/realtime")
@app.websocket("/v1/realtime")
@app.websocket("/realtime")
@app.websocket("/openai/v1/realtime/translations")
@app.websocket("/v1/realtime/translations")
@app.websocket("/realtime/translations")
async def realtime_websocket_endpoint(
@ -13738,12 +13737,7 @@ async def _route_realtime_websocket_session(
is_translation: bool = False,
) -> None:
# Only use explicit parameters, not all query params
query_model: Final = _resolve_realtime_upstream_query_model(
model=model,
intent=intent,
is_translation=is_translation,
route_model=route_model,
)
query_model: Final = route_model if is_translation else model
query_params: Final = cast( # cast-ok: cached tuples contain only the declared realtime query keys
RealtimeQueryParams,
dict(_realtime_query_params_template(query_model, intent)),

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

@ -2,14 +2,17 @@
import json
import time
from collections.abc import Mapping
from collections.abc import Generator, Mapping
from contextlib import contextmanager
from typing import TYPE_CHECKING, Any, Final
import httpx
from fastapi import APIRouter, Depends, HTTPException, Request, Response
from fastapi import status as http_status
from fastapi.responses import PlainTextResponse
from pydantic import ValidationError
import litellm
from litellm._logging import verbose_proxy_logger
from litellm.proxy._types import ProxyException, UserAPIKeyAuth
from litellm.proxy.auth.auth_checks import can_key_call_resolved_model
@ -51,6 +54,14 @@ _NON_BILLABLE_REALTIME_PROTOCOL_MESSAGE: Final = (
)
@contextmanager
def _invalid_request_body_answers_400() -> Generator[None, None, None]:
try:
yield
except ValidationError as error:
raise ProxyException(message=str(error), type="invalid_request_error", param=None, code=400) from error
_TRANSLATION_SECRET_BODY: Final = {
"requestBody": inline_request_body(
RealtimeClientSecretRequest,
@ -151,6 +162,16 @@ def _set_transcription_model_on_session(
}
def nested_transcription_upstream_model(model: str, team_id: str | None, llm_router: "Router | None") -> str:
deployment_model: Final = llm_router.get_routable_upstream_model(model, team_id) if llm_router is not None else None
if deployment_model is None:
return model
try:
return litellm.get_llm_provider(model=deployment_model)[0]
except litellm.BadRequestError:
return deployment_model
async def _authorize_and_bind_nested_transcription_models(
session_data: dict, # mutable-ok: session payload is rewritten in place for provider serialization
user_api_key_dict: UserAPIKeyAuth,
@ -168,7 +189,7 @@ async def _authorize_and_bind_nested_transcription_models(
if nested_models:
_set_transcription_model_on_session(
session=session_data,
model=nested_models[0],
model=nested_transcription_upstream_model(nested_models[0], user_api_key_dict.team_id, llm_router),
)
@ -184,7 +205,7 @@ async def _prepare_client_secret_session(
raise HTTPException(status_code=400, detail="Translation sessions require the translations endpoint")
session_type: Final = forced_session_type or _coerce_realtime_session_type(requested_session_type)
session_data: Final[dict | None] = (
req.session.model_dump(exclude_none=True) if req.session else ({} if session_type == "translation" else None)
req.session.model_dump(exclude_unset=True) if req.session else ({} if session_type == "translation" else None)
)
if session_data is not None:
session_data["type"] = session_type
@ -334,7 +355,8 @@ async def create_realtime_client_secret(
data: dict = {}
try:
body: Final = await read_request_body(request=request)
req: Final = RealtimeClientSecretRequest(**body)
with _invalid_request_body_answers_400():
req: Final = RealtimeClientSecretRequest.model_validate(body)
is_translation_request: Final = "/realtime/translations/client_secrets" in request.url.path
model, session_data, session_type = await _prepare_client_secret_session(
@ -683,7 +705,8 @@ async def create_realtime_transcription_session(
data: dict = {}
try:
body: Final = await read_request_body(request=request)
req: Final = RealtimeTranscriptionSessionRequest(**body)
with _invalid_request_body_answers_400():
req: Final = RealtimeTranscriptionSessionRequest.model_validate(body)
model: Final[str] = req.resolved_model() or "gpt-realtime-whisper"
await can_key_call_resolved_model(

62
litellm/realtime_api/main.py
View file

@ -249,6 +249,14 @@ def _get_realtime_http_extra_headers(
)
def _client_secret_session_body(req: RealtimeClientSecretRequest, model_name: str) -> Mapping[str, object] | None:
if req.session is not None:
return with_resolved_session_model(req.session.model_dump(exclude_unset=True), model_name)
if req.model is not None:
return {"type": "realtime", "model": model_name}
return None
@wrapper_client
async def acreate_realtime_client_secret(
model: str | None = None,
@ -305,20 +313,16 @@ async def acreate_realtime_client_secret(
litellm_params={"api_base": resolved_api_base},
custom_llm_provider=custom_llm_provider,
)
request_data: Final = req.model_dump(exclude_none=True, exclude={"model"})
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"] = {
"type": "realtime",
"model": model_name,
}
session_body: Final = _client_secret_session_body(req, model_name)
return await base_llm_http_handler.async_realtime_client_secret_handler(
api_base=resolved_api_base,
api_key=await _realtime_http_bearer_token(
custom_llm_provider, resolved_api_key, resolved_api_base, litellm_params
),
request_data=request_data,
request_data={
**req.model_dump(exclude_none=True, exclude={"model", "session"}),
**({} if session_body is None else {"session": session_body}),
},
logging_obj=litellm_logging_obj,
timeout=timeout or request_timeout,
provider_config=provider_config,
@ -384,18 +388,16 @@ async def acreate_realtime_translation_client_secret(
litellm_params={"api_base": resolved_api_base},
custom_llm_provider=custom_llm_provider,
)
request_data: Final = req.model_dump(
exclude_none=True,
exclude={"model"},
)
request_data["session"] = with_resolved_session_model(request_data["session"], model_name)
request_data["session"].pop("type", None)
translation_session: Final = with_resolved_session_model(session_config.model_dump(exclude_unset=True), model_name)
return await base_llm_http_handler.async_realtime_translation_client_secret_handler(
api_base=resolved_api_base,
api_key=await _realtime_http_bearer_token(
custom_llm_provider, resolved_api_key, resolved_api_base, litellm_params
),
request_data=request_data,
request_data={
**req.model_dump(exclude_none=True, exclude={"model", "session"}),
"session": {key: value for key, value in translation_session.items() if key != "type"},
},
logging_obj=litellm_logging_obj,
timeout=timeout or request_timeout,
provider_config=provider_config,
@ -406,6 +408,22 @@ async def acreate_realtime_translation_client_secret(
)
def _with_upstream_transcription_model(request_data: Mapping[str, object], model_name: str) -> dict[str, object]:
flat: Final = as_str_mapping(request_data.get("input_audio_transcription"))
audio: Final = as_str_mapping(request_data.get("audio")) or {}
audio_input: Final = as_str_mapping(audio.get("input")) or {}
nested: Final = as_str_mapping(audio_input.get("transcription"))
return {
**request_data,
**({} if flat is None else {"input_audio_transcription": {**flat, "model": model_name}}),
**(
{}
if nested is None
else {"audio": {**audio, "input": {**audio_input, "transcription": {**nested, "model": model_name}}}}
),
}
@wrapper_client
async def acreate_realtime_transcription_session(
model: str | None = None,
@ -464,13 +482,11 @@ async def acreate_realtime_transcription_session(
litellm_params={"api_base": resolved_api_base},
custom_llm_provider=custom_llm_provider,
)
request_data: Final = req.model_dump(exclude_none=True, exclude={"model"})
# Ensure the upstream body's input_audio_transcription.model matches the
# authorized routing model. This prevents a caller from supplying an allowed
# top-level model for auth while sneaking a different model into the nested
# transcription config that gets forwarded to the provider.
if isinstance(request_data.get("input_audio_transcription"), dict):
request_data["input_audio_transcription"]["model"] = model_name
# The upstream transcription model must match the authorized routing model, so a
# caller cannot pass auth with an allowed top-level model and forward a different one.
request_data: Final = _with_upstream_transcription_model(
req.model_dump(exclude_unset=True, exclude={"model"}), model_name
)
return await base_llm_http_handler.async_realtime_transcription_session_handler(
api_base=resolved_api_base,
api_key=await _realtime_http_bearer_token(

7
litellm/types/realtime.py
View file

@ -1,5 +1,5 @@
from collections.abc import Mapping, Sequence
from typing import Any, Literal, TypeAlias
from typing import Any, Literal
from pydantic import ConfigDict
from typing_extensions import ReadOnly, TypedDict
@ -13,7 +13,6 @@ from .llms.openai import (
)
ALL_DELTA_TYPES = Literal["text", "audio"]
RealtimeSessionType: TypeAlias = Literal["realtime", "transcription", "translation"]
class RealtimeResponseTransformInput(TypedDict):
@ -72,7 +71,7 @@ class RealtimeAudioTranscriptionConfig(LiteLLMBaseModel):
model_config = ConfigDict(extra="allow", frozen=True)
model: str | None = None
delay: Literal["minimal", "low", "medium", "high", "xhigh"] | None = None
delay: str | None = None
keywords: Sequence[str] | None = None
language: str | None = None
languages: Sequence[str] | None = None
@ -114,7 +113,7 @@ class RealtimeSessionConfig(LiteLLMBaseModel):
model_config = ConfigDict(extra="allow", frozen=True)
type: RealtimeSessionType | None = None
type: str | None = None
model: str | None = None
instructions: str | None = None
audio: RealtimeSessionAudioConfig | None = None

21
litellm/types/utils.py
View file

@ -44,6 +44,8 @@ from pydantic import (
JsonValue,
PrivateAttr,
SkipValidation,
TypeAdapter,
ValidationError,
field_serializer,
field_validator,
)
@ -775,10 +777,8 @@ MCP_GUARDRAIL_CALL_TYPES: Final[frozenset[str]] = frozenset(
API_ROUTE_TO_CALL_TYPES: Final[Mapping[str, Sequence[CallTypes]]] = {
"/v1/realtime/translations/client_secrets": (CallTypes.acreate_realtime_translation_client_secret,),
"/realtime/translations/client_secrets": (CallTypes.acreate_realtime_translation_client_secret,),
"/openai/v1/realtime/translations/client_secrets": (CallTypes.acreate_realtime_translation_client_secret,),
"/v1/realtime/translations/calls": (CallTypes.arealtime_translation_calls,),
"/realtime/translations/calls": (CallTypes.arealtime_translation_calls,),
"/openai/v1/realtime/translations/calls": (CallTypes.arealtime_translation_calls,),
# Chat Completions
"/chat/completions": [CallTypes.acompletion, CallTypes.completion],
"/v1/chat/completions": [CallTypes.acompletion, CallTypes.completion],
@ -1102,7 +1102,6 @@ API_ROUTE_TO_CALL_TYPES: Final[Mapping[str, Sequence[CallTypes]]] = {
"/openai/v1/realtime": (CallTypes.arealtime,),
"/realtime/translations": (CallTypes.arealtime,),
"/v1/realtime/translations": (CallTypes.arealtime,),
"/openai/v1/realtime/translations": (CallTypes.arealtime,),
"/live/sessions": (CallTypes.arealtime,),
"/v1/live/sessions": (CallTypes.arealtime,),
"/openai/v1/live/sessions": (CallTypes.arealtime,),
@ -2884,6 +2883,9 @@ class TranscriptionDetectedLanguage(LiteLLMBaseModel):
code: str
_DETECTED_LANGUAGES: Final = TypeAdapter(tuple[TranscriptionDetectedLanguage, ...])
class TranscriptionResponse(OpenAIObject):
text: str | None = None
usage: TranscriptionUsageDurationObject | TranscriptionUsageTokensObject | None = None
@ -2907,7 +2909,18 @@ class TranscriptionResponse(OpenAIObject):
languages: Sequence[TranscriptionDetectedLanguage | Mapping[str, object]] | None = None,
**kwargs: object, # kwargs-ok: OpenAI-compatible responses preserve provider extension fields
) -> None:
super().__init__(text=text, usage=usage, languages=languages, **kwargs)
super().__init__(
text=text, usage=usage, languages=TranscriptionResponse.detected_languages(languages), **kwargs
)
if self.languages is None:
del self.languages
@staticmethod
def detected_languages(value: object) -> tuple[TranscriptionDetectedLanguage, ...] | None:
try:
return _DETECTED_LANGUAGES.validate_python(value, from_attributes=True)
except ValidationError:
return None
def set_audio_transcription_duration(self, duration: float) -> None:
self._hidden_params["audio_transcription_duration"] = duration

3
litellm/utils.py
View file

@ -9781,6 +9781,7 @@ class ProviderConfigManager:
def get_provider_audio_transcription_config(
model: str,
provider: LlmProviders,
base_model: str | None = None,
) -> BaseAudioTranscriptionConfig | None:
model_cost_entry: Final = _get_model_cost_entry_for_provider_config(
model=model,
@ -9808,7 +9809,7 @@ class ProviderConfigManager:
return XAIAudioTranscriptionConfig()
elif litellm.LlmProviders.OPENAI == provider:
if model == "gpt-transcribe":
if "gpt-transcribe" in (model, base_model):
from litellm.llms.openai.transcriptions.gpt_transformation import (
OpenAIGPTTranscribeAudioTranscriptionConfig,
)

6
scripts/check_openapi_docs.py
View file

@ -18,6 +18,7 @@ from collections.abc import Iterator, Mapping
from typing import Final, NamedTuple
GATED_METHODS: Final = frozenset({"post"})
SCHEMA_MEDIA_TYPES: Final = ("application/json", "application/sdp")
class Violation(NamedTuple):
@ -32,7 +33,10 @@ def _as_mapping(node: object) -> Mapping[str, object]:
def _content_schema(container: object) -> object:
content: Final = _as_mapping(_as_mapping(container).get("content"))
return next((schema for media in content.values() if (schema := _as_mapping(media).get("schema"))), None)
return next(
(schema for media_type in SCHEMA_MEDIA_TYPES if (schema := _as_mapping(content.get(media_type)).get("schema"))),
None,
)
def _operation_codes(operation: Mapping[str, object]) -> Iterator[str]:

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

@ -3,7 +3,6 @@ import base64
import json
from collections.abc import Coroutine, Mapping, Sequence
from dataclasses import dataclass
from datetime import UTC, datetime
from typing import Final, Literal
from unittest.mock import AsyncMock, MagicMock, patch
@ -20,7 +19,6 @@ from litellm.constants import (
REALTIME_SESSION_SUCCESS_LOGGED_KEY,
)
from litellm.integrations.custom_guardrail import CustomGuardrail
from litellm.litellm_core_utils.litellm_logging import Logging
from litellm.litellm_core_utils.realtime_streaming import (
RealTimeStreaming,
client_sent_openai_beta_realtime_header,
@ -31,14 +29,10 @@ from litellm.llms.xai.realtime.transformation import XAIRealtimeNormalizer
from litellm.llms.xai.realtime.usage_meter import XAIRealtimeUsageMeter
from litellm.proxy._types import UserAPIKeyAuth
from litellm.proxy.litellm_pre_call_utils import _apply_resolved_guardrails_to_metadata
from litellm.realtime_api.main import _build_litellm_metadata
from litellm.types.guardrails import GuardrailEventHooks
from litellm.types.llms.xai import RealtimeMeteredUsage
from litellm.types.realtime import RealtimeResponseTransformInput, RealtimeResponseTypedDict
from litellm.types.utils import GenericGuardrailAPIInputs
from litellm.types.proxy.policy_engine import Policy, PolicyCondition, PolicyMatchContext
from litellm.types.proxy.policy_engine.pipeline_types import GuardrailPipeline, PipelineStep
def _mock_logging() -> MagicMock:
@ -517,6 +511,74 @@ async def test_backend_to_client_beta_receives_normalized_events():
assert sent["output_index"] == 0
@pytest.mark.asyncio
@pytest.mark.parametrize("event_type", ("session.created", "session.updated"))
async def test_beta_client_bridged_to_a_ga_backend_receives_the_beta_session_shape(event_type: str) -> None:
client_ws = MagicMock()
client_ws.scope = {"headers": [(b"openai-beta", b"realtime=v1")]}
client_ws.send_text = AsyncMock()
backend_ws = MagicMock()
backend_ws.recv = AsyncMock(
side_effect=[
json.dumps(
{
"type": event_type,
"event_id": "e1",
"session": {
"type": "realtime",
"object": "realtime.session",
"id": "sess_1",
"model": "gpt-realtime-2.1-mini",
"instructions": "Answer briefly",
"output_modalities": ["audio"],
"max_output_tokens": "inf",
"tools": [],
"audio": {
"input": {
"format": {"type": "audio/pcmu"},
"transcription": {"model": "gpt-transcribe"},
"noise_reduction": None,
"turn_detection": {"type": "server_vad", "create_response": True},
},
"output": {"format": {"type": "audio/pcm", "rate": 24000}, "voice": "marin", "speed": 1.1},
},
},
}
).encode(),
ConnectionClosed(None, None),
]
)
logging_obj: Final = _mock_logging()
logging_obj.async_success_handler = AsyncMock()
logging_obj.success_handler = MagicMock()
streaming: Final = RealTimeStreaming(client_ws, backend_ws, logging_obj, backend_uses_beta_protocol=False)
await streaming.backend_to_client_send_messages()
sent: Final = json.loads(client_ws.send_text.call_args_list[0].args[0])
assert sent["type"] == event_type
assert sent["session"] == {
"object": "realtime.session",
"id": "sess_1",
"model": "gpt-realtime-2.1-mini",
"instructions": "Answer briefly",
"modalities": ["audio", "text"],
"max_response_output_tokens": "inf",
"tools": [],
"input_audio_format": "g711_ulaw",
"input_audio_transcription": {"model": "gpt-transcribe"},
"input_audio_noise_reduction": None,
"turn_detection": {"type": "server_vad", "create_response": True},
"output_audio_format": "pcm16",
"voice": "marin",
"speed": 1.1,
}, (
"Beta and GA session fields, checked 2026-10-20: "
"https://github.com/openai/openai-python/blob/v2.33.0/src/openai/types/beta/realtime/session.py and "
"https://github.com/openai/openai-python/blob/v2.33.0/src/openai/types/realtime/realtime_session_create_response.py"
)
@pytest.mark.asyncio
async def test_backend_to_client_stores_normalized_events_for_logging():
client_ws = MagicMock()
@ -1246,58 +1308,112 @@ async def test_translation_session_update_rejects_disallowed_nested_transcriptio
assert client_ws.close.await_args.kwargs["code"] == 1008
def _translation_transcription_update(transcription: Mapping[str, object]) -> str:
return json.dumps(
{
"type": "session.update",
"session": {"type": "translation", "audio": {"input": {"transcription": transcription}}},
}
)
@pytest.mark.asyncio
async def test_translation_session_update_binds_nested_transcription_model() -> None:
backend_ws = MagicMock()
backend_ws.send = AsyncMock()
streaming = RealTimeStreaming(
async def test_translation_session_update_authorizes_each_new_nested_transcription_model() -> None:
backend_ws: Final = MagicMock(send=AsyncMock())
streaming: Final = RealTimeStreaming(
MagicMock(),
backend_ws,
MagicMock(),
model="gpt-realtime-translate",
user_api_key_dict=UserAPIKeyAuth(
models=["gpt-realtime-translate", "gpt-realtime-whisper", "gpt-live-transcribe"]
),
translation_session=True,
)
for transcription in (
{"model": "gpt-realtime-whisper", "language": "en"},
{"model": "gpt-live-transcribe", "language": "fr"},
{"language": "de"},
):
await streaming._send_to_backend(_translation_transcription_update(transcription))
assert [
json.loads(call.args[0])["session"]["audio"]["input"]["transcription"]
for call in backend_ws.send.await_args_list
] == [
{"model": "gpt-realtime-whisper", "language": "en"},
{"model": "gpt-live-transcribe", "language": "fr"},
{"model": "gpt-live-transcribe", "language": "de"},
]
@pytest.mark.asyncio
async def test_translation_session_update_refuses_a_later_nested_transcription_model_the_key_cannot_call() -> None:
client_ws: Final = MagicMock(send_text=AsyncMock(), close=AsyncMock())
backend_ws: Final = MagicMock(send=AsyncMock())
streaming: Final = RealTimeStreaming(
client_ws,
backend_ws,
MagicMock(),
model="gpt-realtime-translate",
user_api_key_dict=UserAPIKeyAuth(models=["gpt-realtime-translate", "gpt-realtime-whisper"]),
translation_session=True,
)
await streaming._send_to_backend(
json.dumps(
{
"type": "session.update",
"session": {
"type": "translation",
"audio": {
"input": {
"transcription": {"model": "gpt-realtime-whisper", "language": "en"},
}
},
},
}
)
first_sent: Final = await streaming._send_to_backend(
_translation_transcription_update({"model": "gpt-realtime-whisper", "language": "en"})
)
await streaming._send_to_backend(
json.dumps(
{
"type": "session.update",
"session": {
"type": "translation",
"audio": {
"input": {
"transcription": {"model": "gpt-live-transcribe", "language": "fr"},
}
},
},
}
)
second_sent: Final = await streaming._send_to_backend(
_translation_transcription_update({"model": "gpt-live-transcribe", "language": "fr"})
)
first_sent = json.loads(backend_ws.send.await_args_list[0].args[0])
second_sent = json.loads(backend_ws.send.await_args_list[1].args[0])
assert first_sent["session"]["audio"]["input"]["transcription"]["model"] == "gpt-realtime-whisper"
assert second_sent["session"]["audio"]["input"]["transcription"] == {
"model": "gpt-realtime-whisper",
"language": "fr",
}
assert streaming._is_transcription_session is False
assert (first_sent, second_sent) == (True, False)
assert backend_ws.send.await_count == 1
error: Final = json.loads(client_ws.send_text.await_args.args[0])
assert error["error"]["type"] == "permission_error"
assert "gpt-live-transcribe" in error["error"]["message"]
assert client_ws.close.await_args.kwargs["code"] == 1008
@pytest.mark.asyncio
async def test_translation_session_update_sends_the_deployment_model_for_a_nested_transcription_alias(
monkeypatch: pytest.MonkeyPatch,
) -> None:
from litellm.proxy import proxy_server
model_list: Final = [
{
"model_name": "whisper-alias",
"litellm_params": {
"model": "azure/qa-whisper-deployment",
"api_base": "https://realtime-qa.openai.azure.com",
"api_key": "azure-test-key",
},
}
]
monkeypatch.setattr(proxy_server, "llm_router", litellm.Router(model_list=model_list))
monkeypatch.setattr(proxy_server, "llm_model_list", model_list)
backend_ws: Final = MagicMock(send=AsyncMock())
streaming: Final = RealTimeStreaming(
MagicMock(),
backend_ws,
MagicMock(),
model="gpt-realtime-translate",
user_api_key_dict=UserAPIKeyAuth(models=["gpt-realtime-translate", "whisper-alias"]),
translation_session=True,
)
await streaming._send_to_backend(_translation_transcription_update({"model": "whisper-alias", "language": "en"}))
await streaming._send_to_backend(_translation_transcription_update({"language": "fr"}))
assert [
json.loads(call.args[0])["session"]["audio"]["input"]["transcription"]
for call in backend_ws.send.await_args_list
] == [
{"model": "qa-whisper-deployment", "language": "en"},
{"model": "qa-whisper-deployment", "language": "fr"},
]
@pytest.mark.asyncio
@ -3279,80 +3395,7 @@ async def test_translation_rejects_applicable_guardrails_before_forwarding_audio
await streaming.bidirectional_forward()
messages: Final = tuple(json.loads(call.args[0]) for call in client.send_text.await_args_list)
if default_on or requested:
assert messages[0]["error"]["type"] == "guardrail_error"
client.close.assert_awaited_once_with(
code=1008, reason="Translation sessions cannot enforce configured realtime guardrails"
)
backend.send.assert_not_awaited()
else:
assert output_event in messages
backend.send.assert_awaited_once_with(input_event)
@pytest.mark.asyncio
@pytest.mark.parametrize("metadata_bucket", ("metadata", "litellm_metadata"))
@pytest.mark.parametrize("mode", ("pre_call", "post_call"))
@pytest.mark.parametrize("selected,condition_matches", ((True, True), (True, False), (False, True)))
async def test_translation_rejects_selected_pipeline_guardrails_before_forwarding_audio(
monkeypatch: pytest.MonkeyPatch,
metadata_bucket: str,
mode: Literal["pre_call", "post_call"],
selected: bool,
condition_matches: bool,
) -> None:
guardrail: Final = CustomGuardrail(
guardrail_name="speech-policy", event_hook=GuardrailEventHooks(mode), default_on=False
)
monkeypatch.setattr(litellm, "callbacks", [guardrail])
request_data: Final[dict[str, object]] = {"metadata": {}, "litellm_metadata": {"user_id": "qa-user"}}
_apply_resolved_guardrails_to_metadata(
data=request_data,
metadata_variable_name=metadata_bucket,
context=PolicyMatchContext(model="qa-translation"),
policy_names=["translation-policy"] if selected else [],
policies={
"translation-policy": Policy(
condition=PolicyCondition(model="qa-translation" if condition_matches else "unrelated-model"),
pipeline=GuardrailPipeline(mode=mode, steps=[PipelineStep(guardrail="speech-policy", on_fail="block")]),
)
},
)
input_event: Final = json.dumps(
{"type": "session.input_audio_buffer.append", "audio": base64.b64encode(bytes(4800)).decode()}
)
output_event: Final = {"type": "session.output_audio.delta", "delta": base64.b64encode(bytes(4800)).decode()}
client: Final = _ga_client_ws()
client.receive_text = AsyncMock(side_effect=[input_event, ConnectionClosed(None, None)])
client.send_text = AsyncMock()
client.close = AsyncMock()
backend: Final = MagicMock()
backend.send = AsyncMock()
backend.recv = AsyncMock(side_effect=[json.dumps(output_event), ConnectionClosed(None, None)])
logging_obj: Final = Logging(
model="gpt-realtime-translate",
messages=[],
stream=True,
call_type="_arealtime",
start_time=datetime(2026, 1, 1, tzinfo=UTC),
litellm_call_id="pipeline-rejection",
function_id="pipeline-rejection",
)
worker: Final = _InlineLoggingWorker()
streaming: Final = RealTimeStreaming(
websocket=client,
backend_ws=backend,
logging_obj=logging_obj,
translation_session=True,
request_data={"litellm_metadata": _build_litellm_metadata(request_data)},
logging_worker=worker,
)
await streaming.bidirectional_forward()
await worker.drain()
messages: Final = tuple(json.loads(call.args[0]) for call in client.send_text.await_args_list)
if selected and condition_matches:
if (default_on or requested) and event_hook is not GuardrailEventHooks.post_call:
assert messages[0]["error"]["type"] == "guardrail_error"
client.close.assert_awaited_once_with(
code=1008, reason="Translation sessions cannot enforce configured realtime guardrails"

45
tests/unit/llms/azure/test_audio_transcriptions.py
View file

@ -163,7 +163,39 @@ async def test_azure_transcription_parses_response_formats(
assert response.text == expected_text
@pytest.mark.parametrize("model", ["whisper-1", "gpt-transcribe", "custom-transcribe-deployment"])
@pytest.mark.parametrize("api_version", ["2024-06-01", "2025-03-01-preview", "v1", "latest", "preview"])
async def test_azure_router_transcription_sends_the_default_api_version(
respx_mock: respx.MockRouter,
monkeypatch: pytest.MonkeyPatch,
api_version: str,
) -> None:
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
monkeypatch.delenv("AZURE_API_VERSION", raising=False)
route: Final = respx_mock.route(host="transcribe-route.openai.azure.com").mock(
return_value=httpx.Response(200, json={"text": "hello"})
)
router: Final = litellm.Router(
model_list=[
{
"model_name": "transcriber",
"litellm_params": {
"model": "azure/gpt-transcribe",
"api_key": "test-key",
"api_base": "https://transcribe-route.openai.azure.com",
"api_version": api_version,
},
}
]
)
response: Final = await router.atranscription(model="transcriber", file=_audio_file())
sent_url: Final = route.calls.last.request.url
assert sent_url.path == "/openai/deployments/gpt-transcribe/audio/transcriptions"
assert sent_url.params.get("api-version") == litellm.AZURE_DEFAULT_API_VERSION
assert response.text == "hello"
@pytest.mark.parametrize(
("api_version", "expected_api_version"),
[
@ -173,20 +205,19 @@ async def test_azure_transcription_parses_response_formats(
("2025-03-01-preview", "2025-03-01-preview"),
],
)
async def test_azure_transcription_uses_the_deployments_route(
def test_azure_sync_transcription_uses_the_deployments_route(
respx_mock: respx.MockRouter,
monkeypatch: pytest.MonkeyPatch,
model: str,
api_version: str,
expected_api_version: str,
) -> None:
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
monkeypatch.delenv("AZURE_API_VERSION", raising=False)
route: Final = respx_mock.route(host="transcribe-route.openai.azure.com").mock(
return_value=httpx.Response(200, json={"text": "hello"})
)
response: Final = await litellm.atranscription(
model=f"azure/{model}",
response: Final = litellm.transcription(
model="azure/gpt-transcribe",
file=_audio_file(),
api_key="test-key",
api_base="https://transcribe-route.openai.azure.com",
@ -194,6 +225,6 @@ async def test_azure_transcription_uses_the_deployments_route(
)
sent_url: Final = route.calls.last.request.url
assert sent_url.path == f"/openai/deployments/{model}/audio/transcriptions"
assert sent_url.path == "/openai/deployments/gpt-transcribe/audio/transcriptions"
assert sent_url.params.get("api-version") == expected_api_version
assert response.text == "hello"

257
tests/unit/llms/openai/transcriptions/test_gpt_transcribe.py
View file

@ -13,6 +13,7 @@ from openai import AsyncAzureOpenAI, AsyncOpenAI, AsyncStream, AzureOpenAI, Open
import litellm
from litellm.litellm_core_utils.audio_utils.transcription_streaming import wrap_transcription_stream
from litellm.llms.azure.audio_transcriptions import AzureAudioTranscription
from litellm.llms.custom_httpx.http_handler import HTTPHandler
from litellm.llms.openai.transcriptions.gpt_transformation import (
OpenAIGPTTranscribeAudioTranscriptionConfig,
)
@ -64,7 +65,7 @@ def test_gpt_transcribe_optional_params_are_preserved():
def test_transcription_response_preserves_empty_languages():
response = TranscriptionResponse(text="hello", languages=[])
assert response.model_dump()["languages"] == []
assert json.loads(response.model_dump_json())["languages"] == []
def test_sync_transcription_stream_logs_final_text_and_usage_once() -> None:
@ -97,7 +98,7 @@ def test_sync_transcription_stream_logs_final_text_and_usage_once() -> None:
logged_response: Final = logging_obj.success_handler.call_args.args[0]
assert logged_response.text == "hello world"
assert logged_response.usage.model_dump(exclude_none=True) == {"type": "duration", "seconds": 2.5}
assert logged_response.model_dump()["languages"] == [{"code": "en"}]
assert json.loads(logged_response.model_dump_json())["languages"] == [{"code": "en"}]
logging_obj.failure_handler.assert_not_called()
@ -162,38 +163,7 @@ async def test_openai_handler_returns_native_typed_stream():
logging_obj.async_success_handler.assert_awaited_once()
logged_response = logging_obj.async_success_handler.await_args.kwargs["result"]
assert logged_response.text == "hello world"
assert logged_response.languages == []
@pytest.mark.asyncio
async def test_closed_transcription_stream_without_usage_or_duration_does_not_log_success():
async def send_response(request: httpx.Request) -> httpx.Response:
events = (
{"type": "transcript.text.delta", "delta": "hello"},
{"type": "transcript.text.done", "text": "hello", "usage": {"type": "duration", "seconds": 1}},
)
content = "".join(f"data: {json.dumps(event)}\n\n" for event in events).encode()
return httpx.Response(200, content=content, headers={"content-type": "text/event-stream"})
http_client = httpx.AsyncClient(transport=httpx.MockTransport(send_response))
client = AsyncOpenAI(api_key="sk-test", base_url="https://example.com/v1", http_client=http_client)
stream = await client.audio.transcriptions.create(
model="gpt-transcribe", file=("sample.webm", b"audio"), stream=True
)
logging_obj = MagicMock()
logging_obj.async_success_handler = AsyncMock()
logging_obj.async_failure_handler = AsyncMock()
wrapped_stream = wrap_transcription_stream(stream, logging_obj, datetime.now())
async for event in wrapped_stream:
assert event.type == "transcript.text.delta"
break
await wrapped_stream.close()
await client.close()
logging_obj.handle_sync_success_callbacks_for_async_calls.assert_not_called()
logging_obj.async_success_handler.assert_not_awaited()
logging_obj.async_failure_handler.assert_not_awaited()
assert logged_response.languages == ()
@pytest.mark.asyncio
@ -249,19 +219,6 @@ async def test_atranscription_stream_preserves_duration_for_callback_cost():
assert logged_response._hidden_params["audio_transcription_duration"] == pytest.approx(1.0)
def test_gpt_transcribe_rejects_conflicting_language_inputs():
audio_file = io.BytesIO(b"audio")
audio_file.name = "sample.wav"
with pytest.raises(litellm.UnsupportedParamsError, match="cannot be used together"):
litellm.transcription(
model="gpt-transcribe",
file=audio_file,
language="en",
languages=["fr"],
api_key="sk-test",
)
@pytest.mark.parametrize("model", ["gpt-transcribe", "azure/gpt-transcribe"])
def test_gpt_transcribe_rejects_whisper_response_formats(local_model_cost_map: None, model: str) -> None:
audio_file = io.BytesIO(b"audio")
@ -350,16 +307,49 @@ def test_gpt_transcribe_sends_text_response_format_to_the_provider(
assert b'name="response_format"\r\n\r\ntext\r\n' in sent_bodies[0]
def test_gpt_live_transcribe_rejects_file_transcription(local_model_cost_map: None) -> None:
audio_file = io.BytesIO(b"audio")
audio_file.name = "sample.wav"
with pytest.raises(litellm.UnsupportedParamsError, match="Realtime API"):
litellm.transcription(
model="gpt-live-transcribe",
file=audio_file,
api_key="sk-test",
def test_file_transcription_reaches_models_whose_cost_map_lists_other_endpoints(
local_model_cost_map: None, monkeypatch: pytest.MonkeyPatch
) -> None:
sent_requests: Final[list[httpx.Request]] = [] # mutable-ok: collects what the mock upstreams received
def send_gemini_response(request: httpx.Request) -> httpx.Response:
sent_requests.append(request)
return httpx.Response(
200,
json={
"id": "interaction-1",
"status": "completed",
"steps": [
{"type": "model_generation", "content": [{"type": "text", "text": "hello", "annotations": []}]}
],
},
)
def send_whisper_response(request: httpx.Request) -> httpx.Response:
sent_requests.append(request)
return httpx.Response(200, json={"text": "hello"})
gemini_response: Final = litellm.transcription(
model="gemini/gemini-2.5-flash",
file=_one_second_wav(),
api_key="gemini-test-key",
client=HTTPHandler(client=httpx.Client(transport=httpx.MockTransport(send_gemini_response))),
)
monkeypatch.setattr(litellm, "client_session", httpx.Client(transport=httpx.MockTransport(send_whisper_response)))
whisper_response: Final = litellm.transcription(
model="openai/gpt-realtime-whisper",
file=_one_second_wav(),
api_key="self-hosted-whisper-key",
api_base="https://whisper.self-hosted.example/v1",
)
assert gemini_response.text == "hello"
assert whisper_response.text == "hello"
assert tuple(str(request.url.host) for request in sent_requests) == (
"generativelanguage.googleapis.com",
"whisper.self-hosted.example",
)
def test_azure_gpt_transcribe_uses_deployment_scoped_route():
def send_response(request: httpx.Request) -> httpx.Response:
@ -616,3 +606,160 @@ def test_transcription_stream_bills_the_provider_duration_like_the_unstreamed_ca
assert unstreamed_spend.cost > 0
assert unstreamed_spend.usage
assert streamed_spend == unstreamed_spend
class _LoggedResponseRecorder(CustomLogger):
def __init__(self) -> None:
super().__init__()
self.responses: Final[queue.Queue[object]] = queue.Queue()
def log_success_event(self, kwargs, response_obj, start_time, end_time) -> None:
self.responses.put(kwargs["standard_logging_object"]["response"])
@pytest.mark.parametrize(
"provider_body",
({"text": "hello"}, {"text": "hello", "languages": ["en"]}, {"text": "hello", "languages": None}),
)
def test_transcription_without_parsed_languages_has_no_languages_key(
local_model_cost_map: None, monkeypatch: pytest.MonkeyPatch, provider_body: dict[str, object]
) -> None:
recorder: Final = _LoggedResponseRecorder()
monkeypatch.setattr(litellm, "callbacks", [recorder])
with _openai_transcription_client(
httpx.MockTransport(lambda request: httpx.Response(200, json=provider_body))
) as client:
response: Final = litellm.transcription(model="openai/whisper-1", file=_one_second_wav(), client=client)
logged_response: Final = recorder.responses.get(timeout=5)
assert response.text == "hello"
assert "languages" not in json.loads(response.model_dump_json())
assert isinstance(logged_response, dict)
assert "languages" not in logged_response
def test_unstreamed_transcription_sends_no_stream_field(local_model_cost_map: None) -> None:
sent_bodies: Final[list[bytes]] = [] # mutable-ok: collects what the mock upstream received
def send_response(request: httpx.Request) -> httpx.Response:
sent_bodies.append(request.read())
return httpx.Response(200, json={"text": "hello"})
with _openai_transcription_client(httpx.MockTransport(send_response)) as client:
response: Final = litellm.transcription(model="openai/whisper-1", file=_one_second_wav(), client=client)
assert response.text == "hello"
assert len(sent_bodies) == 1
assert b'name="stream"' not in sent_bodies[0]
def test_transcription_forwards_both_language_and_languages(local_model_cost_map: None) -> None:
sent_bodies: Final[list[bytes]] = [] # mutable-ok: collects what the mock upstream received
def send_response(request: httpx.Request) -> httpx.Response:
sent_bodies.append(request.read())
return httpx.Response(200, json={"text": "hello", "usage": {"type": "duration", "seconds": 1}})
with _openai_transcription_client(httpx.MockTransport(send_response)) as client:
response: Final = litellm.transcription(
model="openai/gpt-transcribe", file=_one_second_wav(), language="en", languages=["fr"], client=client
)
assert response.text == "hello"
assert len(sent_bodies) == 1
assert b'name="language"\r\n\r\nen\r\n' in sent_bodies[0]
assert b'name="languages[]"\r\n\r\nfr\r\n' in sent_bodies[0]
def _gpt_transcribe_cost_per_second() -> float:
rate: Final = litellm.model_cost["gpt-transcribe"]["input_cost_per_second"]
assert isinstance(rate, float)
return rate
async def _spend_for_one_second_wav(
monkeypatch: pytest.MonkeyPatch, provider_body: dict[str, object], router: litellm.Router | None
) -> _LoggedSpend:
recorder: Final = _SuccessSpendRecorder()
monkeypatch.setattr(litellm, "callbacks", [recorder])
client: Final = AsyncOpenAI(
api_key="sk-test",
base_url="https://example.com/v1",
http_client=httpx.AsyncClient(
transport=httpx.MockTransport(lambda request: httpx.Response(200, json=provider_body))
),
)
if router is None:
await litellm.atranscription(model="openai/gpt-transcribe", file=_one_second_wav(), client=client)
else:
await router.atranscription(model="my-transcribe", file=_one_second_wav(), client=client)
spend: Final = await asyncio.wait_for(recorder.async_spends.get(), timeout=5)
await client.close()
return spend
@pytest.mark.asyncio
async def test_custom_named_gpt_transcribe_deployment_bills_the_provider_seconds(
local_model_cost_map: None, monkeypatch: pytest.MonkeyPatch
) -> None:
router: Final = litellm.Router(
model_list=[
{
"model_name": "my-transcribe",
"litellm_params": {"model": "openai/my-transcribe", "api_key": "sk-test"},
"model_info": {"base_model": "gpt-transcribe"},
}
]
)
spend: Final = await _spend_for_one_second_wav(
monkeypatch, {"text": "hello", "usage": {"type": "duration", "seconds": 2}}, router
)
assert spend.cost == pytest.approx(2 * _gpt_transcribe_cost_per_second())
@pytest.mark.asyncio
async def test_gpt_transcribe_bills_the_file_length_when_the_provider_reports_zero_seconds(
local_model_cost_map: None, monkeypatch: pytest.MonkeyPatch
) -> None:
spend: Final = await _spend_for_one_second_wav(
monkeypatch, {"text": "hello", "usage": {"type": "duration", "seconds": 0}}, None
)
assert spend.cost == pytest.approx(_gpt_transcribe_cost_per_second())
@pytest.mark.asyncio
async def test_m4a_transcription_stream_that_ends_early_logs_one_spend_row_like_the_unstreamed_call(
local_model_cost_map: None, monkeypatch: pytest.MonkeyPatch
) -> None:
async def send_response(request: httpx.Request) -> httpx.Response:
if not _asks_for_stream(await request.aread()):
return httpx.Response(200, json={"text": "hello"})
return httpx.Response(
200,
content=f"data: {json.dumps({'type': 'transcript.text.delta', 'delta': 'hello'})}\n\n".encode(),
headers={"content-type": "text/event-stream"},
)
recorder: Final = _SuccessSpendRecorder()
monkeypatch.setattr(litellm, "callbacks", [recorder])
client: Final = AsyncOpenAI(
api_key="sk-test",
base_url="https://example.com/v1",
http_client=httpx.AsyncClient(transport=httpx.MockTransport(send_response)),
)
await litellm.atranscription(model="openai/gpt-transcribe", file=("sample.m4a", b"m4a audio"), client=client)
unstreamed_spend: Final = await asyncio.wait_for(recorder.async_spends.get(), timeout=5)
stream: Final = await litellm.atranscription(
model="openai/gpt-transcribe", file=("sample.m4a", b"m4a audio"), stream=True, client=client
)
events: Final = [event.type async for event in stream]
streamed_spend: Final = await asyncio.wait_for(recorder.async_spends.get(), timeout=5)
await asyncio.sleep(0.2)
await client.close()
assert events == ["transcript.text.delta"]
assert streamed_spend.cost == unstreamed_spend.cost
assert recorder.async_spends.empty()

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

@ -476,9 +476,7 @@ def test_managed_live_upgrade_defers_the_model_to_the_session_start_frame(route:
) == {"model": "frame-model"}
@pytest.mark.parametrize(
"route", ("/realtime/translations", "/v1/realtime/translations", "/openai/v1/realtime/translations")
)
@pytest.mark.parametrize("route", ("/realtime/translations", "/v1/realtime/translations"))
@pytest.mark.parametrize("query_model", (None, "requested"))
def test_managed_translation_authorizes_the_same_model_as_the_endpoint(route: str, query_model: str | None) -> None:
from litellm.proxy.agent_endpoints.auth.managed_authorization import managed_inference_request

21
tests/unit/proxy/auth/test_route_checks.py
View file

@ -788,13 +788,10 @@ def test_realtime_webrtc_http_routes_classified_as_llm_api(route):
[
"/realtime/translations",
"/v1/realtime/translations",
"/openai/v1/realtime/translations",
"/realtime/translations/client_secrets",
"/v1/realtime/translations/client_secrets",
"/openai/v1/realtime/translations/client_secrets",
"/realtime/translations/calls",
"/v1/realtime/translations/calls",
"/openai/v1/realtime/translations/calls",
],
)
def test_realtime_translation_routes_allowed_for_openai_virtual_keys(route):
@ -814,6 +811,24 @@ def test_realtime_translation_routes_allowed_for_openai_virtual_keys(route):
)
@pytest.mark.parametrize(
"route",
[
"/openai/v1/realtime/translations",
"/openai/v1/realtime/translations/client_secrets",
"/openai/v1/realtime/translations/calls",
],
)
def test_openai_prefixed_realtime_translation_routes_denied_for_openai_virtual_keys(route: str) -> None:
valid_token: Final = UserAPIKeyAuth(user_id="test_user", allowed_routes=["openai_routes"])
with pytest.raises(HTTPException) as exc_info:
RouteChecks.is_virtual_key_allowed_to_call_route(route=route, valid_token=valid_token)
assert exc_info.value.status_code == 403
assert f"Tried to call route: {route}" in str(exc_info.value.detail)
def test_virtual_key_allowed_routes_with_litellm_routes_member_name_denied():
"""Test that virtual key is denied when route is not in the allowed LiteLLMRoutes group"""

50
tests/unit/proxy/auth/test_user_api_key_auth_request_flow.py
View file

@ -11937,9 +11937,7 @@ async def test_websocket_auth_hands_the_reservation_to_the_socket_state():
@pytest.mark.asyncio
@pytest.mark.parametrize(
"path", ("/realtime/translations", "/v1/realtime/translations", "/openai/v1/realtime/translations")
)
@pytest.mark.parametrize("path", ("/realtime/translations", "/v1/realtime/translations"))
@pytest.mark.parametrize("query", (b"", b"model=gpt-realtime-translate"), ids=("default_model", "explicit_model"))
async def test_translation_websocket_auth_enforces_the_default_model_budget(
monkeypatch: pytest.MonkeyPatch, path: str, query: bytes
@ -11989,6 +11987,52 @@ async def test_translation_websocket_auth_enforces_the_default_model_budget(
assert "budget" in str(denied.value.detail).lower()
@pytest.mark.asyncio
@pytest.mark.parametrize("path", ("/realtime/translations/client_secrets", "/v1/realtime/translations/client_secrets"))
async def test_translation_client_secret_auth_enforces_the_session_model_budget(
monkeypatch: pytest.MonkeyPatch, path: str
) -> None:
from litellm.proxy import proxy_server
from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
model: Final = "gpt-realtime-translate"
token: Final = UserAPIKeyAuth(
token="test-key",
models=[model],
model_max_budget={model: {"max_budget": 5.0, "budget_duration": "1d"}},
)
limiter, _ = await _real_model_budget_limiter(key_spend={model: 10.0})
async def authenticate(request: Request, api_key: str) -> UserAPIKeyAuth:
return token
attrs: Final = {
**_proxy_server_attrs_for_custom_auth(user_custom_auth=authenticate),
"prisma_client": None,
"model_max_budget_limiter": limiter,
}
for name, value in attrs.items():
monkeypatch.setattr(proxy_server, name, value)
monkeypatch.setattr(litellm, "enable_post_custom_auth_checks", True, raising=False)
body: Final = json.dumps({"session": {"type": "translation", "model": model}}).encode()
request: Final = Request(
scope={
"type": "http",
"method": "POST",
"path": path,
"headers": [(b"authorization", b"Bearer sk-test"), (b"content-type", b"application/json")],
"query_string": b"",
},
receive=AsyncMock(return_value={"type": "http.request", "body": body, "more_body": False}),
)
with pytest.raises(ProxyException) as denied:
await user_api_key_auth(request=request, api_key="Bearer sk-test")
assert denied.value.type == "budget_exceeded"
assert model in denied.value.message
@pytest.mark.asyncio
@pytest.mark.parametrize("path", ("/realtime", "/v1/realtime", "/openai/v1/realtime"))
async def test_managed_agent_transcription_websocket_without_a_model_is_refused(

11
tests/unit/proxy/common_utils/test_http_parsing_utils.py
View file

@ -685,6 +685,17 @@ async def test_get_form_data():
assert result["temperature"] == "0.2"
assert result["include"] == ["logprobs"]
assert result["timestamp_granularities"] == ["word", "segment"]
assert tuple(result) == (
"file",
"model",
"include",
"language",
"prompt",
"response_format",
"stream",
"temperature",
"timestamp_granularities",
)
def test_get_tags_from_request_body_with_metadata_tags():

114
tests/unit/proxy/proxy_server/test_routes_audio.py
View file

@ -10,7 +10,8 @@ Pins (PR2):
from __future__ import annotations
import io
from collections.abc import Callable, Mapping
import json
from collections.abc import AsyncIterator, Callable, Mapping
from contextlib import AbstractContextManager
from typing import Final
from unittest.mock import AsyncMock, MagicMock
@ -475,3 +476,114 @@ def test_streamed_transcription_honors_a_deployment_post_call_guardrail(
else:
assert response.headers["content-type"].startswith("text/event-stream")
assert '"text":"my pin is 1234"' in response.text
def _streams_chat_and_transcription(request: httpx.Request) -> httpx.Response:
sse_body: Final = (
b'data: {"id":"chat-1","object":"chat.completion.chunk","created":0,"model":"gpt-4o",'
b'"choices":[{"index":0,"delta":{"content":"hi"},"finish_reason":"stop"}]}\n\ndata: [DONE]\n\n'
if request.url.path.endswith("/chat/completions")
else b'data: {"type":"transcript.text.done","text":"hello world"}\n\n'
)
return httpx.Response(
200,
content=sse_body,
headers={"content-type": "text/event-stream", "x-ratelimit-remaining-requests": "99"},
)
def _litellm_header_names(response: httpx.Response) -> frozenset[str]:
return frozenset(name for name in response.headers if name.startswith("x-litellm-"))
def test_streamed_transcription_carries_the_headers_a_streamed_chat_completion_carries(
client: TestClient,
auth_as: Callable[[], AbstractContextManager[None]],
monkeypatch: pytest.MonkeyPatch,
respx_mock: respx.MockRouter,
) -> None:
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
respx_mock.route(host="stream-upstream.example").mock(side_effect=_streams_chat_and_transcription)
upstream: Final = {"api_key": "sk-test", "api_base": "https://stream-upstream.example/v1"}
monkeypatch.setattr(
proxy_server,
"llm_router",
litellm.Router(
model_list=[
{"model_name": "gpt-4o", "litellm_params": {"model": "openai/gpt-4o", **upstream}},
{"model_name": "gpt-transcribe", "litellm_params": {"model": "openai/gpt-transcribe", **upstream}},
]
),
)
with auth_as():
chat: Final = client.post(
"/v1/chat/completions",
json={"model": "gpt-4o", "stream": True, "messages": [{"role": "user", "content": "hi"}]},
)
transcription: Final = client.post(
"/v1/audio/transcriptions",
files={"file": ("sample.wav", b"audio", "audio/wav")},
data={"model": "gpt-transcribe", "stream": "true"},
)
assert chat.status_code == 200
assert transcription.status_code == 200
assert transcription.headers["content-type"].startswith("text/event-stream")
assert _litellm_header_names(chat) - _litellm_header_names(transcription) <= {"x-litellm-overhead-duration-ms"}
assert transcription.headers["x-litellm-model-name"] == "openai/gpt-transcribe"
assert transcription.headers["llm_provider-x-ratelimit-remaining-requests"] == "99"
assert ("x-litellm-response-cost" in transcription.headers) == ("x-litellm-response-cost" in chat.headers)
class _FailsAfterTwoDeltas(httpx.AsyncByteStream):
async def __aiter__(self) -> AsyncIterator[bytes]:
yield b'data: {"type":"transcript.text.delta","delta":"hello "}\n\n'
yield b'data: {"type":"transcript.text.delta","delta":"world"}\n\n'
raise httpx.ReadError("upstream connection dropped")
@pytest.mark.usefixtures("patched_transcription_stream")
def test_streamed_transcription_error_mid_stream_ends_with_an_error_event_and_runs_the_failure_hook(
client: TestClient,
auth_as: Callable[[], AbstractContextManager[None]],
monkeypatch: pytest.MonkeyPatch,
) -> None:
upstream: Final = AsyncOpenAI(
api_key="sk-test",
base_url="https://upstream.example/v1",
http_client=httpx.AsyncClient(
transport=httpx.MockTransport(
lambda request: httpx.Response(
200, stream=_FailsAfterTwoDeltas(), headers={"content-type": "text/event-stream"}
)
)
),
)
async def route_to_the_sdk(
data: Mapping[str, object], route_type: str, llm_router: object, user_model: object
) -> object:
return litellm.atranscription(model="openai/gpt-transcribe", file=data["file"], stream=True, client=upstream)
failure_hook: Final = AsyncMock(return_value=None)
monkeypatch.setattr(proxy_server.proxy_logging_obj, "post_call_failure_hook", failure_hook)
monkeypatch.setattr(proxy_server, "route_request", route_to_the_sdk)
with auth_as():
response: Final = client.post(
"/v1/audio/transcriptions",
files={"file": ("sample.wav", b"audio", "audio/wav")},
data={"model": "gpt-transcribe", "stream": "true"},
)
frames: Final = tuple(
json.loads(line.removeprefix("data: ")) for line in response.text.splitlines() if line.startswith("data: ")
)
assert response.status_code == 200
assert tuple(frame.get("delta") for frame in frames[:2]) == ("hello ", "world")
assert len(frames) == 3
assert "upstream connection dropped" in frames[2]["error"]["message"]
assert set(frames[2]["error"]) == {"message", "type", "param", "code"}
failure_hook.assert_awaited_once()
assert isinstance(failure_hook.await_args.kwargs["original_exception"], httpx.ReadError)

209
tests/unit/proxy/realtime_endpoints/test_realtime_webrtc_endpoints.py
View file

@ -7,7 +7,7 @@ Tests for LiteLLM proxy realtime WebRTC HTTP endpoints:
import functools
import json
import time
from collections.abc import Awaitable
from collections.abc import Awaitable, Mapping
from typing import Final, Protocol
from unittest.mock import AsyncMock, MagicMock, patch
@ -863,20 +863,6 @@ async def test_realtime_transcription_websocket_default_model_checks_key_scope()
assert "is not available for this API key" in close_kwargs["reason"]
def test_realtime_transcription_upstream_query_omits_model():
from litellm.proxy import proxy_server
assert (
proxy_server._resolve_realtime_upstream_query_model(
model="gpt-live-transcribe",
intent="transcription",
is_translation=False,
route_model="gpt-live-transcribe",
)
is None
)
@pytest.mark.asyncio
async def test_realtime_transcription_websocket_default_model_checks_team_scope():
from litellm.proxy import proxy_server
@ -1619,6 +1605,61 @@ _AZURE_REALTIME_DEPLOYMENTS: Final = (
)
def _post_through_the_azure_deployments(
proxy_app: FastAPI,
mock_add_litellm_data: AddLitellmDataToRequest,
mock_pre_call_hook: PreCallHook,
respx_mock: respx.MockRouter,
monkeypatch: pytest.MonkeyPatch,
path: str,
body: object,
) -> tuple[httpx.Response, respx.Route]:
from litellm.proxy import proxy_server
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
model_list: Final = [ # mutable-ok: Router takes a list of deployments
{
"model_name": alias,
"litellm_params": {
"model": deployment,
"api_base": "https://realtime-qa.openai.azure.com",
"api_key": "azure-test-key",
},
}
for alias, deployment in _AZURE_REALTIME_DEPLOYMENTS
]
monkeypatch.setattr(proxy_server, "llm_router", litellm.Router(model_list=model_list))
monkeypatch.setattr(proxy_server, "llm_model_list", model_list)
upstream: Final = respx_mock.route(host="realtime-qa.openai.azure.com").mock(
return_value=httpx.Response(200, json={"value": "ek_azure", "expires_at": int(time.time()) + 3600})
)
monkeypatch.setitem(
proxy_app.dependency_overrides,
user_api_key_auth,
lambda: UserAPIKeyAuth(user_id="test-user", models=[alias for alias, _ in _AZURE_REALTIME_DEPLOYMENTS]),
)
with (
patch( # test-quality-ok: endpoint test isolates request metadata enrichment
"litellm.proxy.proxy_server.add_litellm_data_to_request", side_effect=mock_add_litellm_data
),
patch( # test-quality-ok: endpoint test isolates the process-wide proxy logger
"litellm.proxy.proxy_server.proxy_logging_obj"
) as logging,
):
logging.pre_call_hook = AsyncMock(side_effect=mock_pre_call_hook)
logging.post_call_failure_hook = AsyncMock()
response: Final = TestClient(proxy_app).post(
path, headers={"Authorization": "Bearer sk-test-master-key"}, json=body
)
return response, upstream
def _upstream_session_value(upstream: respx.Route, session_path: tuple[str, ...]) -> object:
return functools.reduce(
lambda node, key: node[key], session_path, json.loads(upstream.calls.last.request.content)["session"]
)
@pytest.mark.parametrize(
("path", "body", "upstream_model_path", "expected_model"),
(
@ -1651,6 +1692,15 @@ _AZURE_REALTIME_DEPLOYMENTS: Final = (
("model",),
"qa-translate-deployment",
),
(
"/v1/realtime/translations/client_secrets",
{
"model": "azure-translate-alias",
"session": {"audio": {"input": {"transcription": {"model": "azure-transcribe-alias"}}}},
},
("audio", "input", "transcription", "model"),
"qa-whisper-deployment",
),
),
)
def test_client_secrets_mint_a_secret_for_the_azure_deployment_model(
@ -1664,46 +1714,95 @@ def test_client_secrets_mint_a_secret_for_the_azure_deployment_model(
upstream_model_path: tuple[str, ...],
expected_model: str,
) -> None:
from litellm.proxy import proxy_server
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
model_list: Final = [ # mutable-ok: Router takes a list of deployments
{
"model_name": alias,
"litellm_params": {
"model": deployment,
"api_base": "https://realtime-qa.openai.azure.com",
"api_key": "azure-test-key",
},
}
for alias, deployment in _AZURE_REALTIME_DEPLOYMENTS
]
monkeypatch.setattr(proxy_server, "llm_router", litellm.Router(model_list=model_list))
monkeypatch.setattr(proxy_server, "llm_model_list", model_list)
upstream: Final = respx_mock.route(host="realtime-qa.openai.azure.com").mock(
return_value=httpx.Response(200, json={"value": "ek_azure", "expires_at": int(time.time()) + 3600})
response, upstream = _post_through_the_azure_deployments(
proxy_app, mock_add_litellm_data, mock_pre_call_hook, respx_mock, monkeypatch, path, body
)
proxy_app.dependency_overrides[user_api_key_auth] = lambda: UserAPIKeyAuth(
user_id="test-user", models=[alias for alias, _ in _AZURE_REALTIME_DEPLOYMENTS]
)
try:
with (
patch( # test-quality-ok: endpoint test isolates request metadata enrichment
"litellm.proxy.proxy_server.add_litellm_data_to_request", side_effect=mock_add_litellm_data
),
patch( # test-quality-ok: endpoint test isolates the process-wide proxy logger
"litellm.proxy.proxy_server.proxy_logging_obj"
) as logging,
):
logging.pre_call_hook = AsyncMock(side_effect=mock_pre_call_hook)
logging.post_call_failure_hook = AsyncMock()
response: Final = TestClient(proxy_app).post(
path, headers={"Authorization": "Bearer sk-test-master-key"}, json=body
)
finally:
proxy_app.dependency_overrides.pop(user_api_key_auth, None)
assert response.status_code == 200, response.text
upstream_session: Final = json.loads(upstream.calls.last.request.content)["session"]
upstream_model: Final = functools.reduce(lambda node, key: node[key], upstream_model_path, upstream_session)
assert upstream_model == expected_model
assert _upstream_session_value(upstream, upstream_model_path) == expected_model
@pytest.mark.parametrize(
("body", "session_path", "expected"),
(
(
{
"session": {
"type": "realtime",
"model": "azure-realtime-alias",
"audio": {"input": {"turn_detection": None}},
}
},
("audio", "input", "turn_detection"),
None,
),
(
{
"session": {
"type": "transcription",
"audio": {"input": {"transcription": {"model": "azure-transcribe-alias"}, "turn_detection": None}},
}
},
("audio", "input", "turn_detection"),
None,
),
({"session": {"type": "conversation", "model": "azure-realtime-alias"}}, ("type",), "realtime"),
(
{
"session": {
"type": "transcription",
"audio": {"input": {"transcription": {"model": "azure-transcribe-alias", "delay": "auto"}}},
}
},
("audio", "input", "transcription", "delay"),
"auto",
),
),
)
def test_client_secrets_forward_every_session_value_the_caller_set(
proxy_app: FastAPI,
mock_add_litellm_data: AddLitellmDataToRequest,
mock_pre_call_hook: PreCallHook,
respx_mock: respx.MockRouter,
monkeypatch: pytest.MonkeyPatch,
body: Mapping[str, object],
session_path: tuple[str, ...],
expected: object,
) -> None:
response, upstream = _post_through_the_azure_deployments(
proxy_app,
mock_add_litellm_data,
mock_pre_call_hook,
respx_mock,
monkeypatch,
"/v1/realtime/client_secrets",
body,
)
assert response.status_code == 200, response.text
assert _upstream_session_value(upstream, session_path) == expected
@pytest.mark.parametrize(
("path", "body"),
(
("/v1/realtime/client_secrets", {"session": "not-a-session"}),
("/v1/realtime/translations/client_secrets", {"session": {"audio": {"input": "not-an-input"}}}),
("/v1/realtime/transcription_sessions", {"audio": {"input": {"transcription": "not-a-transcription"}}}),
),
)
def test_realtime_session_routes_answer_400_for_a_malformed_body(
proxy_app: FastAPI,
mock_add_litellm_data: AddLitellmDataToRequest,
mock_pre_call_hook: PreCallHook,
respx_mock: respx.MockRouter,
monkeypatch: pytest.MonkeyPatch,
path: str,
body: Mapping[str, object],
) -> None:
response, upstream = _post_through_the_azure_deployments(
proxy_app, mock_add_litellm_data, mock_pre_call_hook, respx_mock, monkeypatch, path, body
)
assert (response.status_code, response.json()["error"]["type"]) == (400, "invalid_request_error")
assert upstream.called is False

86
tests/unit/proxy/test_proxy_server.py
View file

@ -39,6 +39,7 @@ from fastapi import FastAPI, HTTPException, WebSocket
# test /chat/completion request to the proxy
from fastapi.testclient import TestClient
from starlette.testclient import WebSocketTestSession
from litellm.integrations.custom_logger import CustomLogger
from litellm.proxy.common_utils.user_api_key_cache import UserApiKeyCache
@ -3459,6 +3460,91 @@ def test_live_session_pre_call_abort_releases_reservations(
route_call.assert_not_awaited()
def _receive_until_closed(websocket: WebSocketTestSession) -> None:
while True:
websocket.receive_text()
@pytest.mark.parametrize(
("litellm_params", "expected_upstream_url"),
(
pytest.param(
{"model": "openai/gateway-transcribe", "api_base": "https://gateway.example/v1"},
"wss://gateway.example/v1/realtime?model=gateway-transcribe&intent=transcription",
id="gateway",
),
pytest.param(
{"model": "xai/grok-voice", "api_base": "https://api.x.ai/v1"},
"wss://api.x.ai/v1/realtime?model=grok-voice&intent=transcription",
id="xai",
),
pytest.param(
{"model": "openai/gpt-live-transcribe", "api_base": "https://api.openai.com/v1"},
"wss://api.openai.com/v1/realtime?intent=transcription",
id="openai",
),
),
)
def test_realtime_transcription_websocket_keeps_the_model_upstream_except_on_openai_hosts(
client_no_auth: TestClient,
monkeypatch: pytest.MonkeyPatch,
litellm_params: Mapping[str, str],
expected_upstream_url: str,
) -> None:
from starlette.websockets import WebSocketDisconnect
from litellm.proxy import proxy_server
monkeypatch.setattr(
proxy_server,
"llm_router",
litellm.Router(
model_list=[{"model_name": "team-transcriber", "litellm_params": {**litellm_params, "api_key": "fake"}}]
),
)
connect: Final = MagicMock(side_effect=RuntimeError("upstream url captured"))
monkeypatch.setattr("websockets.connect", connect)
with (
client_no_auth.websocket_connect(
"/v1/realtime?model=team-transcriber&intent=transcription",
headers={"Authorization": "Bearer test-api-key"},
) as websocket,
pytest.raises(WebSocketDisconnect),
):
_receive_until_closed(websocket)
assert connect.call_args.args[0] == expected_upstream_url
def test_openai_prefixed_translation_websocket_takes_the_openai_websocket_pass_through(
client_no_auth: TestClient,
monkeypatch: pytest.MonkeyPatch,
) -> None:
from starlette.websockets import WebSocketDisconnect
from litellm.proxy import proxy_server
from litellm.proxy.auth.user_api_key_auth import user_api_key_auth_websocket
from litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints import _OPENAI_WS_DISABLED_REFUSAL
route_call: Final = AsyncMock()
monkeypatch.setattr(proxy_server, "route_request", route_call)
monkeypatch.setattr(proxy_server, "general_settings", {})
monkeypatch.setitem(proxy_server.app.dependency_overrides, user_api_key_auth_websocket, lambda: UserAPIKeyAuth())
with client_no_auth.websocket_connect(
"/openai/v1/realtime/translations?model=gpt-realtime-translate",
headers={"Authorization": "Bearer test-api-key"},
) as websocket:
refusal: Final = websocket.receive_json()
with pytest.raises(WebSocketDisconnect) as closed:
websocket.receive_json()
assert refusal["error"]["message"] == _OPENAI_WS_DISABLED_REFUSAL.message
assert closed.value.code == 1008
route_call.assert_not_awaited()
@pytest.mark.parametrize("path", ["/openai/v1/live/sessions", "/v1/live/sessions", "/live/sessions"])
def test_live_session_invalid_start_is_rejected_without_routing(
path: str,

3
tests/unit/proxy/test_proxy_server_endpoints_and_startup.py
View file

@ -10772,7 +10772,7 @@ async def test_get_current_spend_redis_error_falls_back_to_in_memory():
def test_realtime_websocket_route_aliases_registered():
"""Realtime sessions reach the proxy via six path aliases stacked on
"""Realtime sessions reach the proxy via five path aliases stacked on
`realtime_websocket_endpoint`. Dropping any of them silently 405s
WebSocket upgrades because the catch-all `/openai/{endpoint:path}`
HTTP passthrough only declares HTTP methods. The aliases must also be
@ -10792,7 +10792,6 @@ def test_realtime_websocket_route_aliases_registered():
"/openai/v1/realtime",
"/v1/realtime",
"/realtime",
"/openai/v1/realtime/translations",
"/v1/realtime/translations",
"/realtime/translations",
):

119
tests/unit/realtime_api/test_main.py
View file

@ -1,4 +1,5 @@
import asyncio
import functools
import ipaddress
import json
import ssl
@ -9,7 +10,7 @@ from datetime import datetime, timedelta, timezone
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from types import TracebackType
from typing import Final
from typing import ClassVar, Final
from unittest.mock import AsyncMock, MagicMock, patch
import httpx
@ -793,6 +794,89 @@ async def test_realtime_http_routes_send_a_bare_model_name_to_its_explicit_provi
assert gateway.url == f"https://realtime-gateway.example/v1/realtime/{path}"
class _GatewayThatRecordsTheBody:
body: object = None
def __call__(self, request: httpx.Request) -> httpx.Response:
self.body = json.loads(request.content)
return httpx.Response(200, json={"client_secret": {"value": "ek_test"}})
@pytest.mark.asyncio
async def test_transcription_session_sends_the_deployment_model_upstream_in_every_transcription_shape() -> None:
gateway: Final = _GatewayThatRecordsTheBody()
await realtime_main.acreate_realtime_transcription_session.__wrapped__(
model="openai/gpt-transcribe",
transcription_session={
"audio": {"input": {"transcription": {"model": "team-transcriber", "language": "en"}}},
"input_audio_transcription": {"model": "team-transcriber"},
"input_audio_format": "pcm16",
},
api_base="https://realtime-gateway.example/v1",
api_key="fake-key",
litellm_logging_obj=FakeLogging(),
client=AsyncHTTPHandler(transport=httpx.MockTransport(gateway)),
)
assert gateway.body == {
"audio": {"input": {"transcription": {"model": "gpt-transcribe", "language": "en"}}},
"input_audio_transcription": {"model": "gpt-transcribe"},
"input_audio_format": "pcm16",
}
def _create_realtime_session_with_an_explicit_null(route: str, client: AsyncHTTPHandler) -> Awaitable[httpx.Response]:
if route == "client_secrets":
return realtime_main.acreate_realtime_client_secret.__wrapped__(
session={"type": "realtime", "model": "gpt-realtime", "audio": {"input": {"turn_detection": None}}},
custom_llm_provider="openai",
api_base="https://realtime-gateway.example/v1",
api_key="fake-key",
litellm_logging_obj=FakeLogging(),
client=client,
)
if route == "translation_client_secrets":
return realtime_main.acreate_realtime_translation_client_secret.__wrapped__(
model="gpt-realtime-translate",
session={"audio": {"input": {"noise_reduction": None}}},
custom_llm_provider="openai",
api_base="https://realtime-gateway.example/v1",
api_key="fake-key",
litellm_logging_obj=_realtime_http_logging(route),
client=client,
)
return realtime_main.acreate_realtime_transcription_session.__wrapped__(
model="openai/gpt-transcribe",
transcription_session={"audio": {"input": {"turn_detection": None}}},
api_base="https://realtime-gateway.example/v1",
api_key="fake-key",
litellm_logging_obj=FakeLogging(),
client=client,
)
@pytest.mark.asyncio
@pytest.mark.parametrize(
("route", "body_path"),
(
("client_secrets", ("session", "audio", "input", "turn_detection")),
("translation_client_secrets", ("session", "audio", "input", "noise_reduction")),
("transcription_sessions", ("audio", "input", "turn_detection")),
),
)
async def test_realtime_session_requests_forward_an_explicit_null_upstream(
route: str, body_path: tuple[str, ...]
) -> None:
gateway: Final = _GatewayThatRecordsTheBody()
await _create_realtime_session_with_an_explicit_null(
route, AsyncHTTPHandler(transport=httpx.MockTransport(gateway))
)
parent: Final = functools.reduce(lambda node, key: node[key], body_path[:-1], gateway.body)
assert body_path[-1] in parent
assert parent[body_path[-1]] is None
class _ClientWebSocketWithHeaders:
def __init__(self, headers: tuple[tuple[bytes, bytes], ...]) -> None:
self.scope: Final = {"headers": headers}
@ -1717,6 +1801,39 @@ async def test_translation_websocket_closes_1008_for_a_guarded_request_before_th
]
class _NativeLifecycleGuardrail(CustomGuardrail):
use_native_lifecycle_hooks: ClassVar[bool] = True
@pytest.mark.asyncio
@pytest.mark.parametrize(
("guardrail_class", "event_hook"),
((_NativeLifecycleGuardrail, GuardrailEventHooks.pre_call), (CustomGuardrail, GuardrailEventHooks.post_call)),
)
async def test_translation_websocket_opens_under_a_default_on_guardrail_the_realtime_runner_skips(
guardrail_class: type[CustomGuardrail], event_hook: GuardrailEventHooks, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(
litellm,
"callbacks",
[guardrail_class(guardrail_name="runner-skipped", event_hook=event_hook, default_on=True)],
)
websocket: Final = _RecordingClientWebSocket()
connect: Final = _CapturingConnect()
with patch("websockets.connect", connect):
await realtime_main._arealtime.__wrapped__(
model="openai/gpt-realtime-translate",
websocket=websocket,
api_key="fake-key",
realtime_mode="translation",
litellm_logging_obj=FakeLogging(),
)
assert connect.url == "wss://api.openai.com/v1/realtime/translations?model=gpt-realtime-translate"
assert all(TRANSLATION_GUARDRAILS_UNSUPPORTED_MESSAGE not in event for event in websocket.sent)
@pytest.mark.asyncio
async def test_translation_websocket_closes_1008_for_a_provider_other_than_openai_and_azure() -> None:
websocket: Final = _RecordingClientWebSocket()

19
tests/unit/test_check_openapi_docs.py
View file

@ -1,5 +1,6 @@
"""Tests for scripts/check_openapi_docs.py, the scan behind scripts/openapi_docs_gate.py."""
from collections.abc import Mapping
from typing import Final
import pytest
@ -11,19 +12,25 @@ _RESPONSES: Final = {"200": {"content": {"application/json": {"schema": {"$ref":
_DOCUMENTED: Final = {"description": "Does a thing", "requestBody": _BODY, "responses": _RESPONSES}
@pytest.mark.parametrize(
("request_media_type", "response_media_type"),
(("application/json", "application/json"), ("application/sdp", "application/sdp"), ("multipart/form-data", "audio/wav")),
)
def test_a_fully_documented_post_has_no_findings(request_media_type: str, response_media_type: str) -> None:
operation: Final = {
def _post_with_media_types(request_media_type: str, response_media_type: str) -> Mapping[str, object]:
return {
**_DOCUMENTED,
"requestBody": {"content": {request_media_type: {"schema": {"type": "string"}}}},
"responses": {"200": {"content": {response_media_type: {"schema": {"type": "string"}}}}},
}
@pytest.mark.parametrize("media_type", ("application/json", "application/sdp"))
def test_a_fully_documented_post_has_no_findings(media_type: str) -> None:
operation: Final = _post_with_media_types(media_type, media_type)
assert checker.violations({"paths": {"/v1/thing": {"post": operation}}}) == ()
def test_a_schema_under_another_media_type_still_counts_as_undocumented() -> None:
operation: Final = _post_with_media_types("multipart/form-data", "audio/wav")
assert {v.code for v in checker.violations({"paths": {"/v1/thing": {"post": operation}}})} == {"OA001", "OA003"}
def test_each_missing_piece_is_its_own_rule() -> None:
bare: Final = {"summary": "Thing", "responses": {"200": {"description": "Successful Response"}}}
codes: Final = {v.code for v in checker.violations({"paths": {"/v1/thing": {"post": bare}}})}

86
tests/unit/test_cost_calculator.py
View file

@ -9999,6 +9999,92 @@ def test_realtime_translation_duration_cost_ignores_malformed_provider_usage(
assert output_invalid_cost == pytest.approx(2.0 * litellm.model_cost[model]["input_cost_per_second"])
def _translation_session_cost(deployment_id: str) -> float:
one_minute_session: Final = LiteLLMRealtimeStreamLoggingObject(
usage=Usage(prompt_tokens=0, completion_tokens=0, total_tokens=0),
results=[
{"type": "session.created", "session": {"model": "gpt-realtime-translate"}},
{"type": "session.closed", "usage": {"type": "duration", "input_seconds": 60.0, "output_seconds": 60.0}},
],
)
return completion_cost(
completion_response=one_minute_session,
model="gpt-realtime-translate",
call_type=CallTypes.arealtime.value,
custom_llm_provider="openai",
custom_pricing=True,
router_model_id=deployment_id,
)
def test_realtime_translation_deployment_per_second_rate_replaces_both_public_rates(_local_model_cost_map) -> None:
litellm.Router(
model_list=[
{
"model_name": "translate",
"litellm_params": {
"model": "openai/gpt-realtime-translate",
"api_key": "sk-test",
"input_cost_per_second": 0.001,
},
"model_info": {"id": "translate-input-priced"},
}
]
)
assert _translation_session_cost("translate-input-priced") == pytest.approx(60 * 0.001)
def test_realtime_translation_deployment_without_per_second_rates_bills_public_rates(_local_model_cost_map) -> None:
litellm.Router(
model_list=[
{
"model_name": "translate",
"litellm_params": {
"model": "openai/gpt-realtime-translate",
"api_key": "sk-test",
"input_cost_per_token": 0.0,
"output_cost_per_token": 0.0,
},
"model_info": {"id": "translate-token-priced"},
}
]
)
public_entry: Final = litellm.model_cost["gpt-realtime-translate"]
public_cost: Final = 60 * (public_entry.get("input_cost_per_second") or 0.0) + 60 * (
public_entry.get("output_cost_per_second") or 0.0
)
assert public_cost > 0
assert _translation_session_cost("translate-token-priced") == pytest.approx(public_cost)
def test_realtime_translation_spend_usage_carries_the_billed_output_seconds(_local_model_cost_map) -> None:
logging_obj: Final = Logging(
model="gpt-realtime-translate",
messages=[],
stream=False,
call_type=CallTypes.arealtime.value,
start_time=datetime.datetime.now(),
litellm_call_id="translation-spend-usage",
function_id="translation-spend-usage",
)
logging_obj.update_environment_variables(
model="gpt-realtime-translate", user=None, optional_params={}, litellm_params={}, custom_llm_provider="openai"
)
events: Final[OpenAIRealtimeStreamList] = [
{"type": "session.closed", "usage": {"type": "duration", "input_seconds": 30.0, "output_seconds": 42.5}}
]
logging_obj.success_handler(result=events, start_time=datetime.datetime.now(), end_time=datetime.datetime.now())
payload: Final = logging_obj.model_call_details["standard_logging_object"]
assert payload["metadata"]["usage_object"]["seconds"] == pytest.approx(42.5)
assert payload["response_cost"] == pytest.approx(
42.5 * litellm.model_cost["gpt-realtime-translate"]["output_cost_per_second"]
)
@pytest.mark.parametrize("profile_prefix", ("us", "global"))
@pytest.mark.parametrize(
"model_name",

58
ui/litellm-dashboard/src/lib/http/schema.d.ts generated vendored
View file

@ -10991,7 +10991,7 @@ export interface paths {
* WebSocket: realtime_websocket_endpoint
* @description WebSocket connection endpoint
*/
get: operations["websocket_realtime_websocket_endpoint_get_6"];
get: operations["websocket_realtime_websocket_endpoint_get_5"];
put?: never;
post?: never;
delete?: never;
@ -11064,26 +11064,6 @@ export interface paths {
patch?: never;
trace?: never;
};
"/openai/v1/realtime/translations": {
parameters: {
query?: never;
header?: never;
path?: never;
cookie?: never;
};
/**
* WebSocket: realtime_websocket_endpoint
* @description WebSocket connection endpoint
*/
get: operations["websocket_realtime_websocket_endpoint_get_3"];
put?: never;
post?: never;
delete?: never;
options?: never;
head?: never;
patch?: never;
trace?: never;
};
"/openai/v1/responses": {
parameters: {
query?: never;
@ -13991,7 +13971,7 @@ export interface paths {
* WebSocket: realtime_websocket_endpoint
* @description WebSocket connection endpoint
*/
get: operations["websocket_realtime_websocket_endpoint_get_4"];
get: operations["websocket_realtime_websocket_endpoint_get_3"];
put?: never;
post?: never;
delete?: never;
@ -21759,7 +21739,7 @@ export interface paths {
* WebSocket: realtime_websocket_endpoint
* @description WebSocket connection endpoint
*/
get: operations["websocket_realtime_websocket_endpoint_get_5"];
get: operations["websocket_realtime_websocket_endpoint_get_4"];
put?: never;
post?: never;
delete?: never;
@ -66833,7 +66813,7 @@ export interface operations {
};
};
};
websocket_realtime_websocket_endpoint_get_6: {
websocket_realtime_websocket_endpoint_get_5: {
parameters: {
query?: never;
header?: never;
@ -66911,24 +66891,6 @@ export interface operations {
};
};
};
websocket_realtime_websocket_endpoint_get_3: {
parameters: {
query?: never;
header?: never;
path?: never;
cookie?: never;
};
requestBody?: never;
responses: {
/** @description WebSocket Protocol Switched */
101: {
headers: {
[name: string]: unknown;
};
content?: never;
};
};
};
responses_api_openai_v1_responses_post: {
parameters: {
query?: never;
@ -70206,7 +70168,7 @@ export interface operations {
};
};
};
websocket_realtime_websocket_endpoint_get_4: {
websocket_realtime_websocket_endpoint_get_3: {
parameters: {
query?: never;
header?: never;
@ -70369,7 +70331,7 @@ export interface operations {
} | null;
transcription?: ({
/** Delay */
delay?: ("minimal" | "low" | "medium" | "high" | "xhigh") | null;
delay?: string | null;
/** Keywords */
keywords?: string[] | null;
/** Language */
@ -70432,7 +70394,7 @@ export interface operations {
/** Truncation */
truncation?: unknown | null;
/** Type */
type?: ("realtime" | "transcription" | "translation") | null;
type?: string | null;
} & {
[key: string]: unknown;
}) | null;
@ -81276,7 +81238,7 @@ export interface operations {
};
};
};
websocket_realtime_websocket_endpoint_get_5: {
websocket_realtime_websocket_endpoint_get_4: {
parameters: {
query?: never;
header?: never;
@ -81439,7 +81401,7 @@ export interface operations {
} | null;
transcription?: ({
/** Delay */
delay?: ("minimal" | "low" | "medium" | "high" | "xhigh") | null;
delay?: string | null;
/** Keywords */
keywords?: string[] | null;
/** Language */
@ -81502,7 +81464,7 @@ export interface operations {
/** Truncation */
truncation?: unknown | null;
/** Type */
type?: ("realtime" | "transcription" | "translation") | null;
type?: string | null;
} & {
[key: string]: unknown;
}) | null;