scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-07 08:26:10 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 36

Merge 03af235ad5 into 67c7b97fd2

Browse source
This commit is contained in:
max 2026-08-27 22:55:10 +02:00 • committed by GitHub
commit 1ebc4e0161
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 36 additions and 20 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -84,17 +84,20 @@ class OpenAIRealtime(OpenAIChatCompletion):
"""
Construct the backend websocket URL with all query parameters (including 'model').
"""
from httpx import URL
from httpx import URL, QueryParams
api_base = api_base.replace("https://", "wss://")
api_base = api_base.replace("http://", "ws://")
url = URL(api_base)
# Set the correct path
url = url.copy_with(path="/v1/realtime")
# Include all query parameters including 'model'
if query_params:
url = url.copy_with(params=query_params)
return str(url)
api_base_url: Final = URL(api_base)
websocket_scheme: Final = {"https": "wss", "http": "ws"}.get(api_base_url.scheme, api_base_url.scheme)
base_path: Final = api_base_url.path.rstrip("/")
realtime_path: Final = (
base_path
if base_path.endswith("/v1/realtime")
else f"{base_path}/realtime"
if base_path.endswith("/v1")
else f"{base_path}/v1/realtime"
)
merged_params: Final = QueryParams(query_params).merge(api_base_url.params)
return str(api_base_url.copy_with(scheme=websocket_scheme, path=realtime_path, params=merged_params))
def _make_event_normalizer(self) -> RealtimeEventNormalizer | None:
"""Return a per-session GA event normalizer, or None for passthrough.

33
tests/test_litellm/llms/openai/realtime/test_openai_realtime_handler.py
View file

@ -9,9 +9,16 @@ from litellm.llms.custom_httpx.http_handler import get_shared_realtime_ssl_conte
@pytest.mark.parametrize(
"api_base", ["https://api.openai.com/v1", "https://api.openai.com"]
("api_base", "expected_scheme", "expected_path"),
[
("https://api.openai.com", "wss", "/v1/realtime"),
("https://api.openai.com/v1", "wss", "/v1/realtime"),
("https://api.openai.com/v1/realtime", "wss", "/v1/realtime"),
("http://localhost:8000/gateway/openai", "ws", "/gateway/openai/v1/realtime"),
("https://gateway.example.com/tenant/openai/v1/", "wss", "/tenant/openai/v1/realtime"),
],
)
def test_openai_realtime_handler_url_construction(api_base):
def test_openai_realtime_handler_url_construction(api_base, expected_scheme, expected_path):
from litellm.llms.openai.realtime.handler import OpenAIRealtime
handler = OpenAIRealtime()
@ -21,9 +28,10 @@ def test_openai_realtime_handler_url_construction(api_base):
"model": "gpt-4o-realtime-preview-2024-10-01",
},
)
# Model parameter should be included in the URL
assert url.startswith("wss://api.openai.com/v1/realtime?")
assert "model=gpt-4o-realtime-preview-2024-10-01" in url
parsed_url = httpx.URL(url)
assert parsed_url.scheme == expected_scheme
assert parsed_url.path == expected_path
assert parsed_url.params["model"] == "gpt-4o-realtime-preview-2024-10-01"
def test_openai_realtime_handler_url_with_extra_params():
@ -31,16 +39,21 @@ def test_openai_realtime_handler_url_with_extra_params():
from litellm.types.realtime import RealtimeQueryParams
handler = OpenAIRealtime()
api_base = "https://api.openai.com/v1"
api_base = (
"https://gateway.example.com/tenant/openai/v1"
"?gateway_route=europe&gateway_route=backup"
"&model=gateway-model&intent=gateway-intent"
)
query_params: RealtimeQueryParams = {
"model": "gpt-4o-realtime-preview-2024-10-01",
"intent": "chat",
}
url = handler._construct_url(api_base=api_base, query_params=query_params)
# Both 'model' and other params should be included in the query string
assert url.startswith("wss://api.openai.com/v1/realtime?")
assert "model=gpt-4o-realtime-preview-2024-10-01" in url
assert "intent=chat" in url
parsed_url = httpx.URL(url)
assert parsed_url.path == "/tenant/openai/v1/realtime"
assert parsed_url.params.get_list("gateway_route") == ["europe", "backup"]
assert parsed_url.params["model"] == "gateway-model"
assert parsed_url.params["intent"] == "gateway-intent"
def test_openai_realtime_handler_model_parameter_inclusion():