diff --git a/plugins/telnyx/README.md b/plugins/telnyx/README.md
index dd4564e8b..8f6d5e3ac 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
+- **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
diff --git a/plugins/telnyx/pyproject.toml b/plugins/telnyx/pyproject.toml
index 961aedd0e..2c74735d5 100644
--- a/plugins/telnyx/pyproject.toml
+++ b/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]
diff --git a/plugins/telnyx/tests/test_telnyx_tts.py b/plugins/telnyx/tests/test_telnyx_tts.py
new file mode 100644
index 000000000..f51278aaf
--- /dev/null
+++ b/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)
diff --git a/plugins/telnyx/vision_agents/plugins/telnyx/__init__.py b/plugins/telnyx/vision_agents/plugins/telnyx/__init__.py
index cf5b476c1..5c8973565 100644
--- a/plugins/telnyx/vision_agents/plugins/telnyx/__init__.py
+++ b/plugins/telnyx/vision_agents/plugins/telnyx/__init__.py
@@ -14,6 +14,7 @@
 )
 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
@@ -21,12 +22,14 @@
 __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",
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..cc28ebfdd
--- /dev/null
+++ b/plugins/telnyx/vision_agents/plugins/telnyx/tts.py
@@ -0,0 +1,286 @@
+"""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 binascii
+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]:
+            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()
+                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": " "}))
+                    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
+                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()
+
+        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})}"
+        # 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()
+        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", exc_info=True)
+        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 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:
+                try:
+                    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):
+                    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.
+
+        A corrupt payload is dropped rather than allowed to abort the
+        synthesis; the decoder recovers on the next packet boundary.
+        """
+        chunks: list[PcmData] = []
+        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
diff --git a/uv.lock b/uv.lock
index d0db9afe3..92743d527 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" },
@@ -7549,6 +7550,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" },