scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-03 02:22:24 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 81

Add gpt-realtime-whisper Realtime transcription support (OpenAI + Azure) (#29775)

Browse source 
* Add gpt-realtime-whisper Realtime transcription support (OpenAI + Azure)

Adds first-class support for the gpt-realtime-whisper streaming speech-to-text
model, which uses the Realtime transcription session API rather than the
file-based /audio/transcriptions path.

Model registration: registers gpt-realtime-whisper and azure/gpt-realtime-whisper
with audio-duration pricing (input_cost_per_second = 0.017/60, matching the
published $0.017/minute input audio rate).

REST endpoint: implements POST /v1/realtime/transcription_sessions (plus /realtime
and /openai/v1 aliases) to mint an ephemeral transcription session for the
WebRTC flow. Adds request/response types, OpenAI and Azure URL builders, a shared
base handler (refactored from the client_secrets handler), the
acreate_realtime_transcription_session SDK function, and route registration. The
proxy encrypts the ephemeral key returned under client_secret.value and records
the session type in the token so the follow-up /realtime/calls replays
type=transcription rather than type=realtime.

WebSocket: forwards intent=transcription through to the Azure handler (OpenAI
already received it) with URL-encoding, so gpt-realtime-whisper opens a
transcription session. Transcription-only sessions no longer trigger an
erroneous response.create.

Cost tracking: transcription sessions emit no response.done events; their usage
arrives on conversation.item.input_audio_transcription.completed as
{type: duration, seconds}. That usage is captured out-of-band (usage only, no
transcript duplication) and billed by input_cost_per_second, with a token-billed
fallback for token-priced transcription models.

Adds tests for pricing math, URL builders, request/response types, the proxy
route and SDK function, WebSocket intent forwarding, transcription-session
streaming behavior, and the /realtime/calls session-type replay.

* Address PR review: URL-encode all Azure WS query params; forward query_params through provider_config branch

* Address PR review: session_type validation, model auth fix, cost perf, billing fallback, detail/docs cleanup

* Improve test coverage: detection from backend, error paths, unknown usage type, resolved_model None

* Backport realtime transcription websocket fixes

* Enforce authorized realtime transcription model

* Enforce realtime transcription model access

* Enforce realtime resolved model scopes

* Enforce WebRTC transcription model scope
This commit is contained in:
Emerson Gomes 2026-06-10 22:29:28 -05:00 • committed by GitHub
parent 6068bb7781
commit 8dd7ab7f3f
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
26 changed files with 2541 additions and 33 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
litellm/__init__.py
View file

@ -1373,6 +1373,7 @@ from .search.main import *
from .realtime_api.main import (
_arealtime,
acreate_realtime_client_secret,
acreate_realtime_transcription_session,
arealtime_calls,
)
from .responses.main import _aresponses_websocket

98
litellm/cost_calculator.py
View file

@ -2533,4 +2533,102 @@ def handle_realtime_stream_cost_calculation(
break # exit if we find a valid model
total_cost = input_cost_per_token + output_cost_per_token
if any(r.get("type") == _TRANSCRIPTION_COMPLETED_EVENT_TYPE for r in results):
total_cost += handle_realtime_transcription_cost_calculation(
results=results,
custom_llm_provider=custom_llm_provider,
litellm_model_name=litellm_model_name,
)
return total_cost
_TRANSCRIPTION_COMPLETED_EVENT_TYPE = (
"conversation.item.input_audio_transcription.completed"
)
def handle_realtime_transcription_cost_calculation(
results: OpenAIRealtimeStreamList,
custom_llm_provider: str,
litellm_model_name: str,
) -> float:
"""
Cost for realtime transcription sessions (e.g. gpt-realtime-whisper).
Transcription sessions emit no `response.done` events; instead each
`conversation.item.input_audio_transcription.completed` event carries a
`usage` object billed by the ASR model. The usage is one of:
- {"type": "duration", "seconds": <float>} → priced via input_cost_per_second
- {"type": "tokens", "input_tokens": ...} → priced via input/audio token cost
"""
completed_events = [
cast(dict, result)
for result in results
if result.get("type") == _TRANSCRIPTION_COMPLETED_EVENT_TYPE
]
if not completed_events:
return 0.0
model_name = (
_get_transcription_model_name_from_results(results) or litellm_model_name
)
try:
model_info = litellm.get_model_info(
model=model_name, custom_llm_provider=custom_llm_provider
)
except Exception:
model_info = {}
total_cost = 0.0
for event in completed_events:
usage = event.get("usage") or {}
total_cost += _transcription_usage_cost(usage, model_info)
return total_cost
def _get_transcription_model_name_from_results(
results: OpenAIRealtimeStreamList,
) -> Optional[str]:
"""Resolve the ASR model from a transcription_session.* / session.* event."""
for result in results:
if result.get("type") in (
"transcription_session.created",
"transcription_session.updated",
"session.created",
"session.updated",
):
session = cast(dict, result).get("session", {}) or {}
transcription = (
(session.get("audio", {}) or {}).get("input", {}) or {}
).get("transcription", {}) or session.get("input_audio_transcription", {})
model = (transcription or {}).get("model") or session.get("model")
if model:
return model
return None
def _transcription_usage_cost(usage: dict, model_info: dict) -> float:
usage_type = usage.get("type")
if usage_type == "duration":
seconds = usage.get("seconds") or 0.0
per_second = model_info.get("input_cost_per_second") or 0.0
return float(seconds) * float(per_second)
if usage_type == "tokens":
input_token_details = usage.get("input_token_details") or {}
audio_tokens = input_token_details.get("audio_tokens") or 0
text_tokens = input_token_details.get("text_tokens") or 0
output_tokens = usage.get("output_tokens") or 0
audio_cost = float(audio_tokens) * float(
model_info.get("input_cost_per_audio_token")
or model_info.get("input_cost_per_token")
or 0.0
)
text_cost = float(text_tokens) * float(
model_info.get("input_cost_per_token") or 0.0
)
output_cost = float(output_tokens) * float(
model_info.get("output_cost_per_token") or 0.0
)
return audio_cost + text_cost + output_cost
return 0.0

144
litellm/litellm_core_utils/realtime_streaming.py
View file

@ -47,6 +47,7 @@ class RealTimeStreaming:
user_api_key_dict: Optional[Any] = None,
request_data: Optional[Dict] = None,
backend_uses_beta_protocol: Optional[bool] = None,
force_transcription_model: Optional[str] = None,
):
self.websocket = websocket
self.backend_ws = backend_ws
@ -100,6 +101,11 @@ class RealTimeStreaming:
self._flushing_pending_messages_until_setup: bool = False
self._pending_messages_until_setup: List[str] = []
self._pending_messages_byte_total: int = 0
# Whether this is a transcription-only session (session.type == "transcription",
# e.g. gpt-realtime-whisper). Such sessions must not be sent response.create and
# their input_audio_transcription.completed usage drives duration-based cost.
self._force_transcription_model = force_transcription_model
self._is_transcription_session: bool = force_transcription_model is not None
# Per-connection caps for pre-setup audio frames (message count + total bytes).
_MAX_BUFFERED_MESSAGES: int = 200
@ -209,6 +215,8 @@ class RealTimeStreaming:
self.session_tools = tools
# GA: session.type is required; log it for traceability but no action needed
verbose_logger.debug(f"Realtime session.type: {session.get('type')}")
if session.get("type") == "transcription":
self._is_transcription_session = True
except (json.JSONDecodeError, AttributeError, TypeError):
pass
@ -225,6 +233,55 @@ class RealTimeStreaming:
except (AttributeError, TypeError):
pass
def _detect_transcription_session_from_backend(
self, event_obj: Union[dict, OpenAIRealtimeEvents]
) -> None:
"""Flag transcription-only sessions from backend session events."""
try:
event_type = event_obj.get("type", "")
if event_type in (
"transcription_session.created",
"transcription_session.updated",
):
self._is_transcription_session = True
elif event_type in ("session.created", "session.updated"):
session = cast(dict, event_obj).get("session", {}) or {}
if session.get("type") == "transcription":
self._is_transcription_session = True
except (AttributeError, TypeError):
pass
def _capture_transcription_usage(
self, event_obj: Union[dict, OpenAIRealtimeEvents]
) -> None:
"""
Append a usage-only transcription completed event to the logged results so
the cost calculator can bill it by audio duration. The default logged event
types exclude this event, so it is captured here directly for transcription
sessions rather than widening logging for every realtime session. Only the
type and usage are kept — the transcript is already captured separately in
input_messages, so it is not duplicated into the response log here.
"""
try:
usage = event_obj.get("usage")
if usage is None:
return
# If this event type is already captured by store_message (e.g. the user
# logs all realtime events), don't append a second copy.
if self._should_store_message(event_obj):
return
self.messages.append(
cast(
OpenAIRealtimeEvents,
{
"type": "conversation.item.input_audio_transcription.completed",
"usage": usage,
},
)
)
except (AttributeError, TypeError):
pass
def _collect_tool_calls_from_response_done(
self, event_obj: Union[dict, OpenAIRealtimeEvents]
) -> None:
@ -285,6 +342,7 @@ class RealTimeStreaming:
backend, False if the provider transformation produced no output and
the message was effectively dropped.
"""
message = self._enforce_transcription_session_model(message)
if self.provider_config:
transformed = self.provider_config.transform_realtime_request(
message, self.model, self.session_configuration_request
@ -304,6 +362,80 @@ class RealTimeStreaming:
await self.backend_ws.send(message) # type: ignore[union-attr, attr-defined]
return True
def _enforce_transcription_session_model(self, message: str) -> str:
"""Force client transcription session updates to the authorized model.
`/v1/realtime?intent=transcription` may intentionally omit `model` from
the upstream URL for Azure compatibility, but the proxy still authorizes
a resolved LiteLLM model before opening the backend websocket. If a
client later sends a transcription `session.update`, any model embedded
in that update must be rewritten to the same authorized model instead of
allowing a post-auth model/deployment switch.
Normal realtime sessions keep their independent nested transcription
model behavior because `_force_transcription_model` is only set for
transcription-intent websocket routes.
"""
if self._force_transcription_model is None:
return message
try:
message_obj = json.loads(message)
except (json.JSONDecodeError, TypeError):
return message
if message_obj.get("type") not in (
"session.update",
"transcription_session.update",
):
return message
session = message_obj.get("session")
if not isinstance(session, dict):
return message
if session.get("type") == "transcription":
self._is_transcription_session = True
authorized_model = self._force_transcription_model
changed = False
transcription = session.get("input_audio_transcription")
if (
isinstance(transcription, dict)
and transcription.get("model") != authorized_model
):
session["input_audio_transcription"] = {
**transcription,
"model": authorized_model,
}
changed = True
audio = session.get("audio")
if isinstance(audio, dict):
audio_input = audio.get("input")
if isinstance(audio_input, dict):
nested_transcription = audio_input.get("transcription")
if (
isinstance(nested_transcription, dict)
and nested_transcription.get("model") != authorized_model
):
session["audio"] = {
**audio,
"input": {
**audio_input,
"transcription": {
**nested_transcription,
"model": authorized_model,
},
},
}
changed = True
if not changed:
return message
return json.dumps(message_obj)
def _uses_deferred_backend_setup(self) -> bool:
"""True when setup is deferred until the client's first session.update."""
if self.provider_config is None:
@ -713,6 +845,8 @@ class RealTimeStreaming:
try:
event_obj = json.loads(raw_response)
self._detect_transcription_session_from_backend(event_obj)
# For audio/VAD guardrail path: once the session is ready, tell the backend
# not to auto-respond after VAD detects end-of-speech. We send the
# session.created to the client FIRST so the client is always in sync, then
@ -731,12 +865,20 @@ class RealTimeStreaming:
event_obj.get("type")
== "conversation.item.input_audio_transcription.completed"
):
transcript = event_obj.get("transcript", "")
self._collect_user_input_from_backend_event(event_obj)
## LOGGING — must happen before continue below
self.store_message(raw_response)
# Forward transcript to client so user sees what they said
await self.websocket.send_text(raw_response)
# Transcription-only sessions (e.g. gpt-realtime-whisper) have no
# assistant turn: capture audio-duration usage for cost and never
# trigger response.create.
if self._is_transcription_session:
self._capture_transcription_usage(event_obj)
return True
transcript = event_obj.get("transcript", "")
blocked = await self.run_realtime_guardrails(
transcript,
item_id=event_obj.get("item_id"),

33
litellm/llms/azure/realtime/handler.py
View file

@ -35,6 +35,7 @@ class AzureOpenAIRealtime(AzureChatCompletion):
model: str,
api_version: Optional[str],
realtime_protocol: Optional[str] = None,
query_params: Optional[dict] = None,
) -> str:
"""
Construct Azure realtime WebSocket URL.
@ -46,6 +47,7 @@ class AzureOpenAIRealtime(AzureChatCompletion):
realtime_protocol: Protocol version to use:
- "GA" or "v1": Uses /openai/v1/realtime (GA path)
- "beta" or None: Uses /openai/realtime (beta path, default)
query_params: Extra query params to forward (e.g. intent=transcription).
Returns:
WebSocket URL string
@ -54,6 +56,8 @@ class AzureOpenAIRealtime(AzureChatCompletion):
beta/default: "wss://.../openai/realtime?api-version=2024-10-01-preview&deployment=gpt-4o-realtime-preview"
GA/v1: "wss://.../openai/v1/realtime?model=gpt-realtime-deployment"
"""
from urllib.parse import urlencode
api_base = api_base.replace("https://", "wss://")
# Determine path based on realtime_protocol (case-insensitive)
@ -61,13 +65,25 @@ class AzureOpenAIRealtime(AzureChatCompletion):
"GA",
"V1",
)
intent = (query_params or {}).get("intent")
if _is_ga:
path = "/openai/v1/realtime"
return f"{api_base}{path}?model={model}"
query_parts = []
if intent != "transcription" and (
query_params is None or "model" in query_params
):
query_parts.append(urlencode({"model": model}))
else:
# Default to beta path for backwards compatibility
path = "/openai/realtime"
return f"{api_base}{path}?api-version={api_version}&deployment={model}"
query_parts = [urlencode({"api-version": api_version, "deployment": model})]
if intent:
query_parts.append(urlencode({"intent": intent}))
qs = "&".join(query_parts)
return f"{api_base}{path}?{qs}" if qs else f"{api_base}{path}"
async def async_realtime(
self,
@ -81,6 +97,7 @@ class AzureOpenAIRealtime(AzureChatCompletion):
client: Optional[Any] = None,
timeout: Optional[float] = None,
realtime_protocol: Optional[str] = None,
query_params: Optional[dict] = None,
user_api_key_dict: Optional[Any] = None,
litellm_metadata: Optional[dict] = None,
):
@ -96,7 +113,11 @@ class AzureOpenAIRealtime(AzureChatCompletion):
raise ValueError("api_version is required for Azure OpenAI calls")
url = self._construct_url(
api_base, model, api_version, realtime_protocol=realtime_protocol
api_base,
model,
api_version,
realtime_protocol=realtime_protocol,
query_params=query_params,
)
try:
@ -113,9 +134,15 @@ class AzureOpenAIRealtime(AzureChatCompletion):
websocket,
cast(ClientConnection, backend_ws),
logging_obj,
model=model,
user_api_key_dict=user_api_key_dict,
request_data={"litellm_metadata": litellm_metadata or {}},
backend_uses_beta_protocol=backend_uses_beta_protocol,
force_transcription_model=(
model
if (query_params or {}).get("intent") == "transcription"
else None
),
)
await realtime_streaming.bidirectional_forward()

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

