From b4d43c2c9b8414ff94e2ccd26d690d0427c28480 Mon Sep 17 00:00:00 2001 From: javimp2003uma Date: Sun, 16 Aug 2026 11:29:47 +0200 Subject: [PATCH 1/5] feat(elevenlabs): add WebSocket streaming-input TTS endpoint to proxy Adds support for ElevenLabs' /v1/text-to-speech/{voice_id}/stream-input WebSocket API, which enables low-latency TTS when text arrives token by token (e.g. from an LLM output stream). New proxy WebSocket route: /v1/audio/speech/stream-input - Authenticates via LiteLLM key, connects upstream with xi-api-key - Sends BOS automatically; client streams {"text":"..."} chunks and closes with {"text":""}, matching the ElevenLabs wire protocol - Forwards audio JSON (base64 + alignment fields) back to client as-is - Computes per-character cost from model_prices_and_context_window.json and fires standard aspeech logging/callback chain on session close New module: litellm/llms/elevenlabs/text_to_speech/ws_handler.py - build_elevenlabs_ws_url: constructs wss:// URL with path-encoded voice_id and model/format query params - stream_input_tts: sends BOS, runs bidirectional relay, returns total character count for cost tracking - Pydantic TypeAdapters for both client and server message shapes; no Any in public signatures 15 unit tests covering URL construction, relay character counting, EOS detection, flush/try_trigger_generation passthrough, binary frame handling, and BOS voice/generation-config injection. --- .../elevenlabs/text_to_speech/ws_handler.py | 133 ++++++++ litellm/proxy/proxy_server.py | 140 +++++++++ .../test_elevenlabs_ws_tts_handler.py | 296 ++++++++++++++++++ 3 files changed, 569 insertions(+) create mode 100644 litellm/llms/elevenlabs/text_to_speech/ws_handler.py create mode 100644 tests/test_litellm/llms/elevenlabs/test_elevenlabs_ws_tts_handler.py diff --git a/litellm/llms/elevenlabs/text_to_speech/ws_handler.py b/litellm/llms/elevenlabs/text_to_speech/ws_handler.py new file mode 100644 index 00000000000..12ad5a0c1e8 --- /dev/null +++ b/litellm/llms/elevenlabs/text_to_speech/ws_handler.py @@ -0,0 +1,133 @@ +from __future__ import annotations + +import asyncio +import json +from typing import TYPE_CHECKING, Final, Required + +from pydantic import TypeAdapter +from starlette.websockets import WebSocket +from typing_extensions import ReadOnly, TypedDict + +from litellm.litellm_core_utils.url_utils import encode_url_path_segment +from litellm.secret_managers.main import get_secret_str + +if TYPE_CHECKING: + from websockets.asyncio.client import ClientConnection + +_WS_BASE_URL: Final = "wss://api.elevenlabs.io" +_WS_PATH: Final = "/v1/text-to-speech/{voice_id}/stream-input" + + +class VoiceSettings(TypedDict, total=False): + stability: ReadOnly[float] + similarity_boost: ReadOnly[float] + style: ReadOnly[float] + use_speaker_boost: ReadOnly[bool] + speed: ReadOnly[float] + + +class GenerationConfig(TypedDict, total=False): + chunk_length_schedule: ReadOnly[list[float]] + + +class _TtsClientMessage(TypedDict, total=False): + text: Required[ReadOnly[str]] + flush: ReadOnly[bool] + try_trigger_generation: ReadOnly[bool] + voice_settings: ReadOnly[VoiceSettings] + generator_config: ReadOnly[GenerationConfig] + + +class _TtsServerMessage(TypedDict, total=False): + audio: ReadOnly[str] + isFinal: ReadOnly[bool] # camelCase matches ElevenLabs API field name + + +_CLIENT_MSG_ADAPTER: Final = TypeAdapter(_TtsClientMessage) +_SERVER_MSG_ADAPTER: Final = TypeAdapter(_TtsServerMessage) + + +def build_elevenlabs_ws_url( + model: str, + voice_id: str, + output_format: str, + api_base: str | None = None, +) -> str: + raw_base: Final = (api_base or get_secret_str("ELEVENLABS_API_BASE") or _WS_BASE_URL).rstrip("/") + ws_base: Final = raw_base.replace("https://", "wss://").replace("http://", "ws://") + encoded_voice: Final = encode_url_path_segment(voice_id, field_name="voice_id") + path: Final = _WS_PATH.format(voice_id=encoded_voice) + return f"{ws_base}{path}?model_id={model}&output_format={output_format}" + + +async def _relay_client_to_upstream( + client_ws: WebSocket, + upstream: ClientConnection, +) -> tuple[int, ...]: + chunk_lengths: list[int] = [] # mutable-ok: local accumulator, converted to immutable tuple on return + async for raw in client_ws.iter_text(): + msg = _CLIENT_MSG_ADAPTER.validate_json(raw) + await upstream.send(json.dumps(dict(msg))) + text = msg.get("text", "") + chunk_lengths.append(len(text)) + if not text: + break + return tuple(chunk_lengths) + + +async def _relay_upstream_to_client( + upstream: ClientConnection, + client_ws: WebSocket, +) -> None: + async for raw in upstream: + payload = raw if isinstance(raw, str) else raw.decode() + await client_ws.send_text(payload) + msg = _SERVER_MSG_ADAPTER.validate_json(payload) + if msg.get("isFinal"): + break + + +async def stream_input_tts( + *, + client_ws: WebSocket, + model: str, + voice_id: str, + output_format: str = "mp3_44100_128", + api_key: str | None = None, + api_base: str | None = None, + voice_settings: VoiceSettings | None = None, + generation_config: GenerationConfig | None = None, +) -> int: + """ + Relay a streaming-input TTS session between a client WebSocket and ElevenLabs. + + The proxy sends BOS automatically on connect using the provided voice/generation + settings. The client then sends text chunks as {"text": "..."} and signals end of + stream with {"text": ""}. Audio responses (JSON with base64 audio) are forwarded + back to the client as-is. + + Returns the total number of text characters sent (used for per-character cost tracking). + """ + import websockets + + key: Final = api_key or get_secret_str("ELEVENLABS_API_KEY") + if key is None: + raise ValueError("ElevenLabs API key is required. Set ELEVENLABS_API_KEY.") + + url: Final = build_elevenlabs_ws_url(model, voice_id, output_format, api_base) + + bos: dict[str, object] = {"text": " "} # mutable-ok: fields added conditionally before first send + if voice_settings is not None: + bos["voice_settings"] = voice_settings + if generation_config is not None: + bos["generation_config"] = generation_config + + async with websockets.connect(url, additional_headers={"xi-api-key": key}) as upstream: + await upstream.send(json.dumps(bos)) + results: Final = await asyncio.gather( + _relay_client_to_upstream(client_ws, upstream), + _relay_upstream_to_client(upstream, client_ws), + ) + + chunk_lengths: Final = results[0] + return sum(chunk_lengths) diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index bda6fc25499..3f9f9a572c0 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -10866,6 +10866,146 @@ async def realtime_websocket_endpoint( await websocket.close(code=1011, reason="Internal server error") +###################################################################### + +# /v1/audio/speech/stream-input Endpoint + +###################################################################### + + +@app.websocket("/v1/audio/speech/stream-input") +@app.websocket("/audio/speech/stream-input") +async def elevenlabs_tts_stream_input_endpoint( + websocket: WebSocket, + model: str = fastapi.Query( + ..., + description="Model ID, e.g. 'elevenlabs/eleven_multilingual_v2' or 'eleven_multilingual_v2'.", + ), + voice: str = fastapi.Query( + ..., + description="ElevenLabs voice ID or an OpenAI voice-name alias (alloy, coral, …).", + ), + output_format: str = fastapi.Query( + "mp3_44100_128", + description="Audio output format accepted by ElevenLabs (e.g. 'pcm_44100', 'mp3_44100_128').", + ), + stability: float | None = fastapi.Query(None, description="Voice stability (0-1)."), + similarity_boost: float | None = fastapi.Query(None, description="Voice similarity boost (0-1)."), + style: float | None = fastapi.Query(None, description="Voice style (0-1, v2+ models only)."), + speed: float | None = fastapi.Query(None, description="Speaking speed (0.7-1.2)."), + user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth_websocket), +): + """ + ElevenLabs WebSocket streaming-input TTS endpoint. + + The proxy sends BOS automatically. The client streams text chunks as + {"text": "..."} messages and signals end-of-stream with {"text": ""}. + Audio responses are forwarded back as JSON (same format as ElevenLabs). + + Query parameters map directly to ElevenLabs voice/model options; no + per-message auth is needed because the proxy injects xi-api-key upstream. + """ + from datetime import datetime + from uuid import uuid4 + + import httpx + + from litellm.llms.elevenlabs.text_to_speech.transformation import ( + ElevenLabsTextToSpeechConfig, + ) + from litellm.llms.elevenlabs.text_to_speech.ws_handler import ( + VoiceSettings, + stream_input_tts, + ) + from litellm.litellm_core_utils.litellm_logging import Logging + from litellm.types.llms.openai import HttpxBinaryResponseContent + + elevenlabs_model: Final = model.removeprefix("elevenlabs/") + litellm_model: Final = f"elevenlabs/{elevenlabs_model}" + + try: + await can_key_call_resolved_model( + model=litellm_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() + + elevenlabs_config: Final = ElevenLabsTextToSpeechConfig() + voice_id: Final = elevenlabs_config._extract_voice_id(voice) + + raw_voice_settings: Final = { + k: v + for k, v in { + "stability": stability, + "similarity_boost": similarity_boost, + "style": style, + "speed": speed, + }.items() + if v is not None + } + voice_settings: Final[VoiceSettings | None] = cast(VoiceSettings, raw_voice_settings) if raw_voice_settings else None + + start_time: Final = datetime.now() + litellm_call_id: Final = str(uuid4()) + + logging_obj: Final = Logging( + model=litellm_model, + messages=[], + stream=True, + call_type="aspeech", + start_time=start_time, + litellm_call_id=litellm_call_id, + function_id="elevenlabs_tts_stream_input", + ) + + try: + total_chars: Final = await stream_input_tts( + client_ws=websocket, + model=elevenlabs_model, + voice_id=voice_id, + output_format=output_format, + voice_settings=voice_settings, + ) + + end_time: Final = datetime.now() + + try: + model_info: Final = litellm.get_model_info( + model=litellm_model, custom_llm_provider="elevenlabs" + ) + cost_per_char: Final = model_info.get("input_cost_per_character") or 0.0 + except Exception: + cost_per_char = 0.0 + + response_cost: Final = total_chars * cost_per_char + + mock_http_response: Final = httpx.Response(200, content=b"") + result: Final = HttpxBinaryResponseContent(mock_http_response) + result._hidden_params = { + "response_cost": response_cost, + "model_id": litellm_model, + "litellm_call_id": litellm_call_id, + } + await logging_obj.async_success_handler( + result=result, + start_time=start_time, + end_time=end_time, + cache_hit=False, + ) + except Exception: + verbose_proxy_logger.exception("ElevenLabs TTS stream-input error") + try: + await websocket.close(code=1011, reason="Internal server error") + except Exception: + pass + + ###################################################################### # /v1/assistant Endpoints diff --git a/tests/test_litellm/llms/elevenlabs/test_elevenlabs_ws_tts_handler.py b/tests/test_litellm/llms/elevenlabs/test_elevenlabs_ws_tts_handler.py new file mode 100644 index 00000000000..6f7621a4c9c --- /dev/null +++ b/tests/test_litellm/llms/elevenlabs/test_elevenlabs_ws_tts_handler.py @@ -0,0 +1,296 @@ +""" +Unit tests for litellm/llms/elevenlabs/text_to_speech/ws_handler.py. + +These tests verify the WebSocket URL builder and the relay helpers without +establishing real network connections. +""" +from __future__ import annotations + +import json +from typing import Any +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from litellm.llms.elevenlabs.text_to_speech.ws_handler import ( + _relay_client_to_upstream, + _relay_upstream_to_client, + build_elevenlabs_ws_url, + stream_input_tts, +) + + +class TestBuildElevenLabsWsUrl: + def test_default_base_url(self) -> None: + url = build_elevenlabs_ws_url( + model="eleven_multilingual_v2", + voice_id="21m00Tcm4TlvDq8ikWAM", + output_format="mp3_44100_128", + ) + assert url.startswith("wss://api.elevenlabs.io/v1/text-to-speech/") + assert "?model_id=eleven_multilingual_v2&output_format=mp3_44100_128" in url + assert "21m00Tcm4TlvDq8ikWAM" in url + + def test_custom_api_base_https_converted_to_wss(self) -> None: + url = build_elevenlabs_ws_url( + model="eleven_multilingual_v2", + voice_id="voice123", + output_format="pcm_44100", + api_base="https://custom.elevenlabs.io", + ) + assert url.startswith("wss://custom.elevenlabs.io/") + assert "http" not in url.split("?")[0] + + def test_custom_api_base_http_converted_to_ws(self) -> None: + url = build_elevenlabs_ws_url( + model="eleven_multilingual_v2", + voice_id="voice123", + output_format="pcm_44100", + api_base="http://localhost:8080", + ) + assert url.startswith("ws://localhost:8080/") + + def test_voice_id_with_special_chars_is_encoded(self) -> None: + url = build_elevenlabs_ws_url( + model="eleven_multilingual_v2", + voice_id="voice with spaces", + output_format="mp3_44100_128", + ) + assert " " not in url + assert "voice%20with%20spaces" in url + + def test_query_params_include_model_and_format(self) -> None: + url = build_elevenlabs_ws_url( + model="eleven_turbo_v2", + voice_id="abc", + output_format="ulaw_8000", + ) + assert "model_id=eleven_turbo_v2" in url + assert "output_format=ulaw_8000" in url + + def test_trailing_slash_stripped_from_base(self) -> None: + url = build_elevenlabs_ws_url( + model="eleven_multilingual_v2", + voice_id="abc", + output_format="mp3_44100_128", + api_base="wss://api.elevenlabs.io/", + ) + assert "//" not in url.replace("wss://", "") + + +class TestRelayClientToUpstream: + @pytest.mark.asyncio + async def test_forwards_text_chunks_and_counts_chars(self) -> None: + chunks = [ + json.dumps({"text": "Hello, "}), + json.dumps({"text": "world! "}), + json.dumps({"text": ""}), + ] + client_ws = MagicMock() + client_ws.iter_text = self._make_iter(chunks) + + upstream = AsyncMock() + upstream.send = AsyncMock() + + result = await _relay_client_to_upstream(client_ws, upstream) + + assert result == (7, 7, 0) + assert upstream.send.call_count == 3 + + @pytest.mark.asyncio + async def test_stops_on_empty_text_eos(self) -> None: + chunks = [ + json.dumps({"text": "First "}), + json.dumps({"text": ""}), + json.dumps({"text": "Should not be sent"}), + ] + client_ws = MagicMock() + client_ws.iter_text = self._make_iter(chunks) + + upstream = AsyncMock() + upstream.send = AsyncMock() + + result = await _relay_client_to_upstream(client_ws, upstream) + + assert sum(result) == 6 + assert upstream.send.call_count == 2 + + @pytest.mark.asyncio + async def test_passes_flush_and_try_trigger_fields(self) -> None: + msg = {"text": "Go! ", "flush": True, "try_trigger_generation": True} + chunks = [json.dumps(msg), json.dumps({"text": ""})] + client_ws = MagicMock() + client_ws.iter_text = self._make_iter(chunks) + + upstream = AsyncMock() + sent_payloads: list[dict[str, Any]] = [] + + async def capture_send(payload: str) -> None: + sent_payloads.append(json.loads(payload)) + + upstream.send = capture_send + + await _relay_client_to_upstream(client_ws, upstream) + + assert sent_payloads[0]["flush"] is True + assert sent_payloads[0]["try_trigger_generation"] is True + + @staticmethod + def _make_iter(items: list[str]): + async def _gen(): + for item in items: + yield item + + return lambda: _gen() + + +class TestRelayUpstreamToClient: + @pytest.mark.asyncio + async def test_forwards_audio_messages_to_client(self) -> None: + messages = [ + json.dumps({"audio": "base64audio1"}), + json.dumps({"audio": "base64audio2"}), + json.dumps({"isFinal": True}), + ] + upstream = MagicMock() + upstream.__aiter__ = self._make_iter(messages) + + client_ws = AsyncMock() + client_ws.send_text = AsyncMock() + + await _relay_upstream_to_client(upstream, client_ws) + + assert client_ws.send_text.call_count == 3 + + @pytest.mark.asyncio + async def test_stops_after_is_final(self) -> None: + messages = [ + json.dumps({"isFinal": True}), + json.dumps({"audio": "should_not_be_forwarded"}), + ] + upstream = MagicMock() + upstream.__aiter__ = self._make_iter(messages) + + client_ws = AsyncMock() + client_ws.send_text = AsyncMock() + + await _relay_upstream_to_client(upstream, client_ws) + + assert client_ws.send_text.call_count == 1 + + @pytest.mark.asyncio + async def test_handles_binary_upstream_messages(self) -> None: + messages = [b'{"isFinal": true}'] + upstream = MagicMock() + upstream.__aiter__ = self._make_iter(messages) + + client_ws = AsyncMock() + client_ws.send_text = AsyncMock() + + await _relay_upstream_to_client(upstream, client_ws) + + client_ws.send_text.assert_called_once_with('{"isFinal": true}') + + @staticmethod + def _make_iter(items: list[str | bytes]): + async def _gen(self): + for item in items: + yield item + + return _gen + + +def _make_upstream_mock(messages: list[str], send_capture: list[dict[str, Any]] | None = None): + """Build a minimal async context manager that acts as a websockets connection.""" + + class _FakeUpstream: + def __aiter__(self): + return self._gen() + + async def _gen(self): + for msg in messages: + yield msg + + async def send(self, payload: str) -> None: + if send_capture is not None: + send_capture.append(json.loads(payload)) + + async def __aenter__(self): + return self + + async def __aexit__(self, *args: object) -> None: + pass + + return _FakeUpstream() + + +class TestStreamInputTts: + @pytest.mark.asyncio + async def test_returns_total_char_count(self) -> None: + client_ws = AsyncMock() + client_ws.iter_text = lambda: self._aiter([ + json.dumps({"text": "Hello, "}), + json.dumps({"text": "world! "}), + json.dumps({"text": ""}), + ]) + + upstream_messages = [ + json.dumps({"audio": "dGVzdA=="}), + json.dumps({"isFinal": True}), + ] + + with ( + patch("litellm.llms.elevenlabs.text_to_speech.ws_handler.get_secret_str", return_value="test-key"), + patch("websockets.connect", return_value=_make_upstream_mock(upstream_messages)), + ): + total = await stream_input_tts( + client_ws=client_ws, + model="eleven_multilingual_v2", + voice_id="21m00Tcm4TlvDq8ikWAM", + output_format="mp3_44100_128", + ) + + assert total == 14 + + @pytest.mark.asyncio + async def test_sends_bos_with_voice_settings(self) -> None: + client_ws = AsyncMock() + client_ws.iter_text = lambda: self._aiter([json.dumps({"text": ""})]) + + sent_messages: list[dict[str, Any]] = [] + upstream = _make_upstream_mock([json.dumps({"isFinal": True})], send_capture=sent_messages) + + with ( + patch("litellm.llms.elevenlabs.text_to_speech.ws_handler.get_secret_str", return_value="test-key"), + patch("websockets.connect", return_value=upstream), + ): + await stream_input_tts( + client_ws=client_ws, + model="eleven_multilingual_v2", + voice_id="abc", + output_format="mp3_44100_128", + voice_settings={"stability": 0.5, "similarity_boost": 0.75}, + generation_config={"chunk_length_schedule": [120]}, + ) + + bos = sent_messages[0] + assert bos["text"] == " " + assert bos["voice_settings"] == {"stability": 0.5, "similarity_boost": 0.75} + assert bos["generation_config"] == {"chunk_length_schedule": [120]} + + @pytest.mark.asyncio + async def test_raises_when_api_key_missing(self) -> None: + client_ws = MagicMock() + + with patch("litellm.llms.elevenlabs.text_to_speech.ws_handler.get_secret_str", return_value=None): + with pytest.raises(ValueError, match="ELEVENLABS_API_KEY"): + await stream_input_tts( + client_ws=client_ws, + model="eleven_multilingual_v2", + voice_id="abc", + ) + + @staticmethod + async def _aiter(items: list[str]): + for item in items: + yield item From ee3ddfe91eeb6bd45fc06c4aa72452db018a7fbb Mon Sep 17 00:00:00 2001 From: javimp2003uma Date: Sun, 16 Aug 2026 22:15:18 +0200 Subject: [PATCH 2/5] fix(elevenlabs): resolve CI lint failures and regenerate schema.d.ts - Add noqa suppression comments for B008 (FastAPI Depends), BLE001 (intentional broad-catch in WS session cleanup), and LIT006 (cast of dict built from typed query params) with inline reasons - Replace datetime.now() with datetime.now(tz=timezone.utc) to fix DTZ005 - Replace silent except/pass with a debug log to fix S110 - Regenerate ui/litellm-dashboard/src/lib/http/schema.d.ts to include the two new WebSocket stubs (/v1/audio/speech/stream-input and /audio/speech/stream-input) --- litellm/proxy/proxy_server.py | 18 ++--- ui/litellm-dashboard/src/lib/http/schema.d.ts | 76 +++++++++++++++++++ 2 files changed, 85 insertions(+), 9 deletions(-) diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index 3f9f9a572c0..c8e1c19061a 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -10893,7 +10893,7 @@ async def elevenlabs_tts_stream_input_endpoint( similarity_boost: float | None = fastapi.Query(None, description="Voice similarity boost (0-1)."), style: float | None = fastapi.Query(None, description="Voice style (0-1, v2+ models only)."), speed: float | None = fastapi.Query(None, description="Speaking speed (0.7-1.2)."), - user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth_websocket), + user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth_websocket), # noqa: B008 # FastAPI Depends() is required in WebSocket endpoint signatures ): """ ElevenLabs WebSocket streaming-input TTS endpoint. @@ -10905,7 +10905,7 @@ async def elevenlabs_tts_stream_input_endpoint( Query parameters map directly to ElevenLabs voice/model options; no per-message auth is needed because the proxy injects xi-api-key upstream. """ - from datetime import datetime + from datetime import datetime, timezone from uuid import uuid4 import httpx @@ -10949,9 +10949,9 @@ async def elevenlabs_tts_stream_input_endpoint( }.items() if v is not None } - voice_settings: Final[VoiceSettings | None] = cast(VoiceSettings, raw_voice_settings) if raw_voice_settings else None + voice_settings: Final[VoiceSettings | None] = cast(VoiceSettings, raw_voice_settings) if raw_voice_settings else None # cast-ok: dict built from typed float query params; structural match is guaranteed - start_time: Final = datetime.now() + start_time: Final = datetime.now(tz=timezone.utc) litellm_call_id: Final = str(uuid4()) logging_obj: Final = Logging( @@ -10973,14 +10973,14 @@ async def elevenlabs_tts_stream_input_endpoint( voice_settings=voice_settings, ) - end_time: Final = datetime.now() + end_time: Final = datetime.now(tz=timezone.utc) try: model_info: Final = litellm.get_model_info( model=litellm_model, custom_llm_provider="elevenlabs" ) cost_per_char: Final = model_info.get("input_cost_per_character") or 0.0 - except Exception: + except Exception: # noqa: BLE001 # litellm.get_model_info raises multiple undocumented error types cost_per_char = 0.0 response_cost: Final = total_chars * cost_per_char @@ -10998,12 +10998,12 @@ async def elevenlabs_tts_stream_input_endpoint( end_time=end_time, cache_hit=False, ) - except Exception: + except Exception: # noqa: BLE001 # intentional: catch all session errors to ensure WS cleanup verbose_proxy_logger.exception("ElevenLabs TTS stream-input error") try: await websocket.close(code=1011, reason="Internal server error") - except Exception: - pass + except Exception: # noqa: BLE001 # WS may already be closed; log and discard + verbose_proxy_logger.debug("WebSocket already closed during error cleanup") ###################################################################### diff --git a/ui/litellm-dashboard/src/lib/http/schema.d.ts b/ui/litellm-dashboard/src/lib/http/schema.d.ts index cf37709c377..63ddc43ba66 100644 --- a/ui/litellm-dashboard/src/lib/http/schema.d.ts +++ b/ui/litellm-dashboard/src/lib/http/schema.d.ts @@ -684,6 +684,26 @@ export interface paths { patch?: never; trace?: never; }; + "/audio/speech/stream-input": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + /** + * WebSocket: elevenlabs_tts_stream_input_endpoint + * @description WebSocket connection endpoint + */ + get: operations["websocket_elevenlabs_tts_stream_input_endpoint_get"]; + put?: never; + post?: never; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; "/audio/transcriptions": { parameters: { query?: never; @@ -16062,6 +16082,26 @@ export interface paths { patch?: never; trace?: never; }; + "/v1/audio/speech/stream-input": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + /** + * WebSocket: elevenlabs_tts_stream_input_endpoint + * @description WebSocket connection endpoint + */ + get: operations["websocket_elevenlabs_tts_stream_input_endpoint_get_2"]; + put?: never; + post?: never; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; "/v1/audio/transcriptions": { parameters: { query?: never; @@ -37168,6 +37208,24 @@ export interface operations { }; }; }; + websocket_elevenlabs_tts_stream_input_endpoint_get: { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + requestBody?: never; + responses: { + /** @description WebSocket Protocol Switched */ + 101: { + headers: { + [name: string]: unknown; + }; + content?: never; + }; + }; + }; audio_transcriptions_audio_transcriptions_post: { parameters: { query?: never; @@ -55789,6 +55847,24 @@ export interface operations { }; }; }; + websocket_elevenlabs_tts_stream_input_endpoint_get_2: { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + requestBody?: never; + responses: { + /** @description WebSocket Protocol Switched */ + 101: { + headers: { + [name: string]: unknown; + }; + content?: never; + }; + }; + }; audio_transcriptions_v1_audio_transcriptions_post: { parameters: { query?: never; From d196796c370cff3f38d836dc1cbd11c4e3b6d280 Mon Sep 17 00:00:00 2001 From: javimp2003uma Date: Sun, 16 Aug 2026 22:25:39 +0200 Subject: [PATCH 3/5] fix(elevenlabs): address relay hang, spend attribution, guardrails and URL injection Relay hang: replace asyncio.gather() with asyncio.wait(FIRST_COMPLETED) plus explicit task cancellation so the client-to-upstream relay does not block indefinitely when ElevenLabs sends isFinal before the client closes with {"text": ""}. Spend attribution: call logging_obj.update_environment_variables() with the authenticated key's api_key, key_alias, user_id and team_id so budget enforcement callbacks receive the correct key context. Guardrails / rate limits: run proxy_logging_obj.pre_call_hook() before accepting the WebSocket connection, matching what the batch TTS endpoint does. A guardrail block or rate-limit hit closes the socket with 1008 before the ElevenLabs upstream is opened. URL injection: use urllib.parse.urlencode() for model_id and output_format query parameters so crafted values cannot inject extra query string fields into the upstream URL. --- .../elevenlabs/text_to_speech/ws_handler.py | 30 +++++++++----- litellm/proxy/proxy_server.py | 39 +++++++++++++++++-- 2 files changed, 56 insertions(+), 13 deletions(-) diff --git a/litellm/llms/elevenlabs/text_to_speech/ws_handler.py b/litellm/llms/elevenlabs/text_to_speech/ws_handler.py index 12ad5a0c1e8..12d3b8c6a29 100644 --- a/litellm/llms/elevenlabs/text_to_speech/ws_handler.py +++ b/litellm/llms/elevenlabs/text_to_speech/ws_handler.py @@ -3,6 +3,7 @@ from __future__ import annotations import asyncio import json from typing import TYPE_CHECKING, Final, Required +from urllib.parse import urlencode from pydantic import TypeAdapter from starlette.websockets import WebSocket @@ -57,7 +58,8 @@ def build_elevenlabs_ws_url( ws_base: Final = raw_base.replace("https://", "wss://").replace("http://", "ws://") encoded_voice: Final = encode_url_path_segment(voice_id, field_name="voice_id") path: Final = _WS_PATH.format(voice_id=encoded_voice) - return f"{ws_base}{path}?model_id={model}&output_format={output_format}" + query: Final = urlencode({"model_id": model, "output_format": output_format}) + return f"{ws_base}{path}?{query}" async def _relay_client_to_upstream( @@ -66,9 +68,9 @@ async def _relay_client_to_upstream( ) -> tuple[int, ...]: chunk_lengths: list[int] = [] # mutable-ok: local accumulator, converted to immutable tuple on return async for raw in client_ws.iter_text(): - msg = _CLIENT_MSG_ADAPTER.validate_json(raw) + msg = _CLIENT_MSG_ADAPTER.validate_json(raw) # rebind-ok: loop-body, rebound each iteration await upstream.send(json.dumps(dict(msg))) - text = msg.get("text", "") + text = msg.get("text", "") # rebind-ok: loop-body, rebound each iteration chunk_lengths.append(len(text)) if not text: break @@ -80,9 +82,9 @@ async def _relay_upstream_to_client( client_ws: WebSocket, ) -> None: async for raw in upstream: - payload = raw if isinstance(raw, str) else raw.decode() + payload = raw if isinstance(raw, str) else raw.decode() # rebind-ok: loop-body, rebound each iteration await client_ws.send_text(payload) - msg = _SERVER_MSG_ADAPTER.validate_json(payload) + msg = _SERVER_MSG_ADAPTER.validate_json(payload) # rebind-ok: loop-body, rebound each iteration if msg.get("isFinal"): break @@ -124,10 +126,18 @@ async def stream_input_tts( async with websockets.connect(url, additional_headers={"xi-api-key": key}) as upstream: await upstream.send(json.dumps(bos)) - results: Final = await asyncio.gather( - _relay_client_to_upstream(client_ws, upstream), - _relay_upstream_to_client(upstream, client_ws), - ) - chunk_lengths: Final = results[0] + task_c2u: Final = asyncio.create_task(_relay_client_to_upstream(client_ws, upstream)) + task_u2c: Final = asyncio.create_task(_relay_upstream_to_client(upstream, client_ws)) + + # Wait for whichever relay finishes first, then cancel the other. + # This prevents the client-to-upstream relay from hanging if ElevenLabs + # sends isFinal before the client sends EOS, or if either side disconnects. + await asyncio.wait({task_c2u, task_u2c}, return_when=asyncio.FIRST_COMPLETED) + task_c2u.cancel() + task_u2c.cancel() + results: Final = await asyncio.gather(task_c2u, task_u2c, return_exceptions=True) + + c2u_result: Final = results[0] + chunk_lengths: Final = c2u_result if isinstance(c2u_result, tuple) else () return sum(chunk_lengths) diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index c8e1c19061a..79cfcd6a49e 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -10934,8 +10934,6 @@ async def elevenlabs_tts_stream_input_endpoint( await websocket.close(code=1008, reason=e.message[:120]) return - await websocket.accept() - elevenlabs_config: Final = ElevenLabsTextToSpeechConfig() voice_id: Final = elevenlabs_config._extract_voice_id(voice) @@ -10951,6 +10949,21 @@ async def elevenlabs_tts_stream_input_endpoint( } voice_settings: Final[VoiceSettings | None] = cast(VoiceSettings, raw_voice_settings) if raw_voice_settings else None # cast-ok: dict built from typed float query params; structural match is guaranteed + # Run guardrails and rate-limit checks before opening the upstream connection. + # This mirrors what the batch TTS endpoint does via proxy_logging_obj.pre_call_hook(). + _initial_hook_data: Final[dict[str, object]] = {"model": litellm_model, "user": user_api_key_dict.user_id} + try: + pre_call_data: Final = await proxy_logging_obj.pre_call_hook( + user_api_key_dict=user_api_key_dict, + data=_initial_hook_data, + call_type="aspeech", + ) + except Exception as pre_call_err: # noqa: BLE001 # guardrail block or rate-limit; close before accepting + await websocket.close(code=1008, reason=str(pre_call_err)[:120]) + return + + await websocket.accept() + start_time: Final = datetime.now(tz=timezone.utc) litellm_call_id: Final = str(uuid4()) @@ -10963,6 +10976,21 @@ async def elevenlabs_tts_stream_input_endpoint( litellm_call_id=litellm_call_id, function_id="elevenlabs_tts_stream_input", ) + # Attribute cost to the authenticated key/team so budget enforcement works. + logging_obj.update_environment_variables( + model=litellm_model, + user=user_api_key_dict.user_id, + optional_params={}, + litellm_params={ + "metadata": { + "user_api_key": user_api_key_dict.api_key, + "user_api_key_alias": user_api_key_dict.key_alias, + "user_api_key_user_id": user_api_key_dict.user_id, + "user_api_key_team_id": user_api_key_dict.team_id, + } + }, + custom_llm_provider="elevenlabs", + ) try: total_chars: Final = await stream_input_tts( @@ -10998,8 +11026,13 @@ async def elevenlabs_tts_stream_input_endpoint( end_time=end_time, cache_hit=False, ) - except Exception: # noqa: BLE001 # intentional: catch all session errors to ensure WS cleanup + except Exception as session_err: # noqa: BLE001 # intentional: catch all session errors to ensure WS cleanup verbose_proxy_logger.exception("ElevenLabs TTS stream-input error") + await proxy_logging_obj.post_call_failure_hook( + user_api_key_dict=user_api_key_dict, + original_exception=session_err, + request_data=pre_call_data, + ) try: await websocket.close(code=1011, reason="Internal server error") except Exception: # noqa: BLE001 # WS may already be closed; log and discard From 87bd639bc073372a14f57cada2f775c9d0d5a006 Mon Sep 17 00:00:00 2001 From: javimp2003uma Date: Sun, 16 Aug 2026 22:36:23 +0200 Subject: [PATCH 4/5] fix(elevenlabs): ruff format and cast-ok placement for CI lint --- litellm/proxy/proxy_server.py | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index 79cfcd6a49e..fde4d4cda3a 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -10947,7 +10947,13 @@ async def elevenlabs_tts_stream_input_endpoint( }.items() if v is not None } - voice_settings: Final[VoiceSettings | None] = cast(VoiceSettings, raw_voice_settings) if raw_voice_settings else None # cast-ok: dict built from typed float query params; structural match is guaranteed + _cast_voice_settings: Final[VoiceSettings] = ( + cast( # cast-ok: dict built from typed float query params; structural match is guaranteed + VoiceSettings, + raw_voice_settings, + ) + ) + voice_settings: Final[VoiceSettings | None] = _cast_voice_settings if raw_voice_settings else None # Run guardrails and rate-limit checks before opening the upstream connection. # This mirrors what the batch TTS endpoint does via proxy_logging_obj.pre_call_hook(). @@ -11004,9 +11010,7 @@ async def elevenlabs_tts_stream_input_endpoint( end_time: Final = datetime.now(tz=timezone.utc) try: - model_info: Final = litellm.get_model_info( - model=litellm_model, custom_llm_provider="elevenlabs" - ) + model_info: Final = litellm.get_model_info(model=litellm_model, custom_llm_provider="elevenlabs") cost_per_char: Final = model_info.get("input_cost_per_character") or 0.0 except Exception: # noqa: BLE001 # litellm.get_model_info raises multiple undocumented error types cost_per_char = 0.0 From 357cfb3a79bf37165717d062d2c2231910759055 Mon Sep 17 00:00:00 2001 From: javimp2003uma Date: Mon, 17 Aug 2026 00:43:46 +0200 Subject: [PATCH 5/5] fix(elevenlabs): fix import order, preserve char count across task cancellation Sort lazy imports inside the endpoint to satisfy ruff isort (CI used ruff 0.15.3 which enforces isort within function bodies). Refactor _relay_client_to_upstream to accept an external char_totals list that is appended to before each upstream.send(). This ensures that character counts accumulated before a task cancellation are preserved: when ElevenLabs sends isFinal and the upstream-to-client relay finishes first, the client-to-upstream relay is cancelled but char_totals already reflects every chunk forwarded up to that point, so the session is never recorded at zero cost after real usage. --- .../elevenlabs/text_to_speech/ws_handler.py | 35 ++++++++++++------- litellm/proxy/proxy_server.py | 2 +- .../test_elevenlabs_ws_tts_handler.py | 12 ++++--- 3 files changed, 31 insertions(+), 18 deletions(-) diff --git a/litellm/llms/elevenlabs/text_to_speech/ws_handler.py b/litellm/llms/elevenlabs/text_to_speech/ws_handler.py index 12d3b8c6a29..55503008f28 100644 --- a/litellm/llms/elevenlabs/text_to_speech/ws_handler.py +++ b/litellm/llms/elevenlabs/text_to_speech/ws_handler.py @@ -65,16 +65,21 @@ def build_elevenlabs_ws_url( async def _relay_client_to_upstream( client_ws: WebSocket, upstream: ClientConnection, -) -> tuple[int, ...]: - chunk_lengths: list[int] = [] # mutable-ok: local accumulator, converted to immutable tuple on return + char_totals: list[int], +) -> None: + """Forward client text chunks to the upstream ElevenLabs connection. + + Characters are appended to `char_totals` *before* the upstream send so that + partial counts survive task cancellation (e.g. when ElevenLabs sends isFinal + before the client sends EOS). + """ async for raw in client_ws.iter_text(): msg = _CLIENT_MSG_ADAPTER.validate_json(raw) # rebind-ok: loop-body, rebound each iteration - await upstream.send(json.dumps(dict(msg))) text = msg.get("text", "") # rebind-ok: loop-body, rebound each iteration - chunk_lengths.append(len(text)) + char_totals.append(len(text)) + await upstream.send(json.dumps(dict(msg))) if not text: break - return tuple(chunk_lengths) async def _relay_upstream_to_client( @@ -109,6 +114,8 @@ async def stream_input_tts( back to the client as-is. Returns the total number of text characters sent (used for per-character cost tracking). + Character counts are committed to an external accumulator before each upstream send, + so partial totals are preserved even if the relay is cancelled early. """ import websockets @@ -124,20 +131,24 @@ async def stream_input_tts( if generation_config is not None: bos["generation_config"] = generation_config + char_totals: list[int] = [] # mutable-ok: accumulator written before each send; survives task cancellation + async with websockets.connect(url, additional_headers={"xi-api-key": key}) as upstream: await upstream.send(json.dumps(bos)) - task_c2u: Final = asyncio.create_task(_relay_client_to_upstream(client_ws, upstream)) + task_c2u: Final = asyncio.create_task( + _relay_client_to_upstream(client_ws, upstream, char_totals) + ) task_u2c: Final = asyncio.create_task(_relay_upstream_to_client(upstream, client_ws)) # Wait for whichever relay finishes first, then cancel the other. - # This prevents the client-to-upstream relay from hanging if ElevenLabs - # sends isFinal before the client sends EOS, or if either side disconnects. + # Prevents the client-to-upstream relay from hanging if ElevenLabs sends + # isFinal before the client sends EOS, or if either side disconnects. + # char_totals is written before each upstream.send(), so its contents + # reflect all text actually forwarded, even if task_c2u is cancelled mid-session. await asyncio.wait({task_c2u, task_u2c}, return_when=asyncio.FIRST_COMPLETED) task_c2u.cancel() task_u2c.cancel() - results: Final = await asyncio.gather(task_c2u, task_u2c, return_exceptions=True) + await asyncio.gather(task_c2u, task_u2c, return_exceptions=True) - c2u_result: Final = results[0] - chunk_lengths: Final = c2u_result if isinstance(c2u_result, tuple) else () - return sum(chunk_lengths) + return sum(char_totals) diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index fde4d4cda3a..d09bd06e35c 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -10910,6 +10910,7 @@ async def elevenlabs_tts_stream_input_endpoint( import httpx + from litellm.litellm_core_utils.litellm_logging import Logging from litellm.llms.elevenlabs.text_to_speech.transformation import ( ElevenLabsTextToSpeechConfig, ) @@ -10917,7 +10918,6 @@ async def elevenlabs_tts_stream_input_endpoint( VoiceSettings, stream_input_tts, ) - from litellm.litellm_core_utils.litellm_logging import Logging from litellm.types.llms.openai import HttpxBinaryResponseContent elevenlabs_model: Final = model.removeprefix("elevenlabs/") diff --git a/tests/test_litellm/llms/elevenlabs/test_elevenlabs_ws_tts_handler.py b/tests/test_litellm/llms/elevenlabs/test_elevenlabs_ws_tts_handler.py index 6f7621a4c9c..d06f36ce87b 100644 --- a/tests/test_litellm/llms/elevenlabs/test_elevenlabs_ws_tts_handler.py +++ b/tests/test_litellm/llms/elevenlabs/test_elevenlabs_ws_tts_handler.py @@ -92,9 +92,10 @@ class TestRelayClientToUpstream: upstream = AsyncMock() upstream.send = AsyncMock() - result = await _relay_client_to_upstream(client_ws, upstream) + char_totals: list[int] = [] + await _relay_client_to_upstream(client_ws, upstream, char_totals) - assert result == (7, 7, 0) + assert char_totals == [7, 7, 0] assert upstream.send.call_count == 3 @pytest.mark.asyncio @@ -110,9 +111,10 @@ class TestRelayClientToUpstream: upstream = AsyncMock() upstream.send = AsyncMock() - result = await _relay_client_to_upstream(client_ws, upstream) + char_totals: list[int] = [] + await _relay_client_to_upstream(client_ws, upstream, char_totals) - assert sum(result) == 6 + assert sum(char_totals) == 6 assert upstream.send.call_count == 2 @pytest.mark.asyncio @@ -130,7 +132,7 @@ class TestRelayClientToUpstream: upstream.send = capture_send - await _relay_client_to_upstream(client_ws, upstream) + await _relay_client_to_upstream(client_ws, upstream, []) assert sent_payloads[0]["flush"] is True assert sent_payloads[0]["try_trigger_generation"] is True