Build a voice agent from scratch with Python

What you’ll build

A voice agent you can talk to from your terminal, built on the standard websockets package and about a hundred lines of Python. No SDK, no framework. It comes together in stages: a socket and a single text turn first, then tool calling, then the concurrency that audio needs, and last the microphone itself, where the server decides when you’ve stopped speaking. The first script runs on its own, and each stage after it adds to that code until the full example at the end puts it all in one file. Nothing here assumes you’ve read anything else first. If you’d rather see the whole protocol laid out at once, that’s the reference.

Before you start

You need four things before any of that:

An API key. Sign up or log in at studio.poly.ai and create one in the API Keys tab. The quickstart walks through it. Keep it in POLY_API_KEY in your shell rather than in source control.
Python 3.10 or later.
pip install 'websockets>=14' sounddevice numpy. That version floor matters: connect() took extra_headers before websockets 14 and additional_headers from 14 on, so on an older install the code below raises TypeError rather than failing to authenticate. numpy matters too: sounddevice only declares a dependency on CFFI, so nothing pulls numpy in for you, and the microphone callback later in this guide needs it to convert samples. On Linux, sounddevice also needs the PortAudio library from your package manager (libportaudio2 on Debian and Ubuntu); the macOS and Windows wheels include it.
A microphone, for the last section. The rest runs without one.

Connecting

One connection carries one conversation. You open a WebSocket and keep it open, and every turn after that travels over the same socket as JSON text frames, in both directions. Authentication happens once, when you connect.

connect.py

import asyncio, json, os, websockets

URL = "wss://api.studio.poly.ai/v1/realtime"
API_KEY = os.environ["POLY_API_KEY"]

async def main():
    async with websockets.connect(URL, additional_headers={"X-API-KEY": API_KEY}) as ws:
        print(json.loads(await ws.recv())["type"])  # session.created

asyncio.run(main())

Run that and it prints session.created. The session starts on the server’s default settings, and the next section covers changing them.

Configuring the session

Because the session already exists, configuring it is a message rather than a set of connection parameters. Send session.update and the server answers with session.updated, echoing back the full configuration now in effect. Updates merge into what’s already there, so setting instructions later won’t wipe your tools.

you send: session.update

{
  "type": "session.update",
  "session": {
    "instructions": "You are a helpful voice assistant. Keep responses brief.",
    "output_modalities": ["text"],
    "audio": {
      "input": {
        "format": { "type": "audio/pcm", "rate": 16000 },
        "turn_detection": { "type": "server_vad", "create_response": true }
      }
    }
  }
}

Three of those fields are worth a note. output_modalities is ["text"], because this agent prints the reply as text. A new session starts in ["audio"], where the server speaks the reply and sends the words as response.output_audio_transcript.delta instead of the response.output_text.delta events this guide reads. To hear the agent, leave the default and play response.output_audio.delta, as the quickstart does, or pipe the streamed text into a text-to-speech provider of your own. audio.input.format is PCM16 at the rate you send. 16000 is Dialog-RSN-1’s native rate and the rate the microphone code below captures at. Don’t leave rate out: the server then assumes the standard realtime rate, 24 kHz, and plays 16 kHz audio to the detector and the model 1.5 times too fast. session.updated keeps echoing rate: 24000 either way, and reports the rate you declared as poly_input_rate. turn_detection.create_response is the only field in that object the service reads. The schema’s other timing knobs (silence_duration_ms, threshold and the rest) change nothing, so don’t reach for them to tune turn-taking. Tune it in instructions instead.

The shape of a turn

With the session configured, you can take a turn. Turns share the same two-part shape: adding something to the conversation and asking for a response to it are separate events.

you send: conversation.item.create

{
  "type": "conversation.item.create",
  "item": {
    "type": "message",
    "role": "user",
    "content": [{ "type": "input_text", "text": "What's the capital of France?" }]
  }
}

you send: response.create

{ "type": "response.create" }

