From 5bfbf582aab9cc1663d9636f51845e1e8e8da4ba Mon Sep 17 00:00:00 2001 From: a692570 Date: Wed, 29 Jul 2026 21:02:28 -0700 Subject: [PATCH 01/10] feat(telnyx): add Telnyx LLM plugin Telnyx Inference serves an OpenAI-compatible /v2/ai/chat/completions endpoint, so the LLM is a thin ChatCompletionsLLM subclass pointed at the Telnyx base URL with bearer auth. Follows the plugins/sarvam precedent of one vendor plugin covering LLM, STT, and TTS. --- plugins/telnyx/README.md | 19 ++++ plugins/telnyx/pyproject.toml | 2 + plugins/telnyx/tests/test_telnyx_llm.py | 95 +++++++++++++++++++ .../vision_agents/plugins/telnyx/__init__.py | 4 + .../vision_agents/plugins/telnyx/llm.py | 71 ++++++++++++++ uv.lock | 2 + 6 files changed, 193 insertions(+) create mode 100644 plugins/telnyx/tests/test_telnyx_llm.py create mode 100644 plugins/telnyx/vision_agents/plugins/telnyx/llm.py diff --git a/plugins/telnyx/README.md b/plugins/telnyx/README.md index dd4564e8b..2a8c0c19e 100644 --- a/plugins/telnyx/README.md +++ b/plugins/telnyx/README.md @@ -12,6 +12,7 @@ real-time bidirectional media streaming. - **Audio Conversion**: PCMU, PCMA, and L16 RTP payload conversion - **WebSocket Management**: Handle Telnyx WebSocket media events - **Stream Bridge**: Attach a Telnyx phone participant to a Stream call +- **LLM**: Telnyx Inference via the OpenAI-compatible Chat Completions API ## Installation @@ -46,6 +47,23 @@ call.telnyx_stream = stream await stream.run() ``` +## LLM + +Telnyx Inference is OpenAI-compatible, so the LLM is a thin wrapper over +`ChatCompletionsLLM` pointed at `https://api.telnyx.com/v2/ai`. Streaming and +tool calling work the same as any other Chat Completions provider. + +```python +from vision_agents.plugins import telnyx + +llm = telnyx.LLM(model="openai/gpt-4o") +``` + +Requires `TELNYX_API_KEY` in the environment, or an `api_key` argument. + +Model ids come from the Telnyx catalogue at `GET /v2/ai/models` and are not +validated locally. The default is `meta-llama/Llama-3.3-70B-Instruct`. + ## Examples See [examples/](examples/) for minimal inbound and outbound Telnyx phone @@ -189,5 +207,6 @@ payload = pcm_to_pcmu(pcm) ## Dependencies - vision-agents +- vision-agents-plugins-openai - numpy - fastapi diff --git a/plugins/telnyx/pyproject.toml b/plugins/telnyx/pyproject.toml index 961aedd0e..6a1957830 100644 --- a/plugins/telnyx/pyproject.toml +++ b/plugins/telnyx/pyproject.toml @@ -11,6 +11,7 @@ requires-python = ">=3.10" license = "MIT" dependencies = [ "vision-agents", + "vision-agents-plugins-openai", "numpy>=1.24.0", # capped at <2.0 via workspace override in root pyproject.toml "cryptography>=44.0.0", "fastapi>=0.135.1", @@ -33,6 +34,7 @@ include = ["/vision_agents"] [tool.uv.sources] vision-agents = { workspace = true } +vision-agents-plugins-openai = { workspace = true } [dependency-groups] dev = [ diff --git a/plugins/telnyx/tests/test_telnyx_llm.py b/plugins/telnyx/tests/test_telnyx_llm.py new file mode 100644 index 000000000..2689012a5 --- /dev/null +++ b/plugins/telnyx/tests/test_telnyx_llm.py @@ -0,0 +1,95 @@ +"""Tests for the Telnyx LLM plugin.""" + +import os + +import pytest +from dotenv import load_dotenv +from vision_agents.core.agents.conversation import InMemoryConversation +from vision_agents.plugins.telnyx import LLM +from vision_agents.testing import collect_simple_response + +load_dotenv() + + +class TestTelnyxLLM: + """Unit tests for Telnyx LLM configuration.""" + + def test_requires_api_key(self, monkeypatch): + monkeypatch.delenv("TELNYX_API_KEY", raising=False) + with pytest.raises(ValueError, match="TELNYX_API_KEY"): + LLM() + + async def test_default_model(self): + llm = LLM(api_key="KEY_test") + assert llm.model == "meta-llama/Llama-3.3-70B-Instruct" + + async def test_custom_model(self): + llm = LLM(api_key="KEY_test", model="openai/gpt-4o") + assert llm.model == "openai/gpt-4o" + + async def test_base_url_points_to_telnyx_inference(self): + llm = LLM(api_key="KEY_test") + assert str(llm._client.base_url).startswith("https://api.telnyx.com/v2/ai") + + async def test_bearer_token_used_for_auth(self): + llm = LLM(api_key="KEY_test") + assert llm._client.api_key == "KEY_test" + + async def test_provider_name(self): + llm = LLM(api_key="KEY_test") + assert llm.provider_name == "telnyx" + + async def test_explicit_client_skips_api_key_requirement(self, monkeypatch): + from openai import AsyncOpenAI + + monkeypatch.delenv("TELNYX_API_KEY", raising=False) + client = AsyncOpenAI(api_key="KEY_injected", base_url="https://example.invalid") + llm = LLM(client=client) + assert llm._client is client + + +@pytest.mark.skipif(not os.getenv("TELNYX_API_KEY"), reason="TELNYX_API_KEY not set") +@pytest.mark.integration +class TestTelnyxLLMIntegration: + """Integration tests hitting the real Telnyx Inference endpoint.""" + + @pytest.fixture + async def llm(self): + llm = LLM() + llm.set_conversation(InMemoryConversation("be friendly", [])) + return llm + + async def test_simple_response(self, llm): + deltas, final = await collect_simple_response( + llm.simple_response("Greet the user in English") + ) + assert final.text + assert deltas + + async def test_streaming_chunks(self, llm): + deltas, final = await collect_simple_response( + llm.simple_response("List the first 3 prime numbers, separated by commas.") + ) + assert final.text + assert len(deltas) > 0, f"No chunks emitted. Response text: {final.text!r}" + + async def test_function_calling(self, llm): + calls: list[str] = [] + + @llm.register_function(description="Probe tool that records invocation") + async def probe_tool(ping: str) -> str: + calls.append(ping) + return f"probe_ok:{ping}" + + prompt = ( + "Call the tool named 'probe_tool' with the parameter ping='pong' now. " + "After receiving the tool result, reply by returning ONLY the tool result string." + ) + _, final = await collect_simple_response(llm.simple_response(prompt)) + + assert "pong" in calls, ( + f"probe_tool was not invoked with ping='pong' (got calls={calls})" + ) + assert "probe_ok:pong" in final.text, ( + f"Expected 'probe_ok:pong', got: {final.text}" + ) diff --git a/plugins/telnyx/vision_agents/plugins/telnyx/__init__.py b/plugins/telnyx/vision_agents/plugins/telnyx/__init__.py index cf5b476c1..14df8c06d 100644 --- a/plugins/telnyx/vision_agents/plugins/telnyx/__init__.py +++ b/plugins/telnyx/vision_agents/plugins/telnyx/__init__.py @@ -13,18 +13,22 @@ telnyx_payload_to_pcm, ) from .call_registry import TelnyxCall, TelnyxCallRegistry +from .llm import TelnyxLLM from .media_stream import TelnyxMediaFormat, TelnyxMediaStream, attach_phone_to_call +LLM = TelnyxLLM CallRegistry = TelnyxCallRegistry MediaStream = TelnyxMediaStream __all__ = [ "CallRegistry", + "LLM", "MediaStream", "TELNYX_DEFAULT_SAMPLE_RATE", "TELNYX_L16_SAMPLE_RATE", "TelnyxCall", "TelnyxCallRegistry", + "TelnyxLLM", "TelnyxMediaFormat", "TelnyxMediaStream", "attach_phone_to_call", diff --git a/plugins/telnyx/vision_agents/plugins/telnyx/llm.py b/plugins/telnyx/vision_agents/plugins/telnyx/llm.py new file mode 100644 index 000000000..efda03958 --- /dev/null +++ b/plugins/telnyx/vision_agents/plugins/telnyx/llm.py @@ -0,0 +1,71 @@ +"""Telnyx Inference LLM using the OpenAI-compatible Chat Completions endpoint. + +Telnyx serves ``/v2/ai/chat/completions`` with the same request and response +shape as OpenAI, so we point an ``AsyncOpenAI`` client at the Telnyx base URL +and authenticate with the standard bearer token. Streaming, tool calling, and +conversation history are all inherited from :class:`ChatCompletionsLLM`. + +The catalogue of served models is fetched at runtime from ``/v2/ai/models`` +and changes over time, so model ids are not validated locally. + +Docs: https://developers.telnyx.com/api/inference/inference-embedding/post-chat-completions-public-chat-completions-post +""" + +import logging +import os +from typing import Optional + +from openai import AsyncOpenAI +from vision_agents.plugins.openai import ChatCompletionsLLM + +logger = logging.getLogger(__name__) + +TELNYX_BASE_URL = "https://api.telnyx.com/v2/ai" +DEFAULT_MODEL = "meta-llama/Llama-3.3-70B-Instruct" + + +class TelnyxLLM(ChatCompletionsLLM): + """Telnyx Inference Chat Completions LLM. + + Thin wrapper around :class:`ChatCompletionsLLM` that configures the OpenAI + client for Telnyx's OpenAI-compatible inference endpoint. + + Examples: + + from vision_agents.plugins import telnyx + llm = telnyx.LLM(model="openai/gpt-4o") + """ + + provider_name = "telnyx" + + def __init__( + self, + model: str = DEFAULT_MODEL, + api_key: Optional[str] = None, + base_url: str = TELNYX_BASE_URL, + client: Optional[AsyncOpenAI] = None, + tools_max_rounds: int = 3, + ) -> None: + """Initialize the Telnyx LLM. + + Args: + model: The model id as served by Telnyx Inference, for example + ``meta-llama/Llama-3.3-70B-Instruct`` or ``openai/gpt-4o``. + api_key: Telnyx API key. Defaults to the ``TELNYX_API_KEY`` env var. + base_url: API base URL. Defaults to ``https://api.telnyx.com/v2/ai``. + client: Optional pre-configured ``AsyncOpenAI`` client. Takes + precedence over ``api_key`` / ``base_url``. + tools_max_rounds: Max calling rounds for multi-hop tool calls. + """ + resolved_key = ( + api_key if api_key is not None else os.environ.get("TELNYX_API_KEY") + ) + if client is None and not resolved_key: + raise ValueError( + "TELNYX_API_KEY env var or api_key parameter required for Telnyx LLM" + ) + + if client is None: + client = AsyncOpenAI(api_key=resolved_key, base_url=base_url) + + super().__init__(model=model, client=client, tools_max_rounds=tools_max_rounds) diff --git a/uv.lock b/uv.lock index d0db9afe3..957b77b5a 100644 --- a/uv.lock +++ b/uv.lock @@ -7539,6 +7539,7 @@ dependencies = [ { name = "fastapi" }, { name = "numpy" }, { name = "vision-agents" }, + { name = "vision-agents-plugins-openai" }, ] [package.dev-dependencies] @@ -7553,6 +7554,7 @@ requires-dist = [ { name = "fastapi", specifier = ">=0.135.1" }, { name = "numpy", specifier = ">=1.24.0" }, { name = "vision-agents", editable = "agents-core" }, + { name = "vision-agents-plugins-openai", editable = "plugins/openai" }, ] [package.metadata.requires-dev] From 179deeab3e8e11ca6aa2583e1e8960471e35e731 Mon Sep 17 00:00:00 2001 From: a692570 Date: Wed, 29 Jul 2026 21:07:24 -0700 Subject: [PATCH 02/10] feat(telnyx): add Telnyx TTS plugin Streams text to speech over the Telnyx WebSocket endpoint and decodes the returned MP3 into PcmData as it arrives. Telnyx closes the socket after each stop frame, so the plugin connects per synthesis rather than holding one socket open. Follows the plugins/sarvam precedent of one vendor plugin covering LLM, STT, and TTS. --- plugins/telnyx/README.md | 20 ++ plugins/telnyx/pyproject.toml | 1 + plugins/telnyx/tests/test_telnyx_tts.py | 136 +++++++++ .../vision_agents/plugins/telnyx/__init__.py | 3 + .../vision_agents/plugins/telnyx/tts.py | 258 ++++++++++++++++++ uv.lock | 2 + 6 files changed, 420 insertions(+) create mode 100644 plugins/telnyx/tests/test_telnyx_tts.py create mode 100644 plugins/telnyx/vision_agents/plugins/telnyx/tts.py diff --git a/plugins/telnyx/README.md b/plugins/telnyx/README.md index 2a8c0c19e..55a52e17d 100644 --- a/plugins/telnyx/README.md +++ b/plugins/telnyx/README.md @@ -13,6 +13,7 @@ real-time bidirectional media streaming. - **WebSocket Management**: Handle Telnyx WebSocket media events - **Stream Bridge**: Attach a Telnyx phone participant to a Stream call - **LLM**: Telnyx Inference via the OpenAI-compatible Chat Completions API +- **TTS**: Streaming text to speech over WebSocket ## Installation @@ -64,6 +65,24 @@ Requires `TELNYX_API_KEY` in the environment, or an `api_key` argument. Model ids come from the Telnyx catalogue at `GET /v2/ai/models` and are not validated locally. The default is `meta-llama/Llama-3.3-70B-Instruct`. +## TTS + +```python +from vision_agents.plugins import telnyx + +tts = telnyx.TTS(voice="AWS.Polly.Danielle-Neural") +``` + +Requires `TELNYX_API_KEY` in the environment, or an `api_key` argument. + +Voice ids come from `GET /v2/text-to-speech/voices`. The default is +`Telnyx.KokoroTTS.af_heart`. + +Telnyx serves each synthesis on its own WebSocket and closes the socket after +the stop frame, so the plugin reconnects per `stream_audio` call. Audio arrives +as MP3 and is decoded to `PcmData` as it streams. The output sample rate follows +the voice, so it is taken from the decoder rather than configured. + ## Examples See [examples/](examples/) for minimal inbound and outbound Telnyx phone @@ -208,5 +227,6 @@ payload = pcm_to_pcmu(pcm) - vision-agents - vision-agents-plugins-openai +- aiohttp - numpy - fastapi diff --git a/plugins/telnyx/pyproject.toml b/plugins/telnyx/pyproject.toml index 6a1957830..0ea9a1290 100644 --- a/plugins/telnyx/pyproject.toml +++ b/plugins/telnyx/pyproject.toml @@ -15,6 +15,7 @@ dependencies = [ "numpy>=1.24.0", # capped at <2.0 via workspace override in root pyproject.toml "cryptography>=44.0.0", "fastapi>=0.135.1", + "aiohttp>=3.13.3", ] [project.urls] diff --git a/plugins/telnyx/tests/test_telnyx_tts.py b/plugins/telnyx/tests/test_telnyx_tts.py new file mode 100644 index 000000000..29cd20476 --- /dev/null +++ b/plugins/telnyx/tests/test_telnyx_tts.py @@ -0,0 +1,136 @@ +"""Tests for the Telnyx TTS plugin.""" + +import os + +import pytest +from dotenv import load_dotenv +from vision_agents.plugins.telnyx import TTS +from vision_agents.plugins.telnyx.tts import _Id3Stripper + +load_dotenv() + + +def id3_tag(payload_size: int) -> bytes: + """Build an ID3v2 tag header plus a body of ``payload_size`` bytes.""" + synchsafe = bytes( + [ + (payload_size >> 21) & 0x7F, + (payload_size >> 14) & 0x7F, + (payload_size >> 7) & 0x7F, + payload_size & 0x7F, + ] + ) + return b"ID3\x04\x00\x00" + synchsafe + b"\xaa" * payload_size + + +class TestId3Stripper: + """Unit tests for the streaming ID3v2 tag stripper.""" + + def test_untagged_data_passes_through(self): + stripper = _Id3Stripper() + assert stripper.feed(b"\xff\xf3audio") == b"\xff\xf3audio" + + def test_leading_tag_removed(self): + stripper = _Id3Stripper() + assert stripper.feed(id3_tag(34) + b"\xff\xf3audio") == b"\xff\xf3audio" + + def test_zero_length_tag_removed(self): + stripper = _Id3Stripper() + assert stripper.feed(id3_tag(0) + b"\xff\xf3") == b"\xff\xf3" + + def test_tag_body_spanning_frames(self): + stripper = _Id3Stripper() + blob = id3_tag(40) + b"\xff\xf3audio" + assert stripper.feed(blob[:20]) == b"" + assert stripper.feed(blob[20:]) == b"\xff\xf3audio" + + def test_tag_header_spanning_frames(self): + stripper = _Id3Stripper() + blob = id3_tag(12) + b"\xff\xf3audio" + assert stripper.feed(blob[:4]) == b"" + assert stripper.feed(blob[4:]) == b"\xff\xf3audio" + + def test_tag_at_head_of_later_frame(self): + stripper = _Id3Stripper() + assert stripper.feed(b"\xff\xf3first") == b"\xff\xf3first" + assert stripper.feed(id3_tag(8) + b"\xff\xf3second") == b"\xff\xf3second" + + def test_id3_bytes_inside_audio_are_kept(self): + stripper = _Id3Stripper() + audio = b"\xff\xf3 padding ID3 more audio" + assert stripper.feed(audio) == audio + + def test_tag_consuming_whole_frame(self): + stripper = _Id3Stripper() + blob = id3_tag(100) + b"\xff\xf3audio" + assert stripper.feed(blob[:50]) == b"" + assert stripper.feed(blob[50:]) == b"\xff\xf3audio" + + +class TestTelnyxTTS: + """Unit tests for Telnyx TTS configuration.""" + + async def test_requires_api_key(self, monkeypatch): + monkeypatch.delenv("TELNYX_API_KEY", raising=False) + with pytest.raises(ValueError, match="TELNYX_API_KEY"): + TTS() + + async def test_default_configuration(self): + tts = TTS(api_key="KEY_test") + assert tts.voice == "Telnyx.KokoroTTS.af_heart" + assert tts.provider_name == "telnyx" + + async def test_custom_voice(self): + tts = TTS(api_key="KEY_test", voice="AWS.Polly.Danielle-Neural") + assert tts.voice == "AWS.Polly.Danielle-Neural" + + +@pytest.mark.skipif(not os.getenv("TELNYX_API_KEY"), reason="TELNYX_API_KEY not set") +@pytest.mark.integration +class TestTelnyxTTSIntegration: + """Integration tests against the real Telnyx streaming TTS.""" + + @pytest.fixture + async def tts(self): + instance = TTS(voice="AWS.Polly.Danielle-Neural") + try: + yield instance + finally: + await instance.close() + + async def test_stream_audio_yields_chunks(self, tts): + out = [] + async for item in tts.send_iter( + "This is a test of the Telnyx text to speech API." + ): + out.append(item) + + assert len(out) > 0 + assert out[0].data + assert out[-1].final + + async def test_decoded_audio_is_audible_and_long_enough(self, tts): + chunks = [ + item.data + async for item in tts.send_iter( + "The quick brown fox jumps over the lazy dog, and then keeps " + "running for a good while longer across the field." + ) + if item.data is not None + ] + + assert chunks + rates = {chunk.sample_rate for chunk in chunks} + assert len(rates) == 1 + total_samples = sum(len(chunk.samples) for chunk in chunks) + # A sentence this long spans more than one MP3 file, so a broken ID3 + # strip truncates the audio well before this threshold. + assert total_samples / rates.pop() > 3.0 + assert max(abs(int(chunk.samples.max())) for chunk in chunks) > 0 + + async def test_second_synthesis_reconnects(self, tts): + first = [item async for item in tts.send_iter("First utterance.")] + second = [item async for item in tts.send_iter("Second utterance.")] + + assert any(item.data is not None for item in first) + assert any(item.data is not None for item in second) diff --git a/plugins/telnyx/vision_agents/plugins/telnyx/__init__.py b/plugins/telnyx/vision_agents/plugins/telnyx/__init__.py index 14df8c06d..71cecf320 100644 --- a/plugins/telnyx/vision_agents/plugins/telnyx/__init__.py +++ b/plugins/telnyx/vision_agents/plugins/telnyx/__init__.py @@ -15,6 +15,7 @@ from .call_registry import TelnyxCall, TelnyxCallRegistry from .llm import TelnyxLLM from .media_stream import TelnyxMediaFormat, TelnyxMediaStream, attach_phone_to_call +from .tts import TTS, TelnyxTTSError LLM = TelnyxLLM CallRegistry = TelnyxCallRegistry @@ -24,6 +25,7 @@ "CallRegistry", "LLM", "MediaStream", + "TTS", "TELNYX_DEFAULT_SAMPLE_RATE", "TELNYX_L16_SAMPLE_RATE", "TelnyxCall", @@ -31,6 +33,7 @@ "TelnyxLLM", "TelnyxMediaFormat", "TelnyxMediaStream", + "TelnyxTTSError", "attach_phone_to_call", "l16_to_pcm", "pcma_to_pcm", diff --git a/plugins/telnyx/vision_agents/plugins/telnyx/tts.py b/plugins/telnyx/vision_agents/plugins/telnyx/tts.py new file mode 100644 index 000000000..a83424d61 --- /dev/null +++ b/plugins/telnyx/vision_agents/plugins/telnyx/tts.py @@ -0,0 +1,258 @@ +"""Telnyx Text-to-Speech via WebSocket streaming. + +Docs: https://developers.telnyx.com/api/call-control/text-to-speech + +Two properties of the wire protocol drive this implementation: + +- A synthesis is primed with an init frame, then one or more text frames, then + an empty-text stop frame. Telnyx closes the socket once the stop frame has + been served, so a connection cannot be reused across ``stream_audio`` calls + the way a persistent-socket provider allows. +- The audio frames carry slices of MP3, and a synthesis can span several + concatenated MP3 files, each introduced by its own ID3v2 tag at the head of + a WebSocket frame. Those tags have to be dropped before the bytes reach the + decoder, otherwise decoding fails part way through the utterance. + +The decoded sample rate depends on the voice (Polly voices return 24 kHz, +Kokoro voices 22.05 kHz), so the rate reported by the decoder is used rather +than a configured one. +""" + +import asyncio +import base64 +import json +import logging +import os +from typing import Any, AsyncIterator, Optional, cast +from urllib.parse import urlencode + +import aiohttp +import av +from getstream.video.rtc.track_util import AudioFormat, PcmData +from vision_agents.core import tts + +logger = logging.getLogger(__name__) + +WS_TTS_URL = "wss://api.telnyx.com/v2/text-to-speech/speech" + +DEFAULT_VOICE = "Telnyx.KokoroTTS.af_heart" + +ID3_HEADER_SIZE = 10 + + +class TelnyxTTSError(Exception): + """Raised when Telnyx TTS returns an error frame over WebSocket.""" + + +class _Id3Stripper: + """Removes the ID3v2 tag heading each MP3 file in the audio stream. + + Telnyx concatenates one MP3 file per synthesised segment. Each file starts + with an ID3v2 tag at the head of a WebSocket frame, so only the head of a + frame is inspected. A tag whose header or body spans frames is carried + across calls. + """ + + def __init__(self) -> None: + self._skip = 0 + self._pending = b"" + + def feed(self, data: bytes) -> bytes: + """Return ``data`` with any leading ID3v2 tag removed.""" + if self._skip: + consumed = min(self._skip, len(data)) + data = data[consumed:] + self._skip -= consumed + if not data: + return b"" + + if self._pending: + data = self._pending + data + self._pending = b"" + + if data[:3] != b"ID3": + return data + + if len(data) < ID3_HEADER_SIZE: + self._pending = data + return b"" + + # ID3v2 stores the tag size as four synchsafe bytes (7 bits each). + size = data[6] << 21 | data[7] << 14 | data[8] << 7 | data[9] + body = data[ID3_HEADER_SIZE:] + consumed = min(size, len(body)) + self._skip = size - consumed + return body[consumed:] + + +class TTS(tts.TTS): + """Telnyx streaming Text-to-Speech. + + Opens one WebSocket per synthesis, streams MP3 slices back, and decodes + them into ``PcmData`` as they arrive. + + Examples: + + from vision_agents.plugins import telnyx + tts = telnyx.TTS(voice="AWS.Polly.Danielle-Neural") + """ + + def __init__( + self, + api_key: Optional[str] = None, + voice: str = DEFAULT_VOICE, + idle_timeout: float = 10.0, + ) -> None: + """Initialize Telnyx TTS. + + Args: + api_key: Telnyx API key. Falls back to the ``TELNYX_API_KEY`` env var. + voice: Voice id as listed by ``GET /v2/text-to-speech/voices``, + for example ``Telnyx.KokoroTTS.af_heart`` or + ``AWS.Polly.Danielle-Neural``. + idle_timeout: Seconds of server silence before synthesis is treated + as finished. Normally the server marks the last frame with + ``isFinal``; this is a safety net. + """ + super().__init__(provider_name="telnyx") + + self._api_key = api_key or os.environ.get("TELNYX_API_KEY") + if not self._api_key: + raise ValueError( + "TELNYX_API_KEY env var or api_key parameter required for Telnyx TTS" + ) + + self.voice = voice + self._idle_timeout = idle_timeout + + self._session: Optional[aiohttp.ClientSession] = None + self._ws: Optional[aiohttp.ClientWebSocketResponse] = None + self._lock = asyncio.Lock() + self._stop_event = asyncio.Event() + + async def close(self) -> None: + """Close the current WebSocket and release the aiohttp session.""" + await super().close() + if self._session is not None and not self._session.closed: + await self._session.close() + self._session = None + self._on_disconnected() + + async def stream_audio( + self, text: str, *_: Any, **__: Any + ) -> AsyncIterator[PcmData]: + """Stream TTS audio chunks for ``text``. + + Returns: + Async iterator yielding ``PcmData`` chunks. + """ + + async def _stream() -> AsyncIterator[PcmData]: + self._stop_event.clear() + async with self._lock: + ws = await self._connect() + try: + # Telnyx rejects a text frame that is not preceded by an + # init frame with "Invalid message". + await ws.send_str(json.dumps({"text": " "})) + await ws.send_str(json.dumps({"text": text})) + await ws.send_str(json.dumps({"text": ""})) + async for chunk in self._receive_audio(ws): + yield chunk + finally: + await self._close_ws() + + return _stream() + + async def stop_audio(self) -> None: + """Cancel any in-flight synthesis and drop the connection.""" + self._stop_event.set() + await self._close_ws() + + async def _connect(self) -> aiohttp.ClientWebSocketResponse: + if self._session is None or self._session.closed: + self._session = aiohttp.ClientSession() + + url = f"{WS_TTS_URL}?{urlencode({'voice': self.voice})}" + ws = await self._session.ws_connect( + url, headers={"Authorization": f"Bearer {self._api_key}"} + ) + self._ws = ws + self._on_connected() + logger.debug("Telnyx TTS websocket connected for voice %s", self.voice) + return ws + + async def _close_ws(self) -> None: + if self._ws is not None and not self._ws.closed: + try: + await self._ws.close() + except (aiohttp.ClientError, ConnectionError): + logger.debug("Error closing Telnyx TTS websocket") + self._ws = None + + async def _receive_audio( + self, ws: aiohttp.ClientWebSocketResponse + ) -> AsyncIterator[PcmData]: + """Yield PcmData until the final frame, a stop, an idle timeout, or a close.""" + decoder = cast(av.AudioCodecContext, av.CodecContext.create("mp3", "r")) + resampler = av.AudioResampler(format="s16", layout="mono") + stripper = _Id3Stripper() + + while True: + if self._stop_event.is_set(): + break + try: + msg = await asyncio.wait_for(ws.receive(), timeout=self._idle_timeout) + except asyncio.TimeoutError: + logger.debug("Telnyx TTS idle timeout, ending synthesis") + break + + if msg.type in ( + aiohttp.WSMsgType.CLOSED, + aiohttp.WSMsgType.CLOSING, + aiohttp.WSMsgType.ERROR, + ): + break + if msg.type != aiohttp.WSMsgType.TEXT: + continue + + try: + data = json.loads(msg.data) + except json.JSONDecodeError: + logger.warning("Telnyx TTS sent non-JSON text: %s", msg.data) + continue + + if data.get("error"): + raise TelnyxTTSError(str(data["error"])) + + encoded = data.get("audio") + if encoded: + for pcm in self._decode( + base64.b64decode(encoded), decoder, resampler, stripper + ): + yield pcm + + if data.get("isFinal"): + break + + def _decode( + self, + audio: bytes, + decoder: av.AudioCodecContext, + resampler: av.AudioResampler, + stripper: _Id3Stripper, + ) -> list[PcmData]: + """Decode one WebSocket audio payload into PcmData chunks.""" + chunks: list[PcmData] = [] + for packet in decoder.parse(stripper.feed(audio)): + for frame in decoder.decode(packet): + for resampled in resampler.resample(frame): + chunks.append( + PcmData( + samples=resampled.to_ndarray().reshape(-1), + sample_rate=resampled.sample_rate, + channels=1, + format=AudioFormat.S16, + ) + ) + return chunks diff --git a/uv.lock b/uv.lock index 957b77b5a..fee97f593 100644 --- a/uv.lock +++ b/uv.lock @@ -7535,6 +7535,7 @@ dev = [ name = "vision-agents-plugins-telnyx" source = { editable = "plugins/telnyx" } dependencies = [ + { name = "aiohttp" }, { name = "cryptography" }, { name = "fastapi" }, { name = "numpy" }, @@ -7550,6 +7551,7 @@ dev = [ [package.metadata] requires-dist = [ + { name = "aiohttp", specifier = ">=3.13.3" }, { name = "cryptography", specifier = ">=44.0.0" }, { name = "fastapi", specifier = ">=0.135.1" }, { name = "numpy", specifier = ">=1.24.0" }, From 65cef4ec43e73e4a96b0948fb6f3883cdb2b3791 Mon Sep 17 00:00:00 2001 From: a692570 Date: Wed, 29 Jul 2026 21:30:42 -0700 Subject: [PATCH 03/10] fix(telnyx): guard TTS stop race, handshake timeout, and malformed audio --- plugins/telnyx/tests/test_telnyx_tts.py | 46 +++++++++++++++ .../vision_agents/plugins/telnyx/tts.py | 59 +++++++++++++------ 2 files changed, 87 insertions(+), 18 deletions(-) diff --git a/plugins/telnyx/tests/test_telnyx_tts.py b/plugins/telnyx/tests/test_telnyx_tts.py index 29cd20476..3f215c643 100644 --- a/plugins/telnyx/tests/test_telnyx_tts.py +++ b/plugins/telnyx/tests/test_telnyx_tts.py @@ -1,7 +1,10 @@ """Tests for the Telnyx TTS plugin.""" import os +from types import SimpleNamespace +import aiohttp +import av import pytest from dotenv import load_dotenv from vision_agents.plugins.telnyx import TTS @@ -85,6 +88,49 @@ async def test_custom_voice(self): assert tts.voice == "AWS.Polly.Danielle-Neural" +class TestTelnyxTTSMalformedPayloads: + """The receive loop tolerates junk from the server without aborting.""" + + @staticmethod + def fake_ws(payloads: list[str]) -> object: + messages = [ + SimpleNamespace(type=aiohttp.WSMsgType.TEXT, data=payload) + for payload in payloads + ] + [SimpleNamespace(type=aiohttp.WSMsgType.CLOSED, data=None)] + + class FakeWS: + def __init__(self) -> None: + self._queue = list(messages) + + async def receive(self): + return self._queue.pop(0) + + return FakeWS() + + async def test_non_dict_payload_is_skipped(self): + tts = TTS(api_key="KEY_test") + ws = self.fake_ws(['["not", "a", "dict"]']) + + assert [chunk async for chunk in tts._receive_audio(ws)] == [] + + async def test_invalid_base64_audio_is_skipped(self): + tts = TTS(api_key="KEY_test") + ws = self.fake_ws(['{"audio": "!!!not base64!!!"}']) + + assert [chunk async for chunk in tts._receive_audio(ws)] == [] + + async def test_undecodable_audio_is_dropped(self): + tts = TTS(api_key="KEY_test") + + class FailingDecoder: + def parse(self, data: bytes): + raise av.InvalidDataError(1094995529, "Invalid data") + + assert ( + tts._decode(b"\xff\xf3junk", FailingDecoder(), None, _Id3Stripper()) == [] + ) + + @pytest.mark.skipif(not os.getenv("TELNYX_API_KEY"), reason="TELNYX_API_KEY not set") @pytest.mark.integration class TestTelnyxTTSIntegration: diff --git a/plugins/telnyx/vision_agents/plugins/telnyx/tts.py b/plugins/telnyx/vision_agents/plugins/telnyx/tts.py index a83424d61..1d73cf337 100644 --- a/plugins/telnyx/vision_agents/plugins/telnyx/tts.py +++ b/plugins/telnyx/vision_agents/plugins/telnyx/tts.py @@ -20,6 +20,7 @@ import asyncio import base64 +import binascii import json import logging import os @@ -148,8 +149,11 @@ async def stream_audio( """ async def _stream() -> AsyncIterator[PcmData]: - self._stop_event.clear() async with self._lock: + # Cleared under the lock so a stop_audio() aimed at an + # in-flight synthesis cannot leave the event set for a call + # queued behind it. + self._stop_event.clear() ws = await self._connect() try: # Telnyx rejects a text frame that is not preceded by an @@ -174,8 +178,13 @@ async def _connect(self) -> aiohttp.ClientWebSocketResponse: self._session = aiohttp.ClientSession() url = f"{WS_TTS_URL}?{urlencode({'voice': self.voice})}" - ws = await self._session.ws_connect( - url, headers={"Authorization": f"Bearer {self._api_key}"} + # aiohttp's default client timeout for the handshake is 300s, far past + # the point where a caller waiting on speech should give up. + ws = await asyncio.wait_for( + self._session.ws_connect( + url, headers={"Authorization": f"Bearer {self._api_key}"} + ), + timeout=self._idle_timeout, ) self._ws = ws self._on_connected() @@ -187,7 +196,7 @@ async def _close_ws(self) -> None: try: await self._ws.close() except (aiohttp.ClientError, ConnectionError): - logger.debug("Error closing Telnyx TTS websocket") + logger.debug("Error closing Telnyx TTS websocket", exc_info=True) self._ws = None async def _receive_audio( @@ -222,14 +231,21 @@ async def _receive_audio( logger.warning("Telnyx TTS sent non-JSON text: %s", msg.data) continue + if not isinstance(data, dict): + logger.warning("Telnyx TTS sent unexpected payload: %r", data) + continue + if data.get("error"): raise TelnyxTTSError(str(data["error"])) encoded = data.get("audio") if encoded: - for pcm in self._decode( - base64.b64decode(encoded), decoder, resampler, stripper - ): + try: + raw = base64.b64decode(encoded) + except (binascii.Error, ValueError): + logger.warning("Telnyx TTS sent audio that is not valid base64") + continue + for pcm in self._decode(raw, decoder, resampler, stripper): yield pcm if data.get("isFinal"): @@ -242,17 +258,24 @@ def _decode( resampler: av.AudioResampler, stripper: _Id3Stripper, ) -> list[PcmData]: - """Decode one WebSocket audio payload into PcmData chunks.""" + """Decode one WebSocket audio payload into PcmData chunks. + + A corrupt payload is dropped rather than allowed to abort the + synthesis; the decoder recovers on the next packet boundary. + """ chunks: list[PcmData] = [] - for packet in decoder.parse(stripper.feed(audio)): - for frame in decoder.decode(packet): - for resampled in resampler.resample(frame): - chunks.append( - PcmData( - samples=resampled.to_ndarray().reshape(-1), - sample_rate=resampled.sample_rate, - channels=1, - format=AudioFormat.S16, + try: + for packet in decoder.parse(stripper.feed(audio)): + for frame in decoder.decode(packet): + for resampled in resampler.resample(frame): + chunks.append( + PcmData( + samples=resampled.to_ndarray().reshape(-1), + sample_rate=resampled.sample_rate, + channels=1, + format=AudioFormat.S16, + ) ) - ) + except av.FFmpegError: + logger.warning("Telnyx TTS sent undecodable audio, dropping payload") return chunks From ef6ccb34610c8a4afc0e514c0747134c961b8b70 Mon Sep 17 00:00:00 2001 From: a692570 Date: Thu, 30 Jul 2026 00:16:16 -0700 Subject: [PATCH 04/10] fix(telnyx): treat barge-in socket close as normal TTS end and validate base64 --- plugins/telnyx/tests/test_telnyx_tts.py | 50 +++++++++++++++++++ .../vision_agents/plugins/telnyx/tts.py | 11 ++-- 2 files changed, 58 insertions(+), 3 deletions(-) diff --git a/plugins/telnyx/tests/test_telnyx_tts.py b/plugins/telnyx/tests/test_telnyx_tts.py index 3f215c643..f51278aaf 100644 --- a/plugins/telnyx/tests/test_telnyx_tts.py +++ b/plugins/telnyx/tests/test_telnyx_tts.py @@ -88,6 +88,50 @@ async def test_custom_voice(self): assert tts.voice == "AWS.Polly.Danielle-Neural" +class TestTelnyxTTSBargeIn: + """A socket dropped by stop_audio() ends synthesis instead of raising.""" + + @staticmethod + def tts_with_dropped_socket(stop_before_drop: bool) -> TTS: + instance = TTS(api_key="KEY_test") + + class DroppedWS: + closed = True + + async def send_str(self, data: str) -> None: + if stop_before_drop: + instance._stop_event.set() + raise aiohttp.ClientConnectionResetError("Cannot write to closing") + + async def close(self) -> None: + return None + + class FakeSession: + closed = False + + async def ws_connect(self, url: str, headers: dict[str, str]): + return DroppedWS() + + instance._session = FakeSession() + return instance + + async def test_stop_during_synthesis_ends_quietly(self): + """A socket closed by a concurrent stop_audio() is a barge-in.""" + tts = self.tts_with_dropped_socket(stop_before_drop=True) + + stream = await tts.stream_audio("hello") + assert [chunk async for chunk in stream] == [] + + async def test_connection_drop_without_stop_propagates(self): + """A stale stop must not silence a genuine failure in a new synthesis.""" + tts = self.tts_with_dropped_socket(stop_before_drop=False) + await tts.stop_audio() + + stream = await tts.stream_audio("hello") + with pytest.raises(aiohttp.ClientConnectionError): + [chunk async for chunk in stream] + + class TestTelnyxTTSMalformedPayloads: """The receive loop tolerates junk from the server without aborting.""" @@ -119,6 +163,12 @@ async def test_invalid_base64_audio_is_skipped(self): assert [chunk async for chunk in tts._receive_audio(ws)] == [] + async def test_non_string_audio_is_skipped(self): + tts = TTS(api_key="KEY_test") + ws = self.fake_ws(['{"audio": 12345}']) + + assert [chunk async for chunk in tts._receive_audio(ws)] == [] + async def test_undecodable_audio_is_dropped(self): tts = TTS(api_key="KEY_test") diff --git a/plugins/telnyx/vision_agents/plugins/telnyx/tts.py b/plugins/telnyx/vision_agents/plugins/telnyx/tts.py index 1d73cf337..cc28ebfdd 100644 --- a/plugins/telnyx/vision_agents/plugins/telnyx/tts.py +++ b/plugins/telnyx/vision_agents/plugins/telnyx/tts.py @@ -154,8 +154,8 @@ async def _stream() -> AsyncIterator[PcmData]: # in-flight synthesis cannot leave the event set for a call # queued behind it. self._stop_event.clear() - ws = await self._connect() try: + ws = await self._connect() # Telnyx rejects a text frame that is not preceded by an # init frame with "Invalid message". await ws.send_str(json.dumps({"text": " "})) @@ -163,6 +163,11 @@ async def _stream() -> AsyncIterator[PcmData]: await ws.send_str(json.dumps({"text": ""})) async for chunk in self._receive_audio(ws): yield chunk + except aiohttp.ClientConnectionError: + # stop_audio() closes the socket underneath us, which is a + # normal barge-in rather than a synthesis failure. + if not self._stop_event.is_set(): + raise finally: await self._close_ws() @@ -241,8 +246,8 @@ async def _receive_audio( encoded = data.get("audio") if encoded: try: - raw = base64.b64decode(encoded) - except (binascii.Error, ValueError): + raw = base64.b64decode(encoded, validate=True) + except (binascii.Error, TypeError, ValueError): logger.warning("Telnyx TTS sent audio that is not valid base64") continue for pcm in self._decode(raw, decoder, resampler, stripper): From 2556e2e652c68a559d8708d617c5e9ef79b199ea Mon Sep 17 00:00:00 2001 From: a692570 Date: Wed, 29 Jul 2026 21:11:28 -0700 Subject: [PATCH 05/10] feat(telnyx): add Telnyx STT plugin Streams audio to the Telnyx transcription WebSocket as raw linear16 frames and emits transcripts as they arrive. The configurable sample rate lets telephony audio from TelnyxMediaStream pass through at 8kHz without an upsample. Follows the plugins/sarvam precedent of one vendor plugin covering LLM, STT, and TTS. --- plugins/telnyx/README.md | 19 ++ plugins/telnyx/tests/test_telnyx_stt.py | 164 +++++++++++ .../vision_agents/plugins/telnyx/__init__.py | 2 + .../vision_agents/plugins/telnyx/stt.py | 270 ++++++++++++++++++ 4 files changed, 455 insertions(+) create mode 100644 plugins/telnyx/tests/test_telnyx_stt.py create mode 100644 plugins/telnyx/vision_agents/plugins/telnyx/stt.py diff --git a/plugins/telnyx/README.md b/plugins/telnyx/README.md index 55a52e17d..d570c4f98 100644 --- a/plugins/telnyx/README.md +++ b/plugins/telnyx/README.md @@ -13,6 +13,7 @@ real-time bidirectional media streaming. - **WebSocket Management**: Handle Telnyx WebSocket media events - **Stream Bridge**: Attach a Telnyx phone participant to a Stream call - **LLM**: Telnyx Inference via the OpenAI-compatible Chat Completions API +- **STT**: Streaming speech to text over WebSocket - **TTS**: Streaming text to speech over WebSocket ## Installation @@ -65,6 +66,24 @@ Requires `TELNYX_API_KEY` in the environment, or an `api_key` argument. Model ids come from the Telnyx catalogue at `GET /v2/ai/models` and are not validated locally. The default is `meta-llama/Llama-3.3-70B-Instruct`. +## STT + +```python +from vision_agents.plugins import telnyx + +# 8000 matches the PCMU telephony audio that TelnyxMediaStream decodes, +# so nothing is upsampled on the way to the transcriber. +stt = telnyx.STT(sample_rate=8000) +``` + +Requires `TELNYX_API_KEY` in the environment, or an `api_key` argument. + +Audio is resampled to `sample_rate` and sent as raw `linear16` frames. Pick the +engine with `transcription_engine`; the default is `Telnyx`. + +Telnyx does not send VAD signals on this endpoint, so the plugin emits +transcripts only and leaves turn detection to the agent. + ## TTS ```python diff --git a/plugins/telnyx/tests/test_telnyx_stt.py b/plugins/telnyx/tests/test_telnyx_stt.py new file mode 100644 index 000000000..d8edaf49c --- /dev/null +++ b/plugins/telnyx/tests/test_telnyx_stt.py @@ -0,0 +1,164 @@ +"""Tests for the Telnyx STT plugin.""" + +import os + +import pytest +from dotenv import load_dotenv +from vision_agents.core.edge.types import Participant +from vision_agents.core.stt import Transcript +from vision_agents.plugins.telnyx import STT + +load_dotenv() + + +class TestTelnyxSTT: + """Unit tests for Telnyx STT configuration and message handling.""" + + @pytest.fixture + def participant(self) -> Participant: + return Participant({}, user_id="test-user", id="test-user") + + async def test_requires_api_key(self, monkeypatch): + monkeypatch.delenv("TELNYX_API_KEY", raising=False) + with pytest.raises(ValueError, match="TELNYX_API_KEY"): + STT() + + async def test_default_configuration(self): + stt = STT(api_key="KEY_test") + assert stt.transcription_engine == "Telnyx" + assert stt.language == "en" + assert stt.sample_rate == 16000 + assert stt.interim_results is True + assert stt.provider_name == "telnyx" + + async def test_invalid_engine_rejected(self): + with pytest.raises(ValueError, match="Unsupported Telnyx transcription_engine"): + STT(api_key="KEY_test", transcription_engine="NotAnEngine") + + async def test_url_carries_stream_parameters(self): + stt = STT(api_key="KEY_test", language="es", sample_rate=8000) + url = stt._build_ws_url() + assert url.startswith("wss://api.telnyx.com/v2/speech-to-text/transcription?") + assert "input_format=linear16" in url + assert "sample_rate=8000" in url + assert "language=es" in url + assert "transcription_engine=Telnyx" in url + + async def test_url_uses_interim_results_not_partial_results(self): + on = STT(api_key="KEY_test")._build_ws_url() + off = STT(api_key="KEY_test", interim_results=False)._build_ws_url() + assert "interim_results=true" in on + assert "interim_results=false" in off + assert "partial_results" not in on + + async def test_url_omits_model_when_unset(self): + assert "model=" not in STT(api_key="KEY_test")._build_ws_url() + assert ( + "model=whisper" in STT(api_key="KEY_test", model="whisper")._build_ws_url() + ) + + async def test_final_transcript_emitted(self, participant): + stt = STT(api_key="KEY_test") + stt._current_participant = participant + + stt._handle_message( + {"transcript": "hello world", "confidence": 0.9, "is_final": True} + ) + items = await stt.output.collect(timeout=0) + + transcripts = [i for i in items if isinstance(i, Transcript)] + assert [t.text for t in transcripts] == ["hello world"] + assert transcripts[0].final + assert transcripts[0].confidence == 0.9 + assert transcripts[0].participant == participant + + async def test_interim_transcript_is_not_final(self, participant): + stt = STT(api_key="KEY_test") + stt._current_participant = participant + + stt._handle_message( + {"transcript": "hello", "confidence": None, "is_final": False} + ) + items = await stt.output.collect(timeout=0) + + transcripts = [i for i in items if isinstance(i, Transcript)] + assert [t.text for t in transcripts] == ["hello"] + assert not transcripts[0].final + assert transcripts[0].confidence is None + + async def test_empty_transcript_emits_nothing(self, participant): + stt = STT(api_key="KEY_test") + stt._current_participant = participant + + stt._handle_message({"transcript": "", "confidence": None, "is_final": True}) + stt._handle_message({"transcript": " ", "confidence": None, "is_final": True}) + + assert await stt.output.collect(timeout=0) == [] + + async def test_error_payload_emits_no_transcript(self, participant): + stt = STT(api_key="KEY_test") + stt._current_participant = participant + + stt._handle_message( + { + "errors": [ + { + "code": "40001", + "title": "Invalid Parameter", + "detail": "Unsupported input_format 'zzz'.", + } + ] + } + ) + + assert await stt.output.collect(timeout=0) == [] + + +@pytest.mark.skipif(not os.getenv("TELNYX_API_KEY"), reason="TELNYX_API_KEY not set") +@pytest.mark.integration +class TestTelnyxSTTIntegration: + """Integration tests against the real Telnyx streaming STT.""" + + @pytest.fixture + def participant(self) -> Participant: + return Participant({}, user_id="test-user", id="test-user") + + @pytest.fixture + async def telnyx_stt(self): + stt = STT() + await stt.start() + yield stt + await stt.close() + + @pytest.fixture + async def telnyx_stt_8khz(self): + stt = STT(sample_rate=8000) + await stt.start() + yield stt + await stt.close() + + async def test_transcribe_mia_audio_16khz( + self, telnyx_stt, mia_audio_16khz, participant + ): + await telnyx_stt.process_audio(mia_audio_16khz, participant=participant) + + items = await telnyx_stt.output.collect(timeout=15.0) + + transcripts = [i for i in items if isinstance(i, Transcript)] + assert transcripts, "No Transcript emitted by Telnyx STT" + finals = [t for t in transcripts if t.final] + assert finals, "No final Transcript emitted by Telnyx STT" + assert "forgotten treasures" in " ".join(t.text for t in finals).lower() + assert transcripts[0].participant == participant + + async def test_transcribe_at_telephony_sample_rate( + self, telnyx_stt_8khz, mia_audio_16khz, participant + ): + """8 kHz is what TelnyxMediaStream produces from PCMU telephony audio.""" + await telnyx_stt_8khz.process_audio(mia_audio_16khz, participant=participant) + + items = await telnyx_stt_8khz.output.collect(timeout=15.0) + + finals = [i for i in items if isinstance(i, Transcript) and i.final] + assert finals, "No final Transcript emitted at 8kHz" + assert "forgotten treasures" in " ".join(t.text for t in finals).lower() diff --git a/plugins/telnyx/vision_agents/plugins/telnyx/__init__.py b/plugins/telnyx/vision_agents/plugins/telnyx/__init__.py index 71cecf320..cb17085c3 100644 --- a/plugins/telnyx/vision_agents/plugins/telnyx/__init__.py +++ b/plugins/telnyx/vision_agents/plugins/telnyx/__init__.py @@ -15,6 +15,7 @@ from .call_registry import TelnyxCall, TelnyxCallRegistry from .llm import TelnyxLLM from .media_stream import TelnyxMediaFormat, TelnyxMediaStream, attach_phone_to_call +from .stt import STT from .tts import TTS, TelnyxTTSError LLM = TelnyxLLM @@ -25,6 +26,7 @@ "CallRegistry", "LLM", "MediaStream", + "STT", "TTS", "TELNYX_DEFAULT_SAMPLE_RATE", "TELNYX_L16_SAMPLE_RATE", diff --git a/plugins/telnyx/vision_agents/plugins/telnyx/stt.py b/plugins/telnyx/vision_agents/plugins/telnyx/stt.py new file mode 100644 index 000000000..21dc2be6e --- /dev/null +++ b/plugins/telnyx/vision_agents/plugins/telnyx/stt.py @@ -0,0 +1,270 @@ +"""Telnyx Speech-to-Text via WebSocket streaming. + +Docs: https://developers.telnyx.com/api/call-control/speech-to-text + +Audio is sent as raw binary frames in the encoding named by ``input_format``. +The default is ``linear16``, for which Telnyx requires an explicit +``sample_rate``. Transcripts come back as +``{"transcript": str, "confidence": float | None, "is_final": bool}``. + +The query parameter for partial transcripts is ``interim_results``. +``partial_results`` is accepted by the endpoint but ignored, which silently +yields finals only. +""" + +import asyncio +import json +import logging +import os +import time +from typing import Optional +from urllib.parse import urlencode + +import aiohttp +from getstream.video.rtc.track_util import PcmData +from vision_agents.core import stt +from vision_agents.core.edge.types import Participant +from vision_agents.core.stt import TranscriptResponse + +logger = logging.getLogger(__name__) + +WS_STT_URL = "wss://api.telnyx.com/v2/speech-to-text/transcription" + +SUPPORTED_ENGINES = { + "AssemblyAI", + "Azure", + "Deepgram", + "Google", + "Humain", + "Parakeet", + "Reson8", + "Soniox", + "Speechmatics", + "Telnyx", + "xAI", +} + + +class STT(stt.STT): + """Telnyx streaming Speech-to-Text. + + Uses aiohttp for a fully async WebSocket connection to the Telnyx streaming + endpoint. Audio is resampled to the configured rate and sent as raw + ``linear16`` binary frames. + + Telnyx does not send VAD signals on this endpoint, so turn detection is + left to the agent's own turn detection. + + Examples: + + from vision_agents.plugins import telnyx + stt = telnyx.STT(sample_rate=8000) + """ + + def __init__( + self, + api_key: Optional[str] = None, + transcription_engine: str = "Telnyx", + language: str = "en", + sample_rate: int = 16000, + interim_results: bool = True, + model: Optional[str] = None, + ) -> None: + """Initialize Telnyx STT. + + Args: + api_key: Telnyx API key. Falls back to the ``TELNYX_API_KEY`` env var. + transcription_engine: Engine to transcribe with. Defaults to + ``Telnyx``. + language: Language code, for example ``en``. + sample_rate: Rate in Hz that audio is resampled to before being + sent. Use 8000 to pass telephony audio from + :class:`TelnyxMediaStream` through without upsampling. + interim_results: Emit partial transcripts as they are refined. + model: Optional engine-specific model id. + """ + super().__init__(provider_name="telnyx") + + if transcription_engine not in SUPPORTED_ENGINES: + raise ValueError( + f"Unsupported Telnyx transcription_engine '{transcription_engine}'. " + f"Expected one of: {sorted(SUPPORTED_ENGINES)}" + ) + + self._api_key = api_key or os.environ.get("TELNYX_API_KEY") + if not self._api_key: + raise ValueError( + "TELNYX_API_KEY env var or api_key parameter required for Telnyx STT" + ) + + self.transcription_engine = transcription_engine + self.language = language + self.sample_rate = sample_rate + self.interim_results = interim_results + self.model = model + + self._session: Optional[aiohttp.ClientSession] = None + self._ws: Optional[aiohttp.ClientWebSocketResponse] = None + self._receive_task: Optional[asyncio.Task[None]] = None + self._connection_ready = asyncio.Event() + self._current_participant: Optional[Participant] = None + self._audio_start_time: Optional[float] = None + + async def start(self) -> None: + """Open the Telnyx WebSocket and start the receive loop.""" + await super().start() + + # aiohttp does not attach an Origin header to the handshake. Clients + # that do have to suppress it, because the Telnyx edge rejects a + # WebSocket handshake that carries one. + self._session = aiohttp.ClientSession() + self._ws = await self._session.ws_connect( + self._build_ws_url(), + headers={"Authorization": f"Bearer {self._api_key}"}, + ) + + self._receive_task = asyncio.create_task(self._receive_loop()) + self._connection_ready.set() + self._on_connected() + + async def process_audio( + self, + pcm_data: PcmData, + participant: Participant, + ) -> None: + """Resample a PCM chunk to the configured rate and send it to Telnyx.""" + if self.closed: + logger.warning("Telnyx STT is closed, ignoring audio") + return + + await self._connection_ready.wait() + + if self._ws is None or self._ws.closed: + logger.warning("Telnyx STT WebSocket not open, dropping audio") + return + + resampled = pcm_data.resample(self.sample_rate, 1) + + self._current_participant = participant + if self._audio_start_time is None: + self._audio_start_time = time.perf_counter() + + await self._ws.send_bytes(resampled.samples.tobytes()) + + async def close(self) -> None: + """Close the WebSocket and clean up.""" + await super().close() + + if self._ws is not None and not self._ws.closed: + await self._ws.close() + self._ws = None + + if self._receive_task is not None: + self._receive_task.cancel() + try: + await self._receive_task + except asyncio.CancelledError: + pass + self._receive_task = None + + if self._session is not None and not self._session.closed: + await self._session.close() + self._session = None + + self._connection_ready.clear() + self._on_disconnected() + self._audio_start_time = None + + def _build_ws_url(self) -> str: + params: dict[str, str] = { + "transcription_engine": self.transcription_engine, + "input_format": "linear16", + "sample_rate": str(self.sample_rate), + "language": self.language, + "interim_results": "true" if self.interim_results else "false", + } + if self.model is not None: + params["model"] = self.model + return f"{WS_STT_URL}?{urlencode(params)}" + + async def _receive_loop(self) -> None: + ws = self._ws + if ws is None: + return + try: + async for msg in ws: + if msg.type == aiohttp.WSMsgType.TEXT: + try: + parsed = json.loads(msg.data) + except json.JSONDecodeError: + logger.warning("Telnyx STT sent non-JSON text: %s", msg.data) + continue + if logger.isEnabledFor(logging.DEBUG): + logger.debug("Telnyx STT message: %s", parsed) + self._handle_message(parsed) + elif msg.type in ( + aiohttp.WSMsgType.CLOSED, + aiohttp.WSMsgType.CLOSING, + aiohttp.WSMsgType.ERROR, + ): + break + except asyncio.CancelledError: + raise + except aiohttp.ClientError: + logger.exception("Telnyx STT receive loop error") + + if not self.closed: + self._emit_error_event( + ConnectionError("Telnyx STT WebSocket closed unexpectedly"), + context="telnyx_ws_closed", + ) + + def _handle_message(self, data: dict[str, object]) -> None: + """Dispatch a parsed Telnyx WebSocket message. + + Transcripts arrive as + ``{"transcript": str, "confidence": float | None, "is_final": bool}``. + Parameter rejections arrive as ``{"errors": [{"detail": ...}, ...]}`` + and are followed by the server closing the connection. + """ + errors = data.get("errors") + if isinstance(errors, list) and errors: + details = "; ".join( + str(err.get("detail") or err.get("title")) + for err in errors + if isinstance(err, dict) + ) + self._emit_error_event( + RuntimeError(details or "Telnyx STT error"), + context="telnyx_streaming", + ) + return + + text = data.get("transcript") + if not isinstance(text, str) or not text.strip(): + return + + participant = self._current_participant + if participant is None: + logger.warning("Telnyx transcript received but no participant set") + return + + processing_time_ms: Optional[float] = None + if self._audio_start_time is not None: + processing_time_ms = (time.perf_counter() - self._audio_start_time) * 1000 + + confidence = data.get("confidence") + is_final = bool(data.get("is_final", True)) + + response = TranscriptResponse( + confidence=confidence if isinstance(confidence, float) else None, + language=self.language, + model_name=self.model or self.transcription_engine, + processing_time_ms=processing_time_ms, + ) + + if is_final: + self._audio_start_time = None + self._emit_transcript_event(text, participant, response) + else: + self._emit_transcript_event(text, participant, response, mode="replacement") From c26bd8b0316bd07660dac36cfc3433cae0546e40 Mon Sep 17 00:00:00 2001 From: a692570 Date: Wed, 29 Jul 2026 21:29:02 -0700 Subject: [PATCH 06/10] fix(telnyx): harden STT websocket lifecycle and payload handling --- plugins/telnyx/tests/test_telnyx_stt.py | 30 +++++++++ .../vision_agents/plugins/telnyx/stt.py | 66 ++++++++++++------- 2 files changed, 72 insertions(+), 24 deletions(-) diff --git a/plugins/telnyx/tests/test_telnyx_stt.py b/plugins/telnyx/tests/test_telnyx_stt.py index d8edaf49c..c36c2dded 100644 --- a/plugins/telnyx/tests/test_telnyx_stt.py +++ b/plugins/telnyx/tests/test_telnyx_stt.py @@ -2,8 +2,11 @@ import os +import aiohttp +import numpy as np import pytest from dotenv import load_dotenv +from getstream.video.rtc.track_util import PcmData from vision_agents.core.edge.types import Participant from vision_agents.core.stt import Transcript from vision_agents.plugins.telnyx import STT @@ -95,6 +98,33 @@ async def test_empty_transcript_emits_nothing(self, participant): assert await stt.output.collect(timeout=0) == [] + async def test_send_failure_does_not_propagate(self, participant): + stt = STT(api_key="KEY_test") + + class FailingWS: + closed = False + + async def send_bytes(self, data: bytes) -> None: + raise aiohttp.ClientError("connection reset") + + stt._ws = FailingWS() + stt._connection_ready.set() + + pcm = PcmData( + samples=np.zeros(160, dtype=np.int16), sample_rate=16000, format="s16" + ) + await stt.process_audio(pcm, participant=participant) + + async def test_integer_confidence_is_kept(self, participant): + stt = STT(api_key="KEY_test") + stt._current_participant = participant + + stt._handle_message({"transcript": "hello", "confidence": 1, "is_final": True}) + items = await stt.output.collect(timeout=0) + + transcripts = [i for i in items if isinstance(i, Transcript)] + assert transcripts[0].confidence == 1.0 + async def test_error_payload_emits_no_transcript(self, participant): stt = STT(api_key="KEY_test") stt._current_participant = participant diff --git a/plugins/telnyx/vision_agents/plugins/telnyx/stt.py b/plugins/telnyx/vision_agents/plugins/telnyx/stt.py index 21dc2be6e..6e17dbe95 100644 --- a/plugins/telnyx/vision_agents/plugins/telnyx/stt.py +++ b/plugins/telnyx/vision_agents/plugins/telnyx/stt.py @@ -118,10 +118,15 @@ async def start(self) -> None: # that do have to suppress it, because the Telnyx edge rejects a # WebSocket handshake that carries one. self._session = aiohttp.ClientSession() - self._ws = await self._session.ws_connect( - self._build_ws_url(), - headers={"Authorization": f"Bearer {self._api_key}"}, - ) + try: + self._ws = await self._session.ws_connect( + self._build_ws_url(), + headers={"Authorization": f"Bearer {self._api_key}"}, + ) + except BaseException: + await self._session.close() + self._session = None + raise self._receive_task = asyncio.create_task(self._receive_loop()) self._connection_ready.set() @@ -149,31 +154,39 @@ async def process_audio( if self._audio_start_time is None: self._audio_start_time = time.perf_counter() - await self._ws.send_bytes(resampled.samples.tobytes()) + try: + await self._ws.send_bytes(resampled.samples.tobytes()) + except (aiohttp.ClientError, ConnectionError) as exc: + self._emit_error_event(exc, context="telnyx_send_audio") async def close(self) -> None: """Close the WebSocket and clean up.""" await super().close() - if self._ws is not None and not self._ws.closed: - await self._ws.close() - self._ws = None - - if self._receive_task is not None: - self._receive_task.cancel() + try: + if self._ws is not None and not self._ws.closed: + await self._ws.close() + finally: + self._ws = None try: - await self._receive_task - except asyncio.CancelledError: - pass - self._receive_task = None - - if self._session is not None and not self._session.closed: - await self._session.close() - self._session = None - - self._connection_ready.clear() - self._on_disconnected() - self._audio_start_time = None + if self._receive_task is not None: + self._receive_task.cancel() + try: + await self._receive_task + except asyncio.CancelledError: + pass + except Exception: + logger.exception("Telnyx STT receive task failed on close") + finally: + self._receive_task = None + try: + if self._session is not None and not self._session.closed: + await self._session.close() + finally: + self._session = None + self._connection_ready.clear() + self._on_disconnected() + self._audio_start_time = None def _build_ws_url(self) -> str: params: dict[str, str] = { @@ -199,6 +212,9 @@ async def _receive_loop(self) -> None: except json.JSONDecodeError: logger.warning("Telnyx STT sent non-JSON text: %s", msg.data) continue + if not isinstance(parsed, dict): + logger.warning("Telnyx STT sent unexpected payload: %r", parsed) + continue if logger.isEnabledFor(logging.DEBUG): logger.debug("Telnyx STT message: %s", parsed) self._handle_message(parsed) @@ -254,10 +270,12 @@ def _handle_message(self, data: dict[str, object]) -> None: processing_time_ms = (time.perf_counter() - self._audio_start_time) * 1000 confidence = data.get("confidence") + if isinstance(confidence, bool) or not isinstance(confidence, (int, float)): + confidence = None is_final = bool(data.get("is_final", True)) response = TranscriptResponse( - confidence=confidence if isinstance(confidence, float) else None, + confidence=float(confidence) if confidence is not None else None, language=self.language, model_name=self.model or self.transcription_engine, processing_time_ms=processing_time_ms, From 3c24a2eb7694dfb7ab00a891b91b5ad3100180bd Mon Sep 17 00:00:00 2001 From: "Neevash Ramdial (Nash)" Date: Mon, 10 Aug 2026 17:06:35 -0600 Subject: [PATCH 07/10] fix(telnyx): address review of the LLM, STT and TTS plugins LLM - Drop the hand-built AsyncOpenAI client; ChatCompletionsLLM already builds one from api_key/base_url. Follows the minimax plugin. - Export a single LLM name, matching the sarvam plugin. STT - self.model is str, not Optional[str]. Since #618 the STT base class declares `model: str`, so the Optional assignment failed mypy. - Default interim_results to False. Measured against the live API with the same audio: Speechmatics and Soniox stream partials, while the default Telnyx engine and Deepgram accept the parameter and return finals only. The old default advertised partials that never arrived. - Drop the SUPPORTED_ENGINES allowlist. The catalogue is served by Telnyx and grows over time, and the server already rejects unknown engines with an explicit error. - Flatten close() to the same shape as the sarvam plugin. - Do not emit a second error event for the socket close that an error payload causes. - Drop audio instead of blocking forever when start() has not run. - Declare turn_detection explicitly. TTS - Report the voice as `model` so agent metadata is populated. - Give the handshake its own timeout instead of reusing idle_timeout. - Emit connected/disconnected once per plugin rather than once per utterance; this provider opens a socket per synthesis. - Build PcmData with PcmData.from_av_frame. Tests - Fold the TTS test classes into one and replace static helpers with fixtures. - Cover the interim-transcript path end to end against an engine that emits partials; it had no live coverage. - Assert behaviour rather than private client attributes. --- plugins/telnyx/tests/test_telnyx_llm.py | 15 +- plugins/telnyx/tests/test_telnyx_stt.py | 77 +++++++--- plugins/telnyx/tests/test_telnyx_tts.py | 142 +++++++++--------- .../vision_agents/plugins/telnyx/__init__.py | 4 +- .../vision_agents/plugins/telnyx/llm.py | 11 +- .../vision_agents/plugins/telnyx/stt.py | 96 ++++++------ .../vision_agents/plugins/telnyx/tts.py | 35 +++-- 7 files changed, 213 insertions(+), 167 deletions(-) diff --git a/plugins/telnyx/tests/test_telnyx_llm.py b/plugins/telnyx/tests/test_telnyx_llm.py index 2689012a5..897e595ae 100644 --- a/plugins/telnyx/tests/test_telnyx_llm.py +++ b/plugins/telnyx/tests/test_telnyx_llm.py @@ -4,6 +4,7 @@ import pytest from dotenv import load_dotenv +from openai import AsyncOpenAI from vision_agents.core.agents.conversation import InMemoryConversation from vision_agents.plugins.telnyx import LLM from vision_agents.testing import collect_simple_response @@ -27,25 +28,15 @@ async def test_custom_model(self): llm = LLM(api_key="KEY_test", model="openai/gpt-4o") assert llm.model == "openai/gpt-4o" - async def test_base_url_points_to_telnyx_inference(self): - llm = LLM(api_key="KEY_test") - assert str(llm._client.base_url).startswith("https://api.telnyx.com/v2/ai") - - async def test_bearer_token_used_for_auth(self): - llm = LLM(api_key="KEY_test") - assert llm._client.api_key == "KEY_test" - async def test_provider_name(self): llm = LLM(api_key="KEY_test") assert llm.provider_name == "telnyx" async def test_explicit_client_skips_api_key_requirement(self, monkeypatch): - from openai import AsyncOpenAI - monkeypatch.delenv("TELNYX_API_KEY", raising=False) client = AsyncOpenAI(api_key="KEY_injected", base_url="https://example.invalid") - llm = LLM(client=client) - assert llm._client is client + + assert LLM(client=client).model == "meta-llama/Llama-3.3-70B-Instruct" @pytest.mark.skipif(not os.getenv("TELNYX_API_KEY"), reason="TELNYX_API_KEY not set") diff --git a/plugins/telnyx/tests/test_telnyx_stt.py b/plugins/telnyx/tests/test_telnyx_stt.py index c36c2dded..ef82d5c67 100644 --- a/plugins/telnyx/tests/test_telnyx_stt.py +++ b/plugins/telnyx/tests/test_telnyx_stt.py @@ -1,5 +1,6 @@ """Tests for the Telnyx STT plugin.""" +import asyncio import os import aiohttp @@ -21,6 +22,22 @@ class TestTelnyxSTT: def participant(self) -> Participant: return Participant({}, user_id="test-user", id="test-user") + @pytest.fixture + def stt_with_failing_socket(self) -> STT: + """An started STT whose socket rejects every send.""" + instance = STT(api_key="KEY_test") + + class FailingWS: + closed = False + + async def send_bytes(self, data: bytes) -> None: + raise aiohttp.ClientError("connection reset") + + instance.started = True + instance._ws = FailingWS() + instance._connection_ready.set() + return instance + async def test_requires_api_key(self, monkeypatch): monkeypatch.delenv("TELNYX_API_KEY", raising=False) with pytest.raises(ValueError, match="TELNYX_API_KEY"): @@ -31,12 +48,14 @@ async def test_default_configuration(self): assert stt.transcription_engine == "Telnyx" assert stt.language == "en" assert stt.sample_rate == 16000 - assert stt.interim_results is True + assert stt.interim_results is False assert stt.provider_name == "telnyx" + assert stt.turn_detection is False - async def test_invalid_engine_rejected(self): - with pytest.raises(ValueError, match="Unsupported Telnyx transcription_engine"): - STT(api_key="KEY_test", transcription_engine="NotAnEngine") + async def test_unknown_engine_is_left_to_the_server(self): + """The engine catalogue is served by Telnyx, so it is not pinned here.""" + stt = STT(api_key="KEY_test", transcription_engine="SomeNewEngine") + assert "transcription_engine=SomeNewEngine" in stt._build_ws_url() async def test_url_carries_stream_parameters(self): stt = STT(api_key="KEY_test", language="es", sample_rate=8000) @@ -48,8 +67,8 @@ async def test_url_carries_stream_parameters(self): assert "transcription_engine=Telnyx" in url async def test_url_uses_interim_results_not_partial_results(self): - on = STT(api_key="KEY_test")._build_ws_url() - off = STT(api_key="KEY_test", interim_results=False)._build_ws_url() + on = STT(api_key="KEY_test", interim_results=True)._build_ws_url() + off = STT(api_key="KEY_test")._build_ws_url() assert "interim_results=true" in on assert "interim_results=false" in off assert "partial_results" not in on @@ -98,22 +117,24 @@ async def test_empty_transcript_emits_nothing(self, participant): assert await stt.output.collect(timeout=0) == [] - async def test_send_failure_does_not_propagate(self, participant): - stt = STT(api_key="KEY_test") - - class FailingWS: - closed = False - - async def send_bytes(self, data: bytes) -> None: - raise aiohttp.ClientError("connection reset") - - stt._ws = FailingWS() - stt._connection_ready.set() + async def test_send_failure_does_not_propagate( + self, stt_with_failing_socket, participant + ): + pcm = PcmData( + samples=np.zeros(160, dtype=np.int16), sample_rate=16000, format="s16" + ) + await stt_with_failing_socket.process_audio(pcm, participant=participant) + async def test_audio_before_start_is_dropped(self, participant): + """process_audio must not block forever when start() never ran.""" + stt = STT(api_key="KEY_test") pcm = PcmData( samples=np.zeros(160, dtype=np.int16), sample_rate=16000, format="s16" ) - await stt.process_audio(pcm, participant=participant) + + await asyncio.wait_for( + stt.process_audio(pcm, participant=participant), timeout=1.0 + ) async def test_integer_confidence_is_kept(self, participant): stt = STT(api_key="KEY_test") @@ -192,3 +213,23 @@ async def test_transcribe_at_telephony_sample_rate( finals = [i for i in items if isinstance(i, Transcript) and i.final] assert finals, "No final Transcript emitted at 8kHz" assert "forgotten treasures" in " ".join(t.text for t in finals).lower() + + async def test_interim_results_stream_partial_transcripts( + self, mia_audio_16khz, participant + ): + """Covers the replacement-mode path, which finals-only engines never hit. + + ``interim_results`` is honoured per engine: Speechmatics streams + partials, while the default Telnyx engine returns finals only. + """ + stt = STT(transcription_engine="Speechmatics", interim_results=True) + await stt.start() + try: + await stt.process_audio(mia_audio_16khz, participant=participant) + items = await stt.output.collect(timeout=15.0) + finally: + await stt.close() + + transcripts = [i for i in items if isinstance(i, Transcript)] + assert [t for t in transcripts if not t.final], "No interim Transcript emitted" + assert [t for t in transcripts if t.final], "No final Transcript emitted" diff --git a/plugins/telnyx/tests/test_telnyx_tts.py b/plugins/telnyx/tests/test_telnyx_tts.py index f51278aaf..5744b5499 100644 --- a/plugins/telnyx/tests/test_telnyx_tts.py +++ b/plugins/telnyx/tests/test_telnyx_tts.py @@ -70,73 +70,47 @@ def test_tag_consuming_whole_frame(self): assert stripper.feed(blob[50:]) == b"\xff\xf3audio" -class TestTelnyxTTS: - """Unit tests for Telnyx TTS configuration.""" +@pytest.fixture +def tts() -> TTS: + return TTS(api_key="KEY_test") - async def test_requires_api_key(self, monkeypatch): - monkeypatch.delenv("TELNYX_API_KEY", raising=False) - with pytest.raises(ValueError, match="TELNYX_API_KEY"): - TTS() - async def test_default_configuration(self): - tts = TTS(api_key="KEY_test") - assert tts.voice == "Telnyx.KokoroTTS.af_heart" - assert tts.provider_name == "telnyx" +@pytest.fixture +def dropped_socket_tts(request) -> TTS: + """A TTS whose socket is dropped mid-send, as stop_audio() would. - async def test_custom_voice(self): - tts = TTS(api_key="KEY_test", voice="AWS.Polly.Danielle-Neural") - assert tts.voice == "AWS.Polly.Danielle-Neural" + Parametrised with ``stop_before_drop``: True models a concurrent + ``stop_audio()`` (barge-in), False a genuine connection failure. + """ + stop_before_drop = request.param + instance = TTS(api_key="KEY_test") + class DroppedWS: + closed = True -class TestTelnyxTTSBargeIn: - """A socket dropped by stop_audio() ends synthesis instead of raising.""" + async def send_str(self, data: str) -> None: + if stop_before_drop: + instance._stop_event.set() + raise aiohttp.ClientConnectionResetError("Cannot write to closing") - @staticmethod - def tts_with_dropped_socket(stop_before_drop: bool) -> TTS: - instance = TTS(api_key="KEY_test") + async def close(self) -> None: + return None - class DroppedWS: - closed = True + class FakeSession: + closed = False - async def send_str(self, data: str) -> None: - if stop_before_drop: - instance._stop_event.set() - raise aiohttp.ClientConnectionResetError("Cannot write to closing") + async def ws_connect(self, url: str, headers: dict[str, str]): + return DroppedWS() - async def close(self) -> None: - return None + instance._session = FakeSession() + return instance - class FakeSession: - closed = False - async def ws_connect(self, url: str, headers: dict[str, str]): - return DroppedWS() - - instance._session = FakeSession() - return instance - - async def test_stop_during_synthesis_ends_quietly(self): - """A socket closed by a concurrent stop_audio() is a barge-in.""" - tts = self.tts_with_dropped_socket(stop_before_drop=True) +@pytest.fixture +def ws_serving(): + """Build a websocket stand-in that replays ``payloads`` then closes.""" - stream = await tts.stream_audio("hello") - assert [chunk async for chunk in stream] == [] - - async def test_connection_drop_without_stop_propagates(self): - """A stale stop must not silence a genuine failure in a new synthesis.""" - tts = self.tts_with_dropped_socket(stop_before_drop=False) - await tts.stop_audio() - - stream = await tts.stream_audio("hello") - with pytest.raises(aiohttp.ClientConnectionError): - [chunk async for chunk in stream] - - -class TestTelnyxTTSMalformedPayloads: - """The receive loop tolerates junk from the server without aborting.""" - - @staticmethod - def fake_ws(payloads: list[str]) -> object: + def build(payloads: list[str]) -> object: messages = [ SimpleNamespace(type=aiohttp.WSMsgType.TEXT, data=payload) for payload in payloads @@ -151,27 +125,61 @@ async def receive(self): return FakeWS() - async def test_non_dict_payload_is_skipped(self): - tts = TTS(api_key="KEY_test") - ws = self.fake_ws(['["not", "a", "dict"]']) + return build - assert [chunk async for chunk in tts._receive_audio(ws)] == [] - async def test_invalid_base64_audio_is_skipped(self): - tts = TTS(api_key="KEY_test") - ws = self.fake_ws(['{"audio": "!!!not base64!!!"}']) +class TestTelnyxTTS: + """Unit tests for Telnyx TTS configuration and the receive loop.""" + + async def test_requires_api_key(self, monkeypatch): + monkeypatch.delenv("TELNYX_API_KEY", raising=False) + with pytest.raises(ValueError, match="TELNYX_API_KEY"): + TTS() + + async def test_default_configuration(self, tts): + assert tts.voice == "Telnyx.KokoroTTS.af_heart" + assert tts.provider_name == "telnyx" + + async def test_custom_voice(self): + tts = TTS(api_key="KEY_test", voice="AWS.Polly.Danielle-Neural") + assert tts.voice == "AWS.Polly.Danielle-Neural" + + async def test_voice_is_reported_as_the_model(self): + """Agent metadata reads ``model`` off the component.""" + tts = TTS(api_key="KEY_test", voice="AWS.Polly.Danielle-Neural") + assert tts.model == "AWS.Polly.Danielle-Neural" + + @pytest.mark.parametrize("dropped_socket_tts", [True], indirect=True) + async def test_stop_during_synthesis_ends_quietly(self, dropped_socket_tts): + """A socket closed by a concurrent stop_audio() is a barge-in.""" + stream = await dropped_socket_tts.stream_audio("hello") + assert [chunk async for chunk in stream] == [] + + @pytest.mark.parametrize("dropped_socket_tts", [False], indirect=True) + async def test_connection_drop_without_stop_propagates(self, dropped_socket_tts): + """A stale stop must not silence a genuine failure in a new synthesis.""" + await dropped_socket_tts.stop_audio() + + stream = await dropped_socket_tts.stream_audio("hello") + with pytest.raises(aiohttp.ClientConnectionError): + [chunk async for chunk in stream] + + async def test_non_dict_payload_is_skipped(self, tts, ws_serving): + ws = ws_serving(['["not", "a", "dict"]']) assert [chunk async for chunk in tts._receive_audio(ws)] == [] - async def test_non_string_audio_is_skipped(self): - tts = TTS(api_key="KEY_test") - ws = self.fake_ws(['{"audio": 12345}']) + async def test_invalid_base64_audio_is_skipped(self, tts, ws_serving): + ws = ws_serving(['{"audio": "!!!not base64!!!"}']) assert [chunk async for chunk in tts._receive_audio(ws)] == [] - async def test_undecodable_audio_is_dropped(self): - tts = TTS(api_key="KEY_test") + async def test_non_string_audio_is_skipped(self, tts, ws_serving): + ws = ws_serving(['{"audio": 12345}']) + + assert [chunk async for chunk in tts._receive_audio(ws)] == [] + async def test_undecodable_audio_is_dropped(self, tts): class FailingDecoder: def parse(self, data: bytes): raise av.InvalidDataError(1094995529, "Invalid data") diff --git a/plugins/telnyx/vision_agents/plugins/telnyx/__init__.py b/plugins/telnyx/vision_agents/plugins/telnyx/__init__.py index cb17085c3..50ae8b4ea 100644 --- a/plugins/telnyx/vision_agents/plugins/telnyx/__init__.py +++ b/plugins/telnyx/vision_agents/plugins/telnyx/__init__.py @@ -13,12 +13,11 @@ telnyx_payload_to_pcm, ) from .call_registry import TelnyxCall, TelnyxCallRegistry -from .llm import TelnyxLLM +from .llm import TelnyxLLM as LLM from .media_stream import TelnyxMediaFormat, TelnyxMediaStream, attach_phone_to_call from .stt import STT from .tts import TTS, TelnyxTTSError -LLM = TelnyxLLM CallRegistry = TelnyxCallRegistry MediaStream = TelnyxMediaStream @@ -32,7 +31,6 @@ "TELNYX_L16_SAMPLE_RATE", "TelnyxCall", "TelnyxCallRegistry", - "TelnyxLLM", "TelnyxMediaFormat", "TelnyxMediaStream", "TelnyxTTSError", diff --git a/plugins/telnyx/vision_agents/plugins/telnyx/llm.py b/plugins/telnyx/vision_agents/plugins/telnyx/llm.py index efda03958..3a98ae687 100644 --- a/plugins/telnyx/vision_agents/plugins/telnyx/llm.py +++ b/plugins/telnyx/vision_agents/plugins/telnyx/llm.py @@ -65,7 +65,10 @@ def __init__( "TELNYX_API_KEY env var or api_key parameter required for Telnyx LLM" ) - if client is None: - client = AsyncOpenAI(api_key=resolved_key, base_url=base_url) - - super().__init__(model=model, client=client, tools_max_rounds=tools_max_rounds) + super().__init__( + model=model, + api_key=resolved_key, + base_url=base_url, + client=client, + tools_max_rounds=tools_max_rounds, + ) diff --git a/plugins/telnyx/vision_agents/plugins/telnyx/stt.py b/plugins/telnyx/vision_agents/plugins/telnyx/stt.py index 6e17dbe95..199956059 100644 --- a/plugins/telnyx/vision_agents/plugins/telnyx/stt.py +++ b/plugins/telnyx/vision_agents/plugins/telnyx/stt.py @@ -10,6 +10,11 @@ The query parameter for partial transcripts is ``interim_results``. ``partial_results`` is accepted by the endpoint but ignored, which silently yields finals only. + +``interim_results`` itself is honoured per engine rather than per endpoint. +Measured against the live API with the same audio: ``Speechmatics`` and +``Soniox`` stream partials, while ``Telnyx`` and ``Deepgram`` accept the +parameter and return finals only. """ import asyncio @@ -30,20 +35,6 @@ WS_STT_URL = "wss://api.telnyx.com/v2/speech-to-text/transcription" -SUPPORTED_ENGINES = { - "AssemblyAI", - "Azure", - "Deepgram", - "Google", - "Humain", - "Parakeet", - "Reson8", - "Soniox", - "Speechmatics", - "Telnyx", - "xAI", -} - class STT(stt.STT): """Telnyx streaming Speech-to-Text. @@ -61,36 +52,40 @@ class STT(stt.STT): stt = telnyx.STT(sample_rate=8000) """ + turn_detection: bool = False + def __init__( self, api_key: Optional[str] = None, transcription_engine: str = "Telnyx", language: str = "en", sample_rate: int = 16000, - interim_results: bool = True, - model: Optional[str] = None, + interim_results: bool = False, + model: str = "", ) -> None: """Initialize Telnyx STT. Args: api_key: Telnyx API key. Falls back to the ``TELNYX_API_KEY`` env var. - transcription_engine: Engine to transcribe with. Defaults to - ``Telnyx``. + transcription_engine: Engine to transcribe with, for example + ``Telnyx``, ``Deepgram`` or ``Speechmatics``. The catalogue is + served by Telnyx and grows over time, so the value is not + validated locally; an unknown engine is rejected by the server + with an explicit error. language: Language code, for example ``en``. sample_rate: Rate in Hz that audio is resampled to before being sent. Use 8000 to pass telephony audio from :class:`TelnyxMediaStream` through without upsampling. interim_results: Emit partial transcripts as they are refined. + Honoured per engine, not per endpoint: ``Speechmatics`` and + ``Soniox`` stream partials, while the default ``Telnyx`` engine + accepts the parameter and returns finals only. Defaults to + ``False`` so the default configuration does not advertise + partials it will never emit. model: Optional engine-specific model id. """ super().__init__(provider_name="telnyx") - if transcription_engine not in SUPPORTED_ENGINES: - raise ValueError( - f"Unsupported Telnyx transcription_engine '{transcription_engine}'. " - f"Expected one of: {sorted(SUPPORTED_ENGINES)}" - ) - self._api_key = api_key or os.environ.get("TELNYX_API_KEY") if not self._api_key: raise ValueError( @@ -109,6 +104,7 @@ def __init__( self._connection_ready = asyncio.Event() self._current_participant: Optional[Participant] = None self._audio_start_time: Optional[float] = None + self._error_reported = False async def start(self) -> None: """Open the Telnyx WebSocket and start the receive loop.""" @@ -142,6 +138,10 @@ async def process_audio( logger.warning("Telnyx STT is closed, ignoring audio") return + if not self.started: + logger.warning("Telnyx STT is not started, dropping audio") + return + await self._connection_ready.wait() if self._ws is None or self._ws.closed: @@ -163,30 +163,25 @@ async def close(self) -> None: """Close the WebSocket and clean up.""" await super().close() - try: - if self._ws is not None and not self._ws.closed: - await self._ws.close() - finally: - self._ws = None + if self._ws is not None and not self._ws.closed: + await self._ws.close() + self._ws = None + + if self._receive_task is not None: + self._receive_task.cancel() try: - if self._receive_task is not None: - self._receive_task.cancel() - try: - await self._receive_task - except asyncio.CancelledError: - pass - except Exception: - logger.exception("Telnyx STT receive task failed on close") - finally: - self._receive_task = None - try: - if self._session is not None and not self._session.closed: - await self._session.close() - finally: - self._session = None - self._connection_ready.clear() - self._on_disconnected() - self._audio_start_time = None + await self._receive_task + except asyncio.CancelledError: + pass + self._receive_task = None + + if self._session is not None and not self._session.closed: + await self._session.close() + self._session = None + + self._connection_ready.clear() + self._on_disconnected() + self._audio_start_time = None def _build_ws_url(self) -> str: params: dict[str, str] = { @@ -196,7 +191,7 @@ def _build_ws_url(self) -> str: "language": self.language, "interim_results": "true" if self.interim_results else "false", } - if self.model is not None: + if self.model: params["model"] = self.model return f"{WS_STT_URL}?{urlencode(params)}" @@ -229,7 +224,9 @@ async def _receive_loop(self) -> None: except aiohttp.ClientError: logger.exception("Telnyx STT receive loop error") - if not self.closed: + # The server closes the socket right after an error payload, so the + # close is that error's consequence rather than a separate failure. + if not self.closed and not self._error_reported: self._emit_error_event( ConnectionError("Telnyx STT WebSocket closed unexpectedly"), context="telnyx_ws_closed", @@ -250,6 +247,7 @@ def _handle_message(self, data: dict[str, object]) -> None: for err in errors if isinstance(err, dict) ) + self._error_reported = True self._emit_error_event( RuntimeError(details or "Telnyx STT error"), context="telnyx_streaming", diff --git a/plugins/telnyx/vision_agents/plugins/telnyx/tts.py b/plugins/telnyx/vision_agents/plugins/telnyx/tts.py index cc28ebfdd..b34841720 100644 --- a/plugins/telnyx/vision_agents/plugins/telnyx/tts.py +++ b/plugins/telnyx/vision_agents/plugins/telnyx/tts.py @@ -29,7 +29,7 @@ import aiohttp import av -from getstream.video.rtc.track_util import AudioFormat, PcmData +from getstream.video.rtc.track_util import PcmData from vision_agents.core import tts logger = logging.getLogger(__name__) @@ -103,6 +103,7 @@ def __init__( api_key: Optional[str] = None, voice: str = DEFAULT_VOICE, idle_timeout: float = 10.0, + connect_timeout: float = 10.0, ) -> None: """Initialize Telnyx TTS. @@ -114,6 +115,7 @@ def __init__( idle_timeout: Seconds of server silence before synthesis is treated as finished. Normally the server marks the last frame with ``isFinal``; this is a safety net. + connect_timeout: Seconds to wait for the WebSocket handshake. """ super().__init__(provider_name="telnyx") @@ -124,20 +126,30 @@ def __init__( ) self.voice = voice + # The voice doubles as the model identifier reported in agent metadata; + # Telnyx voice ids carry the provider and model, e.g. AWS.Polly.Danielle-Neural. + self.model = voice self._idle_timeout = idle_timeout + self._connect_timeout = connect_timeout self._session: Optional[aiohttp.ClientSession] = None self._ws: Optional[aiohttp.ClientWebSocketResponse] = None self._lock = asyncio.Lock() self._stop_event = asyncio.Event() + # A socket is opened per synthesis, so connect/disconnect are reported + # once for the life of the plugin rather than once per utterance. + self._connected = False async def close(self) -> None: """Close the current WebSocket and release the aiohttp session.""" await super().close() - if self._session is not None and not self._session.closed: - await self._session.close() + session = self._session self._session = None - self._on_disconnected() + if session is not None and not session.closed: + await session.close() + if self._connected: + self._connected = False + self._on_disconnected() async def stream_audio( self, text: str, *_: Any, **__: Any @@ -189,10 +201,12 @@ async def _connect(self) -> aiohttp.ClientWebSocketResponse: self._session.ws_connect( url, headers={"Authorization": f"Bearer {self._api_key}"} ), - timeout=self._idle_timeout, + timeout=self._connect_timeout, ) self._ws = ws - self._on_connected() + if not self._connected: + self._connected = True + self._on_connected() logger.debug("Telnyx TTS websocket connected for voice %s", self.voice) return ws @@ -273,14 +287,7 @@ def _decode( for packet in decoder.parse(stripper.feed(audio)): for frame in decoder.decode(packet): for resampled in resampler.resample(frame): - chunks.append( - PcmData( - samples=resampled.to_ndarray().reshape(-1), - sample_rate=resampled.sample_rate, - channels=1, - format=AudioFormat.S16, - ) - ) + chunks.append(PcmData.from_av_frame(resampled)) except av.FFmpegError: logger.warning("Telnyx TTS sent undecodable audio, dropping payload") return chunks From 1845ec40218197f695872e2dc3dfd474f2e2f5cc Mon Sep 17 00:00:00 2001 From: "Neevash Ramdial (Nash)" Date: Mon, 10 Aug 2026 17:14:21 -0600 Subject: [PATCH 08/10] docs(telnyx): add an all-Telnyx example, changelog entry, and API notes - Add `examples/voice_agent_call.py`: an inbound call answered by a pipeline that runs entirely on Telnyx, reusing the existing example helpers. Turn detection is local because the transcription endpoint sends no VAD signals. - Add a CHANGELOG entry, as the speechify plugin did when it landed. - Document what the live API actually does: which engines honour `interim_results`, and why the TTS decodes MP3 for every voice even though `audio_format` exists. --- CHANGELOG.md | 13 + plugins/telnyx/README.md | 23 +- plugins/telnyx/examples/README.md | 19 +- plugins/telnyx/examples/voice_agent_call.py | 248 ++++++++++++++++++++ 4 files changed, 299 insertions(+), 4 deletions(-) create mode 100644 plugins/telnyx/examples/voice_agent_call.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 817084927..9da0874de 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,19 @@ ## New Features +### `telnyx` plugin: LLM, STT and TTS (#620, #621, #622) + +The Telnyx plugin, until now a phone transport, also exposes `telnyx.LLM`, +`telnyx.STT`, and `telnyx.TTS`, so a phone agent can run end to end on Telnyx. +`telnyx.LLM` wraps Telnyx Inference's OpenAI-compatible Chat Completions +endpoint and defaults to `meta-llama/Llama-3.3-70B-Instruct`. `telnyx.STT` +streams `linear16` over WebSocket and takes a `sample_rate`, so telephony audio +from `TelnyxMediaStream` can be transcribed at 8 kHz without an upsample; pick +the engine with `transcription_engine`. `telnyx.TTS` streams MP3 over WebSocket +and decodes to `PcmData` as it arrives. All three read `TELNYX_API_KEY` from the +environment. See `plugins/telnyx/examples/voice_agent_call.py` for an inbound +call answered by an all-Telnyx pipeline. + ### `speechify` plugin: Speechify TTS Adds a new `speechify` plugin exposing `speechify.TTS`, backed by Speechify's streaming API. It streams raw PCM audio, defaults to the `simba-3.2` model with the `geffen_32` voice, and reads `SPEECHIFY_API_KEY` from the environment. Install with `vision-agents[speechify]`. diff --git a/plugins/telnyx/README.md b/plugins/telnyx/README.md index d570c4f98..e2ac9b019 100644 --- a/plugins/telnyx/README.md +++ b/plugins/telnyx/README.md @@ -79,11 +79,21 @@ stt = telnyx.STT(sample_rate=8000) Requires `TELNYX_API_KEY` in the environment, or an `api_key` argument. Audio is resampled to `sample_rate` and sent as raw `linear16` frames. Pick the -engine with `transcription_engine`; the default is `Telnyx`. +engine with `transcription_engine`; the default is `Telnyx`. The engine +catalogue is served by Telnyx and is not validated locally. Telnyx does not send VAD signals on this endpoint, so the plugin emits transcripts only and leaves turn detection to the agent. +`interim_results` is honoured per engine rather than per endpoint, and defaults +to `False`. Measured against the live API with the same audio, `Speechmatics` +and `Soniox` stream partial transcripts, while `Telnyx` and `Deepgram` accept +the parameter and return finals only: + +```python +stt = telnyx.STT(transcription_engine="Speechmatics", interim_results=True) +``` + ## TTS ```python @@ -102,6 +112,12 @@ the stop frame, so the plugin reconnects per `stream_audio` call. Audio arrives as MP3 and is decoded to `PcmData` as it streams. The output sample rate follows the voice, so it is taken from the decoder rather than configured. +The endpoint takes an `audio_format` parameter, but it is honoured only by some +voices — `AWS.Polly.*` and `Telnyx.NaturalHD.*` serve raw PCM, while the default +`Telnyx.KokoroTTS.*` returns MP3 regardless. Since the PCM sample rate is not +reported on the wire and differs per voice, the plugin decodes MP3 for every +voice rather than carrying a voice-to-rate table that would go stale. + ## Examples See [examples/](examples/) for minimal inbound and outbound Telnyx phone @@ -118,6 +134,11 @@ uv run plugins/telnyx/examples/outbound_call.py \ uv run plugins/telnyx/examples/inbound_call.py \ --setup-telnyx \ --phone-number +15551234567 + +# Inbound call answered by an all-Telnyx STT/LLM/TTS pipeline +uv run plugins/telnyx/examples/voice_agent_call.py \ + --setup-telnyx \ + --phone-number +15551234567 ``` Telnyx phone calls require a Call Control App. The Call Control App is where diff --git a/plugins/telnyx/examples/README.md b/plugins/telnyx/examples/README.md index c95826e13..4f39daa70 100644 --- a/plugins/telnyx/examples/README.md +++ b/plugins/telnyx/examples/README.md @@ -1,7 +1,12 @@ # Telnyx Phone Examples -Minimal inbound and outbound phone examples for the Telnyx plugin. These examples -use Telnyx Call Control, Telnyx Media Streaming, Stream, and Gemini Realtime. +Minimal inbound and outbound phone examples for the Telnyx plugin. + +- `outbound_call.py` and `inbound_call.py` use Telnyx Call Control, Telnyx Media + Streaming, Stream, and Gemini Realtime. +- `voice_agent_call.py` answers an inbound call with a pipeline that runs + entirely on Telnyx: `telnyx.STT`, `telnyx.LLM`, and `telnyx.TTS`. It needs no + `GOOGLE_API_KEY`. ## Requirements @@ -10,7 +15,7 @@ Create a `.env` file at the repo root or export these variables: ```bash STREAM_API_KEY= STREAM_API_SECRET= -GOOGLE_API_KEY= +GOOGLE_API_KEY= # not needed by voice_agent_call.py TELNYX_API_KEY= TELNYX_PUBLIC_KEY= ``` @@ -57,6 +62,14 @@ uv run plugins/telnyx/examples/inbound_call.py \ --phone-number +15551234567 ``` +Inbound, all Telnyx: + +```bash +uv run plugins/telnyx/examples/voice_agent_call.py \ + --setup-telnyx \ + --phone-number +15551234567 +``` + For inbound calls, `--setup-telnyx` also routes the Telnyx number to the temporary Call Control App and restores the previous routing on normal shutdown. diff --git a/plugins/telnyx/examples/voice_agent_call.py b/plugins/telnyx/examples/voice_agent_call.py new file mode 100644 index 000000000..e2a04a688 --- /dev/null +++ b/plugins/telnyx/examples/voice_agent_call.py @@ -0,0 +1,248 @@ +"""Inbound Telnyx phone example running the whole pipeline on Telnyx. + +Unlike `inbound_call.py`, which bridges the call into a realtime model, this +example uses Telnyx for transport, speech to text, inference, and text to +speech. The only non-Telnyx pieces are Stream for the call and a turn detector, +which Telnyx's transcription endpoint does not provide. + +Run after starting ngrok and routing your Telnyx number to a Call Control App: + + NGROK_URL=example.ngrok-free.app uv run plugins/telnyx/examples/voice_agent_call.py +""" + +import argparse +import asyncio +import contextlib +import logging +import os +import uuid + +import uvicorn +from dotenv import load_dotenv +from fastapi import FastAPI, Request, WebSocket +from fastapi.responses import JSONResponse +from uvicorn.middleware.proxy_headers import ProxyHeadersMiddleware +from vision_agents.core import Agent, User +from vision_agents.plugins import getstream, smart_turn, telnyx +from vision_agents.plugins.getstream.stream_edge_transport import StreamEdge +from vision_agents.plugins.telnyx.example_helpers import ( + TelnyxClient, + TelnyxConfig, + TelnyxSetupError, + cleanup_telnyx_example_setup, + media_stream_url, + parse_verified_telnyx_webhook, + preflight_inbound, + prepare_telnyx_example_setup, + require_env, + require_telnyx_public_key, +) + + +logger = logging.getLogger(__name__) +logging.basicConfig(level=logging.INFO) + +load_dotenv() + +app = FastAPI() +app.add_middleware(ProxyHeadersMiddleware, trusted_hosts=["*"]) +call_registry = telnyx.TelnyxCallRegistry() +telnyx_client: TelnyxClient | None = None +telnyx_config: TelnyxConfig | None = None +telnyx_public_key: str | None = None + + +@app.exception_handler(Exception) +async def global_exception_handler(_request: Request, exc: Exception): + logger.exception("Unhandled exception: %s", exc) + return JSONResponse(status_code=500, content={"detail": "Internal server error"}) + + +async def create_agent() -> tuple[Agent, StreamEdge]: + edge = getstream.Edge() + agent = Agent( + edge=edge, + agent_user=User(id="ai-agent", name="AI Assistant"), + instructions=( + "Speak English. Keep replies short and natural. You are answering " + "an inbound Telnyx test call through Vision Agents. Start by saying " + "this is a quick inbound Telnyx bridge test and ask whether the " + "audio is clear." + ), + # 8000 is what TelnyxMediaStream decodes PCMU telephony audio to, so + # the transcriber receives it without an upsample that adds nothing. + stt=telnyx.STT(sample_rate=8000), + llm=telnyx.LLM(), + tts=telnyx.TTS(), + # Telnyx's transcription endpoint sends no VAD signals, so turns are + # detected locally rather than by the STT. + turn_detection=smart_turn.TurnDetection(), + ) + return agent, edge + + +async def prepare_call(call_id: str): + agent, edge = await create_agent() + phone_user = User( + name=f"Inbound Telnyx call {call_id[:8]}", + id=f"phone-{call_id}", + ) + await edge.create_users([agent.agent_user, phone_user]) + stream_call = await agent.create_call("default", call_id) + logger.info("Prepared Stream call %s", call_id) + return agent, phone_user, stream_call + + +async def wait_for_start( + stream: telnyx.TelnyxMediaStream, timeout: float = 10.0 +) -> None: + deadline = asyncio.get_running_loop().time() + timeout + while not stream.has_started: + if asyncio.get_running_loop().time() >= deadline: + raise TimeoutError("Telnyx media stream did not start in time") + await asyncio.sleep(0.05) + + +@app.post("/telnyx/events") +async def telnyx_events(request: Request): + if telnyx_client is None or telnyx_config is None or telnyx_public_key is None: + raise RuntimeError("Telnyx example was not initialized") + + data = await parse_verified_telnyx_webhook(request, telnyx_public_key) + event_type = data.get("data", {}).get("event_type") + payload = data.get("data", {}).get("payload", {}) + logger.info("Telnyx webhook event: %s", event_type) + + if event_type == "call.initiated" and payload.get("direction") == "incoming": + call_control_id = payload["call_control_id"] + call_id = str(uuid.uuid4()) + telnyx_call = call_registry.create( + call_id, + webhook_data=data, + prepare=lambda: prepare_call(call_id), + ) + stream_url = media_stream_url( + telnyx_config.ngrok_url, + call_id, + telnyx_call.token, + ) + await asyncio.to_thread( + telnyx_client.answer_call, + call_control_id, + stream_url=stream_url, + ) + logger.info("Answered inbound Telnyx call %s", call_id) + + return {"ok": True} + + +@app.websocket("/telnyx/media/{call_id}/{token}") +async def media_stream(websocket: WebSocket, call_id: str, token: str): + telnyx_call = call_registry.validate(call_id, token) + logger.info("Media stream connected for inbound call %s", call_id) + + telnyx_stream = telnyx.TelnyxMediaStream(websocket) + await telnyx_stream.accept() + telnyx_call.telnyx_stream = telnyx_stream + stream_task = asyncio.create_task(telnyx_stream.run()) + + try: + agent, phone_user, stream_call = await telnyx_call.await_prepare() + telnyx_call.stream_call = stream_call + + await telnyx.attach_phone_to_call(stream_call, telnyx_stream, phone_user.id) + await wait_for_start(telnyx_stream) + + async with agent.join(stream_call, participant_wait_timeout=0): + await agent.simple_response( + text="Greet the caller and ask whether the inbound Telnyx audio is clear." + ) + await stream_task + finally: + if not stream_task.done(): + stream_task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await stream_task + call_registry.remove(call_id) + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Run an inbound Telnyx call with a Telnyx STT/LLM/TTS pipeline." + ) + parser.add_argument("--host", default="127.0.0.1") + parser.add_argument("--port", default=8000, type=int) + parser.add_argument( + "--phone-number-id", + default=None, + help="Telnyx phone number resource ID. Defaults to TELNYX_PHONE_NUMBER_ID.", + ) + parser.add_argument( + "--phone-number", + default=None, + help="Telnyx inbound number. Defaults to TELNYX_PHONE_NUMBER.", + ) + parser.add_argument( + "--call-control-app-id", + default=None, + help="Existing Telnyx Call Control App ID. Defaults to TELNYX_CALL_CONTROL_APP_ID.", + ) + parser.add_argument( + "--ngrok-url", + default=None, + help="Public ngrok hostname. Defaults to NGROK_URL or local ngrok autodetection.", + ) + parser.add_argument( + "--setup-telnyx", + action="store_true", + help="Create a temporary Call Control App, route the phone number, and restore it on exit.", + ) + return parser.parse_args() + + +def main() -> None: + global telnyx_client, telnyx_config, telnyx_public_key + + args = parse_args() + values = require_env(["STREAM_API_KEY", "STREAM_API_SECRET", "TELNYX_API_KEY"]) + telnyx_public_key = require_telnyx_public_key() + telnyx_client = TelnyxClient(values["TELNYX_API_KEY"]) + setup = prepare_telnyx_example_setup( + telnyx_client, + api_key=values["TELNYX_API_KEY"], + phone_number=args.phone_number or os.environ.get("TELNYX_PHONE_NUMBER"), + ngrok_url=args.ngrok_url or os.environ.get("NGROK_URL"), + call_control_app_id=( + args.call_control_app_id or os.environ.get("TELNYX_CALL_CONTROL_APP_ID") + ), + phone_number_id=( + args.phone_number_id or os.environ.get("TELNYX_PHONE_NUMBER_ID") + ), + setup_telnyx=args.setup_telnyx, + route_phone_number=True, + ) + telnyx_config = setup.config + resolved_phone_number_id = setup.phone_number_id or ( + args.phone_number_id or os.environ.get("TELNYX_PHONE_NUMBER_ID") + ) + if not resolved_phone_number_id: + raise TelnyxSetupError( + "Missing TELNYX_PHONE_NUMBER_ID. Pass `--setup-telnyx` to discover " + "and route the Telnyx number automatically." + ) + + try: + preflight_inbound( + telnyx_client, + config=telnyx_config, + telnyx_phone_number_id=resolved_phone_number_id, + ) + + logger.info("Telnyx voice agent ready for call %s", resolved_phone_number_id) + uvicorn.run(app, host=args.host, port=args.port) + finally: + cleanup_telnyx_example_setup(telnyx_client, setup) + + +if __name__ == "__main__": + main() From 811f2cdea581ccf55218aa475aac3b6d9428fa46 Mon Sep 17 00:00:00 2001 From: "Neevash Ramdial (Nash)" Date: Mon, 10 Aug 2026 19:20:41 -0600 Subject: [PATCH 09/10] Potential fix for pull request finding 'CodeQL / Clear-text logging of sensitive information' Co-authored-by: Copilot Autofix powered by AI <62310815+github-advanced-security[bot]@users.noreply.github.com> --- plugins/telnyx/examples/voice_agent_call.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/plugins/telnyx/examples/voice_agent_call.py b/plugins/telnyx/examples/voice_agent_call.py index e2a04a688..137b66b35 100644 --- a/plugins/telnyx/examples/voice_agent_call.py +++ b/plugins/telnyx/examples/voice_agent_call.py @@ -238,7 +238,7 @@ def main() -> None: telnyx_phone_number_id=resolved_phone_number_id, ) - logger.info("Telnyx voice agent ready for call %s", resolved_phone_number_id) + logger.info("Telnyx voice agent ready for call") uvicorn.run(app, host=args.host, port=args.port) finally: cleanup_telnyx_example_setup(telnyx_client, setup) From 45d4a6bfa7cee6604044826bd6b370b5371c7217 Mon Sep 17 00:00:00 2001 From: "Neevash Ramdial (Nash)" Date: Wed, 12 Aug 2026 22:52:00 -0600 Subject: [PATCH 10/10] fix(telnyx): stop TTS synthesis on barge-in, not just the socket stop_audio() set the stop event and closed the WebSocket, but the receive loop only checked the event once per WebSocket frame. Telnyx serves a whole synthesis in very few large frames, so a single payload decodes to hundreds of PcmData chunks and the check never ran again: measured against the live API, stopping after 3 chunks still delivered all 445, the same as an uninterrupted synthesis. The Agent's own barge-in was unaffected, because interrupt() also bumps the epoch and send_iter() checks that per chunk. Direct users of stream_audio() plus stop_audio() got the whole utterance regardless. Check the stop event per decoded chunk as well. Stopping after 3 chunks now yields 3. The integration test pins this against an uninterrupted baseline. --- plugins/telnyx/tests/test_telnyx_tts.py | 25 +++++++++++++++++++ .../vision_agents/plugins/telnyx/tts.py | 6 +++++ 2 files changed, 31 insertions(+) diff --git a/plugins/telnyx/tests/test_telnyx_tts.py b/plugins/telnyx/tests/test_telnyx_tts.py index 5744b5499..50209ede1 100644 --- a/plugins/telnyx/tests/test_telnyx_tts.py +++ b/plugins/telnyx/tests/test_telnyx_tts.py @@ -238,3 +238,28 @@ async def test_second_synthesis_reconnects(self, tts): assert any(item.data is not None for item in first) assert any(item.data is not None for item in second) + + async def test_stop_audio_ends_synthesis_promptly(self, tts): + """Barge-in must cut the audio, not just close the socket. + + Telnyx serves a synthesis in very few large frames, so one payload + decodes to hundreds of chunks. Checking the stop only per frame let the + whole utterance play out after a barge-in. + """ + text = ( + "This is a deliberately long utterance so that there is plenty of " + "audio still being streamed when the barge-in arrives, and it keeps " + "going for a while longer to make sure the socket is mid-flight." + ) + baseline = len([chunk async for chunk in await tts.stream_audio(text)]) + assert baseline > 50, f"need a long synthesis to test against, got {baseline}" + + received = 0 + async for _ in await tts.stream_audio(text): + received += 1 + if received == 3: + await tts.stop_audio() + + assert received < baseline / 2, ( + f"stop_audio() did not cut the synthesis: {received} of {baseline} chunks" + ) diff --git a/plugins/telnyx/vision_agents/plugins/telnyx/tts.py b/plugins/telnyx/vision_agents/plugins/telnyx/tts.py index b34841720..a0e75774a 100644 --- a/plugins/telnyx/vision_agents/plugins/telnyx/tts.py +++ b/plugins/telnyx/vision_agents/plugins/telnyx/tts.py @@ -264,7 +264,13 @@ async def _receive_audio( except (binascii.Error, TypeError, ValueError): logger.warning("Telnyx TTS sent audio that is not valid base64") continue + # Telnyx serves a whole synthesis in very few large frames, so + # one payload decodes to hundreds of chunks. Checking the stop + # only per frame would let a barge-in play out the rest of the + # utterance. for pcm in self._decode(raw, decoder, resampler, stripper): + if self._stop_event.is_set(): + return yield pcm if data.get("isFinal"):