Realtime Audio

Use connect_realtime in Python or connectRealtime in Node.js to talk to a model that understands audio and speaks directly. The SDK opens the WebSocket and waits until it is ready. Your app captures the microphone and plays the reply.

For separate speech-to-text and text-to-speech models, use Voice (TTS & STT). Its stream_voice / streamVoice pipeline is a different API; do not mix its tts_audio_chunk events with realtime audio.delta.

​

Pick an example

​

Install

pip install --upgrade 'backboard-sdk[voice]'

Set BACKBOARD_API_KEY in your backend environment. These examples require an SDK build with connect_realtime / connectRealtime. If your installed release does not expose it yet, use the SDK in this repository (pip install -e './sdk/python[voice]' for Python, or build sdk/javascript and install that local package for Node.js). Do not put a long-lived API key in browser or mobile code.

The Python examples below use Python 3.11+. Node.js examples use ES modules; save complete scripts with an .mjs extension. Use a current Node.js LTS release.

​

Start your first conversation

This complete example starts a conversation, sends text, and saves the model’s spoken reply as reply.pcm. No microphone is needed yet.

• Connect with just the provider, model, and any session settings.
• Backboard creates the assistant and thread automatically, like send_message / sendMessage.
• Read both IDs from the returned session if you want to reuse them.
• Send input, receive events, then stop gracefully.

import asyncio
import base64
import os
from backboard import BackboardClient

async def main():
    async with BackboardClient(api_key=os.environ["BACKBOARD_API_KEY"]) as client:
        session = await client.connect_realtime(
            provider="openai", model="gpt-realtime-mini", timeout=45,
            system_prompt="Keep spoken answers short and friendly.",
        )
        async with session:
            thread_id, assistant_id = session.thread_id, session.assistant_id
            print("Conversation IDs:", thread_id, assistant_id)
            fmt = session.output_format
            if fmt != {"encoding": "pcm16", "sample_rate": 24000, "channels": 1}:
                raise ValueError("This example expects 24 kHz mono PCM16 output")
            stopping = False
            await session.send_text("Say hello in one sentence.")
            with open("reply.pcm", "wb") as output:
                async for event in session.events(timeout=90):
                    if event["type"] == "audio.delta":
                        output.write(base64.b64decode(event["data"]))
                    elif event["type"] == "transcript.final":
                        print(event["role"], event["text"])
                    elif event["type"] == "response.done" and not stopping:
                        stopping = True
                        await session.send_json({"type": "stop"})
                    elif event["type"] == "session.ended":
                        print("Billing:", event.get("billing"))
                        break
                    elif event["type"] == "error":
                        raise RuntimeError(event)

asyncio.run(main())

Play reply.pcm using FFmpeg’s ffplay:

ffplay -f s16le -ar 24000 -ac 1 -nodisp -autoexit reply.pcm

The file contains raw audio, not WAV or MP3. The example stops after one reply because it has no tools. In an interactive app, keep receiving after response.done and send the next turn on the same session. Stop when the user ends the conversation.

​

Continue or start fresh

Use the same ID rules as the primary POST /threads/messages API:

After closing the first session, choose one of these alternatives. thread_id and assistant_id below are the IDs returned above (camelCase in Node.js).

# Continue the same conversation.
session = await client.connect_realtime(
    thread_id=thread_id, provider="openai", model="gpt-realtime-mini",
)
# Alternatively, start fresh under the same assistant:
# session = await client.connect_realtime(
#     assistant_id=assistant_id, provider="openai", model="gpt-realtime-mini",
# )

system_prompt / systemPrompt and tools apply to this session only, like per-request overrides on send_message. Re-pass them on a later connection if needed. They do not edit the assistant’s saved defaults. Leaving IDs out means a new conversation, not that Backboard stops saving the conversation.

​

What you send and receive

The SDK’s send methods map to these WebSocket messages:

You receive plain objects. Event fields keep their API spelling, even in Node.js. These illustrative events show the useful fields; the server can include more:

{"type":"audio.delta","data":"BASE64_AUDIO_BYTES","encoding":"pcm16","sample_rate":24000,"channels":1}

{"type":"transcript.final","role":"user","item_id":"user-1","text":"What is gravity?"}

{"type":"transcript.final","role":"assistant","item_id":"reply-1","text":"Gravity pulls objects toward each other."}

{"type":"response.done","response_id":"response-1","usage":{}}

Audio and transcripts may arrive in different orders. Play only audio.delta, not the matching raw provider.event too. Append transcript.delta.text to a draft grouped by role/item ID; replace it with transcript.final.text when ready. BASE64_AUDIO_BYTES is a placeholder, not real audio.

