From 3f01341f0a32c74115774cc732cde8b3f70b70c8 Mon Sep 17 00:00:00 2001 From: "devin-ai-integration[bot]" <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Thu, 29 Oct 2026 00:40:02 -0700 Subject: [PATCH] fix(health)!: openai realtime health check reports healthy with a missing or invalid api key (#41233) * 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> --- litellm/llms/openai/realtime/handler.py | 56 +++++- litellm/realtime_api/main.py | 65 +++++- tests/unit/realtime_api/test_main.py | 256 +++++++++++++++++++++++- tests/unit/test_lazy_imports.py | 2 +- 4 files changed, 362 insertions(+), 17 deletions(-) diff --git a/litellm/llms/openai/realtime/handler.py b/litellm/llms/openai/realtime/handler.py index b7e4cc6868f..5be488f0f6e 100644 --- a/litellm/llms/openai/realtime/handler.py +++ b/litellm/llms/openai/realtime/handler.py @@ -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): diff --git a/litellm/realtime_api/main.py b/litellm/realtime_api/main.py index 7b9068ff95e..e9b534bdd89 100644 --- a/litellm/realtime_api/main.py +++ b/litellm/realtime_api/main.py @@ -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 diff --git a/tests/unit/realtime_api/test_main.py b/tests/unit/realtime_api/test_main.py index 4d89ee497a5..5478b807691 100644 --- a/tests/unit/realtime_api/test_main.py +++ b/tests/unit/realtime_api/test_main.py @@ -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 diff --git a/tests/unit/test_lazy_imports.py b/tests/unit/test_lazy_imports.py index 0416020a9b5..43aee4616d2 100644 --- a/tests/unit/test_lazy_imports.py +++ b/tests/unit/test_lazy_imports.py @@ -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,