@ -40,6 +40,13 @@ class AzureRealtimeHTTPConfig(BaseRealtimeHTTPConfig):
version = api_version or get_secret_str("AZURE_API_VERSION") or "2024-12-17"
return f"{base}/openai/realtime/calls?api-version={version}"
def get_transcription_session_url(
self, api_base: Optional[str], model: str, api_version: Optional[str] = None
) -> str:
base = self.get_api_base(api_base).rstrip("/")
version = api_version or get_secret_str("AZURE_API_VERSION") or "2024-12-17"
return f"{base}/openai/realtime/transcription_sessions?api-version={version}"
def get_realtime_calls_headers(self, ephemeral_key: str) -> dict:
return {
"api-key": ephemeral_key,

9
litellm/llms/base_llm/realtime/http_transformation.py
View file

@ -59,6 +59,15 @@ class BaseRealtimeHTTPConfig(ABC):
) -> str:
"""Return the full URL for POST /realtime/client_secrets."""
def get_transcription_session_url(
self, api_base: Optional[str], model: str, api_version: Optional[str] = None
) -> str:
"""Return the full URL for POST /realtime/transcription_sessions."""
base = (api_base or "").rstrip("/")
if base.endswith("/v1"):
base = base[:-3]
return f"{base}/v1/realtime/transcription_sessions"
@abstractmethod
def validate_environment(
self,

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

@ -5315,6 +5315,21 @@ class BaseLLMHTTPHandler:
headers=error_headers,
)
@staticmethod
def _append_query_params(url: str, query_params: Optional[Dict[str, Any]]) -> str:
"""Append query_params to url, skipping keys already present in the URL."""
if not query_params:
return url
from urllib.parse import parse_qsl, urlencode, urlparse, urlunparse
parsed = urlparse(url)
existing = dict(parse_qsl(parsed.query))
extras = {k: v for k, v in query_params.items() if k not in existing}
if not extras:
return url
new_query = parsed.query + ("&" if parsed.query else "") + urlencode(extras)
return urlunparse(parsed._replace(query=new_query))
async def async_realtime(
self,
model: str,
@ -5328,11 +5343,14 @@ class BaseLLMHTTPHandler:
timeout: Optional[float] = None,
user_api_key_dict: Optional[Any] = None,
litellm_metadata: Optional[Dict[str, Any]] = None,
query_params: Optional[Dict[str, Any]] = None,
):
import websockets
from websockets.asyncio.client import ClientConnection
url = provider_config.get_complete_url(api_base, model, api_key)
url = self._append_query_params(
provider_config.get_complete_url(api_base, model, api_key), query_params
)
headers = provider_config.validate_environment(
headers=headers,
model=model,
@ -5373,6 +5391,11 @@ class BaseLLMHTTPHandler:
model,
user_api_key_dict=user_api_key_dict,
request_data=_request_data,
force_transcription_model=(
model
if (query_params or {}).get("intent") == "transcription"
else None
),
)
if _session_config:
realtime_streaming.session_configuration_request = _session_config
@ -5437,6 +5460,69 @@ class BaseLLMHTTPHandler:
"""
Forward POST /v1/realtime/client_secrets to upstream provider.
Uses provider_config (BaseRealtimeHTTPConfig) for URL construction and
header auth when available; falls back to the legacy OpenAI-style defaults.
"""
return await self._async_realtime_session_post(
endpoint="client_secrets",
api_base=api_base,
api_key=api_key,
request_data=request_data,
logging_obj=logging_obj,
timeout=timeout,
provider_config=provider_config,
model=model,
extra_headers=extra_headers,
client=client,
api_version=api_version,
)
async def async_realtime_transcription_session_handler(
self,
api_base: str,
api_key: str,
request_data: Dict[str, Any],
logging_obj: LiteLLMLoggingObj,
timeout: Union[float, httpx.Timeout],
provider_config: Optional[Any] = None,
model: Optional[str] = None,
extra_headers: Optional[Dict[str, Any]] = None,
client: Optional[Union[HTTPHandler, AsyncHTTPHandler]] = None,
api_version: Optional[str] = None,
) -> httpx.Response:
"""Forward POST /v1/realtime/transcription_sessions to upstream provider."""
return await self._async_realtime_session_post(
endpoint="transcription_sessions",
api_base=api_base,
api_key=api_key,
request_data=request_data,
logging_obj=logging_obj,
timeout=timeout,
provider_config=provider_config,
model=model,
extra_headers=extra_headers,
client=client,
api_version=api_version,
)
async def _async_realtime_session_post(
self,
endpoint: Literal["client_secrets", "transcription_sessions"],
api_base: str,
api_key: str,
request_data: Dict[str, Any],
logging_obj: LiteLLMLoggingObj,
timeout: Union[float, httpx.Timeout],
provider_config: Optional[Any] = None,
model: Optional[str] = None,
extra_headers: Optional[Dict[str, Any]] = None,
client: Optional[Union[HTTPHandler, AsyncHTTPHandler]] = None,
api_version: Optional[str] = None,
) -> httpx.Response:
"""
Shared POST flow for the realtime HTTP session endpoints
(client_secrets and transcription_sessions).
Uses provider_config (BaseRealtimeHTTPConfig) for URL construction and
header auth when available; falls back to the legacy OpenAI-style defaults.
"""
@ -5448,14 +5534,19 @@ class BaseLLMHTTPHandler:
async_httpx_client = client
if provider_config is not None:
url = provider_config.get_complete_url(
api_base=api_base, model=model or "", api_version=api_version
)
if endpoint == "transcription_sessions":
url = provider_config.get_transcription_session_url(
api_base=api_base, model=model or "", api_version=api_version
)
else:
url = provider_config.get_complete_url(
api_base=api_base, model=model or "", api_version=api_version
)
headers: Dict[str, Any] = provider_config.validate_environment(
headers={}, model=model or "", api_key=api_key
)
else:
url = f"{api_base.rstrip('/')}/v1/realtime/client_secrets"
url = f"{api_base.rstrip('/')}/v1/realtime/{endpoint}"
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",

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

@ -157,8 +157,14 @@ class OpenAIRealtime(OpenAIChatCompletion):
websocket,
cast(ClientConnection, backend_ws),
logging_obj,
model=model,
user_api_key_dict=user_api_key_dict,
request_data={"litellm_metadata": litellm_metadata or {}},
force_transcription_model=(
model
if (query_params or {}).get("intent") == "transcription"
else None
),
)
await realtime_streaming.bidirectional_forward()

8
litellm/llms/openai/realtime/http_transformation.py
View file