​

Stream a voice recording

Use the first example as the base, with the replacements below. Keep its receive loop and file writer. Use a no-tools assistant for this one-reply exercise.

First convert a recording to raw PCM16, mono, 24 kHz:

ffmpeg -i question.wav -f s16le -acodec pcm_s16le -ac 1 -ar 24000 question.pcm

Change the connection settings to disable automatic turn detection:

session = await client.connect_realtime(
    provider="openai", model="gpt-realtime-mini", timeout=45,
    provider_options={"audio": {"input": {"turn_detection": None}}},
)

Add this helper above main() in Python, or above the connection in Node.js:

from pathlib import Path

async def send_recording(session):
    expected = {"encoding": "pcm16", "sample_rate": 24000, "channels": 1}
    if session.input_format != expected:
        raise ValueError("question.pcm must be 24 kHz mono PCM16")
    pcm = Path("question.pcm").read_bytes()
    if len(pcm) < 4800 or len(pcm) % 2:
        raise ValueError("Use at least 100 ms of valid PCM16 audio")
    for offset in range(0, len(pcm), 960):
        chunk = pcm[offset:offset + 960]
        await session.send_audio(chunk)
        await asyncio.sleep(len(chunk) / 48000)
    await session.commit()

Replace the first example’s send/receive block with the block below. It uses the helper above and writes reply.pcm. Sending and receiving run at the same time; do not wait until all audio has been sent before receiving.

# Inside `async with session`, after checking the output format.
async with asyncio.TaskGroup() as tasks:
    tasks.create_task(send_recording(session))
    stopping = False
    with open("reply.pcm", "wb") as output:
        async for event in session.events(timeout=90):
            if event["type"] == "audio.delta":
                output.write(base64.b64decode(event["data"]))
            elif event["type"] == "transcript.final":
                print(event["role"], event["text"])
            elif event["type"] == "response.done" and not stopping:
                stopping = True
                await session.send_json({"type": "stop"})
            elif event["type"] == "session.ended":
                break
            elif event["type"] == "error":
                raise RuntimeError(event)

For a fully assembled raw-WebSocket version, see the audio-in/audio-out example. For a microphone, keep automatic turn detection on and do not call commit after every chunk.

​

Use a live microphone

The SDK does not include a device driver. This complete Python, macOS/Linux example uses sounddevice for capture and playback. Install pip install sounddevice and PortAudio if your system requires it (on macOS: brew install portaudio). Use headphones to avoid echo.

No thread ID is required: this example starts a fresh conversation. Pass thread_id when connecting if you want to continue one. It sends 20 ms chunks, plays replies as they arrive, clears buffered playback on interruption, and ends gracefully on Ctrl+C. Your audio device must support the negotiated rates; otherwise add resampling.

import asyncio
import base64
import os
import queue
import signal
import sys
import threading
import sounddevice as sd
from backboard import BackboardClient

