feat(telnyx): add Telnyx TTS plugin

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
+- **TTS**: Streaming text to speech over WebSocket
 
 ## Installation
 
@@ -46,6 +47,24 @@ call.telnyx_stream = stream
 await stream.run()
 ```
 
+## 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
@@ -189,5 +208,6 @@ payload = pcm_to_pcmu(pcm)
 ## Dependencies
 
 - vision-agents
+- aiohttp
 - numpy
 - fastapi

plugins/telnyx/pyproject.toml

@@ -14,6 +14,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]

plugins/telnyx/tests/test_telnyx_tts.py

@@ -0,0 +1,232 @@
+"""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
+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"
+
+
+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."""
+
+    @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_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")
+
+        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:
+    """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)

plugins/telnyx/vision_agents/plugins/telnyx/__init__.py

@@ -14,19 +14,22 @@
 )
 from .call_registry import TelnyxCall, TelnyxCallRegistry
 from .media_stream import TelnyxMediaFormat, TelnyxMediaStream, attach_phone_to_call
+from .tts import TTS, TelnyxTTSError
 
 CallRegistry = TelnyxCallRegistry
 MediaStream = TelnyxMediaStream
 
 __all__ = [
     "CallRegistry",
     "MediaStream",
+    "TTS",
     "TELNYX_DEFAULT_SAMPLE_RATE",
     "TELNYX_L16_SAMPLE_RATE",
     "TelnyxCall",
     "TelnyxCallRegistry",
     "TelnyxMediaFormat",
     "TelnyxMediaStream",
+    "TelnyxTTSError",
     "attach_phone_to_call",
     "l16_to_pcm",
     "pcma_to_pcm",