feat: add deepdub tts extension

ai_agents/agents/ten_packages/extension/deepdub_tts_python/README.md

@@ -0,0 +1,28 @@
+# Deepdub TTS (Streaming) — Python
+
+TEN extension wrapping Deepdub's text-streaming WebSocket API.
+
+- One persistent WebSocket per extension instance, opened and configured during `on_init` (pre-warmed before the first sentence arrives).
+- Each `tts_text_input` fragment is forwarded as a `stream-text` frame, so audio synthesis starts before the LLM has finished producing the full reply.
+- The vendor's `isFinished` boundary is used to emit `tts_audio_end` and release the request lock, naturally serialising back-to-back TEN requests on a single connection.
+
+## Required configuration
+
+Provide via `property.json` `params` or environment variables:
+
+| Property          | Env var                        | Notes                                       |
+| ----------------- | ------------------------------ | ------------------------------------------- |
+| `api_key`         | `DEEPDUB_API_KEY`              | Account API key.                            |
+| `url`             | `DEEPDUB_WS_STREAMING_URL`     | Text-streaming WS endpoint URL.             |
+| `voice_prompt_id` | `DEEPDUB_VOICE_PROMPT_ID`      | Voice prompt to use for the session.        |
+| `model`           | `DEEPDUB_MODEL`                | Defaults to `dd-etts-3.2`.                  |
+| `locale`          | `DEEPDUB_LOCALE`               | Defaults to `en-US`.                        |
+
+## Audio output
+
+Defaults to raw PCM (`s16le`) at 48 kHz, mono — fed directly to the framework's audio pipeline without header parsing.
+
+## Constraints
+
+- The streaming WS binds a single voice/locale/model to a connection (set via `stream-config`). Changing voice mid-session would require a reconnect. v1 assumes a single voice per agent session.
+- `cancel_tts` (TEN flush) drops the in-flight stream and reconnects.

ai_agents/agents/ten_packages/extension/deepdub_tts_python/__init__.py

@@ -0,0 +1 @@
+from . import addon  # noqa: F401

ai_agents/agents/ten_packages/extension/deepdub_tts_python/addon.py

@@ -0,0 +1,10 @@
+from ten_runtime import Addon, register_addon_as_extension, TenEnv
+
+
+@register_addon_as_extension("deepdub_tts_python")
+class DeepdubTTSExtensionAddon(Addon):
+    def on_create_instance(self, ten_env: TenEnv, name: str, context) -> None:
+        from .extension import DeepdubTTSExtension
+
+        ten_env.log_info("DeepdubTTSExtensionAddon on_create_instance")
+        ten_env.on_create_instance_done(DeepdubTTSExtension(name), context)

ai_agents/agents/ten_packages/extension/deepdub_tts_python/config.py

@@ -0,0 +1,72 @@
+from typing import Any, Dict, List, Optional
+
+from pydantic import BaseModel, Field
+
+from ten_ai_base import utils
+
+
+class DeepdubTTSConfig(BaseModel):
+    api_key: str = ""
+    url: str = ""
+    model: str = "dd-etts-3.2"
+    locale: str = "en-US"
+    voice_prompt_id: str = ""
+    sample_rate: int = 48000
+    channels: int = 1
+    # PCM in network byte order, signed 16-bit LE — matches TEN's audio_frame expectations.
+    format: str = "s16le"
+    accept_emojis: bool = False
+    temperature: Optional[float] = None
+    variance: Optional[float] = None
+    tempo: Optional[float] = None
+    prompt_boost: Optional[bool] = None
+    keepalive_interval_seconds: float = 20.0
+    dump: bool = False
+    dump_path: str = ""
+    params: Dict[str, Any] = Field(default_factory=dict)
+    black_list_params: List[str] = Field(default_factory=list)
+
+    def update_params(self) -> None:
+        p = self.params
+        for k in (
+            "api_key",
+            "url",
+            "model",
+            "locale",
+            "voice_prompt_id",
+            "sample_rate",
+            "channels",
+            "format",
+            "accept_emojis",
+            "temperature",
+            "variance",
+            "tempo",
+            "prompt_boost",
+            "keepalive_interval_seconds",
+        ):
+            if k in p:
+                setattr(self, k, p[k])
+                del p[k]
+
+    def validate_params(self) -> None:
+        missing = [
+            n
+            for n in ("api_key", "url", "voice_prompt_id")
+            if not getattr(self, n)
+        ]
+        if missing:
+            raise ValueError(
+                f"required fields are missing or empty: params.{', params.'.join(missing)}"
+            )
+        if self.sample_rate not in (8000, 16000, 22050, 24000, 44100, 48000):
+            raise ValueError(f"unsupported sample_rate: {self.sample_rate}")
+        if self.format not in ("s16le", "wav", "mp3", "opus", "mulaw"):
+            raise ValueError(f"unsupported format: {self.format}")
+
+    def to_str(self, sensitive_handling: bool = False) -> str:
+        if not sensitive_handling:
+            return f"{self}"
+        c = self.copy(deep=True)
+        if c.api_key:
+            c.api_key = utils.encrypt(c.api_key)
+        return f"{c}"

