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..55503008f28 --- /dev/null +++ b/litellm/llms/elevenlabs/text_to_speech/ws_handler.py @@ -0,0 +1,154 @@ +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 +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) + query: Final = urlencode({"model_id": model, "output_format": output_format}) + return f"{ws_base}{path}?{query}" + + +async def _relay_client_to_upstream( + client_ws: WebSocket, + upstream: ClientConnection, + 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 + text = msg.get("text", "") # rebind-ok: loop-body, rebound each iteration + char_totals.append(len(text)) + await upstream.send(json.dumps(dict(msg))) + if not text: + break + + +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() # rebind-ok: loop-body, rebound each iteration + await client_ws.send_text(payload) + msg = _SERVER_MSG_ADAPTER.validate_json(payload) # rebind-ok: loop-body, rebound each iteration + 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). + 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 + + 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 + + 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, 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. + # 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() + await asyncio.gather(task_c2u, task_u2c, return_exceptions=True) + + return sum(char_totals) diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index d9f8e04ebda..b5bae5bf38f 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -12052,6 +12052,183 @@ async def realtime_websocket_endpoint( await _release_realtime_budget_reservation(user_api_key_dict) +###################################################################### + +# /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), # noqa: B008 # FastAPI Depends() is required in WebSocket endpoint signatures +): + """ + 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, timezone + from uuid import uuid4 + + import httpx + + from litellm.litellm_core_utils.litellm_logging import Logging + 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.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 + + 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 + } + _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(). + _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()) + + 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", + ) + # 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( + client_ws=websocket, + model=elevenlabs_model, + voice_id=voice_id, + output_format=output_format, + voice_settings=voice_settings, + ) + + 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: # noqa: BLE001 # litellm.get_model_info raises multiple undocumented error types + 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 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 + verbose_proxy_logger.debug("WebSocket already closed during error cleanup") + + ###################################################################### # /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..d06f36ce87b --- /dev/null +++ b/tests/test_litellm/llms/elevenlabs/test_elevenlabs_ws_tts_handler.py @@ -0,0 +1,298 @@ +""" +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() + + char_totals: list[int] = [] + await _relay_client_to_upstream(client_ws, upstream, char_totals) + + assert char_totals == [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() + + char_totals: list[int] = [] + await _relay_client_to_upstream(client_ws, upstream, char_totals) + + assert sum(char_totals) == 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 diff --git a/ui/litellm-dashboard/src/lib/http/schema.d.ts b/ui/litellm-dashboard/src/lib/http/schema.d.ts index 7eadaa6c991..3e7854743af 100644 --- a/ui/litellm-dashboard/src/lib/http/schema.d.ts +++ b/ui/litellm-dashboard/src/lib/http/schema.d.ts @@ -1082,6 +1082,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; @@ -17456,6 +17476,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; @@ -41587,6 +41627,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; @@ -61758,6 +61816,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;