diff --git a/CHANGELOG.md b/CHANGELOG.md
index 817084927..b38cb9364 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,6 +2,12 @@
 
 ## New Features
 
+### `palabra` plugin: Palabra AI TTS and voice cloning (#627)
+
+Adds a new `palabra` plugin exposing `palabra.TTS`, backed by Palabra's realtime text-to-speech WebSocket API. It is a streaming TTS plugin, so the agent speaks each sentence while the LLM is still writing, and it keeps one session open across utterances — `stop_audio()` cancels synthesis server-side instead of reconnecting. Defaults to the `default_low` voice at 24 kHz and reads `PALABRA_API_KEY` from the environment. Install with `vision-agents[palabra]`.
+
+`palabra.Voices` wraps Palabra's cloned-voice API: `clone()` runs the whole create → upload → poll sequence and returns a `voice_id` that drops straight into `TTS(voice_id=...)`, alongside `get()`, `list()`, `delete()` and `limits()`. `TTS` also gained `idle_timeout` (default 5 s), which abandons a generation the server stops answering so a wedged utterance can't stall the TTS 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/agents-core/pyproject.toml b/agents-core/pyproject.toml
index 708c73683..43b200b4b 100644
--- a/agents-core/pyproject.toml
+++ b/agents-core/pyproject.toml
@@ -89,6 +89,7 @@ tencent = ["vision-agents-plugins-tencent; sys_platform == 'linux'"]
 minimax = ["vision-agents-plugins-minimax"]
 twelvelabs = ["vision-agents-plugins-twelvelabs"]
 speechify = ["vision-agents-plugins-speechify"]
+palabra = ["vision-agents-plugins-palabra"]
 
 
 [tool.hatch.metadata]