@ -41,6 +41,14 @@ class OpenAIRealtimeHTTPConfig(BaseRealtimeHTTPConfig):
base = base[:-3]
return f"{base}/v1/realtime/calls"
def get_transcription_session_url(
self, api_base: Optional[str], model: str, api_version: Optional[str] = None
) -> str:
base = self.get_api_base(api_base).rstrip("/")
if base.endswith("/v1"):
base = base[:-3]
return f"{base}/v1/realtime/transcription_sessions"
def validate_environment(
self,
headers: dict,

34
litellm/model_prices_and_context_window_backup.json
View file

@ -4409,6 +4409,23 @@
"/v1/audio/transcriptions"
]
},
"azure/gpt-realtime-whisper": {
"input_cost_per_second": 0.0002833333333333333,
"litellm_provider": "azure",
"mode": "audio_transcription",
"source": "https://learn.microsoft.com/en-us/azure/ai-foundry/openai/concepts/gpt-realtime-whisper",
"supported_endpoints": [
"/v1/realtime",
"/v1/realtime/transcription_sessions"
],
"supported_modalities": [
"audio"
],
"supported_output_modalities": [
"text"
],
"supports_audio_input": true
},
"azure/gpt-5.1-2025-11-13": {
"cache_read_input_token_cost": 1.25e-07,
"cache_read_input_token_cost_priority": 2.5e-07,
@ -40916,6 +40933,23 @@
"supports_system_messages": true,
"supports_tool_choice": true
},
"gpt-realtime-whisper": {
"input_cost_per_second": 0.0002833333333333333,
"litellm_provider": "openai",
"mode": "audio_transcription",
"source": "https://platform.openai.com/docs/models/gpt-realtime-whisper",
"supported_endpoints": [
"/v1/realtime",
"/v1/realtime/transcription_sessions"
],
"supported_modalities": [
"audio"
],
"supported_output_modalities": [
"text"
],
"supports_audio_input": true
},
"sora-2": {
"litellm_provider": "openai",
"mode": "video_generation",

92
litellm/proxy/auth/auth_checks.py
View file

@ -3155,6 +3155,98 @@ async def can_key_call_model(
raise
async def can_key_call_resolved_model(
model: str,
llm_model_list: Optional[list],
valid_token: UserAPIKeyAuth,
llm_router: Optional[litellm.Router],
) -> None:
from litellm.proxy.proxy_server import (
prisma_client,
proxy_logging_obj,
user_api_key_cache,
)
skip_key_model_check = valid_token.config or (
isinstance(valid_token.models, list)
and SpecialModelNames.all_team_models.value in valid_token.models
)
if not skip_key_model_check:
await can_key_call_model(
model=model,
llm_model_list=llm_model_list,
valid_token=valid_token,
llm_router=llm_router,
)
team_object: Optional[LiteLLM_TeamTableCachedObj] = None
team_object_from_lookup = False
if valid_token.team_id is not None:
try:
team_object = await get_team_object(
team_id=valid_token.team_id,
prisma_client=prisma_client,
user_api_key_cache=user_api_key_cache,
parent_otel_span=valid_token.parent_otel_span,
proxy_logging_obj=proxy_logging_obj,
)
team_object_from_lookup = True
except Exception:
team_object = LiteLLM_TeamTableCachedObj(
team_id=valid_token.team_id,
models=valid_token.team_models,
blocked=valid_token.team_blocked,
team_alias=valid_token.team_alias,
metadata=valid_token.team_metadata,
object_permission_id=valid_token.team_object_permission_id,
object_permission=valid_token.team_object_permission,
)
if team_object is not None:
try:
await can_team_access_model(
model=model,
team_object=team_object,
llm_router=llm_router,
team_model_aliases=valid_token.team_model_aliases,
)
except ProxyException as team_denial:
if team_denial.type != ProxyErrorTypes.team_model_access_denied:
raise
if not await _key_access_group_grants_model(
model=model,
valid_token=valid_token,
team_object=team_object,
llm_router=llm_router,
):
raise
if valid_token.user_id is not None and team_object_from_lookup:
await _check_team_member_model_access(
model=model,
team_object=team_object,
valid_token=valid_token,
llm_router=llm_router,
prisma_client=prisma_client,
user_api_key_cache=user_api_key_cache,
proxy_logging_obj=proxy_logging_obj,
)
if valid_token.project_id is not None:
project_object = await get_project_object(
project_id=valid_token.project_id,
prisma_client=prisma_client,
user_api_key_cache=user_api_key_cache,
proxy_logging_obj=proxy_logging_obj,
)
if project_object is not None and len(project_object.models) > 0:
can_project_access_model(
model=model,
project_object=project_object,
llm_router=llm_router,
)
def can_org_access_model(
model: str,
org_object: Optional[LiteLLM_OrganizationTable],

40
litellm/proxy/proxy_server.py
View file

@ -252,6 +252,7 @@ from litellm.proxy.analytics_endpoints.analytics_endpoints import (
)
from litellm.proxy.auth.auth_checks import (
ExperimentalUIJWTToken,
can_key_call_resolved_model,
get_team_object,
log_db_metrics,
)
@ -9458,13 +9459,15 @@ async def vertex_ai_live_passthrough_endpoint(
@lru_cache(maxsize=_REALTIME_BODY_CACHE_SIZE)
def _realtime_query_params_template(
model: str, intent: Optional[str]
model: Optional[str], intent: Optional[str]
) -> Tuple[Tuple[str, str], ...]:
"""
Build a hashable representation of the realtime query params so we can cache
the repetitive model/intent combinations.
"""
params: List[Tuple[str, str]] = [("model", model)]
params: List[Tuple[str, str]] = []
if model is not None:
params.append(("model", model))
if intent is not None:
params.append(("intent", intent))
return tuple(params)
@ -9475,8 +9478,10 @@ def _realtime_query_params_template(
@app.websocket("/realtime")
async def realtime_websocket_endpoint(
websocket: WebSocket,
model: str,
intent: str = fastapi.Query(
model: Optional[str] = fastapi.Query(
None, description="The model to use for the websocket connection."
),
intent: Optional[str] = fastapi.Query(
None, description="The intent of the websocket connection."
),
guardrails: Optional[str] = fastapi.Query(
@ -9493,6 +9498,27 @@ async def realtime_websocket_endpoint(
accept_kwargs: dict = {}
if requested_protocols:
accept_kwargs["subprotocol"] = requested_protocols[0]
route_model = model
if route_model is None:
if intent == "transcription":
route_model = "gpt-realtime-whisper"
else:
await websocket.close(
code=1008, reason="model query parameter is required"
)
return
assert route_model is not None
try:
await can_key_call_resolved_model(
model=route_model,
llm_model_list=llm_model_list,
valid_token=user_api_key_dict,
llm_router=llm_router,
)
except ProxyException as e:
await websocket.close(code=1008, reason=e.message[:120])
return
await websocket.accept(**accept_kwargs)
# Only use explicit parameters, not all query params
@ -9501,7 +9527,7 @@ async def realtime_websocket_endpoint(
)
data: Dict[str, Any] = {
"model": model,
"model": route_model,
"websocket": websocket,
"query_params": query_params, # Only explicit params
}
@ -9521,7 +9547,7 @@ async def realtime_websocket_endpoint(
request._url = websocket.url
async def return_body():
return _realtime_request_body(model)
return _realtime_request_body(route_model)
request.body = return_body # type: ignore
@ -9547,7 +9573,7 @@ async def realtime_websocket_endpoint(
user_request_timeout=user_request_timeout,
user_max_tokens=user_max_tokens,
user_api_base=user_api_base,
model=model,
model=route_model,
route_type="_arealtime",
)
except Exception as e:

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

@ -10,6 +10,7 @@ from fastapi import status as http_status
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
from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
from litellm.proxy.common_utils.encrypt_decrypt_utils import (
decrypt_value_helper,
@ -19,11 +20,143 @@ from litellm.proxy.common_utils.http_parsing_utils import _read_request_body
from litellm.types.realtime import (
RealtimeClientSecretRequest,
RealtimeClientSecretResponse,
RealtimeTranscriptionSessionRequest,
RealtimeTranscriptionSessionResponse,
)
router = APIRouter()
_REALTIME_TOKEN_VERSION = "realtime_v1"
_DEFAULT_REALTIME_MODEL = "gpt-4o-realtime-preview"
_DEFAULT_TRANSCRIPTION_MODEL = "gpt-realtime-whisper"
_ALLOWED_SESSION_TYPES = ("realtime", "transcription")
def _coerce_realtime_session_type(session_type: Optional[str]) -> str:
if session_type in _ALLOWED_SESSION_TYPES:
return session_type
return "realtime"
def _append_model_candidate(candidates: list[str], model: Any) -> None:
if isinstance(model, str) and model and model not in candidates:
candidates.append(model)
def _transcription_model_candidates_from_session(session: dict) -> list[str]:
candidates: list[str] = []
audio = session.get("audio")
if isinstance(audio, dict):
audio_input = audio.get("input")
if isinstance(audio_input, dict):
nested_transcription = audio_input.get("transcription")
if isinstance(nested_transcription, dict):
_append_model_candidate(
candidates,
nested_transcription.get("model"),
)
flat_transcription = session.get("input_audio_transcription")
if isinstance(flat_transcription, dict):
_append_model_candidate(candidates, flat_transcription.get("model"))
return candidates
def _set_transcription_model_on_session(
session: dict,
model: str,
create_if_missing: bool = False,
) -> None:
updated_existing_config = False
flat_transcription = session.get("input_audio_transcription")
if isinstance(flat_transcription, dict):
session["input_audio_transcription"] = {
**flat_transcription,
"model": model,
}
updated_existing_config = True
audio = session.get("audio")
if isinstance(audio, dict):
audio_input = audio.get("input")
if isinstance(audio_input, dict):
nested_transcription = audio_input.get("transcription")
if isinstance(nested_transcription, dict):
session["audio"] = {
**audio,
"input": {
**audio_input,
"transcription": {
**nested_transcription,
"model": model,
},
},
}
updated_existing_config = True
if updated_existing_config or not create_if_missing:
return
audio = audio if isinstance(audio, dict) else {}
audio_input = audio.get("input")
audio_input = audio_input if isinstance(audio_input, dict) else {}
session["audio"] = {
**audio,
"input": {
**audio_input,
"transcription": {"model": model},
},
}
async def _prepare_client_secret_session(
req: RealtimeClientSecretRequest,
user_api_key_dict: UserAPIKeyAuth,
llm_model_list: Optional[list],
llm_router: Any,
) -> tuple[str, Optional[dict], str]:
session_type = _coerce_realtime_session_type(
req.session.type if req.session else None
)
session_data: Optional[dict] = (
req.session.model_dump(exclude_none=True) if req.session else None
)
if session_data is not None:
session_data["type"] = session_type
session_model = req.session.model if req.session else None
model: str = session_model or req.model or _DEFAULT_REALTIME_MODEL
if session_type != "transcription":
return model, session_data, session_type
transcription_model_candidates = _transcription_model_candidates_from_session(
session_data or {}
)
if not transcription_model_candidates:
_append_model_candidate(transcription_model_candidates, session_model)
_append_model_candidate(transcription_model_candidates, req.model)
if not transcription_model_candidates:
transcription_model_candidates.append(_DEFAULT_TRANSCRIPTION_MODEL)
model = transcription_model_candidates[0]
for transcription_model in transcription_model_candidates:
await can_key_call_resolved_model(
model=transcription_model,
valid_token=user_api_key_dict,
llm_model_list=llm_model_list,
llm_router=llm_router,
)
if session_data is not None:
_set_transcription_model_on_session(
session=session_data,
model=model,
create_if_missing=True,
)
session_data.pop("model", None)
return model, session_data, session_type
def _encode_realtime_token_payload(
@ -32,6 +165,7 @@ def _encode_realtime_token_payload(
user_id: Optional[str],
team_id: Optional[str],
expires_at: Optional[int],
session_type: str = "realtime",
) -> str:
"""
Encode metadata with the upstream ephemeral key so /realtime/calls can
@ -44,6 +178,7 @@ def _encode_realtime_token_payload(
"user_id": user_id or "",
"team_id": team_id or "",
"expires_at": expires_at,
"session_type": session_type,
}
return json.dumps(payload, separators=(",", ":"))
@ -94,6 +229,7 @@ async def create_realtime_client_secret(
add_litellm_data_to_request,
general_settings,
llm_router,
llm_model_list,
proxy_config,
proxy_logging_obj,
route_request,
@ -106,17 +242,18 @@ async def create_realtime_client_secret(
body = await _read_request_body(request=request)
req = RealtimeClientSecretRequest(**body)
model: str = (
(req.session.model if req.session else None)
or req.model
or "gpt-4o-realtime-preview"
model, session_data, session_type = await _prepare_client_secret_session(
req=req,
user_api_key_dict=user_api_key_dict,
llm_model_list=llm_model_list,
llm_router=llm_router,
)
data = {"model": model}
# If session is provided, use it; otherwise create one from model
if req.session:
data["session"] = req.session.model_dump(exclude_none=True)
if session_data is not None:
data["session"] = session_data
elif req.model:
# User provided model at root level, convert to session format
data["session"] = {"type": "realtime", "model": model}
@ -161,6 +298,8 @@ async def create_realtime_client_secret(
"litellm.proxy.realtime_endpoints.webrtc.create_realtime_client_secret(): Exception - %s",
str(e),
)
if isinstance(e, ProxyException):
raise e
if isinstance(e, HTTPException):
raise ProxyException(
message=getattr(e, "message", str(e)),
@ -199,6 +338,7 @@ async def create_realtime_client_secret(
user_id=getattr(user_api_key_dict, "user_id", None),
team_id=getattr(user_api_key_dict, "team_id", None),
expires_at=expires_at if isinstance(expires_at, int) else None,
session_type=session_type,
)
encrypted_token: str = encrypt_value_helper(token_payload)
upstream_json["value"] = encrypted_token
@ -279,16 +419,20 @@ async def proxy_realtime_calls(
model = (
decoded_payload.get("model_id")
or request.query_params.get("model")
or "gpt-4o-realtime-preview"
or _DEFAULT_REALTIME_MODEL
)
user_id = decoded_payload.get("user_id") or None
team_id = decoded_payload.get("team_id") or None
session_type = _coerce_realtime_session_type(
decoded_payload.get("session_type")
)
else:
# Backward compatibility: older tokens contained only encrypted upstream key.
openai_ephemeral_key = decrypted_token_value
model = request.query_params.get("model", "gpt-4o-realtime-preview")
model = request.query_params.get("model", _DEFAULT_REALTIME_MODEL)
user_id = None
team_id = None
session_type = "realtime"
# Build a minimal UserAPIKeyAuth with user/team IDs from the token
# so spend tracking and budget enforcement work correctly.
@ -299,11 +443,17 @@ async def proxy_realtime_calls(
data: dict = {}
try:
# Build session config for the multipart form data
session_config = {
"type": "realtime",
"model": model,
"type": session_type,
}
if session_type == "transcription":
_set_transcription_model_on_session(
session=session_config,
model=model,
create_if_missing=True,
)
else:
session_config["model"] = model
data = {
"model": model,
@ -366,3 +516,145 @@ async def proxy_realtime_calls(
status_code=upstream_resp.status_code,
media_type=upstream_resp.headers.get("content-type", "application/sdp"),
)
@router.post(
"/v1/realtime/transcription_sessions",
dependencies=[Depends(user_api_key_auth)],
tags=["realtime"],
)
@router.post(
"/realtime/transcription_sessions",
dependencies=[Depends(user_api_key_auth)],
tags=["realtime"],
)
@router.post(
"/openai/v1/realtime/transcription_sessions",
dependencies=[Depends(user_api_key_auth)],
tags=["realtime"],
)
async def create_realtime_transcription_session(
request: Request,
fastapi_response: Response,
user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
) -> RealtimeTranscriptionSessionResponse:
"""
Create an ephemeral Realtime transcription session
(POST /v1/realtime/transcription_sessions) for the WebRTC/WebSocket flow.
Mirrors the client_secrets route but targets the transcription_sessions
endpoint and encrypts the ephemeral key returned under `client_secret.value`.
"""
from litellm.proxy.proxy_server import (
add_litellm_data_to_request,
general_settings,
llm_router,
llm_model_list,
proxy_config,
proxy_logging_obj,
route_request,
user_model,
version,
)
data: dict = {}
try:
body = await _read_request_body(request=request)
req = RealtimeTranscriptionSessionRequest(**body)
model: str = req.resolved_model() or "gpt-realtime-whisper"
await can_key_call_resolved_model(
model=model,
valid_token=user_api_key_dict,
llm_model_list=llm_model_list,
llm_router=llm_router,
)
transcription_session = {k: v for k, v in body.items() if k != "model"}
data = {"model": model, "transcription_session": transcription_session}
data = await add_litellm_data_to_request(
data=data,
request=request,
general_settings=general_settings,
user_api_key_dict=user_api_key_dict,
version=version,
proxy_config=proxy_config,
)
data = await proxy_logging_obj.pre_call_hook(
user_api_key_dict=user_api_key_dict,
data=data,
call_type="acreate_realtime_transcription_session",
)
verbose_proxy_logger.debug(
"Realtime: /v1/realtime/transcription_sessions (model=%s)", model
)
llm_call = await route_request(
data=data,
route_type="acreate_realtime_transcription_session",
llm_router=llm_router,
user_model=user_model,
)
upstream_resp: httpx.Response = await llm_call # type: ignore
except Exception as e:
await proxy_logging_obj.post_call_failure_hook(
user_api_key_dict=user_api_key_dict,
original_exception=e,
request_data=data,
)
verbose_proxy_logger.error(
"litellm.proxy.realtime_endpoints.create_realtime_transcription_session(): Exception - %s",
str(e),
)
if isinstance(e, ProxyException):
raise e
if isinstance(e, HTTPException):
raise ProxyException(
message=getattr(e, "detail", getattr(e, "message", str(e))),
type=getattr(e, "type", "None"),
param=getattr(e, "param", "None"),
code=getattr(e, "status_code", http_status.HTTP_400_BAD_REQUEST),
)
raise ProxyException(
message=getattr(e, "message", str(e)),
type=getattr(e, "type", "None"),
param=getattr(e, "param", "None"),
code=getattr(e, "status_code", 500),
)
if upstream_resp.status_code != 200:
verbose_proxy_logger.error(
"Realtime transcription_sessions upstream error %s: %s",
upstream_resp.status_code,
upstream_resp.text,
)
return Response( # type: ignore[return-value]
content=upstream_resp.content,
status_code=upstream_resp.status_code,
media_type="application/json",
)
upstream_json: dict = upstream_resp.json()
# Encrypt the ephemeral key (returned under client_secret.value) with routing
# metadata so the follow-up /realtime/calls request can recover the model.
client_secret = upstream_json.get("client_secret")
if isinstance(client_secret, dict) and "value" in client_secret:
raw_value: str = client_secret.get("value", "")
expires_at = client_secret.get("expires_at")
token_payload = _encode_realtime_token_payload(
ephemeral_key=raw_value,
model_id=model,
user_id=getattr(user_api_key_dict, "user_id", None),
team_id=getattr(user_api_key_dict, "team_id", None),
expires_at=expires_at if isinstance(expires_at, int) else None,
session_type="transcription",
)
client_secret["value"] = encrypt_value_helper(token_payload)
upstream_json["client_secret"] = client_secret
return RealtimeTranscriptionSessionResponse(**upstream_json)

3
litellm/proxy/route_llm_request.py
View file

@ -74,6 +74,7 @@ ROUTE_ENDPOINT_MAPPING = {
"avideo_extension": "/videos/extensions",
"acreate_realtime_client_secret": "/realtime/client_secrets",
"arealtime_calls": "/realtime/calls",
"acreate_realtime_transcription_session": "/realtime/transcription_sessions",
"acreate_container": "/containers",
"alist_containers": "/containers",
"aretrieve_container": "/containers/{container_id}",
@ -261,6 +262,7 @@ async def route_request( # noqa: PLR0915 - Complex routing function, refactorin
"_arealtime", # private function for realtime API
"acreate_realtime_client_secret",
"arealtime_calls",
"acreate_realtime_transcription_session",
"_aresponses_websocket", # private function for responses WebSocket mode
"aimage_edit",
"agenerate_content",
@ -427,6 +429,7 @@ async def route_request( # noqa: PLR0915 - Complex routing function, refactorin
"adelete_run",
"acreate_realtime_client_secret",
"arealtime_calls",
"acreate_realtime_transcription_session",
]:
# If a model is provided, get its credentials from the router
model = data.get("model")

10
litellm/realtime_api/README.md
View file

@ -1 +1,9 @@
Abstraction / Routing logic for OpenAI's `/v1/realtime` endpoint.
Abstraction / Routing logic for OpenAI's `/v1/realtime` endpoints.
Supported endpoints:
- WebSocket: `/v1/realtime` (with `intent=transcription` for transcription-only sessions)
- HTTP: `/v1/realtime/client_secrets`, `/v1/realtime/transcription_sessions`
Supported providers: OpenAI, Azure OpenAI, Bedrock, Vertex AI, xAI.
For user-facing documentation and usage examples, see the litellm-docs repo.

91
litellm/realtime_api/main.py
View file

@ -15,6 +15,7 @@ from litellm.types.realtime import (
RealtimeExpiresAfter,
RealtimeQueryParams,
RealtimeSessionConfig,
RealtimeTranscriptionSessionRequest,
)
from litellm.types.router import GenericLiteLLMParams
from litellm.types.utils import LlmProviders
@ -159,6 +160,78 @@ async def acreate_realtime_client_secret(
)
@wrapper_client
async def acreate_realtime_transcription_session(
model: Optional[str] = None,
transcription_session: Optional[Dict[str, Any]] = None,
timeout: Optional[float] = None,
**kwargs,
):
"""
Create an ephemeral transcription session via POST
/v1/realtime/transcription_sessions.
``transcription_session`` is the upstream request body (input_audio_format,
input_audio_transcription, turn_detection, …). ``model`` is a LiteLLM-only
routing hint; the provider model lives in
``transcription_session.input_audio_transcription.model``.
"""
req = RealtimeTranscriptionSessionRequest(
model=model,
**(transcription_session or {}),
)
model_name = req.resolved_model() or "gpt-realtime-whisper"
litellm_logging_obj: LiteLLMLogging = kwargs.get("litellm_logging_obj") # type: ignore
litellm_params = GenericLiteLLMParams(**kwargs)
(
model_name,
custom_llm_provider,
dynamic_api_key,
dynamic_api_base,
) = get_llm_provider(
model=model_name,
api_base=litellm_params.api_base,
api_key=litellm_params.api_key,
)
(
provider_config,
resolved_api_base,
resolved_api_key,
) = _get_realtime_http_provider_config(
custom_llm_provider=custom_llm_provider,
dynamic_api_base=dynamic_api_base,
dynamic_api_key=dynamic_api_key,
litellm_params=litellm_params,
)
litellm_logging_obj.update_from_kwargs(
kwargs=kwargs,
model=model_name,
optional_params={"transcription_session": transcription_session},
litellm_params={"api_base": resolved_api_base},
custom_llm_provider=custom_llm_provider,
)
request_data = 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
return await base_llm_http_handler.async_realtime_transcription_session_handler(
api_base=resolved_api_base,
api_key=resolved_api_key,
request_data=request_data,
logging_obj=litellm_logging_obj,
timeout=timeout or request_timeout,
provider_config=provider_config,
model=model_name,
extra_headers=kwargs.get("extra_headers"),
client=kwargs.get("client"),
api_version=litellm_params.api_version,
)
@wrapper_client
async def arealtime_calls(
openai_ephemeral_key: str,
@ -246,9 +319,13 @@ async def _arealtime( # noqa: PLR0915
api_key=api_key,
)
# Ensure query params use the normalized provider model (no proxy aliases).
# If the client supplied `model` in the URL, ensure it uses the normalized
# provider model (no proxy aliases). If they omitted it, preserve that shape
# for transcription-only sessions like OpenAI's `?intent=transcription`.
if query_params is not None:
query_params = {**query_params, "model": model}
query_params = {**query_params}
if "model" in query_params:
query_params["model"] = model
litellm_logging_obj.update_from_kwargs(
kwargs=kwargs,
@ -278,6 +355,7 @@ async def _arealtime( # noqa: PLR0915
headers=headers,
user_api_key_dict=kwargs.get("user_api_key_dict"),
litellm_metadata=_build_litellm_metadata(kwargs),
query_params=query_params,
)
elif _custom_llm_provider == "azure":
api_base = (
@ -300,8 +378,13 @@ async def _arealtime( # noqa: PLR0915
kwargs.get("realtime_protocol")
or litellm_params.get("realtime_protocol")
or os.environ.get("LITELLM_AZURE_REALTIME_PROTOCOL")
or "beta"
)
if (
realtime_protocol is None
and (query_params or {}).get("intent") == "transcription"
):
realtime_protocol = "GA"
realtime_protocol = realtime_protocol or "beta"
await azure_realtime.async_realtime(
model=model,
websocket=websocket,
@ -313,6 +396,7 @@ async def _arealtime( # noqa: PLR0915
timeout=timeout,
logging_obj=litellm_logging_obj,
realtime_protocol=realtime_protocol,
query_params=query_params,
user_api_key_dict=kwargs.get("user_api_key_dict"),
litellm_metadata=_build_litellm_metadata(kwargs),
)
@ -450,6 +534,7 @@ async def _arealtime( # noqa: PLR0915
headers=headers,
user_api_key_dict=kwargs.get("user_api_key_dict"),
litellm_metadata=_build_litellm_metadata(kwargs),
query_params=query_params,
)
else:
raise ValueError(f"Unsupported model: {model}")

37
litellm/types/realtime.py
View file

@ -115,3 +115,40 @@ class RealtimeClientSecretResponse(BaseModel):
expires_at: Optional[int] = None
value: str
session: Optional[Dict[str, Any]] = None
class RealtimeTranscriptionSessionRequest(BaseModel):
"""
Request body for POST /v1/realtime/transcription_sessions.
Mirrors OpenAI's RealtimeTranscriptionSessionCreateRequest. The model used
for routing is taken from the LiteLLM-only top-level `model` hint, falling
back to `input_audio_transcription.model`. All other fields pass through
unchanged to the provider.
"""
model_config = {"extra": "allow"}
# LiteLLM-only routing hint — stripped before forwarding upstream.
model: Optional[str] = None
input_audio_transcription: Optional[Dict[str, Any]] = None
def resolved_model(self) -> Optional[str]:
if self.model:
return self.model
if self.input_audio_transcription:
return self.input_audio_transcription.get("model")
return None
class RealtimeTranscriptionSessionResponse(BaseModel):
"""
Response from POST /v1/realtime/transcription_sessions.
`client_secret.value` contains the encrypted token instead of the raw
ephemeral key. Unknown fields pass through unchanged.
"""
model_config = {"extra": "allow"}
client_secret: Optional[Dict[str, Any]] = None

1
litellm/types/utils.py
View file

@ -533,6 +533,7 @@ CallTypesLiteral = Literal[
"acreate_skill",
"acreate_realtime_client_secret",
"arealtime_calls",
"acreate_realtime_transcription_session",
]
# Mapping of API routes to their corresponding call types

34
model_prices_and_context_window.json
View file

@ -4409,6 +4409,23 @@
"/v1/audio/transcriptions"
]
},
"azure/gpt-realtime-whisper": {
"input_cost_per_second": 0.0002833333333333333,
"litellm_provider": "azure",
"mode": "audio_transcription",
"source": "https://learn.microsoft.com/en-us/azure/ai-foundry/openai/concepts/gpt-realtime-whisper",
"supported_endpoints": [
"/v1/realtime",
"/v1/realtime/transcription_sessions"
],
"supported_modalities": [
"audio"
],
"supported_output_modalities": [
"text"
],
"supports_audio_input": true
},
"azure/gpt-5.1-2025-11-13": {
"cache_read_input_token_cost": 1.25e-07,
"cache_read_input_token_cost_priority": 2.5e-07,
@ -40956,6 +40973,23 @@
"supports_system_messages": true,
"supports_tool_choice": true
},
"gpt-realtime-whisper": {
"input_cost_per_second": 0.0002833333333333333,
"litellm_provider": "openai",
"mode": "audio_transcription",
"source": "https://platform.openai.com/docs/models/gpt-realtime-whisper",
"supported_endpoints": [
"/v1/realtime",
"/v1/realtime/transcription_sessions"
],
"supported_modalities": [
"audio"
],
"supported_output_modalities": [
"text"
],
"supports_audio_input": true
},
"sora-2": {
"litellm_provider": "openai",
"mode": "video_generation",

35
tests/llm_translation/realtime/test_openai_realtime.py
View file

@ -393,3 +393,38 @@ async def test_realtime_query_params_use_normalized_model_name(monkeypatch):
called_kwargs = mock_async_realtime.call_args.kwargs
assert called_kwargs["query_params"]["model"] == "gpt-4o-realtime-preview"
assert called_kwargs["query_params"]["intent"] == "chat"
@pytest.mark.asyncio
async def test_realtime_query_params_preserve_missing_model(monkeypatch):
"""
OpenAI-compatible transcription clients can connect with only
?intent=transcription and send the model in session.update. Do not add
model= back into the upstream query params when the client omitted it.
"""
from litellm.realtime_api import main as realtime_main
mock_async_realtime = AsyncMock()
monkeypatch.setattr(
realtime_main,
"openai_realtime",
MagicMock(async_realtime=mock_async_realtime),
)
def fake_get_llm_provider(model, api_base=None, api_key=None):
return ("gpt-realtime-whisper", "openai", None, None)
monkeypatch.setattr(realtime_main, "get_llm_provider", fake_get_llm_provider)
query_params: RealtimeQueryParams = {"intent": "transcription"}
await realtime_main._arealtime(
model="gpt-realtime-whisper",
websocket=MagicMock(),
api_key="sk-test",
query_params=query_params,
litellm_logging_obj=MagicMock(),
)
called_kwargs = mock_async_realtime.call_args.kwargs
assert called_kwargs["query_params"] == {"intent": "transcription"}

4
tests/proxy_unit_tests/test_realtime_cache.py
View file

@ -44,10 +44,14 @@ def test_realtime_query_params_template_caches_each_pair_separately():
params_with_intent_first = _realtime_query_params_template("gpt-4o", "intent-a")
params_with_intent_second = _realtime_query_params_template("gpt-4o", "intent-a")
params_without_intent = _realtime_query_params_template("gpt-4o", None)
params_transcription_without_model = _realtime_query_params_template(
None, "transcription"
)
assert params_with_intent_first is params_with_intent_second
assert params_with_intent_first == (("model", "gpt-4o"), ("intent", "intent-a"))
assert params_without_intent == (("model", "gpt-4o"),)
assert params_transcription_without_model == (("intent", "transcription"),)
assert params_with_intent_first is not params_without_intent

284
tests/test_litellm/litellm_core_utils/test_realtime_streaming.py
View file

@ -521,6 +521,290 @@ async def test_transcription_captured_in_backend_to_client():
assert logging_obj.model_call_details["messages"] == streaming.input_messages
@pytest.mark.asyncio
async def test_transcription_session_captures_usage_and_skips_response_create():
"""
For a transcription-only session (session.type == "transcription", e.g.
gpt-realtime-whisper), the completed event's audio-duration usage must be
captured for cost and response.create must NOT be sent to the backend.
"""
client_ws = MagicMock()
client_ws.send_text = AsyncMock()
session_created = json.dumps(
{
"type": "session.created",
"session": {
"type": "transcription",
"audio": {
"input": {"transcription": {"model": "gpt-realtime-whisper"}}
},
},
}
).encode()
completed = json.dumps(
{
"type": "conversation.item.input_audio_transcription.completed",
"transcript": "hello world",
"item_id": "item_1",
"usage": {"type": "duration", "seconds": 12.0},
}
).encode()
backend_ws = MagicMock()
backend_ws.recv = AsyncMock(
side_effect=[session_created, completed, ConnectionClosed(None, None)]
)
backend_ws.send = AsyncMock()
logging_obj = MagicMock()
logging_obj.model_call_details = {}
logging_obj.async_success_handler = AsyncMock()
logging_obj.success_handler = MagicMock()
streaming = RealTimeStreaming(client_ws, backend_ws, logging_obj)
await streaming.backend_to_client_send_messages()
assert streaming._is_transcription_session is True
captured = [
m
for m in streaming.messages
if m.get("type") == "conversation.item.input_audio_transcription.completed"
]
assert len(captured) == 1, "completed usage event must be captured for cost"
assert captured[0]["usage"]["seconds"] == 12.0
# Transcript still forwarded to the client.
client_ws.send_text.assert_any_call(completed.decode())
# No response.create — transcription sessions have no assistant turn.
sent_to_backend = [
json.loads(c.args[0]) for c in backend_ws.send.call_args_list if c.args
]
assert all(
e.get("type") != "response.create" for e in sent_to_backend
), f"transcription session must not trigger response.create, got: {sent_to_backend}"
@pytest.mark.asyncio
async def test_non_transcription_completed_event_still_triggers_response_create():
"""
Regression guard: a normal (non-transcription) session with no guardrails must
keep triggering response.create on a completed transcription event.
"""
client_ws = MagicMock()
client_ws.send_text = AsyncMock()
completed = json.dumps(
{
"type": "conversation.item.input_audio_transcription.completed",
"transcript": "hi",
"item_id": "item_1",
}
).encode()
backend_ws = MagicMock()
backend_ws.recv = AsyncMock(side_effect=[completed, ConnectionClosed(None, None)])
backend_ws.send = AsyncMock()
logging_obj = MagicMock()
logging_obj.async_success_handler = AsyncMock()
logging_obj.success_handler = MagicMock()
streaming = RealTimeStreaming(client_ws, backend_ws, logging_obj)
await streaming.backend_to_client_send_messages()
assert streaming._is_transcription_session is False
sent_to_backend = [
json.loads(c.args[0]) for c in backend_ws.send.call_args_list if c.args
]
assert any(e.get("type") == "response.create" for e in sent_to_backend)
def test_client_session_update_marks_transcription_session():
"""A client session.update with type=transcription flags the session."""
streaming = RealTimeStreaming(MagicMock(), MagicMock(), MagicMock())
assert streaming._is_transcription_session is False
streaming._collect_user_input_from_client_event(
json.dumps({"type": "session.update", "session": {"type": "transcription"}})
)
assert streaming._is_transcription_session is True
@pytest.mark.asyncio
async def test_transcription_session_update_enforces_authorized_flat_model():
backend_ws = MagicMock()
backend_ws.send = AsyncMock()
streaming = RealTimeStreaming(
MagicMock(),
backend_ws,
MagicMock(),
model="gpt-realtime-whisper",
force_transcription_model="gpt-realtime-whisper",
)
await streaming._send_to_backend(
json.dumps(
{
"type": "session.update",
"session": {
"type": "transcription",
"input_audio_transcription": {
"model": "restricted-transcription-model",
"language": "en",
},
},
}
)
)
sent = json.loads(backend_ws.send.await_args.args[0])
assert sent["session"]["input_audio_transcription"] == {
"model": "gpt-realtime-whisper",
"language": "en",
}
assert streaming._is_transcription_session is True
@pytest.mark.asyncio
async def test_transcription_session_update_enforces_authorized_nested_model():
backend_ws = MagicMock()
backend_ws.send = AsyncMock()
streaming = RealTimeStreaming(
MagicMock(),
backend_ws,
MagicMock(),
model="gpt-realtime-whisper",
force_transcription_model="gpt-realtime-whisper",
)
await streaming._send_to_backend(
json.dumps(
{
"type": "session.update",
"session": {
"type": "transcription",
"audio": {
"input": {
"transcription": {
"model": "restricted-transcription-model",
"prompt": "domain words",
},
"format": {"type": "audio/pcm", "rate": 24000},
}
},
},
}
)
)
sent = json.loads(backend_ws.send.await_args.args[0])
assert sent["session"]["audio"]["input"]["transcription"] == {
"model": "gpt-realtime-whisper",
"prompt": "domain words",
}
assert sent["session"]["audio"]["input"]["format"] == {
"type": "audio/pcm",
"rate": 24000,
}
assert streaming._is_transcription_session is True
@pytest.mark.asyncio
async def test_normal_realtime_session_keeps_nested_transcription_model():
backend_ws = MagicMock()
backend_ws.send = AsyncMock()
streaming = RealTimeStreaming(
MagicMock(),
backend_ws,
MagicMock(),
model="gpt-4o-realtime-preview",
)
await streaming._send_to_backend(
json.dumps(
{
"type": "session.update",
"session": {
"type": "realtime",
"audio": {
"input": {
"transcription": {
"model": "whisper-1",
"language": "en",
}
}
},
},
}
)
)
sent = json.loads(backend_ws.send.await_args.args[0])
assert sent["session"]["audio"]["input"]["transcription"] == {
"model": "whisper-1",
"language": "en",
}
assert streaming._is_transcription_session is False
def test_detect_transcription_session_from_backend_transcription_session_events():
"""Backend transcription_session.created/updated events flag the session."""
streaming = RealTimeStreaming(MagicMock(), MagicMock(), MagicMock())
assert streaming._is_transcription_session is False
streaming._detect_transcription_session_from_backend(
{"type": "transcription_session.created"}
)
assert streaming._is_transcription_session is True
streaming2 = RealTimeStreaming(MagicMock(), MagicMock(), MagicMock())
streaming2._detect_transcription_session_from_backend(
{"type": "transcription_session.updated"}
)
assert streaming2._is_transcription_session is True
def test_detect_transcription_session_from_backend_session_created_with_type():
"""Backend session.created with type=transcription flags the session."""
streaming = RealTimeStreaming(MagicMock(), MagicMock(), MagicMock())
streaming._detect_transcription_session_from_backend(
{"type": "session.created", "session": {"type": "transcription"}}
)
assert streaming._is_transcription_session is True
def test_detect_transcription_session_from_backend_ignores_non_transcription():
"""Backend session.created without type=transcription does not flag the session."""
streaming = RealTimeStreaming(MagicMock(), MagicMock(), MagicMock())
streaming._detect_transcription_session_from_backend(
{"type": "session.created", "session": {"model": "gpt-4o-realtime-preview"}}
)
assert streaming._is_transcription_session is False
def test_capture_transcription_usage_deduplicates_when_already_stored():
"""
When the event is already in messages (logged via store_message), it must not
be appended a second time by _capture_transcription_usage.
"""
import litellm
streaming = RealTimeStreaming(MagicMock(), MagicMock(), MagicMock())
# Add the event type to the default logged list so _should_store_message returns True.
streaming.logged_real_time_event_types = [
"conversation.item.input_audio_transcription.completed"
]
event = {
"type": "conversation.item.input_audio_transcription.completed",
"usage": {"type": "duration", "seconds": 5.0},
}
streaming.store_message(json.dumps(event))
initial_count = len(streaming.messages)
streaming._capture_transcription_usage(event)
assert len(streaming.messages) == initial_count # no duplicate
@pytest.mark.asyncio
async def test_client_ack_caches_setup_to_prevent_duplicate_session_update_setup():
websocket = MagicMock()

136
tests/test_litellm/llms/azure/realtime/test_azure_realtime_handler.py
View file

@ -147,6 +147,103 @@ async def test_construct_url_ga_protocol():
assert "deployment" not in url
@pytest.mark.asyncio
async def test_construct_url_forwards_transcription_intent_ga():
"""
Transcription sessions connect with intent=transcription. The Azure handler
must forward that query param so gpt-realtime-whisper opens a transcription
session instead of a normal realtime session.
"""
from litellm.llms.azure.realtime.handler import AzureOpenAIRealtime
handler = AzureOpenAIRealtime()
url = handler._construct_url(
api_base="https://my-endpoint.openai.azure.com",
model="gpt-realtime-whisper",
api_version="2025-04-01-preview",
realtime_protocol="GA",
query_params={"model": "gpt-realtime-whisper", "intent": "transcription"},
)
assert "/openai/v1/realtime?" in url
assert "intent=transcription" in url
assert "model=" not in url
@pytest.mark.asyncio
async def test_construct_url_forwards_transcription_intent_ga_without_model_query():
"""
OpenAI-compatible transcription clients may connect with only
intent=transcription and send the transcription model in session.update.
Preserve that query shape instead of forcing model= into the upstream URL.
"""
from litellm.llms.azure.realtime.handler import AzureOpenAIRealtime
handler = AzureOpenAIRealtime()
url = handler._construct_url(
api_base="https://my-endpoint.openai.azure.com",
model="gpt-realtime-whisper",
api_version="2025-04-01-preview",
realtime_protocol="GA",
query_params={"intent": "transcription"},
)
assert url == (
"wss://my-endpoint.openai.azure.com/openai/v1/realtime"
"?intent=transcription"
)
@pytest.mark.asyncio
async def test_construct_url_forwards_transcription_intent_beta():
from litellm.llms.azure.realtime.handler import AzureOpenAIRealtime
handler = AzureOpenAIRealtime()
url = handler._construct_url(
api_base="https://my-endpoint.openai.azure.com",
model="whisper-deploy",
api_version="2024-10-01-preview",
query_params={"intent": "transcription"},
)
assert "/openai/realtime?" in url
assert "deployment=whisper-deploy" in url
assert "intent=transcription" in url
@pytest.mark.asyncio
async def test_construct_url_encodes_intent_value():
"""A crafted intent value must be URL-encoded, not injected as raw query params."""
from litellm.llms.azure.realtime.handler import AzureOpenAIRealtime
handler = AzureOpenAIRealtime()
url = handler._construct_url(
api_base="https://my-endpoint.openai.azure.com",
model="gpt-realtime-whisper",
api_version="2025-04-01-preview",
realtime_protocol="GA",
query_params={"intent": "transcription&foo=bar"},
)
assert "intent=transcription%26foo%3Dbar" in url
assert "&foo=bar" not in url
@pytest.mark.asyncio
async def test_construct_url_no_intent_when_absent():
"""No intent param leaks into the URL when not provided."""
from litellm.llms.azure.realtime.handler import AzureOpenAIRealtime
handler = AzureOpenAIRealtime()
url = handler._construct_url(
api_base="https://my-endpoint.openai.azure.com",
model="gpt-4o-realtime-preview",
api_version="2024-10-01-preview",
realtime_protocol="GA",
query_params={"model": "gpt-4o-realtime-preview"},
)
assert "intent=" not in url
@pytest.mark.asyncio
async def test_construct_url_v1_protocol():
"""
@ -368,6 +465,45 @@ async def test_realtime_protocol_from_litellm_params():
assert litellm_params.get("realtime_protocol") == "GA"
@pytest.mark.asyncio
async def test_arealtime_transcription_intent_defaults_to_ga(monkeypatch):
"""
Azure gpt-realtime-whisper transcription connects on the GA /openai/v1/realtime
path. If the DB model lacks realtime_protocol, infer GA from intent=transcription.
"""
from litellm.realtime_api import main as realtime_main
mock_async_realtime = AsyncMock()
monkeypatch.setattr(
realtime_main,
"azure_realtime",
MagicMock(async_realtime=mock_async_realtime),
)
def fake_get_llm_provider(model, api_base=None, api_key=None):
return (
"gpt-realtime-whisper",
"azure",
"test-key",
"https://my-endpoint.openai.azure.com",
)
monkeypatch.setattr(realtime_main, "get_llm_provider", fake_get_llm_provider)
await realtime_main._arealtime(
model="azure/gpt-realtime-whisper",
websocket=MagicMock(),
api_key="test-key",
api_version="2025-04-01-preview",
query_params={"intent": "transcription"},
litellm_logging_obj=MagicMock(),
)
called_kwargs = mock_async_realtime.call_args.kwargs
assert called_kwargs["realtime_protocol"] == "GA"
assert called_kwargs["query_params"] == {"intent": "transcription"}
@pytest.mark.asyncio
async def test_async_realtime_default_maintains_backwards_compatibility():
"""

226
tests/test_litellm/llms/openai/realtime/test_transcription_sessions.py Normal file
View file

@ -0,0 +1,226 @@
"""
Tests for the Realtime transcription_sessions surface used by gpt-realtime-whisper:
- OpenAI / Azure URL construction (POST /v1/realtime/transcription_sessions)
- RealtimeTranscriptionSessionRequest model-resolution + passthrough
- BaseLLMHTTPHandler.async_realtime_transcription_session_handler targeting
"""
import os
import sys
from unittest.mock import AsyncMock, MagicMock
import httpx
import pytest
sys.path.insert(0, os.path.abspath("../../../../.."))
from litellm.llms.azure.realtime.http_transformation import AzureRealtimeHTTPConfig
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler
from litellm.llms.custom_httpx.llm_http_handler import BaseLLMHTTPHandler
from litellm.llms.openai.realtime.http_transformation import OpenAIRealtimeHTTPConfig
from litellm.types.realtime import RealtimeTranscriptionSessionRequest
def test_openai_transcription_session_url():
cfg = OpenAIRealtimeHTTPConfig()
assert (
cfg.get_transcription_session_url(
api_base="https://api.openai.com", model="gpt-realtime-whisper"
)
== "https://api.openai.com/v1/realtime/transcription_sessions"
)
def test_openai_transcription_session_url_strips_trailing_v1():
"""A /v1 suffix must not be duplicated in the path."""
cfg = OpenAIRealtimeHTTPConfig()
assert (
cfg.get_transcription_session_url(
api_base="https://api.openai.com/v1", model="gpt-realtime-whisper"
)
== "https://api.openai.com/v1/realtime/transcription_sessions"
)
def test_azure_transcription_session_url_uses_deployment_and_api_version():
cfg = AzureRealtimeHTTPConfig()
url = cfg.get_transcription_session_url(
api_base="https://my.openai.azure.com",
model="whisper-deploy",
api_version="2025-04-01-preview",
)
assert (
url
== "https://my.openai.azure.com/openai/realtime/transcription_sessions?api-version=2025-04-01-preview"
)
def test_request_resolves_model_returns_none_when_both_absent():
req = RealtimeTranscriptionSessionRequest(input_audio_format="pcm16")
assert req.resolved_model() is None
req = RealtimeTranscriptionSessionRequest(
model="openai/gpt-realtime-whisper",
input_audio_transcription={"model": "gpt-realtime-whisper"},
)
assert req.resolved_model() == "openai/gpt-realtime-whisper"
def test_request_resolves_model_from_input_audio_transcription():
req = RealtimeTranscriptionSessionRequest(
input_audio_transcription={"model": "gpt-realtime-whisper", "language": "en"},
)
assert req.resolved_model() == "gpt-realtime-whisper"
def test_request_passthrough_excludes_routing_hint():
"""Unknown fields pass through; the litellm-only `model` hint is not forwarded."""
req = RealtimeTranscriptionSessionRequest(
model="openai/gpt-realtime-whisper",
input_audio_format="pcm16",
input_audio_transcription={"model": "gpt-realtime-whisper"},
turn_detection=None,
)
forwarded = req.model_dump(exclude_none=True, exclude={"model"})
assert "model" not in forwarded
assert forwarded["input_audio_format"] == "pcm16"
assert forwarded["input_audio_transcription"] == {"model": "gpt-realtime-whisper"}
@pytest.mark.asyncio
async def test_handler_posts_to_transcription_sessions_url():
handler = BaseLLMHTTPHandler()
mock_response = MagicMock(spec=httpx.Response)
mock_client = MagicMock(spec=AsyncHTTPHandler)
mock_client.post = AsyncMock(return_value=mock_response)
logging_obj = MagicMock()
logging_obj.pre_call = MagicMock()
request_body = {"input_audio_transcription": {"model": "gpt-realtime-whisper"}}
result = await handler.async_realtime_transcription_session_handler(
api_base="https://api.openai.com",
api_key="sk-test",
request_data=request_body,
logging_obj=logging_obj,
timeout=10.0,
provider_config=OpenAIRealtimeHTTPConfig(),
model="gpt-realtime-whisper",
client=mock_client,
)
assert result is mock_response
_, kwargs = mock_client.post.call_args
assert kwargs["url"] == "https://api.openai.com/v1/realtime/transcription_sessions"
assert kwargs["json"] == request_body
assert kwargs["headers"]["Authorization"] == "Bearer sk-test"
@pytest.mark.asyncio
async def test_client_secret_handler_still_targets_client_secrets_url():
"""Refactor regression: the client_secrets handler must keep its own URL."""
handler = BaseLLMHTTPHandler()
mock_response = MagicMock(spec=httpx.Response)
mock_client = MagicMock(spec=AsyncHTTPHandler)
mock_client.post = AsyncMock(return_value=mock_response)
logging_obj = MagicMock()
logging_obj.pre_call = MagicMock()
await handler.async_realtime_client_secret_handler(
api_base="https://api.openai.com",
api_key="sk-test",
request_data={"session": {"type": "realtime"}},
logging_obj=logging_obj,
timeout=10.0,
provider_config=OpenAIRealtimeHTTPConfig(),
model="gpt-4o-realtime-preview",
client=mock_client,
)
_, kwargs = mock_client.post.call_args
assert kwargs["url"] == "https://api.openai.com/v1/realtime/client_secrets"
@pytest.mark.asyncio
async def test_sdk_fn_routes_openai_transcription_session(monkeypatch):
"""
litellm.acreate_realtime_transcription_session resolves the OpenAI provider
from the transcription model and POSTs to the OpenAI transcription_sessions URL.
"""
import litellm
monkeypatch.setenv("OPENAI_API_KEY", "sk-unit-test")
mock_response = MagicMock(spec=httpx.Response)
mock_client = MagicMock(spec=AsyncHTTPHandler)
mock_client.post = AsyncMock(return_value=mock_response)
result = await litellm.acreate_realtime_transcription_session(
model="openai/gpt-realtime-whisper",
transcription_session={
"input_audio_format": "pcm16",
"input_audio_transcription": {"model": "gpt-realtime-whisper"},
},
client=mock_client,
)
assert result is mock_response
_, kwargs = mock_client.post.call_args
assert kwargs["url"].endswith("/v1/realtime/transcription_sessions")
# The litellm-only routing hint must not be forwarded upstream.
assert "model" not in kwargs["json"]
assert kwargs["json"]["input_audio_transcription"] == {
"model": "gpt-realtime-whisper"
}
def test_append_query_params_skips_existing_keys():
from litellm.llms.custom_httpx.llm_http_handler import BaseLLMHTTPHandler
url = "wss://example.com/v1/realtime?model=gpt-4o"
result = BaseLLMHTTPHandler._append_query_params(
url, {"model": "ignored", "intent": "transcription"}
)
assert "model=ignored" not in result
assert "intent=transcription" in result
def test_append_query_params_no_params_returns_unchanged():
from litellm.llms.custom_httpx.llm_http_handler import BaseLLMHTTPHandler
url = "wss://example.com/v1/realtime?model=gpt-4o"
assert BaseLLMHTTPHandler._append_query_params(url, None) == url
assert BaseLLMHTTPHandler._append_query_params(url, {}) == url
def test_append_query_params_encodes_special_chars():
from litellm.llms.custom_httpx.llm_http_handler import BaseLLMHTTPHandler
url = "wss://example.com/v1/realtime"
result = BaseLLMHTTPHandler._append_query_params(url, {"intent": "a&b=c"})
assert "intent=a%26b%3Dc" in result
assert "&b=c" not in result
def test_azure_construct_url_encodes_model_and_api_version():
"""model and api-version must be URL-encoded to prevent query-string injection."""
from litellm.llms.azure.realtime.handler import AzureOpenAIRealtime
h = AzureOpenAIRealtime()
url = h._construct_url(
"https://x.openai.azure.com",
"deploy&evil=1",
"2024-10-01-preview",
)
assert "evil=1" not in url.split("?", 1)[1]
url_ga = h._construct_url(
"https://x.openai.azure.com",
"deploy&evil=1",
None,
realtime_protocol="GA",
)
assert "evil=1" not in url_ga.split("?", 1)[1]

680
tests/test_litellm/proxy/realtime_endpoints/test_realtime_webrtc_endpoints.py
View file

@ -241,6 +241,142 @@ async def test_client_secrets_success_with_mock(
proxy_app.dependency_overrides.pop(user_api_key_auth, None)
@pytest.mark.asyncio
async def test_client_secrets_transcription_rejects_disallowed_nested_model(
proxy_app,
):
proxy_app.dependency_overrides[user_api_key_auth] = lambda: UserAPIKeyAuth(
user_id="test-user",
models=["gpt-4o-realtime-preview"],
)
try:
client = TestClient(proxy_app, raise_server_exceptions=False)
with (
patch("litellm.proxy.proxy_server.route_request") as mock_route_request,
patch("litellm.proxy.proxy_server.proxy_logging_obj") as mock_logging,
):
mock_logging.post_call_failure_hook = AsyncMock()
response = client.post(
"/v1/realtime/client_secrets",
headers={"Authorization": "Bearer sk-test-master-key"},
json={
"model": "gpt-4o-realtime-preview",
"session": {
"type": "transcription",
"model": "gpt-4o-realtime-preview",
"audio": {
"input": {
"transcription": {
"model": "gpt-realtime-whisper"
}
}
},
},
},
)
assert response.status_code == 403
assert "Tried to access gpt-realtime-whisper" in response.text
mock_route_request.assert_not_called()
finally:
proxy_app.dependency_overrides.pop(user_api_key_auth, None)
@pytest.mark.asyncio
async def test_client_secrets_transcription_routes_on_nested_model(
proxy_app,
mock_add_litellm_data,
mock_pre_call_hook,
):
proxy_app.dependency_overrides[user_api_key_auth] = lambda: UserAPIKeyAuth(
user_id="test-user",
models=["gpt-4o-realtime-preview", "gpt-realtime-whisper"],
)
captured = {}
future_expires_at = int(time.time()) + 3600
async def _capturing_route(*args, **kwargs):
captured["data"] = kwargs.get("data")
async def _inner():
resp = MagicMock(spec=httpx.Response)
resp.status_code = 200
resp.text = (
f'{{"value":"upstream_ephemeral_key","expires_at":{future_expires_at}}}'
)
resp.content = (
f'{{"value":"upstream_ephemeral_key","expires_at":{future_expires_at}}}'
).encode()
resp.headers = {}
resp.json.return_value = {
"value": "upstream_ephemeral_key",
"expires_at": future_expires_at,
}
return resp
return _inner()
try:
client = TestClient(proxy_app)
with (
patch(
"litellm.proxy.proxy_server.route_request",
side_effect=_capturing_route,
),
patch(
"litellm.proxy.proxy_server.add_litellm_data_to_request",
side_effect=mock_add_litellm_data,
),
patch("litellm.proxy.proxy_server.proxy_logging_obj") as mock_logging,
):
mock_logging.pre_call_hook = AsyncMock(side_effect=mock_pre_call_hook)
mock_logging.post_call_failure_hook = AsyncMock()
response = client.post(
"/v1/realtime/client_secrets",
headers={"Authorization": "Bearer sk-test-master-key"},
json={
"model": "gpt-4o-realtime-preview",
"session": {
"type": "transcription",
"model": "gpt-4o-realtime-preview",
"audio": {
"input": {
"transcription": {
"model": "gpt-realtime-whisper"
}
}
},
},
},
)
assert response.status_code == 200
assert captured["data"]["model"] == "gpt-realtime-whisper"
session = captured["data"]["session"]
assert session["type"] == "transcription"
assert "model" not in session
assert (
session["audio"]["input"]["transcription"]["model"]
== "gpt-realtime-whisper"
)
encrypted_value = response.json()["value"]
decoded = _decode_realtime_token_payload(
decrypt_value_helper(
encrypted_value,
key="client_secret.value",
exception_type="debug",
)
or ""
)
assert decoded is not None
assert decoded["model_id"] == "gpt-realtime-whisper"
assert decoded["session_type"] == "transcription"
finally:
proxy_app.dependency_overrides.pop(user_api_key_auth, None)
def test_realtime_calls_requires_auth(proxy_app):
"""POST /v1/realtime/calls returns 401 without Authorization.
@ -311,3 +447,547 @@ async def test_realtime_calls_success_with_valid_encrypted_token(
assert response.status_code == 201
assert response.content.startswith(b"v=0")
assert b"application/sdp" in response.headers.get("content-type", "").encode()
def test_token_payload_carries_session_type():
"""The encrypted token records the session kind so /realtime/calls can replay it."""
payload = _encode_realtime_token_payload(
ephemeral_key="epk",
model_id="gpt-realtime-whisper",
user_id=None,
team_id=None,
expires_at=None,
session_type="transcription",
)
decoded = _decode_realtime_token_payload(payload)
assert decoded is not None
assert decoded["session_type"] == "transcription"
@pytest.mark.asyncio
async def test_realtime_calls_replays_transcription_session_type(
proxy_app,
mock_add_litellm_data,
mock_pre_call_hook,
):
"""
A token minted for a transcription session must drive /realtime/calls to send
session.type == "transcription" upstream, not the default "realtime".
"""
captured = {}
async def _capturing_route(*args, **kwargs):
captured["session"] = kwargs.get("data", {}).get("session")
async def _inner():
resp = MagicMock(spec=httpx.Response)
resp.status_code = 201
resp.content = b"v=0\r\n"
resp.headers = {"content-type": "application/sdp"}
return resp
return _inner()
token_payload = _encode_realtime_token_payload(
ephemeral_key="epk",
model_id="gpt-realtime-whisper",
user_id=None,
team_id=None,
expires_at=int(time.time()) + 3600,
session_type="transcription",
)
encrypted_token = encrypt_value_helper(token_payload)
client = TestClient(proxy_app)
with (
patch(
"litellm.proxy.proxy_server.route_request",
side_effect=_capturing_route,
),
patch(
"litellm.proxy.proxy_server.add_litellm_data_to_request",
side_effect=mock_add_litellm_data,
),
patch("litellm.proxy.proxy_server.proxy_logging_obj") as mock_logging,
):
mock_logging.pre_call_hook = AsyncMock(side_effect=mock_pre_call_hook)
mock_logging.post_call_failure_hook = AsyncMock()
client.post(
"/v1/realtime/calls",
headers={"Authorization": f"Bearer {encrypted_token}"},
content=b"v=0\r\n",
)
assert captured["session"]["type"] == "transcription"
assert (
captured["session"]["audio"]["input"]["transcription"]["model"]
== "gpt-realtime-whisper"
)
# --- transcription_sessions endpoint ---
@pytest.fixture
def mock_route_request_transcription_sessions():
"""Mock route_request to return a fake transcription_sessions upstream response."""
future_expires_at = int(time.time()) + 3600
body = {
"id": "sess_abc",
"object": "realtime.transcription_session",
"client_secret": {
"value": "upstream_ephemeral_key",
"expires_at": future_expires_at,
},
}
mock_resp = MagicMock(spec=httpx.Response)
mock_resp.status_code = 200
mock_resp.text = json.dumps(body)
mock_resp.content = json.dumps(body).encode()
mock_resp.headers = {}
mock_resp.json.return_value = body
async def _mock_route(*args, **kwargs):
async def _inner():
return mock_resp
return _inner()
return _mock_route
def test_transcription_sessions_requires_auth(proxy_app):
"""POST /v1/realtime/transcription_sessions returns 401 without Authorization."""
from fastapi import HTTPException
def _raise_401():
raise HTTPException(status_code=401, detail="Unauthorized")
proxy_app.dependency_overrides[user_api_key_auth] = _raise_401
try:
client = TestClient(proxy_app, raise_server_exceptions=False)
response = client.post(
"/v1/realtime/transcription_sessions",
json={"input_audio_transcription": {"model": "gpt-realtime-whisper"}},
)
assert response.status_code == 401
finally:
proxy_app.dependency_overrides.pop(user_api_key_auth, None)
@pytest.mark.asyncio
async def test_transcription_sessions_rejects_disallowed_resolved_model(
proxy_app,
):
proxy_app.dependency_overrides[user_api_key_auth] = lambda: UserAPIKeyAuth(
user_id="test-user",
models=["gpt-4o-realtime-preview"],
)
try:
client = TestClient(proxy_app, raise_server_exceptions=False)
with (
patch("litellm.proxy.proxy_server.route_request") as mock_route_request,
patch("litellm.proxy.proxy_server.proxy_logging_obj") as mock_logging,
):
mock_logging.post_call_failure_hook = AsyncMock()
response = client.post(
"/v1/realtime/transcription_sessions",
headers={"Authorization": "Bearer sk-test-master-key"},
json={
"input_audio_transcription": {"model": "gpt-realtime-whisper"}
},
)
assert response.status_code == 403
assert "Tried to access gpt-realtime-whisper" in response.text
mock_route_request.assert_not_called()
finally:
proxy_app.dependency_overrides.pop(user_api_key_auth, None)
@pytest.mark.asyncio
async def test_transcription_sessions_rejects_disallowed_team_model_scope(
proxy_app,
):
from litellm.proxy._types import LiteLLM_TeamTableCachedObj
team = LiteLLM_TeamTableCachedObj(
team_id="team-a",
models=["gpt-4o-realtime-preview"],
)
proxy_app.dependency_overrides[user_api_key_auth] = lambda: UserAPIKeyAuth(
user_id="test-user",
team_id="team-a",
models=["*"],
)
try:
client = TestClient(proxy_app, raise_server_exceptions=False)
with (
patch("litellm.proxy.proxy_server.route_request") as mock_route_request,
patch("litellm.proxy.proxy_server.proxy_logging_obj") as mock_logging,
patch(
"litellm.proxy.auth.auth_checks.get_team_object",
new=AsyncMock(return_value=team),
),
patch(
"litellm.proxy.auth.auth_checks.get_team_membership",
new=AsyncMock(return_value=None),
),
):
mock_logging.post_call_failure_hook = AsyncMock()
response = client.post(
"/v1/realtime/transcription_sessions",
headers={"Authorization": "Bearer sk-test-master-key"},
json={
"input_audio_transcription": {"model": "gpt-realtime-whisper"}
},
)
assert response.status_code == 403
assert "team" in response.text.lower()
assert "Tried to access gpt-realtime-whisper" in response.text
mock_route_request.assert_not_called()
finally:
proxy_app.dependency_overrides.pop(user_api_key_auth, None)
@pytest.mark.asyncio
async def test_transcription_sessions_rejects_disallowed_project_model_scope(
proxy_app,
):
from litellm.proxy._types import LiteLLM_ProjectTableCachedObj
project = LiteLLM_ProjectTableCachedObj(
project_id="project-a",
models=["gpt-4o-realtime-preview"],
created_by="test-user",
updated_by="test-user",
)
proxy_app.dependency_overrides[user_api_key_auth] = lambda: UserAPIKeyAuth(
user_id="test-user",
project_id="project-a",
models=["*"],
)
try:
client = TestClient(proxy_app, raise_server_exceptions=False)
with (
patch("litellm.proxy.proxy_server.route_request") as mock_route_request,
patch("litellm.proxy.proxy_server.proxy_logging_obj") as mock_logging,
patch(
"litellm.proxy.auth.auth_checks.get_project_object",
new=AsyncMock(return_value=project),
),
):
mock_logging.post_call_failure_hook = AsyncMock()
response = client.post(
"/v1/realtime/transcription_sessions",
headers={"Authorization": "Bearer sk-test-master-key"},
json={
"input_audio_transcription": {"model": "gpt-realtime-whisper"}
},
)
assert response.status_code == 403
assert "project" in response.text.lower()
assert "Tried to access gpt-realtime-whisper" in response.text
mock_route_request.assert_not_called()
finally:
proxy_app.dependency_overrides.pop(user_api_key_auth, None)
@pytest.mark.asyncio
async def test_transcription_sessions_rejects_disallowed_team_member_model_scope(
proxy_app,
):
from litellm.proxy._types import (
LiteLLM_BudgetTable,
LiteLLM_TeamMembership,
LiteLLM_TeamTableCachedObj,
)
team = LiteLLM_TeamTableCachedObj(team_id="team-a", models=["*"])
membership = LiteLLM_TeamMembership(
user_id="test-user",
team_id="team-a",
litellm_budget_table=LiteLLM_BudgetTable(
allowed_models=["gpt-4o-realtime-preview"],
),
)
proxy_app.dependency_overrides[user_api_key_auth] = lambda: UserAPIKeyAuth(
user_id="test-user",
team_id="team-a",
models=["*"],
)
try:
client = TestClient(proxy_app, raise_server_exceptions=False)
with (
patch("litellm.proxy.proxy_server.route_request") as mock_route_request,
patch("litellm.proxy.proxy_server.proxy_logging_obj") as mock_logging,
patch(
"litellm.proxy.auth.auth_checks.get_team_object",
new=AsyncMock(return_value=team),
),
patch(
"litellm.proxy.auth.auth_checks.get_team_membership",
new=AsyncMock(return_value=membership),
),
):
mock_logging.post_call_failure_hook = AsyncMock()
response = client.post(
"/v1/realtime/transcription_sessions",
headers={"Authorization": "Bearer sk-test-master-key"},
json={
"input_audio_transcription": {"model": "gpt-realtime-whisper"}
},
)
assert response.status_code == 403
assert "Team member not allowed to access model" in response.text
mock_route_request.assert_not_called()
finally:
proxy_app.dependency_overrides.pop(user_api_key_auth, None)
@pytest.mark.asyncio
async def test_realtime_transcription_websocket_default_model_checks_key_scope():
from litellm.proxy import proxy_server
websocket = MagicMock()
websocket.headers = {}
websocket.close = AsyncMock()
websocket.accept = AsyncMock()
await proxy_server.realtime_websocket_endpoint(
websocket=websocket,
model=None,
intent="transcription",
user_api_key_dict=UserAPIKeyAuth(models=["gpt-4o-realtime-preview"]),
)
websocket.accept.assert_not_awaited()
websocket.close.assert_awaited_once()
_, close_kwargs = websocket.close.call_args
assert close_kwargs["code"] == 1008
assert "not allowed to access model" in close_kwargs["reason"]
@pytest.mark.asyncio
async def test_realtime_transcription_websocket_default_model_checks_team_scope():
from litellm.proxy import proxy_server
from litellm.proxy._types import LiteLLM_TeamTableCachedObj
team = LiteLLM_TeamTableCachedObj(
team_id="team-a",
models=["gpt-4o-realtime-preview"],
)
websocket = MagicMock()
websocket.headers = {}
websocket.close = AsyncMock()
websocket.accept = AsyncMock()
with (
patch(
"litellm.proxy.auth.auth_checks.get_team_object",
new=AsyncMock(return_value=team),
),
patch(
"litellm.proxy.auth.auth_checks.get_team_membership",
new=AsyncMock(return_value=None),
),
):
await proxy_server.realtime_websocket_endpoint(
websocket=websocket,
model=None,
intent="transcription",
user_api_key_dict=UserAPIKeyAuth(
user_id="test-user",
team_id="team-a",
models=["*"],
),
)
websocket.accept.assert_not_awaited()
websocket.close.assert_awaited_once()
_, close_kwargs = websocket.close.call_args
assert close_kwargs["code"] == 1008
assert "not allowed to access model" in close_kwargs["reason"]
@pytest.mark.asyncio
async def test_transcription_sessions_encrypts_client_secret(
proxy_app,
mock_route_request_transcription_sessions,
mock_add_litellm_data,
mock_pre_call_hook,
):
"""
POST /v1/realtime/transcription_sessions returns 200 and the ephemeral key
under client_secret.value must be encrypted (never the raw upstream key).
"""
proxy_app.dependency_overrides[user_api_key_auth] = lambda: UserAPIKeyAuth(
user_id="test-user", team_id="test-team"
)
captured_route_type = {}
async def _capturing_route(*args, **kwargs):
captured_route_type["route_type"] = kwargs.get("route_type")
return await mock_route_request_transcription_sessions(*args, **kwargs)
try:
client = TestClient(proxy_app)
with (
patch(
"litellm.proxy.proxy_server.route_request",
side_effect=_capturing_route,
),
patch(
"litellm.proxy.proxy_server.add_litellm_data_to_request",
side_effect=mock_add_litellm_data,
),
patch("litellm.proxy.proxy_server.proxy_logging_obj") as mock_logging,
):
mock_logging.pre_call_hook = AsyncMock(side_effect=mock_pre_call_hook)
mock_logging.post_call_failure_hook = AsyncMock()
response = client.post(
"/v1/realtime/transcription_sessions",
headers={"Authorization": "Bearer sk-test-master-key"},
json={
"input_audio_format": "pcm16",
"input_audio_transcription": {"model": "gpt-realtime-whisper"},
},
)
assert response.status_code == 200
data = response.json()
assert data["client_secret"]["value"] != "upstream_ephemeral_key"
# The encrypted value must decrypt back to a payload carrying the raw key.
decrypted = decrypt_value_helper(
data["client_secret"]["value"],
key="client_secret.value",
exception_type="debug",
)
assert decrypted is not None
assert "upstream_ephemeral_key" in decrypted
# Routed through the dedicated transcription_sessions route type.
assert (
captured_route_type["route_type"]
== "acreate_realtime_transcription_session"
)
finally:
proxy_app.dependency_overrides.pop(user_api_key_auth, None)
def test_session_type_coerced_for_unknown_value():
"""An unrecognized session_type in the token falls back to 'realtime'."""
payload = _encode_realtime_token_payload(
ephemeral_key="epk",
model_id="gpt-4o",
user_id=None,
team_id=None,
expires_at=None,
session_type="INJECTED_TYPE",
)
# Force-deserialize and check the coercion that happens in proxy_realtime_calls.
decoded = json.loads(payload)
session_type = decoded.get("session_type") or "realtime"
if session_type not in ("realtime", "transcription"):
session_type = "realtime"
assert session_type == "realtime"
@pytest.mark.asyncio
async def test_transcription_sessions_returns_upstream_error_verbatim(
proxy_app,
mock_add_litellm_data,
mock_pre_call_hook,
):
"""Non-200 upstream response is forwarded unchanged (no encryption attempted)."""
mock_resp = MagicMock(spec=httpx.Response)
mock_resp.status_code = 400
mock_resp.content = b'{"error":"bad_request"}'
mock_resp.headers = {}
mock_resp.json.return_value = {"error": "bad_request"}
mock_resp.text = '{"error":"bad_request"}'
async def _mock_route(*args, **kwargs):
async def _inner():
return mock_resp
return _inner()
proxy_app.dependency_overrides[user_api_key_auth] = lambda: UserAPIKeyAuth(
user_id="test-user", team_id="test-team"
)
try:
client = TestClient(proxy_app)
with (
patch(
"litellm.proxy.proxy_server.route_request",
side_effect=_mock_route,
),
patch(
"litellm.proxy.proxy_server.add_litellm_data_to_request",
side_effect=mock_add_litellm_data,
),
patch("litellm.proxy.proxy_server.proxy_logging_obj") as mock_logging,
):
mock_logging.pre_call_hook = AsyncMock(side_effect=mock_pre_call_hook)
mock_logging.post_call_failure_hook = AsyncMock()
response = client.post(
"/v1/realtime/transcription_sessions",
headers={"Authorization": "Bearer sk-test-master-key"},
json={"input_audio_transcription": {"model": "gpt-realtime-whisper"}},
)
assert response.status_code == 400
assert response.content == b'{"error":"bad_request"}'
finally:
proxy_app.dependency_overrides.pop(user_api_key_auth, None)
@pytest.mark.asyncio
async def test_transcription_sessions_wraps_route_exception(
proxy_app,
mock_add_litellm_data,
mock_pre_call_hook,
):
"""A route exception is wrapped in a ProxyException with a human-readable message."""
from fastapi import HTTPException
async def _raise_http(*args, **kwargs):
raise HTTPException(status_code=403, detail="Model not allowed")
proxy_app.dependency_overrides[user_api_key_auth] = lambda: UserAPIKeyAuth(
user_id="test-user"
)
try:
client = TestClient(proxy_app, raise_server_exceptions=False)
with (
patch(
"litellm.proxy.proxy_server.route_request",
side_effect=_raise_http,
),
patch(
"litellm.proxy.proxy_server.add_litellm_data_to_request",
side_effect=mock_add_litellm_data,
),
patch("litellm.proxy.proxy_server.proxy_logging_obj") as mock_logging,
):
mock_logging.pre_call_hook = AsyncMock(side_effect=mock_pre_call_hook)
mock_logging.post_call_failure_hook = AsyncMock()
response = client.post(
"/v1/realtime/transcription_sessions",
headers={"Authorization": "Bearer sk-test-master-key"},
json={"input_audio_transcription": {"model": "gpt-realtime-whisper"}},
)
assert response.status_code == 403
assert "Model not allowed" in response.text
finally:
proxy_app.dependency_overrides.pop(user_api_key_auth, None)

146
tests/test_litellm/test_cost_calculator.py
View file

@ -463,12 +463,154 @@ def test_realtime_logging_object_allows_null_transcript_in_conversation_item_add
usage=usage,
results=results,
)
assert logging_result.usage.total_tokens == 18
assert logging_result.results[0]["item"]["content"][0]["transcript"] is None
assert logging_result.results[0]["item"]["content"][0]["transcript"] is None
def test_custom_pricing_with_router_model_id():
def test_realtime_transcription_duration_cost(monkeypatch):
"""
gpt-realtime-whisper transcription sessions are billed by input audio duration
($0.017/min). The .completed events carry usage {type: duration, seconds: N};
cost must equal total_seconds * input_cost_per_second.
"""
monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True")
monkeypatch.setattr(litellm, "model_cost", litellm.get_model_cost_map(url=""))
from litellm.cost_calculator import RealtimeAPITokenUsageProcessor
results: OpenAIRealtimeStreamList = [
{
"type": "session.created",
"session": {
"type": "transcription",
"audio": {
"input": {"transcription": {"model": "gpt-realtime-whisper"}}
},
},
},
{
"type": "conversation.item.input_audio_transcription.completed",
"transcript": "hello",
"usage": {"type": "duration", "seconds": 60.0},
},
{
"type": "conversation.item.input_audio_transcription.completed",
"transcript": "world",
"usage": {"type": "duration", "seconds": 30.0},
},
]
combined = RealtimeAPITokenUsageProcessor.collect_and_combine_usage_from_realtime_stream_results(
results=results
)
cost = handle_realtime_stream_cost_calculation(
results=results,
combined_usage_object=combined,
custom_llm_provider="openai",
litellm_model_name="gpt-realtime-whisper",
)
# 90 seconds at $0.017/minute.
expected = 90.0 * (0.017 / 60)
assert abs(cost - expected) < 1e-9
assert cost > 0 # guards against the duration branch being dropped
def test_realtime_transcription_duration_cost_resolves_model_from_litellm_name(
monkeypatch,
):
"""When no session event carries the ASR model, the litellm_model_name is used."""
monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True")
monkeypatch.setattr(litellm, "model_cost", litellm.get_model_cost_map(url=""))
results: OpenAIRealtimeStreamList = [
{
"type": "conversation.item.input_audio_transcription.completed",
"usage": {"type": "duration", "seconds": 120.0},
},
]
cost = handle_realtime_stream_cost_calculation(
results=results,
combined_usage_object=Usage(),
custom_llm_provider="azure",
litellm_model_name="azure/gpt-realtime-whisper",
)
assert abs(cost - 120.0 * (0.017 / 60)) < 1e-9
def test_realtime_transcription_no_completed_events_is_zero(monkeypatch):
"""A realtime stream without transcription completed events adds no extra cost."""
monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True")
monkeypatch.setattr(litellm, "model_cost", litellm.get_model_cost_map(url=""))
from litellm.cost_calculator import handle_realtime_transcription_cost_calculation
results: OpenAIRealtimeStreamList = [
{"type": "session.created", "session": {"model": "gpt-realtime-whisper"}},
{"type": "response.done", "response": {"usage": {}}},
]
assert (
handle_realtime_transcription_cost_calculation(
results=results,
custom_llm_provider="openai",
litellm_model_name="gpt-realtime-whisper",
)
== 0.0
)
def test_realtime_transcription_token_billed_fallback(monkeypatch):
"""
Token-billed transcription models price by audio/text tokens. Verify the
fallback path multiplies audio tokens by the model's audio token cost.
"""
monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True")
monkeypatch.setattr(litellm, "model_cost", litellm.get_model_cost_map(url=""))
from litellm.cost_calculator import _transcription_usage_cost
# gpt-4o-transcribe: input_cost_per_audio_token = 2.5e-06, input_cost_per_token = 2.5e-06,
# output_cost_per_token = 1e-05
model_info = litellm.get_model_info(
model="gpt-4o-transcribe", custom_llm_provider="openai"
)
usage = {
"type": "tokens",
"input_tokens": 40,
"output_tokens": 10,
"total_tokens": 50,
"input_token_details": {"audio_tokens": 30, "text_tokens": 10},
}
cost = _transcription_usage_cost(usage, model_info)
expected = (
30 * 2.5e-06 # audio tokens
+ 10 * 2.5e-06 # text tokens
+ 10 * 1e-05 # output tokens
)
assert abs(cost - expected) < 1e-12
def test_transcription_usage_cost_returns_zero_for_unknown_type():
"""An unrecognized usage type yields 0 (safe fallback, no exception)."""
from litellm.cost_calculator import _transcription_usage_cost
assert _transcription_usage_cost({"type": "future_billing_type"}, {}) == 0.0
assert _transcription_usage_cost({}, {}) == 0.0
def test_get_transcription_model_falls_back_to_session_model(monkeypatch):
"""session.model is used when transcription-specific model fields are absent."""
monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True")
monkeypatch.setattr(litellm, "model_cost", litellm.get_model_cost_map(url=""))
from litellm.cost_calculator import _get_transcription_model_name_from_results
results: OpenAIRealtimeStreamList = [
{"type": "session.created", "session": {"model": "gpt-realtime-whisper"}},
]
assert _get_transcription_model_name_from_results(results) == "gpt-realtime-whisper"
from litellm import Router
router = Router(