ai_agents/agents/ten_packages/extension/deepdub_tts_python/deepdub_tts.py

@@ -0,0 +1,221 @@
+import asyncio
+import base64
+import json
+import time
+from typing import Awaitable, Callable, Optional
+
+import websockets
+
+from ten_runtime import AsyncTenEnv
+
+from .config import DeepdubTTSConfig
+
+
+class DeepdubTTSException(Exception):
+    def __init__(self, message: str, code: int = -1):
+        super().__init__(message)
+        self.message = message
+        self.code = code
+
+
+AudioCb = Callable[[bytes], Awaitable[None]]
+FinishCb = Callable[[], Awaitable[None]]
+ErrorCb = Callable[[DeepdubTTSException], Awaitable[None]]
+
+
+class DeepdubStreamingClient:
+    """One persistent text-streaming WebSocket per instance.
+
+    Connect → recv status → send `stream-config` → push `stream-text` frames as
+    they arrive → receive interleaved audio and `isFinished` boundaries.
+    """
+
+    def __init__(
+        self,
+        config: DeepdubTTSConfig,
+        ten_env: Optional[AsyncTenEnv],
+        on_audio: AudioCb,
+        on_finish: FinishCb,
+        on_error: ErrorCb,
+    ):
+        self.config = config
+        self.ten_env = ten_env
+        self.on_audio = on_audio
+        self.on_finish = on_finish
+        self.on_error = on_error
+
+        self._ws: Optional[websockets.ClientConnection] = None
+        self._ready = asyncio.Event()
+        self._stopping = False
+        self._discarding = False
+        self._reader_task: Optional[asyncio.Task] = None
+        self._supervisor_task: Optional[asyncio.Task] = None
+        self._keepalive_task: Optional[asyncio.Task] = None
+        self._stopped = asyncio.Event()
+        self._send_lock = asyncio.Lock()
+
+    # ---------- lifecycle ----------
+
+    async def start(self) -> None:
+        self._supervisor_task = asyncio.create_task(self._supervise())
+
+    async def stop(self) -> None:
+        self._stopping = True
+        await self._close_ws()
+        if self._supervisor_task:
+            await self._stopped.wait()
+
+    async def cancel(self) -> None:
+        """Drop the in-flight stream; reconnect from scratch."""
+        self._discarding = True
+        await self._close_ws()
+        # supervisor will reconnect
+
+    # ---------- public send API ----------
+
+    async def wait_ready(self, timeout: float = 10.0) -> None:
+        await asyncio.wait_for(self._ready.wait(), timeout=timeout)
+
+    async def send_text(self, text: str) -> None:
+        if not text:
+            return
+        await self.wait_ready()
+        async with self._send_lock:
+            assert self._ws is not None
+            await self._ws.send(
+                json.dumps({"action": "stream-text", "data": {"text": text}})
+            )
+
+    async def send_cancel(self) -> None:
+        async with self._send_lock:
+            if self._ws is not None:
+                try:
+                    await self._ws.send(json.dumps({"action": "cancel"}))
+                except Exception:
+                    pass
+
+    # ---------- internals ----------
+
+    async def _close_ws(self) -> None:
+        ws = self._ws
+        self._ws = None
+        self._ready.clear()
+        if ws is not None:
+            try:
+                await ws.close()
+            except Exception:
+                pass
+        if self._keepalive_task:
+            self._keepalive_task.cancel()
+            self._keepalive_task = None
+
+    async def _supervise(self) -> None:
+        backoff = 0.5
+        while not self._stopping:
+            try:
+                await self._connect_and_run()
+                backoff = 0.5
+            except Exception as e:
+                if self.ten_env:
+                    self.ten_env.log_warn(f"deepdub stream loop error: {e}")
+                try:
+                    await self.on_error(
+                        e
+                        if isinstance(e, DeepdubTTSException)
+                        else DeepdubTTSException(str(e))
+                    )
+                except Exception:
+                    pass
+                if self._stopping:
+                    break
+                await asyncio.sleep(backoff)
+                backoff = min(backoff * 2, 5.0)
+            finally:
+                self._discarding = False
+                await self._close_ws()
+        self._stopped.set()
+
+    async def _connect_and_run(self) -> None:
+        headers = {"x-api-key": self.config.api_key}
+        if self.ten_env:
+            self.ten_env.log_info("deepdub stream WS connecting")
+        t0 = time.time()
+        async with websockets.connect(
+            self.config.url,
+            additional_headers=headers,
+            max_size=1024 * 1024 * 16,
+        ) as ws:
+            self._ws = ws
+            # Initial status frame.
+            status_raw = await ws.recv()
+            status = json.loads(status_raw)
+            if status.get("action") == "error":
+                raise DeepdubTTSException(
+                    status.get("message", "connection rejected")
+                )
+            # Send stream-config (pre-warm).
+            cfg = {
+                "model": self.config.model,
+                "locale": self.config.locale,
+                "voicePromptId": self.config.voice_prompt_id,
+                "format": self.config.format,
+                "sampleRate": self.config.sample_rate,
+                "acceptEmojis": self.config.accept_emojis,
+                "temperature": self.config.temperature,
+                "variance": self.config.variance,
+                "tempo": self.config.tempo,
+                "promptBoost": self.config.prompt_boost,
+            }
+            await ws.send(
+                json.dumps({"action": "stream-config", "config": cfg})
+            )
+            self._ready.set()
+            if self.ten_env:
+                self.ten_env.log_info(
+                    f"deepdub stream WS ready ({int((time.time()-t0)*1000)}ms)"
+                )
+            self._keepalive_task = asyncio.create_task(self._keepalive())
+            await self._read_loop(ws)
+
+    async def _keepalive(self) -> None:
+        try:
+            while not self._stopping and self._ws is not None:
+                await asyncio.sleep(self.config.keepalive_interval_seconds)
+                if self._ws is None:
+                    return
+                async with self._send_lock:
+                    if self._ws is not None:
+                        try:
+                            await self._ws.send(json.dumps({"action": "ping"}))
+                        except Exception:
+                            return
+        except asyncio.CancelledError:
+            return
+
+    async def _read_loop(self, ws: websockets.ClientConnection) -> None:
+        while not self._stopping:
+            raw = await ws.recv()
+            try:
+                msg = json.loads(raw)
+            except Exception:
+                continue
+            action = msg.get("action")
+            if action in ("pong", "status"):
+                continue
+            err = msg.get("error")
+            if err and not msg.get("generationId"):
+                await self.on_error(DeepdubTTSException(str(err)))
+                continue
+            if self._discarding:
+                # Drop frames belonging to the cancelled run.
+                continue
+            data = msg.get("data")
+            if data:
+                try:
+                    pcm = base64.b64decode(data)
+                except Exception:
+                    pcm = b""
+                if pcm:
+                    await self.on_audio(pcm)
+            if msg.get("isFinished"):
+                await self.on_finish()