diff --git a/plugins/palabra/README.md b/plugins/palabra/README.md
new file mode 100644
index 000000000..4b1b12efc
--- /dev/null
+++ b/plugins/palabra/README.md
@@ -0,0 +1,144 @@
+# Palabra AI
+
+[Palabra AI](https://palabra.ai) provides a realtime Text-to-Speech (TTS) API built for streaming: text is accepted
+incrementally over a WebSocket and audio comes back as raw PCM within a few hundred milliseconds, which makes it a good
+fit for voice AI agents.
+
+The Palabra plugin for Vision Agents lets you give your agent a Palabra voice, in 25 languages and with your own cloned
+voices.
+
+## Features
+
+- Streaming TTS over a single persistent WebSocket – the session is opened once and reused for every utterance
+- Sentence-level streaming (`streaming = True`), so the agent starts speaking while the LLM is still writing
+- Instant barge-in: `stop_audio()` cancels synthesis server-side without dropping the connection
+- Raw PCM output at any sample rate between 8 kHz and 48 kHz
+- 25 languages
+- Voice cloning from an audio sample, via `palabra.Voices`
+
+## Installation
+
+```bash
+uv add "vision-agents[palabra]"
+# or directly
+uv add vision-agents-plugins-palabra
+```
+
+## Usage
+
+```python
+from vision_agents.plugins import palabra
+
+tts = palabra.TTS()
+```
+
+Use it in an agent:
+
+```python
+from vision_agents.core.agents import Agent
+from vision_agents.core.edge.types import User
+from vision_agents.plugins import deepgram, gemini, getstream, palabra
+
+agent = Agent(
+    edge=getstream.Edge(),
+    agent_user=User(name="Palabra Voice Bot", id="agent"),
+    instructions="You're a helpful voice AI assistant.",
+    stt=deepgram.STT(),
+    llm=gemini.LLM(),
+    tts=palabra.TTS(voice_id="default_high", language="en"),
+)
+```
+
+<Warning>
+  To initialise without passing in the API key, make sure `PALABRA_API_KEY` is available as an environment variable.
+  You can do this either by defining it in a `.env` file or exporting it directly in your terminal. Create a key in the
+  [Palabra platform](https://platform.palabra.ai/api-keys).
+</Warning>
+
+## Examples
+
+Check out our [Palabra example](https://github.com/GetStream/Vision-Agents/tree/main/plugins/palabra/example) to see
+working code:
+
+- [main.py](https://github.com/GetStream/Vision-Agents/blob/main/plugins/palabra/example/main.py) – a voice bot that
+  uses Palabra TTS in a Stream call
+- [tts_smoke.py](https://github.com/GetStream/Vision-Agents/blob/main/plugins/palabra/example/tts_smoke.py) – synthesize
+  a few sentences to a WAV file and report the time to first audio chunk
+- [clone_voice.py](https://github.com/GetStream/Vision-Agents/blob/main/plugins/palabra/example/clone_voice.py) – clone a
+  voice from an audio sample and speak with it
+
+## Configuration
+
+| Name                | Type              | Default           | Description                                                                                              |
+|---------------------|-------------------|-------------------|----------------------------------------------------------------------------------------------------------|
+| `api_key`           | `str` or `None`   | `None`            | Your Palabra API key. Falls back to the `PALABRA_API_KEY` environment variable.                          |
+| `voice_id`          | `str`             | `"default_low"`   | Voice to synthesize with: `default_low`, `default_high`, or the id of a [cloned voice](https://platform.palabra.ai/docs/assets/voices). |
+| `language`          | `str`             | `"en"`            | BCP-47 language code of the text, e.g. `en`, `en-gb`, `de`, `pt-eu`, `ko`.                               |
+| `model`             | `str`             | `"auto"`          | TTS model id. `auto` lets Palabra pick the model.                                                        |
+| `sample_rate`       | `int`             | `24000`           | Output sample rate in Hz. Must be between `8000` and `48000`.                                            |
+| `speed`             | `float` or `None` | `None`            | Speech speed multiplier between `0.0` and `2.0`. `None` uses the server default.                         |
+| `deaccent_strength` | `float` or `None` | `None`            | Accent reduction for cloned voices, between `0.0` and `1.0`. `None` uses the server default.             |
+| `ws_url`            | `str`             | `WS_URL_EU`       | Palabra endpoint. Pass `palabra.WS_URL_US` to use the US region.                                          |
+| `idle_timeout`      | `float`           | `5.0`             | Seconds to wait for the next audio frame before abandoning a generation. Guards against a server that stops answering without sending an error. |
+
+## Functionality
+
+### Send text to convert to speech
+
+`send_iter()` sends the text to Palabra and yields `TTSOutputChunk`s carrying the produced PCM audio:
+
+```python
+async for chunk in tts.send_iter("Demo text you want the AI voice to say"):
+    pass
+```
+
+Text longer than Palabra's 1024-character per-message limit is split across several messages automatically, on word
+boundaries.
+
+### Stop speaking
+
+```python
+await tts.stop_audio()
+```
+
+This sends a `cancel` to Palabra and drops any audio still in flight. The WebSocket session stays open, so the next
+utterance does not pay for a new handshake.
+
+## Voice cloning
+
+`palabra.Voices` wraps Palabra's [cloned voice API](https://platform.palabra.ai/docs/assets/voices). `clone()` runs the
+whole sequence — reserve the voice, upload the sample to the presigned target, poll until Palabra reports it `ready` —
+and returns a `voice_id` you pass straight to `TTS`:
+
+```python
+from vision_agents.plugins import palabra
+
+async with palabra.Voices() as voices:
+    voice = await voices.clone("Narrator", "sample.wav", lang_code="en")
+
+tts = palabra.TTS(voice_id=voice.voice_id, deaccent_strength=0.7)
+```
+
+`deaccent_strength` exists specifically for cloned voices: lower it to keep more of the speaker's accent, raise it to
+neutralise it.
+
+Palabra needs **at least 30 seconds** of clean, single-speaker audio, at most **10 MB**, as MP3, WAV, FLAC, WEBM, MP4,
+MPEG or MPG. Cloning usually finishes in under a minute; `clone()` polls until then, or pass `wait=False` to return
+immediately and check `voices.get(voice_id).ready` yourself.
+
+| Method                                    | Description                                                                       |
+|-------------------------------------------|-----------------------------------------------------------------------------------|
+| `clone(name, sample, ...)`                | Clone a voice from an audio or video sample. Returns a `ClonedVoice`.              |
+| `get(voice_id)`                           | Fetch one voice, including `processing_status` and any errors or warnings.         |
+| `list(search=..., lang=..., page_size=...)`| List cloned voices.                                                              |
+| `delete(voice_id)`                        | Permanently delete a cloned voice. Irreversible.                                  |
+| `limits()`                                | Cloned voice quota for the account (`total`, `limit`, `remaining`, …).             |
+
+Cloned voices count against your account quota, so delete the ones you no longer need. Only clone a person's voice with
+their explicit consent.
+
+## Dependencies
+
+- [`vision-agents`](https://pypi.org/project/vision-agents/)
+- [`websockets`](https://pypi.org/project/websockets/)
+- [`httpx`](https://pypi.org/project/httpx/) – for the voice cloning REST API
diff --git a/plugins/palabra/example/.env.example b/plugins/palabra/example/.env.example
new file mode 100644
index 000000000..9c262fb9e
--- /dev/null
+++ b/plugins/palabra/example/.env.example
@@ -0,0 +1,12 @@
+# Stream API credentials
+STREAM_API_KEY=your_stream_api_key_here
+STREAM_API_SECRET=your_stream_api_secret_here
+
+# Palabra TTS
+PALABRA_API_KEY=your_palabra_api_key_here
+
+# Deepgram STT
+DEEPGRAM_API_KEY=your_deepgram_api_key_here
+
+# Gemini LLM
+GOOGLE_API_KEY=your_google_api_key_here
diff --git a/plugins/palabra/example/README.md b/plugins/palabra/example/README.md
new file mode 100644
index 000000000..a634d6203
--- /dev/null
+++ b/plugins/palabra/example/README.md
@@ -0,0 +1,73 @@
+# Stream + Palabra Voice Bot Example
+
+This example demonstrates how to build a voice bot that joins a Stream video call, transcribes participants with
+Deepgram STT, and speaks responses with Palabra AI TTS.
+
+## What it does
+
+- Creates a voice bot that joins a Stream video call
+- Uses Deepgram for realtime STT and turn detection
+- Uses Palabra for streaming TTS responses
+- Uses Gemini for the LLM response
+
+## Prerequisites
+
+1. **Stream Account**: Get your API credentials from [Stream Dashboard](https://getstream.io/try-for-free/?utm_source=github.com&utm_medium=referral&utm_campaign=vision_agents)
+2. **Palabra Account**: Create an API key at [platform.palabra.ai/api-keys](https://platform.palabra.ai/api-keys)
+3. **Deepgram Account**: Set a `DEEPGRAM_API_KEY` for STT.
+4. **Google AI Account**: Set a `GOOGLE_API_KEY` for the example LLM.
+5. **Python 3.10+**: Required for running the example
+
+## Installation
+
+You can use your preferred package manager, but we recommend [`uv`](https://docs.astral.sh/uv/).
+
+1. **Navigate to this directory:**
+   ```bash
+   cd plugins/palabra/example
+   ```
+
+2. **Install dependencies:**
+   ```bash
+   uv sync
+   ```
+
+3. **Set up environment variables:**
+   Copy `.env.example` to `.env` and fill in `STREAM_API_KEY`, `STREAM_API_SECRET`, `PALABRA_API_KEY`,
+   `DEEPGRAM_API_KEY`, and `GOOGLE_API_KEY`.
+
+## Usage
+
+Run the voice bot:
+
+```bash
+uv run main.py run
+```
+
+Join the generated call, speak into your microphone, and the bot should answer out loud.
+
+## Checking TTS on its own
+
+`tts_smoke.py` drives the plugin without a call: it synthesizes a few sentences over one WebSocket session, prints the
+time to first audio chunk for each, and writes the result to `palabra_smoke.wav`.
+
+```bash
+uv run tts_smoke.py
+uv run tts_smoke.py "Anything else you would like to hear"
+```
+
+Only `PALABRA_API_KEY` is needed for this one.
+
+## Cloning a voice
+
+`clone_voice.py` clones a voice and then speaks with it. Pass your own recording, or pass nothing and it reads a passage
+with Palabra's stock voice to produce the sample itself, so the script works with no extra files:
+
+```bash
+uv run clone_voice.py                    # synthesize a sample, then clone it
+uv run clone_voice.py my_recording.wav   # clone from your own recording
+uv run clone_voice.py --keep             # don't delete the voice afterwards
+```
+
+Palabra needs at least 30 seconds of clean, single-speaker audio. Cloned voices count against your quota, so the script
+deletes the voice on the way out unless you pass `--keep`. Only clone someone's voice with their explicit consent.
diff --git a/plugins/palabra/example/clone_voice.py b/plugins/palabra/example/clone_voice.py
new file mode 100644
index 000000000..f1e582c02
--- /dev/null
+++ b/plugins/palabra/example/clone_voice.py
@@ -0,0 +1,126 @@
+#!/usr/bin/env python3
+"""
+Clone a voice with Palabra, then speak with it.
+
+Palabra needs at least 30 seconds of clean, single-speaker audio to clone from.
+Pass your own recording, or pass nothing and the script records a sample by
+reading a passage with Palabra's stock voice — which makes the example
+self-contained, at the cost of cloning a synthetic voice rather than a human one.
+
+Usage::
+    uv run clone_voice.py                      # synthesize a sample, then clone it
+    uv run clone_voice.py my_recording.wav     # clone from your own recording
+    uv run clone_voice.py --keep               # don't delete the voice afterwards
+
+Requires ``PALABRA_API_KEY`` (see `.env.example`). Cloned voices count against
+your account quota, so the script deletes the voice on the way out unless you
+pass ``--keep``.
+
+Only clone someone's voice with their explicit consent.
+"""
+
+import asyncio
+import sys
+import wave
+from pathlib import Path
+
+from dotenv import load_dotenv
+from vision_agents.plugins import palabra
+
+load_dotenv()
+
+SAMPLE_PATH = Path("palabra_voice_sample.wav")
+OUTPUT_PATH = Path("palabra_cloned_voice.wav")
+
+# ~40 seconds of speech: comfortably over Palabra's 30 second minimum.
+SAMPLE_SCRIPT = [
+    "Every morning the harbour wakes slowly, one boat at a time.",
+    "The fishermen speak in short sentences, mostly about the weather.",
+    "By seven the market is loud, and the gulls have taken the high walls.",
+    "A woman sells coffee from a cart she has pushed to the same corner for years.",
+    "She knows every regular by the way they hold their cup.",
+    "Later the tide turns and the water goes flat and grey.",
+    "Children run along the pier, daring each other to look over the edge.",
+    "In the evening the boats come back heavier than they left.",
+    "Someone always sings on the way in, badly, and nobody minds.",
+    "The harbour sleeps again before the town does.",
+]
+
+
+async def write_sample(tts: palabra.TTS, path: Path) -> float:
+    """Synthesize the passage into a WAV file and return its duration."""
+    audio = bytearray()
+    for line in SAMPLE_SCRIPT:
+        async for chunk in tts.send_iter(line):
+            if chunk.data is not None:
+                audio += chunk.data.samples.tobytes()
+
+    with wave.open(str(path), "wb") as wav:
+        wav.setnchannels(1)
+        wav.setsampwidth(2)
+        wav.setframerate(tts.sample_rate)
+        wav.writeframes(audio)
+    return len(audio) / 2 / tts.sample_rate
+
+
+async def speak(voice_id: str, path: Path) -> float:
+    """Say a line with the cloned voice and write it to ``path``."""
+    tts = palabra.TTS(voice_id=voice_id, deaccent_strength=0.7)
+    audio = bytearray()
+    try:
+        async for chunk in tts.send_iter(
+            "This is my cloned voice, generated from a short audio sample."
+        ):
+            if chunk.data is not None:
+                audio += chunk.data.samples.tobytes()
+        rate = tts.sample_rate
+    finally:
+        await tts.close()
+
+    with wave.open(str(path), "wb") as wav:
+        wav.setnchannels(1)
+        wav.setsampwidth(2)
+        wav.setframerate(rate)
+        wav.writeframes(audio)
+    return len(audio) / 2 / rate
+
+
+async def main() -> None:
+    args = [a for a in sys.argv[1:] if a != "--keep"]
+    keep = "--keep" in sys.argv
+    sample = Path(args[0]) if args else SAMPLE_PATH
+
+    async with palabra.Voices() as voices:
+        quota = await voices.limits()
+        print(f"voice quota: {quota.total}/{quota.limit} used, {quota.remaining} left")
+        if quota.remaining < 1:
+            print("no quota left — delete a voice first (voices.delete(voice_id))")
+            return
+
+        if not args:
+            tts = palabra.TTS()
+            try:
+                duration = await write_sample(tts, sample)
+            finally:
+                await tts.close()
+            print(f"recorded {duration:.1f}s sample -> {sample}")
+
+        print("cloning (this takes a moment)...")
+        voice = await voices.clone("Vision Agents demo", sample, lang_code="en")
+        print(f"voice {voice.voice_id} is {voice.processing_status}")
+        if voice.warnings:
+            print(f"warnings: {voice.warnings}")
+
+        try:
+            duration = await speak(voice.voice_id, OUTPUT_PATH)
+            print(f"spoke {duration:.1f}s with the cloned voice -> {OUTPUT_PATH}")
+        finally:
+            if keep:
+                print(f"keeping voice {voice.voice_id}")
+            else:
+                await voices.delete(voice.voice_id)
+                print(f"deleted voice {voice.voice_id}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
diff --git a/plugins/palabra/example/main.py b/plugins/palabra/example/main.py
new file mode 100644
index 000000000..3033824de
--- /dev/null
+++ b/plugins/palabra/example/main.py
@@ -0,0 +1,72 @@
+#!/usr/bin/env python3
+"""
+Example: Text-to-Speech with Palabra AI using the Agent class
+
+This minimal example shows how to:
+1. Create an Agent with Deepgram STT, a Gemini LLM and Palabra TTS
+2. Join a Stream video call
+3. Greet users and respond to spoken input
+
+Palabra TTS is a streaming plugin: the agent forwards each finished sentence of
+the LLM response over an already-open WebSocket, so the bot starts speaking
+before the model has finished writing.
+
+Run it, join the call in your browser, and speak to the bot.
+
+Usage::
+    uv run main.py run
+
+The script looks for the following env vars (see `.env.example`):
+    STREAM_API_KEY / STREAM_API_SECRET
+    PALABRA_API_KEY
+    DEEPGRAM_API_KEY
+    GOOGLE_API_KEY
+"""
+
+import asyncio
+import logging
+
+from dotenv import load_dotenv
+from vision_agents.core import Runner
+from vision_agents.core.agents import Agent, AgentLauncher
+from vision_agents.core.edge.types import User
+from vision_agents.plugins import deepgram, gemini, getstream, palabra
+
+logger = logging.getLogger(__name__)
+
+load_dotenv()
+
+
+async def create_agent(**kwargs) -> Agent:
+    """Create an agent with Deepgram STT, a Gemini LLM and Palabra TTS."""
+    agent = Agent(
+        edge=getstream.Edge(),
+        agent_user=User(name="Palabra Voice Bot", id="agent"),
+        instructions=(
+            "You're a helpful voice AI assistant. "
+            "Keep replies short and conversational."
+        ),
+        stt=deepgram.STT(),
+        llm=gemini.LLM(),
+        tts=palabra.TTS(voice_id="default_low", language="en"),
+    )
+
+    return agent
+
+
+async def join_call(agent: Agent, call_type: str, call_id: str, **kwargs) -> None:
+    call = await agent.create_call(call_type, call_id)
+
+    logger.info("Starting Palabra TTS voice bot")
+
+    async with agent.join(call):
+        logger.info("Joined call")
+        await asyncio.sleep(3)
+        await agent.simple_response(
+            "Hello! I'm listening. What would you like to talk about?"
+        )
+        await agent.finish()
+
+
+if __name__ == "__main__":
+    Runner(AgentLauncher(create_agent=create_agent, join_call=join_call)).cli()
diff --git a/plugins/palabra/example/palabra_smoke.wav b/plugins/palabra/example/palabra_smoke.wav
new file mode 100644
index 000000000..79982e704
Binary files /dev/null and b/plugins/palabra/example/palabra_smoke.wav differ
diff --git a/plugins/palabra/example/pyproject.toml b/plugins/palabra/example/pyproject.toml
new file mode 100644
index 000000000..175fa9947
--- /dev/null
+++ b/plugins/palabra/example/pyproject.toml
@@ -0,0 +1,20 @@
+[project]
+name = "palabra-example"
+version = "0.0.0"
+requires-python = ">=3.10"
+
+dependencies = [
+    "vision-agents",
+    "python-dotenv>=1.0",
+    "vision-agents-plugins-palabra",
+    "vision-agents-plugins-deepgram",
+    "vision-agents-plugins-getstream",
+    "vision-agents-plugins-gemini",
+]
+
+[tool.uv.sources]
+"vision-agents" = { path = "../../../agents-core", editable = true }
+"vision-agents-plugins-palabra" = { path = "..", editable = true }
+"vision-agents-plugins-deepgram" = { path = "../../deepgram", editable = true }
+"vision-agents-plugins-getstream" = { path = "../../getstream", editable = true }
+"vision-agents-plugins-gemini" = { path = "../../gemini", editable = true }
diff --git a/plugins/palabra/example/tts_smoke.py b/plugins/palabra/example/tts_smoke.py
new file mode 100644
index 000000000..88d51b797
--- /dev/null
+++ b/plugins/palabra/example/tts_smoke.py
@@ -0,0 +1,80 @@
+#!/usr/bin/env python3
+"""
+Smoke test: synthesize a few sentences with Palabra TTS and write a WAV file.
+
+Unlike the full agent example, this does not join a Stream call – it drives
+``palabra.TTS`` directly so you can confirm the API key works, listen to the
+result, and see the latency you get.
+
+It also shows the reason the plugin keeps one WebSocket open: ``start()`` pays
+for the handshake and session ``init`` up front, and every sentence after that
+reuses the same connection. The reported time-to-first-audio is therefore
+synthesis latency only, not connection setup.
+
+Usage::
+    uv run tts_smoke.py
+    uv run tts_smoke.py "Some other text to speak"
+
+Requires ``PALABRA_API_KEY`` (see `.env.example`). The output is written to
+``palabra_smoke.wav`` (mono, 16-bit PCM).
+"""
+
+import asyncio
+import sys
+import time
+import wave
+
+from dotenv import load_dotenv
+from vision_agents.plugins import palabra
+
+load_dotenv()
+
+OUTPUT_PATH = "palabra_smoke.wav"
+
+SENTENCES = [
+    "The sun was setting over the mountains, casting long golden shadows.",
+    "Birds were returning to their nests, filling the air with evening songs.",
+    "A gentle breeze moved through the tall grass toward the horizon.",
+]
+
+
+async def main() -> None:
+    sentences = sys.argv[1:] or SENTENCES
+
+    tts = palabra.TTS()
+    # Open the WebSocket before the first sentence so the handshake and the
+    # session `init` are not counted in the latency below.
+    await tts.start()
+
+    audio = bytearray()
+    try:
+        for sentence in sentences:
+            started = time.perf_counter()
+            first_chunk_at = None
+            chunks = 0
+
+            async for chunk in tts.send_iter(sentence):
+                if chunk.data is None:
+                    continue
+                if first_chunk_at is None:
+                    first_chunk_at = time.perf_counter() - started
+                audio += chunk.data.samples.tobytes()
+                chunks += 1
+
+            latency = f"{first_chunk_at * 1000:.0f}ms" if first_chunk_at else "n/a"
+            print(f"{sentence[:48]!r}: {chunks} chunks, first audio in {latency}")
+    finally:
+        await tts.close()
+
+    with wave.open(OUTPUT_PATH, "wb") as wav:
+        wav.setnchannels(1)
+        wav.setsampwidth(2)
+        wav.setframerate(tts.sample_rate)
+        wav.writeframes(audio)
+
+    duration = len(audio) / 2 / tts.sample_rate
+    print(f"wrote {duration:.2f}s of audio to {OUTPUT_PATH}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
diff --git a/plugins/palabra/py.typed b/plugins/palabra/py.typed
new file mode 100644
index 000000000..e69de29bb
diff --git a/plugins/palabra/pyproject.toml b/plugins/palabra/pyproject.toml
new file mode 100644
index 000000000..46ff4f912
--- /dev/null
+++ b/plugins/palabra/pyproject.toml
@@ -0,0 +1,41 @@
+[build-system]
+requires = ["hatchling", "hatch-vcs"]
+build-backend = "hatchling.build"
+
+[project]
+name = "vision-agents-plugins-palabra"
+dynamic = ["version"]
+description = "Palabra AI TTS integration for Vision Agents"
+readme = "README.md"
+keywords = ["palabra", "TTS", "text-to-speech", "AI", "voice agents", "agents"]
+requires-python = ">=3.10"
+license = "MIT"
+dependencies = [
+    "vision-agents",
+    "websockets>=15.0,<16",
+    "httpx>=0.28,<1",
+]
+
+[project.urls]
+Documentation = "https://visionagents.ai/"
+Website = "https://visionagents.ai/"
+Source = "https://github.com/GetStream/Vision-Agents"
+
+[tool.hatch.version]
+source = "vcs"
+raw-options = { root = "..", search_parent_directories = true, fallback_version = "0.0.0" }
+
+[tool.hatch.build.targets.wheel]
+packages = ["vision_agents"]
+
+[tool.hatch.build.targets.sdist]
+include = ["/vision_agents"]
+
+[tool.uv.sources]
+vision-agents = { workspace = true }
+
+[dependency-groups]
+dev = [
+    "pytest>=8.4.1",
+    "pytest-asyncio>=1.0.0",
+]
diff --git a/plugins/palabra/tests/__init__.py b/plugins/palabra/tests/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/plugins/palabra/tests/test_tts.py b/plugins/palabra/tests/test_tts.py
new file mode 100644
index 000000000..29f56f0b7
--- /dev/null
+++ b/plugins/palabra/tests/test_tts.py
@@ -0,0 +1,196 @@
+import asyncio
+import json
+import os
+from typing import AsyncIterator
+
+import pytest
+from dotenv import load_dotenv
+from vision_agents.plugins import palabra
+from vision_agents.plugins.palabra.tts import MAX_TEXT_LENGTH, _split_text
+
+load_dotenv()
+
+
+class TestPalabraTTS:
+    def test_defaults(self) -> None:
+        tts = palabra.TTS(api_key="fake")
+        assert tts.voice_id == "default_low"
+        assert tts.language == "en"
+        assert tts.model == "auto"
+        assert tts.sample_rate == 24000
+        assert tts.streaming is True
+
+    def test_missing_api_key_raises(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.delenv("PALABRA_API_KEY", raising=False)
+        with pytest.raises(ValueError):
+            palabra.TTS()
+
+    def test_out_of_range_sample_rate_raises(self) -> None:
+        with pytest.raises(ValueError):
+            palabra.TTS(api_key="fake", sample_rate=96000)
+
+    def test_out_of_range_speed_raises(self) -> None:
+        with pytest.raises(ValueError):
+            palabra.TTS(api_key="fake", speed=3.0)
+
+    def test_out_of_range_deaccent_strength_raises(self) -> None:
+        with pytest.raises(ValueError):
+            palabra.TTS(api_key="fake", deaccent_strength=1.5)
+
+    def test_non_positive_idle_timeout_raises(self) -> None:
+        with pytest.raises(ValueError):
+            palabra.TTS(api_key="fake", idle_timeout=0)
+
+    def test_init_message_requests_pcm_with_configured_voice(self) -> None:
+        tts = palabra.TTS(
+            api_key="fake",
+            voice_id="default_high",
+            language="de",
+            sample_rate=16000,
+            speed=1.2,
+            deaccent_strength=0.5,
+        )
+        message = json.loads(tts._init_message)
+
+        assert message == {
+            "type": "init",
+            "language": "de",
+            "model": "auto",
+            "voice_options": {
+                "voice_id": "default_high",
+                "speed": 1.2,
+                "deaccent_strength": 0.5,
+            },
+            "output": {"format": "pcm", "sample_rate": 16000},
+        }
+
+    def test_init_message_omits_unset_voice_options(self) -> None:
+        tts = palabra.TTS(api_key="fake")
+        message = json.loads(tts._init_message)
+
+        assert message["voice_options"] == {"voice_id": "default_low"}
+
+    def test_short_text_is_sent_as_a_single_message(self) -> None:
+        assert _split_text("Hello there.") == ["Hello there."]
+
+    def test_long_text_is_split_on_word_boundaries(self) -> None:
+        text = " ".join(["word"] * 600)
+        chunks = _split_text(text)
+
+        assert len(chunks) > 1
+        assert all(len(chunk) <= MAX_TEXT_LENGTH for chunk in chunks)
+        assert " ".join(chunks) == text
+
+    def test_long_text_without_spaces_is_split_at_the_limit(self) -> None:
+        text = "a" * (MAX_TEXT_LENGTH + 10)
+        chunks = _split_text(text)
+
+        assert chunks == ["a" * MAX_TEXT_LENGTH, "a" * 10]
+
+
+@pytest.mark.skipif(
+    os.getenv("PALABRA_API_KEY") is None, reason="PALABRA_API_KEY not set"
+)
+@pytest.mark.integration
+class TestPalabraTTSIntegration:
+    @pytest.fixture
+    async def tts(self) -> AsyncIterator[palabra.TTS]:
+        tts = palabra.TTS()
+        yield tts
+        await tts.close()
+
+    async def test_convert_text_to_audio(self, tts: palabra.TTS) -> None:
+        out = []
+        async for item in tts.send_iter("Hello from Palabra AI."):
+            out.append(item)
+
+        assert len(out) > 0
+        assert out[0].data
+        assert out[0].data.sample_rate == tts.sample_rate
+        assert out[-1].final
+
+    async def test_synthesizes_text_longer_than_one_message(
+        self, tts: palabra.TTS
+    ) -> None:
+        text = "The quick brown fox jumps over the lazy dog. " * 40
+        assert len(text) > MAX_TEXT_LENGTH
+
+        chunks = [pcm async for pcm in await tts.stream_audio(text)]
+
+        assert len(chunks) > 0
+
+    async def test_stop_audio_keeps_session_usable(self, tts: palabra.TTS) -> None:
+        long_text = (
+            "This is a fairly long sentence that the server should synthesize "
+            "across many audio chunks before it completes. " * 4
+        )
+
+        chunks = []
+        async for pcm in await tts.stream_audio(long_text):
+            chunks.append(pcm)
+            if len(chunks) == 2:
+                await tts.stop_audio()
+
+        assert len(chunks) >= 2
+
+        follow_up = [pcm async for pcm in await tts.stream_audio("Hello again.")]
+        assert len(follow_up) > 0
+
+    async def test_stop_audio_releases_a_reader_waiting_on_the_server(
+        self, tts: palabra.TTS
+    ) -> None:
+        """Barge-in arrives on a different task than the one reading audio.
+
+        Palabra does not acknowledge `cancel`, so the reader has to be released
+        by the plugin itself or the agent stays stuck mid-utterance.
+        """
+        long_text = (
+            "This is a fairly long sentence that the server should synthesize "
+            "across many audio chunks before it completes. " * 4
+        )
+
+        async def drain() -> int:
+            count = 0
+            async for _ in await tts.stream_audio(long_text):
+                count += 1
+            return count
+
+        reader = asyncio.ensure_future(drain())
+        await asyncio.sleep(0.5)
+        await tts.stop_audio()
+
+        await asyncio.wait_for(reader, timeout=5)
+
+        follow_up = [pcm async for pcm in await tts.stream_audio("Hello again.")]
+        assert len(follow_up) > 0
+
+    async def test_idle_timeout_ends_a_generation_that_never_produces_audio(
+        self,
+    ) -> None:
+        """A wedged generation must not stall the pipeline forever.
+
+        Driven with an unreachably short idle timeout so the guard trips on the
+        first read instead of waiting on a real server fault.
+        """
+        tts = palabra.TTS(idle_timeout=0.001)
+        try:
+            chunks = [pcm async for pcm in await tts.stream_audio("Hello there.")]
+            assert chunks == []
+
+            tts._idle_timeout = 5.0
+            recovered = [pcm async for pcm in await tts.stream_audio("Hello again.")]
+            assert len(recovered) > 0
+        finally:
+            await tts.close()
+
+    async def test_reuses_a_single_connection_across_utterances(
+        self, tts: palabra.TTS
+    ) -> None:
+        async for _ in await tts.stream_audio("First utterance."):
+            pass
+        websocket = tts._websocket
+
+        async for _ in await tts.stream_audio("Second utterance."):
+            pass
+
+        assert tts._websocket is websocket
diff --git a/plugins/palabra/tests/test_voices.py b/plugins/palabra/tests/test_voices.py
new file mode 100644
index 000000000..a78e73afc
--- /dev/null
+++ b/plugins/palabra/tests/test_voices.py
@@ -0,0 +1,174 @@
+import os
+import wave
+from pathlib import Path
+from typing import AsyncIterator
+
+import pytest
+from dotenv import load_dotenv
+from vision_agents.plugins import palabra
+from vision_agents.plugins.palabra.voices import (
+    MAX_SAMPLE_BYTES,
+    PalabraVoiceError,
+    describe_problems,
+)
+
+load_dotenv()
+
+# Long enough to satisfy Palabra's 30 second minimum when cloned for real.
+SAMPLE_SCRIPT = [
+    "The harbour wakes slowly every morning, one boat at a time.",
+    "The fishermen speak in short sentences, mostly about the weather.",
+    "By seven the market is loud and the gulls have taken the high walls.",
+    "A woman sells coffee from a cart on the same corner every day.",
+    "She knows every regular by the way they hold their cup.",
+    "Later the tide turns and the water goes flat and grey.",
+    "Children run along the pier, daring each other to look over the edge.",
+    "In the evening the boats come back heavier than they left.",
+    "Someone always sings on the way in, badly, and nobody minds.",
+    "The harbour sleeps again before the town does.",
+]
+
+
+class TestVoices:
+    def test_missing_api_key_raises(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.delenv("PALABRA_API_KEY", raising=False)
+        with pytest.raises(ValueError):
+            palabra.Voices()
+
+    def test_trailing_slash_is_stripped_from_base_url(self) -> None:
+        voices = palabra.Voices(api_key="fake", base_url="https://example.com/")
+        assert voices.base_url == "https://example.com"
+
+    async def test_clone_rejects_a_missing_sample(self) -> None:
+        voices = palabra.Voices(api_key="fake")
+        with pytest.raises(ValueError, match="not found"):
+            await voices.clone("Missing", "does-not-exist.wav")
+
+    async def test_clone_rejects_an_unsupported_format(self, tmp_path: Path) -> None:
+        sample = tmp_path / "sample.ogg"
+        sample.write_bytes(b"not audio")
+
+        voices = palabra.Voices(api_key="fake")
+        with pytest.raises(ValueError, match="Unsupported voice sample format"):
+            await voices.clone("Wrong format", sample)
+
+    async def test_clone_rejects_an_oversized_sample(self, tmp_path: Path) -> None:
+        sample = tmp_path / "sample.wav"
+        sample.write_bytes(b"\0" * (MAX_SAMPLE_BYTES + 1))
+
+        voices = palabra.Voices(api_key="fake")
+        with pytest.raises(ValueError, match="at most"):
+            await voices.clone("Too big", sample)
+
+    def test_problems_are_summarized_from_palabra_error_objects(self) -> None:
+        """Palabra reports errors as RFC 7807 objects, not strings."""
+        assert (
+            describe_problems(
+                [
+                    {"title": "Failed Dependency", "detail": "Depended action failed."},
+                    {"title": "Only a title"},
+                ]
+            )
+            == "Depended action failed.; Only a title"
+        )
+        assert describe_problems([]) == ""
+
+    def test_ready_reflects_processing_status(self) -> None:
+        pending = palabra.ClonedVoice(
+            voice_id="v1", name="v", processing_status="pending", errors=[], warnings=[]
+        )
+        ready = palabra.ClonedVoice(
+            voice_id="v1", name="v", processing_status="ready", errors=[], warnings=[]
+        )
+
+        assert pending.ready is False
+        assert ready.ready is True
+
+
+@pytest.mark.skipif(
+    os.getenv("PALABRA_API_KEY") is None, reason="PALABRA_API_KEY not set"
+)
+@pytest.mark.integration
+class TestVoicesIntegration:
+    @pytest.fixture
+    async def voices(self) -> AsyncIterator[palabra.Voices]:
+        async with palabra.Voices() as voices:
+            yield voices
+
+    async def test_limits_reports_the_account_quota(
+        self, voices: palabra.Voices
+    ) -> None:
+        limits = await voices.limits()
+
+        assert limits.limit > 0
+        assert limits.remaining == limits.limit - limits.total
+
+    async def test_list_returns_cloned_voices(self, voices: palabra.Voices) -> None:
+        listed = await voices.list(page_size=5)
+
+        assert isinstance(listed, list)
+        assert all(voice.voice_id for voice in listed)
+
+    @pytest.mark.timeout(300)
+    async def test_a_rejected_sample_does_not_strand_quota(
+        self, voices: palabra.Voices, tmp_path: Path
+    ) -> None:
+        """Palabra reserves the voice before the sample is judged.
+
+        A sample it cannot use must leave the quota exactly as it was, otherwise
+        repeated failures eventually block cloning altogether.
+        """
+        sample = tmp_path / "too_short.wav"
+        with wave.open(str(sample), "wb") as wav:
+            wav.setnchannels(1)
+            wav.setsampwidth(2)
+            wav.setframerate(24000)
+            wav.writeframes(b"\0" * 2 * 24000)  # 1s of silence, well under 30s
+
+        before = await voices.limits()
+        with pytest.raises(PalabraVoiceError):
+            await voices.clone("Rejected sample", sample, timeout=180, poll_interval=5)
+
+        assert (await voices.limits()).total == before.total
+
+    @pytest.mark.timeout(300)
+    async def test_clone_produces_a_voice_usable_for_synthesis(
+        self, voices: palabra.Voices, tmp_path: Path
+    ) -> None:
+        """Full round trip: record a sample, clone it, speak with it, delete it."""
+        sample = tmp_path / "sample.wav"
+        tts = palabra.TTS()
+        audio = bytearray()
+        try:
+            for line in SAMPLE_SCRIPT:
+                async for chunk in tts.send_iter(line):
+                    if chunk.data is not None:
+                        audio += chunk.data.samples.tobytes()
+            rate = tts.sample_rate
+        finally:
+            await tts.close()
+
+        with wave.open(str(sample), "wb") as wav:
+            wav.setnchannels(1)
+            wav.setsampwidth(2)
+            wav.setframerate(rate)
+            wav.writeframes(audio)
+        assert len(audio) / 2 / rate > 30
+
+        voice = await voices.clone("Vision Agents test", sample, lang_code="en")
+        try:
+            assert voice.ready
+            assert (await voices.get(voice.voice_id)).voice_id == voice.voice_id
+
+            cloned_tts = palabra.TTS(voice_id=voice.voice_id)
+            try:
+                chunks = [
+                    pcm async for pcm in await cloned_tts.stream_audio("Hello again.")
+                ]
+            finally:
+                await cloned_tts.close()
+            assert len(chunks) > 0
+        finally:
+            await voices.delete(voice.voice_id)
+
+        assert all(v.voice_id != voice.voice_id for v in await voices.list())
diff --git a/plugins/palabra/vision_agents/plugins/palabra/__init__.py b/plugins/palabra/vision_agents/plugins/palabra/__init__.py
new file mode 100644
index 000000000..c809ed0c3
--- /dev/null
+++ b/plugins/palabra/vision_agents/plugins/palabra/__init__.py
@@ -0,0 +1,16 @@
+from .tts import TTS, WS_URL_EU, WS_URL_US, PalabraTTSError
+from .voices import ClonedVoice, PalabraVoiceError, VoiceLimits, Voices
+
+# Re-export under the new namespace for convenience
+__path__ = __import__("pkgutil").extend_path(__path__, __name__)
+
+__all__ = [
+    "TTS",
+    "WS_URL_EU",
+    "WS_URL_US",
+    "ClonedVoice",
+    "PalabraTTSError",
+    "PalabraVoiceError",
+    "VoiceLimits",
+    "Voices",
+]
diff --git a/plugins/palabra/vision_agents/plugins/palabra/tts.py b/plugins/palabra/vision_agents/plugins/palabra/tts.py
new file mode 100644
index 000000000..ac0d9860e
--- /dev/null
+++ b/plugins/palabra/vision_agents/plugins/palabra/tts.py
@@ -0,0 +1,334 @@
+"""Palabra AI Text-to-Speech via the realtime WebSocket API.
+
+Docs: https://platform.palabra.ai/docs/text-to-speech/realtime-tts
+
+A single WebSocket stays open across ``stream_audio`` calls: the session is
+initialised once, then every utterance is a ``text`` message tagged with its own
+``generation_id``. Audio chunks echo that id, so chunks left over from a
+cancelled utterance are dropped without reconnecting.
+"""
+
+import asyncio
+import base64
+import json
+import logging
+import os
+import uuid
+from typing import AsyncIterator, Optional
+
+import websockets
+import websockets.exceptions
+from getstream.video.rtc.track_util import AudioFormat, PcmData
+from vision_agents.core import tts
+
+logger = logging.getLogger(__name__)
+
+WS_URL_EU = "wss://stream.palabra.ai/tts-api/v1/text-to-speech/stream"
+WS_URL_US = "wss://stream.us.palabra.ai/tts-api/v1/text-to-speech/stream"
+
+# Palabra rejects text messages above this length, so longer utterances are
+# split across several messages.
+MAX_TEXT_LENGTH = 1024
+
+CANCEL_MESSAGE = json.dumps({"type": "cancel"})
+
+
+class PalabraTTSError(Exception):
+    """Raised when Palabra reports an error over the WebSocket."""
+
+    def __init__(self, code: str, desc: str) -> None:
+        super().__init__(f"{code}: {desc}")
+        self.code = code
+        self.desc = desc
+
+
+def _split_text(text: str) -> list[str]:
+    """Split ``text`` into messages of at most ``MAX_TEXT_LENGTH`` characters."""
+    if len(text) <= MAX_TEXT_LENGTH:
+        return [text]
+
+    chunks: list[str] = []
+    remaining = text
+    while len(remaining) > MAX_TEXT_LENGTH:
+        split_at = remaining.rfind(" ", 0, MAX_TEXT_LENGTH + 1)
+        if split_at <= 0:
+            split_at = MAX_TEXT_LENGTH
+        chunks.append(remaining[:split_at])
+        remaining = remaining[split_at:].lstrip()
+    if remaining:
+        chunks.append(remaining)
+    return chunks
+
+
+class TTS(tts.TTS):
+    """Palabra AI streaming Text-to-Speech."""
+
+    streaming = True
+
+    def __init__(
+        self,
+        api_key: Optional[str] = None,
+        voice_id: str = "default_low",
+        language: str = "en",
+        model: str = "auto",
+        sample_rate: int = 24000,
+        speed: Optional[float] = None,
+        deaccent_strength: Optional[float] = None,
+        ws_url: str = WS_URL_EU,
+        idle_timeout: float = 5.0,
+    ) -> None:
+        """Initialize Palabra TTS.
+
+        Args:
+            api_key: Palabra API key. Falls back to ``PALABRA_API_KEY`` env var.
+            voice_id: Voice to synthesize with – ``default_low``, ``default_high``
+                or the id of a cloned voice.
+            language: BCP-47 language code of the text (e.g. ``en``, ``de``, ``pt-eu``).
+            model: TTS model id. ``auto`` lets Palabra pick.
+            sample_rate: Output sample rate in Hz (8000–48000).
+            speed: Speech speed multiplier (0.0–2.0). ``None`` uses the server default.
+            deaccent_strength: Accent reduction for cloned voices (0.0–1.0).
+                ``None`` uses the server default.
+            ws_url: Palabra TTS endpoint. Defaults to the EU region; pass
+                ``WS_URL_US`` for the US region.
+            idle_timeout: Seconds to wait for the next frame of a generation
+                before giving up on it. Chunks normally arrive ~200 ms apart, so
+                this only trips when the server stops answering without sending
+                an error, which would otherwise stall the whole TTS pipeline.
+        """
+        super().__init__(provider_name="palabra")
+
+        api_key = api_key or os.getenv("PALABRA_API_KEY")
+        if not api_key:
+            raise ValueError(
+                "PALABRA_API_KEY env var or api_key parameter required for Palabra TTS"
+            )
+        if not 8000 <= sample_rate <= 48000:
+            raise ValueError(
+                f"Palabra TTS sample_rate must be between 8000 and 48000 Hz; got {sample_rate}"
+            )
+        if speed is not None and not 0.0 <= speed <= 2.0:
+            raise ValueError(
+                f"Palabra TTS speed must be between 0.0 and 2.0; got {speed}"
+            )
+        if deaccent_strength is not None and not 0.0 <= deaccent_strength <= 1.0:
+            raise ValueError(
+                "Palabra TTS deaccent_strength must be between 0.0 and 1.0; "
+                f"got {deaccent_strength}"
+            )
+        if idle_timeout <= 0:
+            raise ValueError(
+                f"Palabra TTS idle_timeout must be greater than 0; got {idle_timeout}"
+            )
+
+        self.voice_id = voice_id
+        self.language = language
+        self.model = model
+        self.sample_rate = sample_rate
+
+        self._api_key = api_key
+        self._ws_url = ws_url
+        self._idle_timeout = idle_timeout
+
+        voice_options: dict[str, object] = {"voice_id": voice_id}
+        if speed is not None:
+            voice_options["speed"] = speed
+        if deaccent_strength is not None:
+            voice_options["deaccent_strength"] = deaccent_strength
+        # Settings apply to the whole session and cannot be changed once sent
+        # (the server answers CONFLICT), so the payload is built once here.
+        self._init_message = json.dumps(
+            {
+                "type": "init",
+                "language": language,
+                "model": model,
+                "voice_options": voice_options,
+                "output": {"format": "pcm", "sample_rate": sample_rate},
+            }
+        )
+
+        self._websocket: websockets.ClientConnection | None = None
+        self._pending_recv: asyncio.Future[bytes] | None = None
+        self._generation = 0
+
+    async def start(self) -> None:
+        """Open the WebSocket up front so the first utterance skips the handshake."""
+        await self._ensure_connection()
+
+    async def close(self) -> None:
+        """Close the WebSocket and release resources."""
+        await self._reset_connection()
+        self._on_disconnected()
+        await super().close()
+
+    async def stream_audio(self, text: str, *_, **__) -> AsyncIterator[PcmData]:
+        """Synthesize ``text`` over the persistent WebSocket.
+
+        Args:
+            text: The text to convert to speech.
+
+        Returns:
+            Async iterator yielding ``PcmData`` chunks.
+        """
+        generation_id = uuid.uuid4().hex
+        self._generation += 1
+        generation = self._generation
+
+        try:
+            websocket = await self._send_text(text, generation_id)
+        except (websockets.exceptions.WebSocketException, OSError):
+            logger.warning("Palabra TTS websocket dropped; reconnecting")
+            await self._reset_connection()
+            websocket = await self._send_text(text, generation_id)
+
+        return self._receive_audio(websocket, generation_id, generation)
+
+    async def stop_audio(self) -> None:
+        """Cancel in-flight synthesis. The session stays open for the next utterance."""
+        self._generation += 1
+
+        # Palabra does not confirm the cancel, so a reader parked in recv() would
+        # sit there until the *next* utterance produced a frame. Cancelling the
+        # pending read releases it now; websockets guarantees the connection
+        # stays usable and no message is lost. Detaching it first is how the
+        # reader tells our cancel apart from its own task being cancelled.
+        pending_recv = self._pending_recv
+        if pending_recv is not None:
+            self._pending_recv = None
+            pending_recv.cancel()
+
+        websocket = self._websocket
+        if websocket is None:
+            return
+        try:
+            await websocket.send(CANCEL_MESSAGE)
+        except (websockets.exceptions.WebSocketException, OSError):
+            await self._reset_connection()
+
+    async def _send_text(
+        self, text: str, generation_id: str
+    ) -> websockets.ClientConnection:
+        websocket = await self._ensure_connection()
+        chunks = _split_text(text)
+        last = len(chunks) - 1
+        for index, chunk in enumerate(chunks):
+            await websocket.send(
+                json.dumps(
+                    {
+                        "type": "text",
+                        "text": chunk,
+                        "generation_id": generation_id,
+                        "is_eos": index == last,
+                    }
+                )
+            )
+        return websocket
+
+    async def _receive_audio(
+        self,
+        websocket: websockets.ClientConnection,
+        generation_id: str,
+        generation: int,
+    ) -> AsyncIterator[PcmData]:
+        while self._generation == generation:
+            # decode=False keeps the frame as bytes, which json parses directly.
+            recv = asyncio.ensure_future(websocket.recv(decode=False))
+            self._pending_recv = recv
+            try:
+                message = await asyncio.wait_for(recv, self._idle_timeout)
+            except asyncio.CancelledError:
+                # stop_audio() detaches the read before cancelling it; anything
+                # still attached means our own task is being cancelled, which
+                # must propagate.
+                if self._pending_recv is not recv:
+                    return
+                raise
+            # Must precede OSError: asyncio.TimeoutError *is* builtin
+            # TimeoutError, which subclasses OSError.
+            except asyncio.TimeoutError:
+                logger.warning(
+                    "Palabra TTS sent no audio for %.1fs; abandoning generation %s",
+                    self._idle_timeout,
+                    generation_id,
+                )
+                return
+            except (websockets.exceptions.ConnectionClosed, OSError):
+                await self._reset_connection()
+                raise
+            finally:
+                if self._pending_recv is recv:
+                    self._pending_recv = None
+
+            try:
+                data = json.loads(message)
+            except json.JSONDecodeError:
+                logger.warning("Skipping non-JSON Palabra TTS websocket message")
+                continue
+
+            message_type = data.get("message_type")
+            payload = data.get("data") or {}
+
+            if message_type == "error":
+                raise PalabraTTSError(
+                    payload.get("code", "UNKNOWN_ERROR"), payload.get("desc", "")
+                )
+            if message_type != "audio_chunk":
+                continue
+            # Chunks of a cancelled or abandoned utterance the server is still
+            # draining onto the shared socket.
+            if payload.get("generation_id") != generation_id:
+                continue
+
+            audio = payload.get("audio")
+            if audio:
+                yield PcmData.from_bytes(
+                    base64.b64decode(audio),
+                    sample_rate=self.sample_rate,
+                    channels=1,
+                    format=AudioFormat.S16,
+                )
+            if payload.get("last_chunk"):
+                return
+
+    async def _ensure_connection(self) -> websockets.ClientConnection:
+        websocket = self._websocket
+        if websocket is not None and websocket.state is websockets.State.OPEN:
+            return websocket
+
+        await self._reset_connection()
+        websocket = await websockets.connect(
+            self._ws_url,
+            additional_headers={"Authorization": f"Bearer {self._api_key}"},
+            # Base64-encoded PCM barely compresses; deflating every audio frame
+            # only adds CPU work and latency on the playback path.
+            compression=None,
+        )
+        # Until it is stored, _reset_connection() cannot reach this socket, so a
+        # failed init would leak it.
+        try:
+            await websocket.send(self._init_message)
+        except (
+            asyncio.CancelledError,
+            websockets.exceptions.WebSocketException,
+            OSError,
+        ):
+            await websocket.close()
+            raise
+        self._websocket = websocket
+        self._on_connected()
+        logger.debug(
+            "Palabra TTS websocket connected at %dHz with voice %s",
+            self.sample_rate,
+            self.voice_id,
+        )
+        return websocket
+
+    async def _reset_connection(self) -> None:
+        websocket = self._websocket
+        self._websocket = None
+        if websocket is None:
+            return
+        try:
+            await websocket.close()
+        except (websockets.exceptions.WebSocketException, OSError):
+            logger.debug("Error closing Palabra TTS websocket")
diff --git a/plugins/palabra/vision_agents/plugins/palabra/voices.py b/plugins/palabra/vision_agents/plugins/palabra/voices.py
new file mode 100644
index 000000000..2c54a1ad4
--- /dev/null
+++ b/plugins/palabra/vision_agents/plugins/palabra/voices.py
@@ -0,0 +1,382 @@
+"""Palabra AI voice cloning.
+
+Docs: https://platform.palabra.ai/docs/assets/voices
+
+Cloning is a three step dance: ``POST /saas/voice/clone`` reserves a voice and
+returns a presigned upload target, the sample is uploaded there, and the voice
+then moves through ``created -> pending -> ready``. ``Voices.clone`` runs all
+three and hands back a ``voice_id`` usable as ``TTS(voice_id=...)``.
+"""
+
+import asyncio
+import logging
+import mimetypes
+import os
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Optional
+
+import httpx
+
+logger = logging.getLogger(__name__)
+
+API_BASE_URL = "https://api.palabra.ai"
+
+# Palabra's documented sample limits: at least 30 seconds of clean, single
+# speaker audio, at most 10 MB.
+MAX_SAMPLE_BYTES = 10 * 1024 * 1024
+SUPPORTED_SAMPLE_SUFFIXES = {
+    ".mp3",
+    ".wav",
+    ".flac",
+    ".webm",
+    ".mp4",
+    ".mpeg",
+    ".mpg",
+}
+
+READY = "ready"
+FAILED = "failed"
+
+
+class PalabraVoiceError(Exception):
+    """Raised when the Palabra voice API rejects a request or cloning fails."""
+
+
+@dataclass
+class ClonedVoice:
+    """A cloned voice. ``voice_id`` is what ``TTS(voice_id=...)`` expects."""
+
+    voice_id: str
+    name: str
+    processing_status: str
+    # Palabra reports these as RFC 7807 problem objects, not plain strings.
+    errors: list[dict]
+    warnings: list[dict]
+
+    @property
+    def ready(self) -> bool:
+        """True once the voice can be used for synthesis."""
+        return self.processing_status == READY
+
+
+@dataclass
+class VoiceLimits:
+    """Cloned voice quota for the account."""
+
+    total: int
+    limit: int
+    remaining: int
+    ready: int
+    pending: int
+    failed: int
+
+
+def _read_sample(path: Path) -> tuple[bytes, str]:
+    """Validate a sample and read it.
+
+    Runs in a worker thread: samples are up to 10 MB and the event loop is on a
+    realtime audio path.
+
+    Returns:
+        The file contents and its MIME type.
+    """
+    if not path.is_file():
+        raise ValueError(f"Voice sample not found: {path}")
+
+    suffix = path.suffix.lower()
+    if suffix not in SUPPORTED_SAMPLE_SUFFIXES:
+        raise ValueError(
+            f"Unsupported voice sample format '{suffix}'. "
+            f"Expected one of: {sorted(SUPPORTED_SAMPLE_SUFFIXES)}"
+        )
+
+    size = path.stat().st_size
+    if size > MAX_SAMPLE_BYTES:
+        raise ValueError(
+            f"Voice sample is {size} bytes; Palabra accepts at most {MAX_SAMPLE_BYTES}"
+        )
+
+    mime_type = mimetypes.guess_type(path.name)[0] or "application/octet-stream"
+    return path.read_bytes(), mime_type
+
+
+def describe_problems(problems: list[dict]) -> str:
+    """Summarize Palabra's problem objects into one human-readable line."""
+    return "; ".join(str(p.get("detail") or p.get("title") or p) for p in problems)
+
+
+def _parse_voice(data: dict) -> ClonedVoice:
+    result = data.get("processing_result") or {}
+    return ClonedVoice(
+        voice_id=data["voice_id"],
+        name=data.get("name", ""),
+        processing_status=data.get("processing_status", ""),
+        errors=list(result.get("errors") or []),
+        warnings=list(result.get("warnings") or []),
+    )
+
+
+class Voices:
+    """Client for Palabra's cloned voice API.
+
+    Example:
+        >>> async with palabra.Voices() as voices:
+        ...     voice = await voices.clone("Narrator", "sample.wav")
+        >>> tts = palabra.TTS(voice_id=voice.voice_id)
+    """
+
+    def __init__(
+        self,
+        api_key: Optional[str] = None,
+        base_url: str = API_BASE_URL,
+        client: Optional[httpx.AsyncClient] = None,
+    ) -> None:
+        """Initialize the voice API client.
+
+        Args:
+            api_key: Palabra API key. Falls back to ``PALABRA_API_KEY`` env var.
+            base_url: Palabra REST API base URL.
+            client: Optional pre-configured ``httpx.AsyncClient``.
+        """
+        api_key = api_key or os.getenv("PALABRA_API_KEY")
+        if not api_key:
+            raise ValueError(
+                "PALABRA_API_KEY env var or api_key parameter required for Palabra voices"
+            )
+
+        self.base_url = base_url.rstrip("/")
+        self._api_key = api_key
+        self._client = client or httpx.AsyncClient(timeout=60.0)
+        self._owns_client = client is None
+
+    async def clone(
+        self,
+        name: str,
+        sample: str | Path,
+        *,
+        lang_code: str = "en",
+        description: Optional[str] = None,
+        denoise: bool = False,
+        speech_normalization: bool = True,
+        labels: Optional[dict[str, str]] = None,
+        wait: bool = True,
+        timeout: float = 300.0,
+        poll_interval: float = 3.0,
+    ) -> ClonedVoice:
+        """Clone a voice from an audio sample.
+
+        Args:
+            name: Name to file the voice under.
+            sample: Path to the audio or video sample. Palabra wants at least
+                30 seconds of clean, single speaker audio, at most 10 MB.
+            lang_code: Language spoken in the sample (e.g. ``en``, ``uk``).
+            description: Optional free-form description.
+            denoise: Ask Palabra to denoise the sample before training.
+            speech_normalization: Normalize loudness of the sample.
+            labels: Optional metadata, e.g. ``{"gender": "Female"}``. Used for
+                filtering in ``list``.
+            wait: Poll until the voice is ready (or failed) before returning.
+            timeout: Seconds to wait for processing when ``wait`` is True.
+            poll_interval: Seconds between status polls.
+
+        Returns:
+            The cloned voice. When ``wait`` is False it is still processing and
+            ``ready`` will be False.
+
+        Raises:
+            PalabraVoiceError: If the API rejects the request, processing fails,
+                or processing does not finish within ``timeout``.
+        """
+        path = Path(sample)
+        data, mime_type = await asyncio.to_thread(_read_sample, path)
+
+        payload: dict[str, object] = {
+            "name": name,
+            "samples": [
+                {
+                    "filename": path.name,
+                    "mime_type": mime_type,
+                    "lang_code": lang_code,
+                    "speech_normalization": speech_normalization,
+                    "denoise": denoise,
+                }
+            ],
+        }
+        if description is not None:
+            payload["description"] = description
+        if labels is not None:
+            payload["labels"] = labels
+
+        # The API wraps request bodies in a `data` envelope, mirroring responses.
+        created = await self._request(
+            "POST", "/saas/voice/clone", json={"data": payload}
+        )
+        voice = _parse_voice(created)
+
+        samples = created.get("samples") or []
+        if not samples:
+            raise PalabraVoiceError(
+                f"Palabra returned no upload target for voice {voice.voice_id}"
+            )
+        # The voice is reserved and counts against the account quota from here
+        # on, so anything that goes wrong has to release it again.
+        try:
+            await self._upload_sample(samples[0], path.name, data, mime_type)
+            logger.debug("Uploaded sample for Palabra voice %s", voice.voice_id)
+
+            if not wait:
+                return voice
+            return await self._wait_until_processed(
+                voice.voice_id, timeout, poll_interval
+            )
+        except BaseException:
+            # Deliberately unfiltered: the quota is only 50 voices, and *any*
+            # escape from here - HTTP error, rejected sample, cancellation, or a
+            # bug in our own parsing - strands a reservation on the account.
+            await self._discard(voice.voice_id)
+            raise
+
+    async def get(self, voice_id: str) -> ClonedVoice:
+        """Fetch a single voice, including its processing status."""
+        return _parse_voice(await self._request("GET", f"/saas/voice/m/{voice_id}"))
+
+    async def list(
+        self,
+        *,
+        search: Optional[str] = None,
+        lang: Optional[str] = None,
+        page_size: Optional[int] = None,
+    ) -> list[ClonedVoice]:
+        """List cloned voices.
+
+        Args:
+            search: Case-insensitive name search (at least 2 characters).
+            lang: Filter by sample language code.
+            page_size: Voices per page (1–100). Only the first page is returned.
+        """
+        params: dict[str, str | int] = {}
+        if search is not None:
+            params["search"] = search
+        if lang is not None:
+            params["lang"] = lang
+        if page_size is not None:
+            params["page_size"] = page_size
+
+        data = await self._request("GET", "/saas/voice", params=params)
+        return [_parse_voice(item) for item in data.get("items") or []]
+
+    async def delete(self, voice_id: str) -> None:
+        """Permanently delete a cloned voice."""
+        await self._request("DELETE", f"/saas/voice/m/{voice_id}")
+
+    async def limits(self) -> VoiceLimits:
+        """Return the account's cloned voice quota."""
+        data = await self._request("GET", "/saas/voice/limits")
+        return VoiceLimits(
+            total=data["total"],
+            limit=data["limit"],
+            remaining=data["remaining"],
+            ready=data["ready"],
+            pending=data["pending"],
+            failed=data["failed"],
+        )
+
+    async def close(self) -> None:
+        """Close the HTTP client if this instance created it."""
+        if self._owns_client:
+            await self._client.aclose()
+
+    async def _upload_sample(
+        self, sample: dict, filename: str, data: bytes, mime_type: str
+    ) -> None:
+        """POST the sample to the presigned target Palabra handed back."""
+        url = sample.get("url")
+        if not url:
+            raise PalabraVoiceError("Palabra upload target is missing its URL")
+
+        response = await self._client.post(
+            url,
+            data=sample.get("form_data") or {},
+            files={"file": (filename, data, mime_type)},
+        )
+        if response.is_error:
+            raise PalabraVoiceError(
+                f"Uploading the voice sample failed with HTTP {response.status_code}: "
+                f"{response.text[:200]}"
+            )
+
+    async def _wait_until_processed(
+        self, voice_id: str, timeout: float, poll_interval: float
+    ) -> ClonedVoice:
+        deadline = asyncio.get_running_loop().time() + timeout
+        while True:
+            voice = await self.get(voice_id)
+            if voice.ready:
+                logger.debug("Palabra voice %s is ready", voice_id)
+                return voice
+            if voice.processing_status == FAILED:
+                raise PalabraVoiceError(
+                    f"Palabra failed to clone voice {voice_id}: "
+                    f"{describe_problems(voice.errors) or 'no reason given'}"
+                )
+            if asyncio.get_running_loop().time() >= deadline:
+                raise PalabraVoiceError(
+                    f"Palabra voice {voice_id} was still '{voice.processing_status}' "
+                    f"after {timeout:.0f}s"
+                )
+            await asyncio.sleep(poll_interval)
+
+    async def _discard(self, voice_id: str) -> None:
+        """Release a voice that never finished cloning, best effort."""
+        try:
+            await self.delete(voice_id)
+        except (PalabraVoiceError, httpx.HTTPError):
+            logger.exception("Could not delete incomplete Palabra voice %s", voice_id)
+
+    async def _request(
+        self,
+        method: str,
+        path: str,
+        *,
+        json: Optional[dict] = None,
+        params: Optional[dict] = None,
+    ) -> dict:
+        """Call the Palabra REST API and unwrap its ``{ok, data}`` envelope."""
+        response = await self._client.request(
+            method,
+            f"{self.base_url}{path}",
+            json=json,
+            params=params,
+            headers={"Authorization": f"Bearer {self._api_key}"},
+        )
+        if response.is_error:
+            raise PalabraVoiceError(
+                f"Palabra {method} {path} failed with HTTP {response.status_code}: "
+                f"{response.text[:200]}"
+            )
+
+        # DELETE-style endpoints may answer with no body at all.
+        if not response.content.strip():
+            return {}
+        try:
+            body = response.json()
+        except ValueError as exc:
+            raise PalabraVoiceError(
+                f"Palabra {method} {path} returned a non-JSON body: "
+                f"{response.text[:200]}"
+            ) from exc
+
+        if not isinstance(body, dict):
+            raise PalabraVoiceError(
+                f"Palabra {method} {path} returned {type(body).__name__}, expected an object"
+            )
+        if not body.get("ok", False):
+            raise PalabraVoiceError(f"Palabra {method} {path} returned {body}")
+        return body.get("data") or {}
+
+    async def __aenter__(self) -> "Voices":
+        return self
+
+    async def __aexit__(self, *_) -> None:
+        await self.close()
diff --git a/pyproject.toml b/pyproject.toml
index 18f7fd7a6..1970a4975 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -45,6 +45,7 @@ vision-agents-plugins-liveavatar = { workspace = true }
 vision-agents-plugins-tencent = { workspace = true }
 vision-agents-plugins-twelvelabs = { workspace = true }
 vision-agents-plugins-speechify = { workspace = true }
+vision-agents-plugins-palabra = { workspace = true }
 
 [tool.uv]
 # Workspace-level override to resolve numpy version conflicts
@@ -97,6 +98,7 @@ members = [
     "plugins/tencent",
     "plugins/twelvelabs",
     "plugins/speechify",
+    "plugins/palabra",
 ]
 exclude = [
     "**/__pycache__",
diff --git a/uv.lock b/uv.lock
index d0db9afe3..1585a1fa0 100644
--- a/uv.lock
+++ b/uv.lock
@@ -35,6 +35,7 @@ members = [
     "vision-agents-plugins-nvidia",
     "vision-agents-plugins-openai",
     "vision-agents-plugins-openrouter",
+    "vision-agents-plugins-palabra",
     "vision-agents-plugins-pocket",
     "vision-agents-plugins-qdrant",
     "vision-agents-plugins-qwen",
@@ -6511,6 +6512,9 @@ openai = [
 openrouter = [
     { name = "vision-agents-plugins-openrouter" },
 ]
+palabra = [
+    { name = "vision-agents-plugins-palabra" },
+]
 pocket = [
     { name = "vision-agents-plugins-pocket" },
 ]
@@ -6608,6 +6612,7 @@ requires-dist = [
     { name = "vision-agents-plugins-nvidia", marker = "extra == 'nvidia'", editable = "plugins/nvidia" },
     { name = "vision-agents-plugins-openai", marker = "extra == 'openai'", editable = "plugins/openai" },
     { name = "vision-agents-plugins-openrouter", marker = "extra == 'openrouter'", editable = "plugins/openrouter" },
+    { name = "vision-agents-plugins-palabra", marker = "extra == 'palabra'", editable = "plugins/palabra" },
     { name = "vision-agents-plugins-pocket", marker = "extra == 'pocket'", editable = "plugins/pocket" },
     { name = "vision-agents-plugins-qdrant", marker = "extra == 'qdrant'", editable = "plugins/qdrant" },
     { name = "vision-agents-plugins-qwen", marker = "extra == 'qwen'", editable = "plugins/qwen" },
@@ -6625,7 +6630,7 @@ requires-dist = [
     { name = "vision-agents-plugins-wizper", marker = "extra == 'wizper'", editable = "plugins/wizper" },
     { name = "vision-agents-plugins-xai", marker = "extra == 'xai'", editable = "plugins/xai" },
 ]
-provides-extras = ["anam", "anthropic", "assemblyai", "aws", "cartesia", "decart", "deepgram", "dev", "elevenlabs", "fast-whisper", "fish", "gemini", "getstream", "huggingface", "inworld", "kokoro", "lemonslice", "liveavatar", "local", "minimax", "mistral", "moondream", "nvidia", "openai", "openrouter", "pocket", "qdrant", "qwen", "redis", "roboflow", "sarvam", "smart-turn", "speechify", "telnyx", "tencent", "turbopuffer", "twelvelabs", "twilio", "ultralytics", "vogent", "wizper", "xai"]
+provides-extras = ["anam", "anthropic", "assemblyai", "aws", "cartesia", "decart", "deepgram", "dev", "elevenlabs", "fast-whisper", "fish", "gemini", "getstream", "huggingface", "inworld", "kokoro", "lemonslice", "liveavatar", "local", "minimax", "mistral", "moondream", "nvidia", "openai", "openrouter", "palabra", "pocket", "qdrant", "qwen", "redis", "roboflow", "sarvam", "smart-turn", "speechify", "telnyx", "tencent", "turbopuffer", "twelvelabs", "twilio", "ultralytics", "vogent", "wizper", "xai"]
 
 [[package]]
 name = "vision-agents-plugins-anam"
@@ -7329,6 +7334,34 @@ dev = [
     { name = "pytest-asyncio", specifier = ">=1.0.0" },
 ]
 
+[[package]]
+name = "vision-agents-plugins-palabra"
+source = { editable = "plugins/palabra" }
+dependencies = [
+    { name = "httpx" },
+    { name = "vision-agents" },
+    { name = "websockets" },
+]
+
+[package.dev-dependencies]
+dev = [
+    { name = "pytest" },
+    { name = "pytest-asyncio" },
+]
+
+[package.metadata]
+requires-dist = [
+    { name = "httpx", specifier = ">=0.28,<1" },
+    { name = "vision-agents", editable = "agents-core" },
+    { name = "websockets", specifier = ">=15.0,<16" },
+]
+
+[package.metadata.requires-dev]
+dev = [
+    { name = "pytest", specifier = ">=8.4.1" },
+    { name = "pytest-asyncio", specifier = ">=1.0.0" },
+]
+
 [[package]]
 name = "vision-agents-plugins-pocket"
 source = { editable = "plugins/pocket" }