Real-Time Streaming API

{
  "action": "status",
  "connectionId": "b6f1c2d0-1e2f-4a3b-8c9d-0e1f2a3b4c5d",
  "message": "connected"
}

{
  "action": "...",
  "data": { ... }
}

{ "action": "error", "message": "please send 'stream-config' action with model before streaming text" }

{
  "action": "stream-config",
  "data": {
    "model": "dd-etts-3.0",
    "voicePromptId": "59da0f21-63de-4aef-9ade-e5cabfe639ab",
    "locale": "en-US",
    "format": "s16le",
    "sampleRate": 16000,
    "realtime": true
  }
}

{
  "action": "stream-text",
  "data": { "text": "Hello, how are you today?" }
}

{
  "action": "stream-text",
  "text": "Hello, how are you today?"
}

{ "action": "stream-text", "data": { "text": "Hello! <config locale=\"es-MX\" /> ¡Hola, mundo!" } }

{ "action": "stream-text", "data": { "text": "Cheerful tone. <config voicePromptId=\"new-uuid\" /> Now with a different voice." } }

<config locale="es-MX" voicePromptId="some-uuid" targetGender="female" />

{ "action": "end-stream" }

{ "action": "cancel" }

{
  "generationId": "4da9902b-9141-4fb7-9efb-d616ce266ed9",
  "index": 0,
  "isFinished": false,
  "data": "//uQxAAAAAANIAAAAAExBTUUzLjEwMFVVVVVV..."
}

{
  "generationId": "4da9902b-9141-4fb7-9efb-d616ce266ed9",
  "index": 5,
  "isFinished": true
}

{
  "generationId": "4da9902b-9141-4fb7-9efb-d616ce266ed9",
  "index": 5,
  "isFinished": true,
  "isFinal": true
}

{
  "generationId": "4da9902b-9141-4fb7-9efb-d616ce266ed9",
  "isFinished": true,
  "isCancelled": true
}

{
  "action": "error",
  "message": "please send 'stream-config' action with model before streaming text",
  "time": 1753512960
}

{
  "generationId": "4da9902b-9141-4fb7-9efb-d616ce266ed9",
  "error": "error message"
}

import asyncio
import base64
import json
import os

import websockets

WS_URL = "wss://wss.deepdub.ai/ws"
API_KEY = os.environ["DEEPDUB_API_KEY"]

TOKENS = ["Hello, ", "this ", "is ", "a ", "streaming ", "test."]


async def main():
    async with websockets.connect(
        WS_URL,
        additional_headers={"x-api-key": API_KEY},
        ping_timeout=90,
    ) as ws:
        # 1) Welcome message
        hello = json.loads(await ws.recv())
        print(hello)  # {"action": "status", "connectionId": "...", "message": "connected"}

        # 2) Configure the session
        await ws.send(json.dumps({
            "action": "stream-config",
            "data": {
                "model": "dd-etts-3.0",
                "voicePromptId": "59da0f21-63de-4aef-9ade-e5cabfe639ab",
                "locale": "en-US",
                "format": "s16le",
                "sampleRate": 16000,
                "realtime": True,
            },
        }))

        # 3) Stream text as it becomes available
        for token in TOKENS:
            await ws.send(json.dumps({"action": "stream-text", "data": {"text": token}}))

        # 4) Signal end of the turn
        await ws.send(json.dumps({"action": "end-stream"}))

        # 5) Collect audio chunks
        pcm = bytearray()
        while True:
            msg = json.loads(await ws.recv())
            if msg.get("action") == "error":
                raise RuntimeError(msg["message"])
            if msg.get("data"):
                pcm += base64.b64decode(msg["data"])
            if msg.get("isFinished") and msg.get("isFinal"):
                break  # whole turn is done

        with open("output.pcm", "wb") as f:
            f.write(pcm)
        print(f"Received {len(pcm)} bytes of audio")


asyncio.run(main())

import asyncio
import base64
import json
import os

import websockets
from openai import AsyncOpenAI

WS_URL = "wss://wss.deepdub.ai/ws"
API_KEY = os.environ["DEEPDUB_API_KEY"]

openai = AsyncOpenAI()


async def produce(ws, prompt):
    """Forward LLM tokens into the TTS stream, then close the turn."""
    stream = await openai.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        stream=True,
    )
    async for chunk in stream:
        token = chunk.choices[0].delta.content
        if token:
            # Send each token as it arrives; the server batches them for you.
            await ws.send(json.dumps({"action": "stream-text", "text": token}))

    # The LLM is done — flush the buffer and close the turn.
    await ws.send(json.dumps({"action": "end-stream"}))


async def consume(ws):
    """Receive audio chunks until the turn is final."""
    pcm = bytearray()
    while True:
        msg = json.loads(await ws.recv())
        if msg.get("action") == "error":
            raise RuntimeError(msg["message"])
        if msg.get("data"):
            audio = base64.b64decode(msg["data"])
            pcm += audio
            # play(audio)  # feed to your audio output here for real-time playback
        if msg.get("isFinished") and msg.get("isFinal"):
            break  # the assistant has finished speaking this turn
    return bytes(pcm)


async def main():
    async with websockets.connect(
        WS_URL,
        additional_headers={"x-api-key": API_KEY},
        ping_timeout=90,
    ) as ws:
        # Wait for the welcome message, then configure the session once.
        await ws.recv()
        await ws.send(json.dumps({
            "action": "stream-config",
            "data": {
                "model": "dd-etts-3.0",
                "voicePromptId": "59da0f21-63de-4aef-9ade-e5cabfe639ab",
                "locale": "en-US",
                "format": "s16le",
                "sampleRate": 16000,
                "realtime": True,
            },
        }))

        # Run the LLM producer and the audio consumer concurrently.
        producer = asyncio.create_task(produce(ws, "Tell me a fun fact about the ocean."))
        audio = await consume(ws)
        await producer

        print(f"Received {len(audio)} bytes of audio")


asyncio.run(main())