Realtime Text-to-Speech (TTS) API

{
  "type": "init",
  "language": "en",
  "model": "auto",
  "voice_options": {
    "voice_id": "default_low",
    "speed": 1.0,
    "deaccent_strength": 1.0
  },
  "output": {
    "format": "pcm",
    "sample_rate": 24000
  }
}

{
  "type": "text",
  "text": "Hello, how can I help you today?",
  "is_eos": true
}

{
  "type": "text",
  "text": "This part is spoken faster.",
  "is_eos": false,
  "voice_options": {
    "speed": 1.2
  }
}

{
  "type": "cancel"
}

{
  "message_type": "audio_chunk",
  "data": {
    "audio": "Y3VyaW91cyBtaW5kcyB0aGluayBhbGlrZQ==",
    "size": 9600,
    "generation_id": "1a2b3c4d",
    "last_chunk": false,
    "chunk_generation_delta": 120,
    "audio_len": 0.2
  }
}

{
  "message_type": "error",
  "data": {
    "code": "SERVICE_UNAVAILABLE",
    "desc": "Speech synthesis timed out. Please try again."
  }
}

import asyncio, json, base64, websockets

WS_TTS_URL = "wss://stream.palabra.ai/tts-api/v1/text-to-speech/stream"
API_KEY = "<your-api-key>"  # create one at https://platform.palabra.ai/api-keys

SENTENCES = [
    (
        "The sun was setting over the mountains,",
        "casting long golden shadows across the valley below.",
    ),
    (
        "Birds were returning to their nests,",
        "filling the air with their evening songs.",
    ),
    (
        "A gentle breeze moved through the tall grass,",
        "creating waves that rippled toward the horizon.",
    ),
]

async def main():
    url = f"{WS_TTS_URL}?token={API_KEY}"

    async with websockets.connect(url) as ws:
        # 1. Initialise the session
        await ws.send(json.dumps({
            "type": "init",
            "language": "en",
            "model": "auto",
            "voice_options": {
                "voice_id": "default_low",
                "speed": 1.0,
                "deaccent_strength": 1.0,
            },
            "output": {
                "format": "pcm",
                "sample_rate": 24000,
            }
        }))

        # 2. Send each sentence chunk by chunk, is_eos=True on the last chunk of each sentence
        sent = 0
        for i, sentence in enumerate(SENTENCES):
            for k, chunk in enumerate(sentence):
                msg = {
                    "type": "text",
                    "text": chunk,
                    "is_eos": k == len(sentence) - 1,
                }
                if k == 0:  # override speed only at the start of each sentence
                    msg["voice_options"] = {"speed": min(0.5 + i / 10, 1.0)}
                await ws.send(json.dumps(msg))
                sent += 1

        # 3. Collect audio until every generation has signalled last_chunk
        audio = bytearray()
        done = 0
        async for raw in ws:
            data = json.loads(raw)
            if data.get("message_type") == "audio_chunk":
                chunk = data["data"]
                audio.extend(base64.b64decode(chunk["audio"]))
                print(f"Chunk: {chunk['size']} bytes, total: {len(audio)}")
                if chunk.get("last_chunk"):
                    done += 1
                    if done == sent:   # all sentences finished
                        break
            elif data.get("message_type") == "error":
                print(f"Error: {data['data']['code']} — {data['data']['desc']}")
                break

        print(f"Total: {len(audio)} bytes")
        with open("output.pcm", "wb") as f:
            f.write(audio)
        # Play: ffplay -f s16le -ar 24000 -ac 1 output.pcm

asyncio.run(main())