Live TTS (WebSocket)

wss://client.camb.ai/apis/live-tts/ws

import asyncio
import json
import websockets


async def synthesize(api_key: str, text: str, out_path: str = "out.mp3") -> None:
    url = "wss://client.camb.ai/apis/live-tts/ws"
    async with websockets.connect(
        url,
        additional_headers=[("x-api-key", api_key)],
    ) as ws:
        # 1. First frame: session config.
        await ws.send(json.dumps({
            "type": "session.start",
            "voice_id": 6460,
            "language": "en-us",
            "output_format": "mp3",
            "word_timestamps": True,
        }))

        # 2. Server confirms with session.ready.
        ready = json.loads(await ws.recv())
        assert ready["type"] == "session.ready", ready
        print(f"session {ready['session_id']} run_id={ready['run_id']}")

        # 3. Stream text. You can call text.chunk many times; the server
        #    segments based on content (and on a 1s idle flush).
        await ws.send(json.dumps({"type": "text.chunk", "text": text}))
        await ws.send(json.dumps({"type": "text.done"}))

        # 4. Receive ordered audio + json frames until the session ends.
        with open(out_path, "wb") as f:
            async for msg in ws:
                if isinstance(msg, bytes):
                    f.write(msg)
                    continue

                frame = json.loads(msg)
                kind = frame["type"]
                if kind == "segment.start":
                    print(f"  segment {frame['segment_id']}: {frame['text']!r}")
                    for w in frame.get("word_timestamps", []):
                        print(f"    {w['start']:6.2f}s → {w['end']:6.2f}s  {w['word']}")
                elif kind == "segment.skipped":
                    print(f"  ! skipped segment {frame['segment_id']}: {frame['text']!r}")
                elif kind == "session.done":
                    print("done")
                    break
                elif kind == "session.error":
                    raise RuntimeError(frame["error"])


asyncio.run(synthesize(
    api_key="your-camb-api-key",
    text="Hello, world. This is a streaming text-to-speech demo.",
))

async for token in llm_stream():
    await ws.send(json.dumps({"type": "text.chunk", "text": token}))

# LLM finished; flush any tail and end cleanly.
await ws.send(json.dumps({"type": "text.done"}))

buffers: dict[int, bytearray] = {}
current_segment: int | None = None

async for msg in ws:
    if isinstance(msg, bytes):
        if current_segment is not None:
            buffers.setdefault(current_segment, bytearray()).extend(msg)
        continue

    frame = json.loads(msg)
    if frame["type"] == "segment.start":
        current_segment = frame["segment_id"]
    elif frame["type"] == "segment.done":
        sid = frame["segment_id"]
        player.enqueue(bytes(buffers.pop(sid)))   # play this segment
        current_segment = None
    elif frame["type"] == "session.done":
        break

if frame["type"] == "segment.skipped":
    await ws.send(json.dumps({"type": "text.chunk", "text": frame["text"]}))

{
  "type": "segment.start",
  "segment_id": 0,
  "text": "Hello, world.",
  "word_timestamps": [
    {"word": "Hello", "start": 0.04, "end": 0.32},
    {"word": "world", "start": 0.38, "end": 0.71}
  ]
}