The split is useful in two places later on. You can seed a conversation with prior history without firing a reply at every item, and once tools are in play you add the tool’s result first and ask for the follow-up second, since a tool result on its own doesn’t end a turn. The reply comes back as a stream of events rather than one lump of JSON. Append every response.output_text.delta and you get the text as it types out. response.done closes the turn, with the outcome in response.status. Connection, session and turn together make a first script:

first_turn.py

# pip install 'websockets>=14'
import asyncio, json, os, websockets

URL = "wss://api.studio.poly.ai/v1/realtime"
API_KEY = os.environ["POLY_API_KEY"]

async def main():
    async with websockets.connect(URL, additional_headers={"X-API-KEY": API_KEY}) as ws:
        await ws.recv()  # session.created

        await ws.send(json.dumps({
            "type": "session.update",
            "session": {"instructions": "Be concise.", "output_modalities": ["text"]},
        }))
        updated = json.loads(await ws.recv())
        if updated["type"] != "session.updated":
            # A refused update applies nothing, so the session would run on its defaults.
            raise RuntimeError(updated["error"]["message"])

        await ws.send(json.dumps({
            "type": "conversation.item.create",
            "item": {
                "type": "message", "role": "user",
                "content": [{"type": "input_text", "text": "What's the capital of France?"}],
            },
        }))
        await ws.send(json.dumps({"type": "response.create"}))

        reply = ""
        async for raw in ws:
            event = json.loads(raw)
            if event["type"] == "response.output_text.delta":
                reply += event["delta"]
            elif event["type"] == "error":
                # The connection stays open after an error, so a turn can fail without the
                # socket closing. Without this branch the script prints an empty reply.
                raise RuntimeError(event["error"]["message"])
            elif event["type"] == "response.done":
                status = event["response"]["status"]
                if status != "completed":
                    raise RuntimeError(f"response {status}")
                print(reply)
                break

asyncio.run(main())

The error branch is worth keeping. Failures on this API arrive as an error event and leave the connection open, so a turn can fail without the socket closing. Without the branch, a failed generation prints an empty string and exits zero.

Adding tool calling

So far the agent only answers questions. Tools let it look things up and act on them. Declare them on the session, then dispatch with a dict rather than a chain of if/elif, which stays flat as the list grows:

TOOLS = {
    "get_time": lambda args: {"time": "11:04 AM", "timezone": "CDT"},
    "get_weather": lambda args: {
        "location": args["location"],
        "temperature_f": 68,
        "condition": "partly cloudy",
    },
}

SESSION_TOOLS = [
    {"type": "function", "name": "get_time", "description": "Get the current time",
     "parameters": {"type": "object", "properties": {}}},
    {"type": "function", "name": "get_weather", "description": "Get the current weather for a location",
     "parameters": {"type": "object", "properties": {"location": {"type": "string"}},
                    "required": ["location"]}},
]

async def handle_tool_call(ws, name, arguments_json, call_id):
    arguments = json.loads(arguments_json) if arguments_json else {}
    tool = TOOLS.get(name)
    result = tool(arguments) if tool else {"error": f"unknown tool {name!r}"}
    await ws.send(json.dumps({
        "type": "conversation.item.create",
        "item": {"type": "function_call_output", "call_id": call_id, "output": json.dumps(result)},
    }))
    await ws.send(json.dumps({"type": "response.create"}))

Send SESSION_TOOLS as session.tools in your session.update and you’re wired up. A call then arrives as response.function_call_arguments.done, carrying the name, the full arguments as a JSON string, and a server-minted call_id your reply has to echo back. handle_tool_call is your half of the exchange, and it’s the two-part move again with a tool result standing in for a user message: look the function up, run it, add the result as a function_call_output item, then ask for the follow-up turn with a fresh response.create. The second response.create is easy to leave out, and nothing else triggers the follow-up, since a tool result carries no audio and ends no turn by itself. Look it up with .get rather than indexing TOOLS[name] directly, too: a model can invent a name that isn’t in the dict, and an unhandled KeyError there takes down the whole event loop instead of just failing that one call. The reference’s function calling section has the full wire shape.

Structuring concurrent tasks

