Protobuf quickstart

Stream two sentences to Coda with Protobuf messages and save the audio to output.pcm. You can run it without an audio device, then convert the file to WAV to hear it.

Prepare the client

Install uv and Python 3.11 or later. Create a Rime API key and set RIME_API_KEY in your server’s environment. Create a project and install the transport client and message definitions:

uv init --python 3.11 coda-binary
cd coda-binary
uv add 'websockets>=15,<16' 'rime-api==0.0.1'

rime-api supplies generated Protobuf classes and schema files. Your application still uses a standard WebSocket client to connect, authenticate, send messages, and manage retries. The LiveKit plugin uses these same message classes.

Stream audio to a file

Save this as stream_binary.py. It connects to wss://coda.api.rime.ai/ws with rime.v1.binary and overwrites output.pcm in the current directory. The synthesis parameters and lifecycle are the same as in the JSON quickstart. The script decodes each server message as WebSocketResponse and reads its audio field. The full message includes Protobuf metadata, so it cannot go directly to a player.

import asyncio
import os

from rime_api import text_to_speech_pb2 as proto
from websockets.asyncio.client import connect

URL = "wss://coda.api.rime.ai/ws"
CONTEXT_ID = "turn-1"


async def write_audio(output, chunk):
    task = asyncio.create_task(asyncio.to_thread(output.write, chunk))
    try:
        await asyncio.shield(task)
    except asyncio.CancelledError:
        # A worker thread cannot be cancelled. Finish the write before closing.
        while not task.done():
            try:
                await asyncio.shield(task)
            except asyncio.CancelledError:
                pass
        task.result()
        raise


async def receive_event(connection):
    raw = await connection.recv()
    if not isinstance(raw, bytes):
        raise RuntimeError("Expected a Protobuf binary message")
    event = proto.WebSocketResponse.FromString(raw)
    if event.WhichOneof("payload") == "error":
        raise RuntimeError(f"Rime error: {event.error}")
    return event


async def send_sentences(connection):
    for sentence in ["Hello from Coda. ", "This audio uses the binary protocol. "]:
        request = proto.WebSocketRequest(context_id=CONTEXT_ID, text=sentence)
        await connection.send(request.SerializeToString())
    request = proto.WebSocketRequest(context_id=CONTEXT_ID)
    request.end.SetInParent()
    await connection.send(request.SerializeToString())


async def receive_audio(connection, output):
    while True:
        event = await receive_event(connection)
        if event.context_id != CONTEXT_ID:
            raise RuntimeError("Unexpected context ID")
        payload = event.WhichOneof("payload")
        if payload == "started":
            print("Request ID:", event.started.request_id)
        elif payload == "audio":
            await write_audio(output, event.audio)
        elif payload == "done":
            return
        elif payload == "cancelled":
            raise RuntimeError("Synthesis was cancelled")


async def main():
    headers = {"Authorization": f"Bearer {os.environ['RIME_API_KEY']}"}
    async with asyncio.timeout(120):
        async with connect(
            URL, additional_headers=headers, subprotocols=["rime.v1.binary"]
        ) as connection:
            if connection.subprotocol != "rime.v1.binary":
                raise RuntimeError("Binary subprotocol was not selected")
            event = await receive_event(connection)
            if event.WhichOneof("payload") != "ready" or event.ready.protocol != 1:
                raise RuntimeError("Expected WebSocket protocol version 1")
            request = proto.WebSocketRequest(
                context_id=CONTEXT_ID,
                start=proto.SynthesisRequest(
                    speaker="lyra",
                    language="en",
                    audio_parameters=proto.AudioParameters(
                        audio_format="audio/pcm", sampling_rate=24000
                    ),
                ),
            )
            await connection.send(request.SerializeToString())
            with open("output.pcm", "wb", buffering=0) as output:
                async with asyncio.TaskGroup() as tasks:
                    tasks.create_task(send_sentences(connection))
                    tasks.create_task(receive_audio(connection, output))
    print("Saved output.pcm")


if __name__ == "__main__":
    asyncio.run(main())

The script has four functions:

send_sentences: sends each sentence as a text message, then sends end. To stream input over time, replace its fixed list with complete sentences from your application.
receive_audio: reads events until done, and writes each audio chunk. To route audio to a browser, phone system, or player, replace the await write_audio(...) call with your audio handler.
write_audio: writes each chunk on a worker thread. The receiver waits for each write before it reads the next message, so chunks stay in order. On cancellation, it waits for the active write before it closes the file.
main: connects with rime.v1.binary, waits for ready, sends start, and runs the sender and receiver together.

end and cancel are empty messages, so Protobuf does not select the payload until you mark it as set. Call request.end.SetInParent() or request.cancel.SetInParent() to select it, because a request with only a context_id sends no operation. Run the script after setting your API key:

uv run stream_binary.py

A successful run prints a request ID and Saved output.pcm. The file contains raw mono, signed 16-bit little-endian PCM at 24 kHz, with no WAV header, and a failed run can leave a partial file. To hear the audio, convert it to WAV.

Listen to the result

Most players do not open headerless PCM. This command wraps output.pcm in a WAV header and saves output.wav, which any audio player opens:

uv run python - <<'PY'
import wave
from pathlib import Path

pcm = Path("output.pcm").read_bytes()
with wave.open("output.wav", "wb") as out:
    out.setnchannels(1)
    out.setsampwidth(2)
    out.setframerate(24000)
    out.writeframes(pcm)
PY

Generate bindings for another language

The installed package includes schema/rime/text_to_speech.proto and schema/text_to_speech.asyncapi.yaml. Use the Protobuf source to generate messages for your language. The AsyncAPI file describes the JSON encoding. This command copies the installed Protobuf schema to your project:

uv run python - <<'PY'
from importlib.resources import files
from pathlib import Path

schema = files("rime_api").joinpath("schema/rime/text_to_speech.proto")
Path("text_to_speech.proto").write_bytes(schema.read_bytes())
PY

Generate rime.WebSocketRequest and rime.WebSocketResponse with your language’s Protobuf compiler. Each WebSocket message carries one serialized message. Do not add a gRPC header, length prefix, base64 layer, or custom framing. The schema defines gRPC services too, but this WebSocket client only needs the message types. For message fields, connection reuse, cancellation, and errors, see the Coda WebSocket API reference.

Overview

Coda API reference

Mist v3 API reference

Speech-to-text API reference

API Metadata

Other APIs

Protobuf quickstart

Prepare the client

Stream audio to a file

Listen to the result

Generate bindings for another language

​Prepare the client

​Stream audio to a file

​Listen to the result

​Generate bindings for another language

Prepare the client

Stream audio to a file

Listen to the result

Generate bindings for another language