scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-30 06:43:27 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 249

fix(health)!: openai realtime health check reports healthy with a missing or invalid api key (#41233)

Browse source 
* fix(health): report openai realtime deployments with a missing or invalid api key as unhealthy

* fix(health): time out a silent openai realtime handshake and keep the first-event parsing in the openai handler

* fix(health): treat invalid_model as healthy, read the first event only on OpenAI hosts, map rate limits to 429

* fix(health): respect the global api_base, report a missing key as 401, keep invalid_model healthy only for known models

Drop the separate first-event timeout so model_info.health_check_timeout governs the wait, and keep the shared stream error table unchanged by raising AuthenticationError for invalid_api_key in the realtime check

* fix(health): keep invalid_model healthy unless the cost map lists the model outside realtime, and load realtime types lazily

* fix(health): read the cost map's listed endpoints, not its mode, since registering a realtime deployment overwrites the mode

* fix(health): read a realtime gateway's first event and keep Live deployments listing /v1/live healthy

Behind a custom api_base the OpenAI realtime health check now waits up to
3 seconds for the first event, so a gateway that passes on OpenAI's
invalid_api_key error reports unhealthy while a silent gateway stays
healthy. /v1/live counts as a realtime endpoint for the invalid_model rule.

---------

Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
This commit is contained in:
devin-ai-integration[bot] 2026-10-29 00:40:02 -07:00 • committed by GitHub
parent 904c092ffe
commit 3f01341f0a
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 362 additions and 17 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -4,11 +4,17 @@ This file contains the calling OpenAI's `/v1/realtime` endpoint.
This requires websockets, and is currently only supported on LiteLLM Proxy.
"""
import asyncio
import ssl
from typing import TYPE_CHECKING, Final, cast
from collections.abc import Mapping
from typing import TYPE_CHECKING, Final, Literal, cast
from pydantic import TypeAdapter
from litellm._logging import redact_string, verbose_logger
from litellm.constants import REALTIME_WEBSOCKET_MAX_MESSAGE_SIZE_BYTES
from litellm.exceptions import AuthenticationError
from litellm.responses.streaming_iterator import map_stream_error_to_exception
from litellm.types.realtime import RealtimeQueryParams
from ....litellm_core_utils.litellm_logging import Logging as LiteLLMLogging
@ -25,6 +31,54 @@ from ..openai import OpenAIChatCompletion
if TYPE_CHECKING:
from fastapi import WebSocket
from openai.types.realtime import RealtimeError
from websockets.asyncio.client import ClientConnection
_SERVER_EVENT_FIELDS: Final = TypeAdapter(Mapping[str, object])
_CREDENTIALS_ACCEPTED_FOR_A_NON_REALTIME_MODEL: Final = "invalid_model"
_CREDENTIALS_REJECTED: Final = "invalid_api_key"
def first_event_error(first_event: str | bytes) -> "RealtimeError | None":
event: Final = _SERVER_EVENT_FIELDS.validate_json(first_event)
if event.get("type") != "error":
return None
from openai.types.realtime import RealtimeErrorEvent
return RealtimeErrorEvent.model_validate(event).error
def _confirm_first_event(
first_event: str | bytes, model: str, sent_credentials: bool, model_listed_outside_realtime: bool
) -> Literal[True]:
error: Final = first_event_error(first_event)
if error is None or (
error.code == _CREDENTIALS_ACCEPTED_FOR_A_NON_REALTIME_MODEL and not model_listed_outside_realtime
):
return True
if error.code == _CREDENTIALS_REJECTED or not sent_credentials:
raise AuthenticationError(message=error.message, llm_provider="openai", model=model)
raise map_stream_error_to_exception(error, model=model, custom_llm_provider="openai")
async def confirm_session_started(
connection: "ClientConnection", model: str, sent_credentials: bool, model_listed_outside_realtime: bool
) -> Literal[True]:
return _confirm_first_event(await connection.recv(), model, sent_credentials, model_listed_outside_realtime)
async def confirm_gateway_session_started(
connection: "ClientConnection",
model: str,
sent_credentials: bool,
model_listed_outside_realtime: bool,
first_event_wait_seconds: float,
) -> Literal[True]:
try:
first_event: Final = await asyncio.wait_for(connection.recv(), first_event_wait_seconds)
except asyncio.TimeoutError:
return True
return _confirm_first_event(first_event, model, sent_credentials, model_listed_outside_realtime)
class OpenAIRealtime(OpenAIChatCompletion):

65
litellm/realtime_api/main.py
View file

@ -31,7 +31,12 @@ from litellm.types.realtime import (
)
from litellm.types.router import GenericLiteLLMParams
from litellm.types.utils import CallTypes, LlmProviders
from litellm.utils import ProviderConfigManager, resolve_api_key_secret_reference
from litellm.utils import (
ModelInfoLookupMiss,
ProviderConfigManager,
lookup_model_info_for_registration,
resolve_api_key_secret_reference,
)
from ..litellm_core_utils.core_helpers import as_str_mapping, normalize_drop_params
from ..litellm_core_utils.credential_accessor import CredentialAccessor
@ -42,8 +47,13 @@ from ..llms.azure.common_utils import get_azure_ad_token
from ..llms.azure.realtime.handler import AzureOpenAIRealtime, azure_realtime_protocol_for_client
from ..llms.bedrock.realtime.handler import BedrockRealtime, raise_if_session_committed
from ..llms.custom_httpx.http_handler import realtime_ssl_for_url
from ..llms.openai.common_utils import is_openai_backed_api_base
from ..llms.openai.live.handler import OpenAILiveSessions
from ..llms.openai.realtime.handler import OpenAIRealtime
from ..llms.openai.realtime.handler import (
OpenAIRealtime,
confirm_gateway_session_started,
confirm_session_started,
)
from ..llms.openai.workload_identity import openai_static_api_key, resolve_openai_bearer_token_async
from ..llms.vertex_ai.audio_transcription.realtime_transformation import is_vertex_speech_to_text_model
from ..llms.vertex_ai.realtime.transformation import VertexAIRealtimeConfig, vertex_realtime_config
@ -740,6 +750,18 @@ def _is_transcription_only_realtime_model(model: str, custom_llm_provider: str)
return "/v1/realtime/transcription_sessions" in (model_info.get("supported_endpoints") or ())
_REALTIME_SERVED_ENDPOINTS: Final = ("/v1/realtime", "/v1/audio/transcriptions", "/v1/live")
_GATEWAY_FIRST_EVENT_WAIT_SECONDS: Final = 3.0
def _cost_map_lists_outside_realtime(model: str) -> bool:
model_info: Final = lookup_model_info_for_registration(model, "openai")
endpoints: Final = (
() if isinstance(model_info, ModelInfoLookupMiss) else model_info.get("supported_endpoints") or ()
)
return bool(endpoints) and not any(endpoint.startswith(_REALTIME_SERVED_ENDPOINTS) for endpoint in endpoints)
_TRANSCRIPTION_QUERY_PARAMS: Final[RealtimeQueryParams] = {"intent": "transcription"}
@ -785,6 +807,7 @@ async def realtime_health_check(
api_version: str | None = None,
realtime_protocol: str | None = None,
model_params: Mapping[str, object] | None = None,
gateway_first_event_wait_seconds: float = _GATEWAY_FIRST_EVENT_WAIT_SECONDS,
) -> bool:
"""
Health check for realtime API - tries connection to the realtime API websocket
@ -800,9 +823,16 @@ async def realtime_health_check(
without the OpenAI-Beta header is bridged to, with transcription-only models adding intent=transcription
Returns:
bool - True if connection is successful, False otherwise
bool - True once the connection is open, and on OpenAI's own hosts once the first server event is not an
error, since OpenAI accepts the websocket handshake with missing or invalid credentials and only reports
the failure in its first server event. Behind any other OpenAI api_base the first event is read for
`gateway_first_event_wait_seconds` only, and a gateway that stays silent that long counts as healthy.
An `invalid_model` error also counts as healthy unless litellm's
cost map lists only endpoints realtime never serves for the model: OpenAI checks the key first, and Live
and transcription models are served through other endpoints. The listed `mode` is not read, since a
deployment's own `mode: realtime` replaces it when the router registers the deployment
Raises:
Exception - if the connection is not successful
Exception - if the connection is not successful or if OpenAI's first server event is any other error
"""
import websockets
@ -822,10 +852,11 @@ async def realtime_health_check(
str | None, api_version or resolved_params.get("api_version")
)
url: str | None = None
openai_api_base: Final = resolved_api_base or litellm.api_base or "https://api.openai.com/"
health_check_api_key: Final = (
await resolve_openai_bearer_token_async(
static_api_key=openai_static_api_key(resolved_api_key),
api_base=resolved_api_base,
api_base=openai_api_base,
litellm_params=resolved_params,
)
if custom_llm_provider == LlmProviders.OPENAI.value
@ -850,10 +881,7 @@ async def realtime_health_check(
query_params=azure_query_params,
)
elif custom_llm_provider == "openai":
url = openai_realtime.construct_url(
api_base=resolved_api_base or "https://api.openai.com/",
query_params={"model": model},
)
url = openai_realtime.construct_url(api_base=openai_api_base, query_params={"model": model})
elif custom_llm_provider == "xai":
url = xai_realtime.construct_url(
api_base=resolved_api_base or "https://api.x.ai/v1", query_params={"model": model}
@ -899,8 +927,23 @@ async def realtime_health_check(
additional_headers=auth_headers,
max_size=REALTIME_WEBSOCKET_MAX_MESSAGE_SIZE_BYTES,
ssl=realtime_ssl_for_url(url),
):
return True
) as connection:
if custom_llm_provider != "openai":
return True
if not is_openai_backed_api_base(url):
return await confirm_gateway_session_started(
connection,
model,
sent_credentials=bool(auth_headers),
model_listed_outside_realtime=_cost_map_lists_outside_realtime(model),
first_event_wait_seconds=gateway_first_event_wait_seconds,
)
return await confirm_session_started(
connection,
model,
sent_credentials=bool(auth_headers),
model_listed_outside_realtime=_cost_map_lists_outside_realtime(model),
)
_realtime_health_check = realtime_health_check

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

@ -4,12 +4,15 @@ import time
from collections.abc import Awaitable, Callable, Mapping
from pathlib import Path
from types import TracebackType
from typing import Final
from typing import Final, Protocol
from unittest.mock import AsyncMock, MagicMock, patch
import httpx
import openai
import pytest
import respx
from websockets.exceptions import ConnectionClosedError
from websockets.frames import Close
import litellm
from litellm.integrations.custom_guardrail import CustomGuardrail
@ -439,18 +442,96 @@ async def test_arealtime_without_live_start_preserves_openai_realtime_dispatch(
assert captured.values["api_base"] == "https://realtime-gateway.example/v1"
class _ServerEvents(Protocol):
async def recv(self) -> str: ...
class _ScriptedConnection:
def __init__(self, *frames: str) -> None:
self._frames: Final = iter(frames)
async def recv(self) -> str:
return next(self._frames)
class _SilentConnection:
async def recv(self) -> str:
await asyncio.Event().wait()
raise AssertionError("unreachable")
class _ConnectionClosedBeforeAnyEvent:
async def recv(self) -> str:
raise ConnectionClosedError(Close(3000, "invalid_api_key"), None)
_OPENAI_INVALID_API_KEY_EVENT: Final = (
'{"type": "error", "event_id": "event_1", "error": {"type": "invalid_request_error", "code": "invalid_api_key", '
'"message": "Incorrect API key provided: sk-proj-****0000. You can find your API key at '
'https://platform.openai.com/account/api-keys.", "param": null, "event_id": null}}'
)
_OPENAI_MISSING_AUTH_EVENT: Final = (
'{"type": "error", "event_id": "event_2", "error": {"type": "invalid_request_error", "code": null, '
'"message": "Missing bearer or basic authentication in header", "param": null, "event_id": null}}'
)
_OPENAI_SERVER_ERROR_EVENT: Final = (
'{"type": "error", "event_id": "event_3", '
'"error": {"type": "server_error", "code": null, "message": "The server had an error", "param": null}}'
)
_OPENAI_RATE_LIMIT_EVENT: Final = (
'{"type": "error", "event_id": "event_5", "error": {"type": "tokens", "code": "rate_limit_exceeded", '
'"message": "Rate limit reached for gpt-realtime", "param": null, "event_id": null}}'
)
_OPENAI_INSUFFICIENT_QUOTA_EVENT: Final = (
'{"type": "error", "event_id": "event_6", "error": {"type": "insufficient_quota", "code": "insufficient_quota", '
'"message": "You exceeded your current quota", "param": null, "event_id": null}}'
)
_OPENAI_LIVE_MODEL_ON_REALTIME_EVENT: Final = (
'{"type": "error", "event_id": "event_7", "error": {"type": "invalid_request_error", "code": "invalid_model", '
'"message": "Model \\"gpt-live-1\\" is not supported in realtime mode.", "param": null, "event_id": null}}'
)
_OPENAI_CHAT_MODEL_ON_REALTIME_EVENT: Final = (
'{"type": "error", "event_id": "event_8", "error": {"type": "invalid_request_error", "code": "invalid_model", '
'"message": "Model \\"gpt-5.5\\" is not supported in realtime mode.", "param": null, "event_id": null}}'
)
_OPENAI_MODEL_NOT_FOUND_EVENT: Final = (
'{"type": "error", "event_id": "event_9", "error": {"type": "invalid_request_error", "code": "model_not_found", '
'"message": "The model gpt-realtime-2.1x does not exist or you do not have access to it.", "param": null, '
'"event_id": null}}'
)
_OPENAI_SESSION_CREATED_EVENT: Final = (
'{"type": "session.created", "event_id": "event_4", "session": {"type": "realtime", "model": "gpt-realtime"}}'
)
def _register_as_realtime_deployment(model: str, *supported_endpoints: str) -> None:
litellm.Router(
model_list=[
{
"model_name": "realtime-deployment",
"litellm_params": {"model": f"openai/{model}", "api_key": "sk-real"},
"model_info": {
"mode": "realtime",
**({"supported_endpoints": list(supported_endpoints)} if supported_endpoints else {}),
},
}
]
)
class _CapturingConnect:
def __init__(self) -> None:
def __init__(self, connection: _ServerEvents | None = None) -> None:
self.url: str | None = None
self.kwargs: dict[str, object] = {}
self._connection: Final = connection or _ScriptedConnection(_OPENAI_SESSION_CREATED_EVENT)
def __call__(self, url: str, **kwargs: object) -> "_CapturingConnect":
self.url = url
self.kwargs = kwargs
return self
async def __aenter__(self) -> MagicMock:
return MagicMock()
async def __aenter__(self) -> _ServerEvents:
return self._connection
async def __aexit__(
self,
@ -652,6 +733,173 @@ async def test_azure_health_check_honors_env_realtime_protocol(monkeypatch):
assert connect.url == _AZURE_BETA_HEALTH_URL
@pytest.mark.asyncio
@pytest.mark.parametrize(
("first_event", "expected_exception", "expected_status", "expected_message"),
[
(_OPENAI_INVALID_API_KEY_EVENT, litellm.AuthenticationError, 401, "Incorrect API key provided"),
(_OPENAI_SERVER_ERROR_EVENT, litellm.InternalServerError, 500, "The server had an error"),
(_OPENAI_RATE_LIMIT_EVENT, litellm.RateLimitError, 429, "Rate limit reached"),
(_OPENAI_INSUFFICIENT_QUOTA_EVENT, litellm.RateLimitError, 429, "You exceeded your current quota"),
],
)
async def test_openai_health_check_reports_the_first_error_event_as_unhealthy(
first_event: str, expected_exception: type[openai.APIStatusError], expected_status: int, expected_message: str
):
connect: Final = _CapturingConnect(_ScriptedConnection(first_event))
with patch("websockets.connect", connect), pytest.raises(expected_exception) as raised:
await realtime_main.realtime_health_check(
model="gpt-realtime", custom_llm_provider="openai", api_key="sk-wrong"
)
assert raised.value.status_code == expected_status
assert expected_message in str(raised.value)
assert connect.url == "wss://api.openai.com/v1/realtime?model=gpt-realtime"
@pytest.mark.asyncio
async def test_openai_health_check_without_an_api_key_sends_no_auth_header_and_fails_authentication():
connect: Final = _CapturingConnect(_ScriptedConnection(_OPENAI_MISSING_AUTH_EVENT))
with patch("websockets.connect", connect), pytest.raises(litellm.AuthenticationError) as raised:
await realtime_main.realtime_health_check(model="gpt-realtime", custom_llm_provider="openai", api_key=None)
assert connect.kwargs["additional_headers"] == {}
assert raised.value.status_code == 401
assert "Missing bearer or basic authentication" in str(raised.value)
@pytest.mark.asyncio
async def test_openai_health_check_is_healthy_once_session_created_arrives():
connect: Final = _CapturingConnect(_ScriptedConnection(_OPENAI_SESSION_CREATED_EVENT))
with patch("websockets.connect", connect):
assert await realtime_main.realtime_health_check(
model="gpt-realtime", custom_llm_provider="openai", api_key="sk-real"
)
@pytest.mark.asyncio
@pytest.mark.parametrize("model", ["gpt-live-1", "gpt-4o-transcribe", "gpt-audio", "gpt-live-unreleased"])
async def test_openai_health_check_is_healthy_for_a_model_served_outside_realtime_once_the_key_is_accepted(
local_model_cost_map, model: str
):
connect: Final = _CapturingConnect(_ScriptedConnection(_OPENAI_LIVE_MODEL_ON_REALTIME_EVENT))
with patch("websockets.connect", connect):
assert await realtime_main.realtime_health_check(model=model, custom_llm_provider="openai", api_key="sk-real")
@pytest.mark.asyncio
async def test_openai_health_check_is_healthy_for_a_live_deployment_that_lists_the_live_sessions_endpoint(
local_model_cost_map,
):
_register_as_realtime_deployment("gpt-live-1", "/v1/live/sessions")
connect: Final = _CapturingConnect(_ScriptedConnection(_OPENAI_LIVE_MODEL_ON_REALTIME_EVENT))
with patch("websockets.connect", connect):
assert await realtime_main.realtime_health_check(
model="gpt-live-1", custom_llm_provider="openai", api_key="sk-real"
)
@pytest.mark.asyncio
@pytest.mark.parametrize(
("model", "first_event", "expected_exception", "expected_message"),
[
(
"gpt-5.5",
_OPENAI_CHAT_MODEL_ON_REALTIME_EVENT,
litellm.BadRequestError,
'Model "gpt-5.5" is not supported in realtime mode',
),
(
"gpt-realtime-2.1x",
_OPENAI_MODEL_NOT_FOUND_EVENT,
litellm.NotFoundError,
"The model gpt-realtime-2.1x does not exist",
),
],
)
async def test_openai_health_check_is_unhealthy_for_a_model_name_every_session_would_fail_on(
local_model_cost_map, model: str, first_event: str, expected_exception: type[Exception], expected_message: str
):
_register_as_realtime_deployment(model)
connect: Final = _CapturingConnect(_ScriptedConnection(first_event))
with patch("websockets.connect", connect), pytest.raises(expected_exception) as raised:
await realtime_main.realtime_health_check(model=model, custom_llm_provider="openai", api_key="sk-real")
assert expected_message in str(raised.value)
@pytest.mark.asyncio
@pytest.mark.parametrize(
("deployment_api_base", "global_api_base"),
[("https://realtime-gateway.example/v1", None), (None, "https://realtime-gateway.example/v1")],
)
async def test_openai_health_check_through_a_compatible_gateway_is_healthy_when_it_sends_no_first_event(
monkeypatch: pytest.MonkeyPatch, deployment_api_base: str | None, global_api_base: str | None
):
monkeypatch.setattr(litellm, "api_base", global_api_base)
connect: Final = _CapturingConnect(_SilentConnection())
with patch("websockets.connect", connect):
assert await asyncio.wait_for(
realtime_main.realtime_health_check(
model="gpt-realtime",
custom_llm_provider="openai",
api_key="sk-real",
api_base=deployment_api_base,
gateway_first_event_wait_seconds=0.01,
),
timeout=5,
)
assert connect.url == "wss://realtime-gateway.example/v1/realtime?model=gpt-realtime"
@pytest.mark.asyncio
@pytest.mark.parametrize(
("connection", "expected_exception"),
[
(_ScriptedConnection(_OPENAI_INVALID_API_KEY_EVENT), litellm.AuthenticationError),
(_ConnectionClosedBeforeAnyEvent(), ConnectionClosedError),
],
)
async def test_openai_health_check_through_a_compatible_gateway_is_unhealthy_when_it_rejects_the_session(
connection: _ServerEvents, expected_exception: type[Exception]
):
connect: Final = _CapturingConnect(connection)
with patch("websockets.connect", connect), pytest.raises(expected_exception):
await realtime_main.realtime_health_check(
model="gpt-realtime",
custom_llm_provider="openai",
api_key="sk-wrong",
api_base="https://realtime-gateway.example/v1",
)
assert connect.url == "wss://realtime-gateway.example/v1/realtime?model=gpt-realtime"
@pytest.mark.asyncio
async def test_openai_health_check_never_reports_healthy_before_the_first_event_arrives():
connect: Final = _CapturingConnect(_SilentConnection())
with patch("websockets.connect", connect), pytest.raises(asyncio.TimeoutError):
await asyncio.wait_for(
realtime_main.realtime_health_check(model="gpt-realtime", custom_llm_provider="openai", api_key="sk-real"),
timeout=0.05,
)
@pytest.mark.asyncio
async def test_openai_health_check_is_unhealthy_when_the_socket_closes_before_any_event():
connect: Final = _CapturingConnect(_ConnectionClosedBeforeAnyEvent())
with patch("websockets.connect", connect), pytest.raises(ConnectionClosedError):
await realtime_main.realtime_health_check(
model="gpt-realtime", custom_llm_provider="openai", api_key="sk-wrong"
)
@pytest.mark.asyncio
async def test_xai_health_check_trusts_the_handshake_without_reading_a_first_event():
connect: Final = _CapturingConnect(_ScriptedConnection())
with patch("websockets.connect", connect):
assert await realtime_main.realtime_health_check(
model="grok-voice-latest", custom_llm_provider="xai", api_key="sk-wrong"
)
assert connect.url == "wss://api.x.ai/v1/realtime?model=grok-voice-latest"
class _ConnectThatStopsAfterCapturingTheUrl:
url: str | None = None

2
tests/unit/test_lazy_imports.py
View file

@ -47,7 +47,7 @@ def test_import_litellm_does_not_load_fastapi_or_bpe_table():
[
sys.executable,
"-c",
"import sys, litellm; print(','.join(m for m in ('fastapi','starlette','litellm.proxy.proxy_cli','litellm.litellm_core_utils.default_encoding') if m in sys.modules))",
"import sys, litellm; print(','.join(m for m in ('fastapi','starlette','litellm.proxy.proxy_cli','litellm.litellm_core_utils.default_encoding','openai.types.realtime') if m in sys.modules))",
],
check=True,
capture_output=True,