Handling one turn at a time is fine while you’re typing. With audio, the agent needs to push frames up, watch for a turn to start, and handle what comes back at the same time, so it’s easier to restructure around concurrent tasks before adding the microphone.

async def send_audio_task(ws, mic_queue):
    while True:
        frame = await mic_queue.get()
        await ws.send(json.dumps({
            "type": "input_audio_buffer.append",
            "audio": base64.b64encode(frame).decode(),
        }))

async def event_handler_task(ws):
    reply = ""
    async for raw in ws:
        event = json.loads(raw)
        event_type = event["type"]
        if event_type == "response.output_text.delta":
            reply += event["delta"]
        elif event_type == "response.output_text.done":
            print(reply)
            reply = ""
        elif event_type == "response.function_call_arguments.done":
            await handle_tool_call(ws, event["name"], event["arguments"], event["call_id"])
        elif event_type == "input_audio_buffer.speech_started":
            print("(listening...)")
        elif event_type == "error":
            print(f"error: {event['error']['message']}")
        elif event_type == "response.done" and event["response"]["status"] == "failed":
            print(f"response failed: {event['response']}")

async def run(ws, mic_queue):
    tasks = [
        asyncio.create_task(send_audio_task(ws, mic_queue)),
        asyncio.create_task(event_handler_task(ws)),
    ]
    try:
        await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
        for task in tasks:              # re-raise whichever task actually ended the race
            if task.done() and not task.cancelled():
                task.result()
    finally:
        for task in tasks:
            task.cancel()
        await asyncio.gather(*tasks, return_exceptions=True)

asyncio.wait(..., return_when=asyncio.FIRST_COMPLETED) returns as soon as one task ends, and the finally cancels the rest rather than leaving them running. Without that, a Ctrl-C can leave tasks behind. asyncio.wait doesn’t re-raise the way asyncio.gather does, though, so if a task ends because of a genuine bug rather than a handled protocol error, calling .result() on it afterward is what surfaces that exception instead of run returning as if nothing happened. The error branch prints and carries on here, where the one-shot script earlier raised. A long-running agent shouldn’t fall over because one turn failed, and the connection survives an error, so the next turn can still succeed. The full example at the end of this guide adds a stop_after_turn flag for its one-shot --say mode, and that path does need to raise: a script with no microphone feeding it more audio would otherwise hang forever waiting for a turn that’s never coming.

Adding real audio input and server VAD

Streaming audio in means feeding frames into the queue send_audio_task already drains, base64-encoded PCM16, appended continuously. Set blocksize explicitly rather than leaving it at whatever sounddevice picks by default. Every callback becomes its own WebSocket message, and an unset default lets the host choose the block size, which can be very small. The server’s voice activity detector reads 32 ms frames and rebuffers whatever you send, so the block doesn’t have to match, but 512 samples at 16kHz is exactly one frame:

import sounddevice as sd

SAMPLE_RATE = 16000  # Dialog-RSN-1's native rate; send it as `audio.input.format.rate` in session.update
BLOCKSIZE = 512  # 512 samples @ 16kHz = 32ms, matching the detector's frame size

def start_microphone(mic_queue, loop):
    def callback(indata, frames, time_info, status):
        pcm16 = (indata * 32767).astype("int16").tobytes()
        loop.call_soon_threadsafe(mic_queue.put_nowait, pcm16)

    return sd.InputStream(samplerate=SAMPLE_RATE, channels=1, dtype="float32",
                          blocksize=BLOCKSIZE, callback=callback)

sounddevice runs that callback on its own thread, so frames reach the event loop through loop.call_soon_threadsafe rather than by touching the queue from that thread. SAMPLE_RATE = 16000 asks the device for Dialog-RSN-1’s native rate, which most microphones support. If yours doesn’t, sd.InputStream fails with PortAudioError: Invalid sample rate rather than choosing a nearby rate. In that case, capture at the device’s own rate. The full example sends SAMPLE_RATE as audio.input.format.rate, so the server knows the rate and resamples on the way in:

SAMPLE_RATE = int(sd.query_devices(kind="input")["default_samplerate"])  # 44100, 48000, whatever it reports