async def main():
    if sys.byteorder != "little":
        raise RuntimeError("This example requires a little-endian machine")
    async with BackboardClient(api_key=os.environ["BACKBOARD_API_KEY"]) as client:
        session = await client.connect_realtime(
            provider="openai",
            model="gpt-realtime-mini", timeout=45,
        )
        async with session:
            print("Conversation IDs:", session.thread_id, session.assistant_id)
            for fmt in (session.input_format, session.output_format):
                if fmt["encoding"] != "pcm16" or fmt["channels"] != 1:
                    raise ValueError("This player needs mono PCM16")
            input_rate = session.input_format["sample_rate"]
            output_rate = session.output_format["sample_rate"]
            microphone = queue.Queue(maxsize=50)  # About one second of input.
            overflow = threading.Event()
            playback = bytearray()
            lock = threading.Lock()
            stopping = asyncio.Event()
            loop = asyncio.get_running_loop()
            loop.add_signal_handler(signal.SIGINT, stopping.set)

            def capture(data, frames, timing, status):
                try:
                    microphone.put_nowait(bytes(data))
                except queue.Full:
                    overflow.set()

            def play(outdata, frames, timing, status):
                size = frames * 2
                with lock:
                    chunk = bytes(playback[:size])
                    del playback[:size]
                outdata[:] = chunk + bytes(size - len(chunk))

            async def send_microphone():
                while not stopping.is_set():
                    if overflow.is_set():
                        raise RuntimeError("Microphone queue overflow: connection is too slow")
                    try:
                        chunk = await asyncio.to_thread(microphone.get, True, 0.1)
                    except queue.Empty:
                        continue
                    if not stopping.is_set():
                        await session.send_audio(chunk)

            async def stop_on_request():
                await stopping.wait()
                await session.send_json({"type": "stop"})

            async def receive():
                async for event in session.events():
                    kind = event["type"]
                    if kind == "audio.delta":
                        chunk = base64.b64decode(event["data"])
                        with lock:
                            if len(playback) + len(chunk) > output_rate * 2 * 5:
                                raise RuntimeError("Playback fell more than five seconds behind")
                            playback.extend(chunk)
                    elif kind == "interrupted":
                        with lock:
                            playback.clear()
                    elif kind == "transcript.final":
                        print(event["role"], event["text"])
                    elif kind == "tool.call" and not event.get("hosted"):
                        raise RuntimeError("Add the custom tool handler below for this assistant")
                    elif kind == "error":
                        raise RuntimeError(event)
                    elif kind == "session.ended":
                        return

            try:
                with sd.RawInputStream(samplerate=input_rate, channels=1, dtype="int16",
                                       blocksize=input_rate // 50, callback=capture), \
                     sd.RawOutputStream(samplerate=output_rate, channels=1, dtype="int16",
                                        blocksize=output_rate // 50, callback=play):
                    print("Speak now. Ctrl+C ends the conversation.")
                    async with asyncio.TaskGroup() as tasks:
                        sender = tasks.create_task(send_microphone())
                        stopper = tasks.create_task(stop_on_request())
                        try:
                            await receive()
                        finally:
                            stopping.set()
                            sender.cancel()
                            stopper.cancel()
            finally:
                loop.remove_signal_handler(signal.SIGINT)

asyncio.run(main())

In a Node.js app, feed your capture library’s PCM buffers to session.sendAudio and decoded audio.delta.data buffers to a streaming audio device. Use the same bounded queues and interruption behavior. The SDK transports bytes; it does not capture, resample, or play them. Browser capture needs a different device layer; see browser connections.

​

Voices and turn detection

Pass native settings as provider_options (Python) or providerOptions (Node.js). Do not add an extra session, setup, or provider-name wrapper. Nested provider fields keep their original spelling in both SDKs.

Use one of these connection recipes with your existing thread:

# OpenAI: 24 kHz input/output; silence-based automatic turns.
session = await client.connect_realtime(
    thread_id, provider="openai", model="gpt-realtime-mini",
    provider_options={
        "audio": {
            "output": {"voice": "cedar", "speed": 1.1},
            "input": {"turn_detection": {"type": "server_vad"}},
        },
    },
)

For Google or xAI, use these provider-specific options instead. Select a supported model from GET /models?model_type=realtime; do not reuse an OpenAI model name.

# Google: 16 kHz input, 24 kHz output.
session = await client.connect_realtime(
    thread_id, provider="google", model="gemini-3.8-live",
    provider_options={"generationConfig": {"speechConfig": {
        "voiceConfig": {"prebuiltVoiceConfig": {"voiceName": "Kore"}},
    }}},
)

# xAI: 24 kHz input/output. Voice is a top-level native setting.
session = await client.connect_realtime(
    thread_id, provider="xai", model="grok-voice-latest",
    provider_options={"voice": "ara", "turn_detection": {"type": "server_vad"}},
)

These are alternatives, not three simultaneous connections to the same thread. Close the old session gracefully before trying another recipe.

For manual mode, send all chunks for the turn, then call commit() once. Python uses None in place of JSON null. Text input does not need a commit.

​

Instructions, transcription, and updates

Assistant instructions are used by default. Override them through native instructions on OpenAI/xAI, or Google’s systemInstruction.parts. Overrides replace the session instructions, including Backboard’s appended tool guidance.

OpenAI input transcription defaults to gpt-4o-mini-transcribe and is separately billed. Disable it with {"audio":{"input":{"transcription":null}}} in native options. The model still hears your speech, but spoken user turns will not be saved as text history. Google enables input/output transcription by default.

For settings the provider allows changing live, use provider.event. For example, with OpenAI:

await session.send_json({
    "type": "provider.event",
    "event": {"type": "session.update", "session": {"instructions": "Answer in Spanish."}},
})

Provider/model changes and Backboard tool selection need a new connection. See provider settings for more options. Gemini Extended Thinking can emit several response.done events per interaction; use provider.event.event.serverContent.interactionStatus === "IDLE" when you need to detect the full interaction’s end.

​

Tools

Tool calls work whether the user types or speaks. Backboard runs internal tools. Your application runs custom functions. Keep receiving audio while either runs.

​

Internal tools

​

Memory and web search

Add these connection options to make memory and search available:

session = await client.connect_realtime(
    thread_id, provider="openai", model="gpt-realtime-mini",
    memory="Auto", web_search="Auto", timeout=90,
)
await session.send_text("Remember that I like short answers. Search today's space news.")

Use Readonly for memory searches without writes. Use memory_pro="Auto" / memoryPro: "Auto" instead of memory for PRO memory. Memory belongs to the assistant and is shared by its threads; history belongs to this conversation.

​

Image and video generation

Choose supported media models from the image/video catalog. For this example, set BACKBOARD_IMAGE_MODEL and BACKBOARD_VIDEO_MODEL to models available through OpenRouter. Remove either tool’s settings if you do not need it.

session = await client.connect_realtime(
    thread_id, provider="openai", model="gpt-realtime-mini",
    image_generation="auto", image_model_provider="openrouter",
    image_model_name=os.environ["BACKBOARD_IMAGE_MODEL"],
    video_generation="auto", video_model_provider="openrouter",
    video_model_name=os.environ["BACKBOARD_VIDEO_MODEL"], timeout=90,
)
await session.send_text("Create an image of a blue robot and describe it briefly.")

Pass selected-model settings as image_config / imageConfig or video_config / videoConfig. See image settings and video settings.

​

Handle hosted events

Use these branches in your receive loop, alongside audio/transcript/error handling:

if event["type"] == "tool.call" and event.get("hosted"):
    print("Backboard is running:", event["name"])
elif event["type"] == "tool.completed":
    print("Tool handling finished:", event["tool_call_id"])
elif event["type"] == "media.generated":
    print("Generated attachments:", event["generated_media"])

Do not submit outputs for hosted: true. The model receives the result from Backboard and continues speaking. tool.completed does not itself prove success; it can follow an error result. Media success emits media.generated attachments. Do not stop on the first response.done in these tool examples. Keep the session open until the user ends it. Paid media jobs can continue after disconnect; do not automatically retry them.

​

Custom functions

Here is the full tool flow: define → attach → receive → execute → return → listen. The demo looks up an order from a small local map. Replace the map with your authorized database query or API call in a real app.

​

Register the function

Use the following tool definition in both languages (it is valid JSON):

{
  "type": "function",
  "function": {
    "name": "get_order_status",
    "description": "Look up the delivery status of an order.",
    "parameters": {
      "type": "object",
      "properties": {"order_id": {"type": "string"}},
      "required": ["order_id"],
      "additionalProperties": false
    }
  }
}

Save it as order-tool.json, then pass it when connecting. No assistant or thread creation call is needed. This replaces the connection portion of the first example. Keep the same client.

import json
from pathlib import Path

tool = json.loads(Path("order-tool.json").read_text())
session = await client.connect_realtime(
    provider="openai", model="gpt-realtime-mini", timeout=45, tools=[tool],
    system_prompt="Use get_order_status for order questions. Never invent a status.",
)

​

Execute calls without blocking audio

tools replaces the assistant’s function list for this session. Omit it to inherit saved tools, or pass [] to clear them. Hosted tools selected through memory, search, or generation settings are still added separately. Re-pass session-only tools when reconnecting, or save reusable tools through the assistant API.

Add this helper above your connection code. It validates the function and arguments, tracks each call, and ignores cancelled work. It does not use eval or dynamically execute a function name from the model.

class OrderTools:
    def __init__(self, session):
        self.session = session
        self.tasks = {}
        self.errors = []

    async def run(self, event):
        args = event.get("arguments")
        if event["name"] != "get_order_status":
            result = {"error": "Unknown tool"}
        elif not isinstance(args, dict) or set(args) != {"order_id"} or not isinstance(args["order_id"], str):
            result = {"error": "Expected one string order_id"}
        else:
            # Replace with an async, user-authorized database/API lookup.
            result = {"order_id": args["order_id"], "status": {"A123": "shipped"}.get(args["order_id"], "not found")}
        await self.session.submit_tool_outputs([{
            "tool_call_id": event["tool_call_id"], "name": event["name"], "output": result,
        }])

    def handle(self, event):
        if event["type"] == "tool.call" and not event.get("hosted"):
            call_id = event["tool_call_id"]
            if call_id in self.tasks:
                return
            task = asyncio.create_task(self.run(event))
            self.tasks[call_id] = task
            def finished(done):
                self.tasks.pop(call_id, None)
                if not done.cancelled() and done.exception():
                    self.errors.append(done.exception())
            task.add_done_callback(finished)
        elif event["type"] == "tool.cancelled":
            for call_id in event["tool_call_ids"]:
                if call_id in self.tasks:
                    self.tasks[call_id].cancel()

    async def close(self):
        tasks = list(self.tasks.values())
        for task in tasks:
            task.cancel()
        await asyncio.gather(*tasks, return_exceptions=True)

​

Ask the question and receive the answer

Use this receive loop instead of the first example’s loop. It keeps receiving through every tool round and spoken reply. Press Ctrl+C after the answer to stop gracefully (Python signal handling below is for macOS/Linux). In a UI, wire the same stop action to your End button. These snippets reuse base64/asyncio or Node’s open import from the first example and the connected session above.

import signal

tools = OrderTools(session)
loop = asyncio.get_running_loop()
# Keep a reference so shutdown errors are observed and repeated Ctrl+C is harmless.
stop_tasks = []
def on_interrupt():
    if not stop_tasks:
        stop_tasks.append(asyncio.create_task(session.send_json({"type": "stop"})))

loop.add_signal_handler(signal.SIGINT, on_interrupt)
try:
    async with session:
        await session.send_text("Where is order A123? Use the order tool.")
        with open("reply.pcm", "wb") as output:
            async for event in session.events(timeout=120):
                tools.handle(event)
                if tools.errors:
                    raise RuntimeError("Tool result could not be submitted") from tools.errors[0]
                if event["type"] == "audio.delta":
                    output.write(base64.b64decode(event["data"]))
                elif event["type"] == "transcript.final":
                    print(event["role"], event["text"])
                elif event["type"] == "session.ended":
                    break
                elif event["type"] == "error":
                    raise RuntimeError(event)
finally:
    loop.remove_signal_handler(signal.SIGINT)
    await tools.close()
    if stop_tasks:
        await asyncio.gather(*stop_tasks)

An example call and result on the wire:

{"type":"tool.call","tool_call_id":"order-1","name":"get_order_status","arguments":{"order_id":"A123"},"hosted":false}

{"type":"tool.outputs","outputs":[{"tool_call_id":"order-1","name":"get_order_status","output":{"order_id":"A123","status":"shipped"}}]}

Use the realtime session’s submit_tool_outputs / submitToolOutputs, not the REST tool-output method. Calls can arrive in parallel or in several rounds. Match each result’s ID and name exactly. Do not submit cancelled or already completed calls, and never submit a hosted call’s result. tool.completed is only for Backboard-hosted calls, not an acknowledgement of your custom outputs.

For slow custom tools, use asynchronous I/O or a worker rather than blocking the event loop. Add application timeouts, enforce user permissions, and return an error object when the business operation fails. Cancellation does not necessarily undo an operation already performed by an external service.

​

Stop and continue later

Send stop, keep receiving until session.ended or an error, then close. The Python context manager and Node.js close() only close the socket; they do not send stop for you. Stop microphone capture before requesting shutdown.

Final transcripts and tool results are saved to the thread; raw native audio is not archived. To continue, connect again with the same thread ID. To start a new conversation, omit the thread ID (optionally keep the assistant ID). To switch providers, finish the old session and reconnect with a different provider/model. There is no automatic reconnect.

Handle history_incomplete and billing_pending errors during shutdown. Graceful stop waits only a bounded time for late transcripts/usage and does not wait for every tool. Billing amounts in session.ended.billing are decimal strings.

​

Browser connections

The Node.js SDK examples run on a trusted server. For a browser, have your backend authorize the user and issue a single-use ticket. Use a browser WebSocket with that ticket; never embed the API key.

Your browser audio layer must capture/resample PCM and schedule playback. A MediaRecorder WebM/Opus blob is not the default PCM input. Use an AudioWorklet or another real-time capture layer. See the API flow.

​

Session methods and options

begin_event / beginEvent holds session.begin; input_format / inputFormat and output_format / outputFormat hold the initial audio formats. Provider setup events received before session.begin remain available through the event loop. Use one receive loop; multiple consumers do not each receive a copy of every event. thread_id / threadId and assistant_id / assistantId expose the resolved conversation IDs, including IDs automatically created by the server.

Default setup timeout: Python 15 seconds; Node.js 15,000 ms, or 90,000 ms when image/video generation is enabled. Set Python timeout=90 for media setup. Receive timeouts are separate. Neither changes the server’s session limits.

For the protocol without an SDK, see Core Concepts.