The rest of the code stays the same either way. Once frames are flowing, the server takes care of turn-taking. There’s no silence threshold to pick and no timer to tune on the client side; it appends frames and waits to be told what happened. The server owns voice activity detection, and Dialog-RSN-1 judges what the detector proposes. When nothing is playing, input_audio_buffer.speech_started arrives as soon as the detector hears sound. While a reply is playing, Dialog-RSN-1 confirms the caller is really talking before it sends one. input_audio_buffer.speech_stopped always waits for Dialog-RSN-1’s judgment that they’ve finished, then the server commits the turn and, with create_response: true, answers it. Pause mid-sentence and the detector may read it as silence, but if Dialog-RSN-1 judges the caller isn’t finished, you don’t see a turn boundary. If it judges the sound was noise, it drops the turn, so a speech_started can arrive with no speech_stopped after it. Start talking again while a reply is still streaming and Dialog-RSN-1 cancels that reply, once confirmed. It arrives as a response.done with status cancelled that you didn’t ask for, so the event handler needs to expect one. The reference’s turn-taking section covers the mechanics.

Full example

Everything above, in one file, plus one thing not shown yet: a --say flag that sends a single text turn, the exact add-then-respond shape from the quickstart, instead of opening the mic. Without it, tool calling, the one part of this guide that doesn’t need working audio hardware, is also the one part a reader can’t reach without a microphone to talk into.

Full code on GitHub

This file with its requirements, ready to clone and run.

voice_agent.py

# pip install 'websockets>=14' sounddevice numpy
import argparse
import asyncio
import base64
import json
import os

import sounddevice as sd
import websockets

URL = "wss://api.studio.poly.ai/v1/realtime"
API_KEY = os.environ["POLY_API_KEY"]
SAMPLE_RATE = 16000
BLOCKSIZE = 512  # 512 samples @ 16kHz = 32ms, matching the detector's frame size

TOOLS = {
    "get_time": lambda args: {"time": "11:04 AM", "timezone": "CDT"},
    "get_weather": lambda args: {
        "location": args["location"],
        "temperature_f": 68,
        "condition": "partly cloudy",
    },
}

SESSION_TOOLS = [
    {"type": "function", "name": "get_time", "description": "Get the current time",
     "parameters": {"type": "object", "properties": {}}},
    {"type": "function", "name": "get_weather", "description": "Get the current weather for a location",
     "parameters": {"type": "object", "properties": {"location": {"type": "string"}},
                    "required": ["location"]}},
]


async def handle_tool_call(ws, name, arguments_json, call_id):
    arguments = json.loads(arguments_json) if arguments_json else {}
    tool = TOOLS.get(name)
    result = tool(arguments) if tool else {"error": f"unknown tool {name!r}"}
    await ws.send(json.dumps({
        "type": "conversation.item.create",
        "item": {"type": "function_call_output", "call_id": call_id, "output": json.dumps(result)},
    }))
    await ws.send(json.dumps({"type": "response.create"}))


async def send_audio_task(ws, mic_queue):
    while True:
        frame = await mic_queue.get()
        await ws.send(json.dumps({
            "type": "input_audio_buffer.append",
            "audio": base64.b64encode(frame).decode(),
        }))


async def event_handler_task(ws, stop_after_turn=False):
    reply = ""
    pending_tool = False
    async for raw in ws:
        event = json.loads(raw)
        event_type = event["type"]
        if event_type == "response.output_text.delta":
            reply += event["delta"]
        elif event_type == "response.output_text.done":
            print(reply)
            reply = ""
            pending_tool = False
        elif event_type == "response.function_call_arguments.done":
            pending_tool = True
            await handle_tool_call(ws, event["name"], event["arguments"], event["call_id"])
        elif event_type == "input_audio_buffer.speech_started":
            print("(listening...)")
        elif event_type == "error":
            print(f"error: {event['error']['message']}")
            if stop_after_turn:
                raise RuntimeError(event["error"]["message"])
        elif event_type == "response.done":
            status = event["response"]["status"]
            if status == "failed":
                print(f"response failed: {event['response']}")
                if stop_after_turn:
                    raise RuntimeError(f"response failed: {event['response']}")
            # a tool call means another response follows, so this isn't the final turn yet
            if stop_after_turn and status == "completed" and not pending_tool:
                return


def start_microphone(mic_queue, loop):
    def callback(indata, frames, time_info, status):
        pcm16 = (indata * 32767).astype("int16").tobytes()
        loop.call_soon_threadsafe(mic_queue.put_nowait, pcm16)

    return sd.InputStream(samplerate=SAMPLE_RATE, channels=1, dtype="float32",
                          blocksize=BLOCKSIZE, callback=callback)


async def send_text_turn(ws, text):
    await ws.send(json.dumps({
        "type": "conversation.item.create",
        "item": {"type": "message", "role": "user",
                 "content": [{"type": "input_text", "text": text}]},
    }))
    await ws.send(json.dumps({"type": "response.create"}))


async def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--say", help="send one text turn instead of opening the mic")
    args = parser.parse_args()

    headers = {"X-API-KEY": API_KEY}
    async with websockets.connect(URL, additional_headers=headers) as ws:
        await ws.recv()  # session.created

        await ws.send(json.dumps({
            "type": "session.update",
            "session": {
                "instructions": "You are a helpful voice assistant. Keep responses brief.",
                "output_modalities": ["text"],
                "tools": SESSION_TOOLS,
                "audio": {
                    "input": {
                        "format": {"type": "audio/pcm", "rate": SAMPLE_RATE},
                        "turn_detection": {"type": "server_vad", "create_response": True},
                    },
                },
            },
        }))
        updated = json.loads(await ws.recv())
        if updated["type"] != "session.updated":
            # A refused update applies nothing, so the session would run on its defaults.
            raise RuntimeError(updated["error"]["message"])

        if args.say:
            await send_text_turn(ws, args.say)
            await event_handler_task(ws, stop_after_turn=True)
            return

        mic_queue = asyncio.Queue()
        loop = asyncio.get_running_loop()
        mic = start_microphone(mic_queue, loop)
        mic.start()

        print("Talk to it. Ctrl-C to quit.")
        tasks = [
            asyncio.create_task(send_audio_task(ws, mic_queue)),
            asyncio.create_task(event_handler_task(ws)),
        ]
        try:
            await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
            for task in tasks:              # re-raise whichever task actually ended the race
                if task.done() and not task.cancelled():
                    task.result()
        finally:
            mic.stop()
            mic.close()
            for task in tasks:
                task.cancel()
            await asyncio.gather(*tasks, return_exceptions=True)


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

Run it and say something. Asking for the time, or the weather somewhere, exercises the tool round trip. Most of what you’d add from here goes in TOOLS and instructions; the protocol layer stays as it is.

Next steps

Overview

What realtime streaming is, and how turn-taking, generation, and prompting fit together.

Reference

Every event, the session object, function calling, and error codes.

Browser (JavaScript)

The same protocol, from a browser tab, with microphone capture in the client.

Streaming to TTS

Where that “pipe the streamed text into TTS” line above actually goes: barge-in, unspeakable text, and playback reports.

Overview

Quickstart

Prompt guide

Prompt cookbook

Evaluations

Reference

Guides

Build a voice agent from scratch with Python

What you’ll build

Before you start

Connecting

Configuring the session

The shape of a turn

Adding tool calling

Structuring concurrent tasks

Adding real audio input and server VAD

Full example

Full code on GitHub

Next steps

Overview

Reference

Browser (JavaScript)

Streaming to TTS

​What you’ll build

​Before you start

​Connecting

​Configuring the session

​The shape of a turn

​Adding tool calling

​Structuring concurrent tasks

​Adding real audio input and server VAD

​Full example

Full code on GitHub

​Next steps

Overview

Reference

Browser (JavaScript)

Streaming to TTS

What you’ll build

Before you start

Connecting

Configuring the session

The shape of a turn

Adding tool calling

Structuring concurrent tasks

Adding real audio input and server VAD

Full example

Next steps