Build a real-time voice AI agent in Python with the AssemblyAI Voice Agent API

You can build a working real-time voice agent in Python in well under 100 lines of code—if you pick the right primitive. This tutorial walks through building one on the AssemblyAI Voice Agent API: a single WebSocket that wraps streaming speech-to-text, an LLM, text-to-speech, turn detection, and tool calling at $4.50/hour flat, billed by the minute. No three-provider pipeline to wire up, no separate STT WebSocket plus LLM HTTP call plus TTS stream to babysit. Audio in, audio out, tool calls in between.

By the end you'll have a runnable Python voice agent that listens through your microphone, holds a real conversation, and calls Python functions to take actions. The full repo is linked at the end—fork it and swap in your own tools.

Want to hear what you're about to build before you write a line of code? Talk to a live agent and barge in on it, cut it off, spell out an email. That's the bar we're building to.

If you'd rather chain streaming STT, an LLM, and a TTS provider yourself, our chained Python voice agent tutorial covers that path. And if you already run an orchestrator, the Voice Agent API ships as a drop-in plugin for both LiveKit and Pipecat—more on those routes at the end.

Why use the Voice Agent API for a Python voice agent

The classic "voice agent in Python" tutorial stitches together a streaming STT API, an LLM HTTP endpoint, and a TTS streaming connection. That's three providers, three sets of credentials, three latency budgets to tune, and your own turn-detection logic to write and maintain. It works. It's also a lot of plumbing, and every seam between vendors is a place where things drift out of sync.

The Voice Agent API replaces all of that with one WebSocket. You connect once, send audio frames, and receive both audio output and tool-call events on the same stream. It works natively with Claude Code and needs no SDK—it's just a WebSocket and JSON. Three properties make it worth it for production Python voice agents:

• One bill, one set of logs. $4.50/hour of session time (billed by the minute) covers STT, LLM inference, TTS, turn detection, and tool calling. You're not reconciling three invoices in a spreadsheet, and you're not debugging across three vendors' status pages at 2am.
• Speech accuracy that survives real conversations. The STT foundation is Universal-3.5 Pro Realtime, released June 23, 2026. On Pipecat's open STT benchmark—real agent conversations, not read-aloud audiobooks—it posts a 6.99% pooled word error rate, versus 9.04% for Google Chirp3, 9.76% for ElevenLabs Scribe v2, and 15.58% for Deepgram Flux. That gap is the difference between an agent that hears "my account is 4-9-3-A" and one that doesn't.
• Tool calling that maps to Python functions cleanly. Define tools as JSON schemas, the LLM calls them, results stream back into the conversation. No separate function-calling API, no second LLM provider to manage.

The part most STT APIs miss: agent_context

Here's the capability that changed how we think about accuracy for voice agents. Universal-3.5 Pro Realtime is the first streaming STT that takes the agent's own question as input—we call it agent_context.

Think about what a voice agent actually asks: "What's the email on file?" "Can you read me your account ID?" "Was that a yes?" The replies are short, often mumbled, and frequently full of things that wreck a general transcription model—spelled-out emails, alphanumeric IDs, one-word confirmations. A model transcribing that reply in isolation is guessing. A model that knows the agent just asked for an email knows to expect an email, and resolves "j-dot-smith at gmail" correctly instead of turning it into word soup.

Passing the agent's question as context cut WER by 10.2% across 20,000 real voice-agent audio files. It also keeps a rolling conversation memory (we call it Context Carryover), so a name mentioned in turn two is still spelled right in turn nine. On the structured stuff that actually breaks agents, the numbers hold up: entity error rate of 15.31%, names at 16.92%, and phone numbers at just 3.55%.

On the Voice Agent API, this runs under the hood—you get it for free. If you're building the chained architecture instead, agent_context is exposed directly on streaming speech-to-text.

How this compares to the OpenAI Realtime API

The Voice Agent API's most direct competitor is OpenAI's Realtime API, so let's be straight about the tradeoffs.

OpenAI's Realtime API is a single speech-to-speech model—audio goes in, audio comes out, and one model does everything in between. It's elegant, and for some use cases it's the right call. The Voice Agent API takes the opposite bet: dedicated, best-in-class models for each step—Universal-3.5 Pro Realtime for listening, a frontier LLM for reasoning, a purpose-built TTS for speaking. Three places to be excellent instead of one place to be good at everything.

That architecture shows up in three ways that matter for a production agent:

• Accuracy on structured data. A speech-to-speech model transcribes and reasons in one pass, which means there's no dedicated STT layer tuned for the account numbers, emails, and confirmation codes that voice agents live and die on. A specialized STT foundation with agent_context is built exactly for that.
• Neural turn detection, not VAD. Knowing when the user is done talking is half of what makes an agent feel human. The Voice Agent API uses a neural turn-detection model that understands semantic completeness—it won't cut you off mid-thought during a pause, and it won't sit there waiting after you've clearly finished. Simple voice-activity detection can't tell the difference between "I'm thinking" and "I'm done."
• Predictable pricing. The Voice Agent API is a flat $4.50/hour, billed by the minute. That's roughly 4x cheaper than the OpenAI Realtime API's ~$18/hour effective rate, and it's a number you can put in a budget without modeling token usage per turn.

Both are good products. If you want one model to own the whole loop, OpenAI's approach is clean. If you want the best listener, the best turn detection, and a bill you can predict, use dedicated models.

Architecture

Microphone
   │  PCM16 24kHz mono
   ▼
Your Python script
   │  WebSocket: input.audio frames
   ▼
AssemblyAI Voice Agent API
 ┌────────────────────────────────┐
 │  STT + Turn detection           │
 │      ↓                          │
 │  LLM + tool calling             │
 │      ↓                          │
 │  TTS                            │
 └────────────────────────────────┘
   │
   │  WebSocket: reply.audio + tool.call events
   ▼
Your Python script
   ├─► Speaker playback
   └─► Dispatch tool calls back to LLM

Audio flows in both directions on the same WebSocket. Your script captures mic audio, base64-encodes it, and sends it as input.audio events. The API returns audio playback chunks as reply.audio events and structured tool.call events when the LLM decides to invoke one of your tools. You dispatch the tool, send back a tool.result, and the conversation continues.

Before you start

You'll need:

• An AssemblyAI account with Voice Agent API access
• Python 3.11+
• A working microphone and speakers—use headphones for clean barge-in, because desktop mics pick up the agent's own voice and cause it to interrupt itself
• portaudio installed system-wide (brew install portaudio on macOS, apt install portaudio19-dev on Debian/Ubuntu)

Install the dependencies:

pip install "websockets>=14" python-dotenv pyaudio

Drop your API key into a .env file:

ASSEMBLYAI_API_KEY=your_key_here

Get your free API key if you don't have one yet—it takes about a minute and comes with credit to run this tutorial.

Step 1: Capture microphone audio

PyAudio captures raw PCM audio. The Voice Agent API's default audio/pcm encoding is 24 kHz, 16-bit, mono—the audio-format docs recommend ~50 ms chunks for low latency.

# audio.py
import threading
from queue import Queue
import pyaudio

SAMPLE_RATE = 24000
CHUNK_SIZE = 1200  # 50ms at 24kHz 16-bit mono

class Mic:
    def __init__(self):
        self._pa = pyaudio.PyAudio()
        self.queue = Queue()
        self._running = False

    def start(self):
        self._running = True
        self._stream = self._pa.open(
            format=pyaudio.paInt16, channels=1, rate=SAMPLE_RATE,
            input=True, frames_per_buffer=CHUNK_SIZE,
        )
        threading.Thread(target=self._loop, daemon=True).start()

    def _loop(self):
        while self._running:
            self.queue.put(self._stream.read(CHUNK_SIZE,
                exception_on_overflow=False))

    def stop(self):
        self._running = False
        self._stream.stop_stream(); self._stream.close()
        self._pa.terminate()

class Speaker:
    def __init__(self):
        self._pa = pyaudio.PyAudio()
        self._stream = self._open()

    def _open(self):
        return self._pa.open(
            format=pyaudio.paInt16, channels=1, rate=SAMPLE_RATE, output=True,
        )

    def play(self, audio_bytes):
        self._stream.write(audio_bytes)

    def flush_and_restart(self):
        # Called on barge-in: drop any queued speech and reopen the stream.
        try:
            self._stream.stop_stream(); self._stream.close()
        except Exception:
            pass
        self._stream = self._open()

    def close(self):
        self._stream.stop_stream(); self._stream.close()
        self._pa.terminate()

Step 2: Open the Voice Agent API session

The connection starts with a session.update message that declares your system prompt, the tools you want available, the agent's voice, and an opening greeting. The API picks audio/pcm (24 kHz) by default, so you don't need to specify input/output format explicitly.

# agent.py
import asyncio, base64, json, os
import websockets
from dotenv import load_dotenv

from audio import Mic, Speaker
from tools import TOOLS, dispatch_tool

load_dotenv()

VOICE_AGENT_WS = "wss://agents.assemblyai.com/v1/ws"

SYSTEM_PROMPT = """You are a helpful voice assistant.
Keep replies short and conversational — one or two sentences.
Use the available tools to answer questions when relevant."""

async def open_session(ws):
    await ws.send(json.dumps({
        "type": "session.update",
        "session": {
            "system_prompt": SYSTEM_PROMPT,
            "greeting": "Hi! How can I help?",
            "tools": TOOLS,
            "output": {"voice": "ivy"},
        },
    }))

A few details worth flagging up front, because they're the easy ones to get wrong:

• The auth header uses Authorization: Bearer YOUR_KEY—note the Bearer prefix. This differs from every other AssemblyAI endpoint, which accepts the raw API key with no prefix.
• The first message you send is session.update, not session.start. All config nests under a session object.
• The voice field is a named voice from the Voice Agent API catalog (e.g. ivy, james, sophie)—not an ElevenLabs voice ID. See the voices reference for the full list.
• Wait for the server's session.ready event before sending any audio.

Step 3: Pump audio in, route events out

Two coroutines run concurrently: one sends mic chunks once the session is ready, the other reads events as they arrive.

async def run_agent():
    mic = Mic()
    speaker = Speaker()

    async with websockets.connect(
        VOICE_AGENT_WS,
        additional_headers={"Authorization": f"Bearer {os.environ['ASSEMBLYAI_API_KEY']}"},
    ) as ws:
        await open_session(ws)

        ready = asyncio.Event()
        pending_tools = []
        loop = asyncio.get_event_loop()

        async def send_audio():
            await ready.wait()
            mic.start()
            while True:
                chunk = await loop.run_in_executor(None, mic.queue.get)
                await ws.send(json.dumps({
                    "type": "input.audio",
                    "audio": base64.b64encode(chunk).decode(),
                }))

        async def receive():
            async for raw in ws:
                event = json.loads(raw)
                kind = event["type"]

                if kind == "session.ready":
                    ready.set()
                    print(f"Session ready: {event.get('session_id')}")

                elif kind == "reply.audio":
                    speaker.play(base64.b64decode(event["data"]))

                elif kind == "tool.call":
                    # Accumulate — flush on reply.done, not now.
                    result = dispatch_tool(event["name"], event.get("arguments", {}))
                    pending_tools.append({"call_id": event["call_id"], "result":
result})

                elif kind == "reply.done":
                    if event.get("status") == "interrupted":
                        pending_tools.clear()
                        speaker.flush_and_restart()
                    elif pending_tools:
                        for tool in pending_tools:
                            value = tool["result"]
                            if not isinstance(value, str):
                                value = json.dumps(value)
                            await ws.send(json.dumps({
                                "type": "tool.result",
                                "call_id": tool["call_id"],
                                "result": value,
                            }))
                        pending_tools.clear()

                elif kind == "transcript.user":
                    print(f"You:   {event['text']}")

                elif kind == "transcript.agent":
                    print(f"Agent: {event['text']}")

        await asyncio.gather(send_audio(), receive())

That's the entire voice agent loop. The Voice Agent API handles every layer of the pipeline—STT, LLM, TTS, turn detection—inside the WebSocket. Your job is to feed it audio, play what comes back, and dispatch tool calls.

Two more easy-to-miss details:

• Tool-result timing. Per the tool-calling docs, accumulate tool results when tool.call fires and send them inside the reply.done handler—not immediately. The agent generates a short transition phrase ("let me check on that") while the tools run; sending results too early can cause timing issues.
• Interruption handling. When the user barges in, the server sends reply.done with status: "interrupted". Drop any queued tool results and flush the speaker so the caller doesn't keep hearing the previous reply.

Step 4: Implement the tools

The dispatch_tool function is where your agent does real work. The Voice Agent API delivers tool.call events with arguments already parsed as a Python dict—no json.loads() needed.

# tools.py
TOOLS = [
    {
        "type": "function",
        "name": "get_weather",
        "description": "Get the current weather for a city.",
        "parameters": {
            "type": "object",
            "properties": {"city": {"type": "string"}},
            "required": ["city"],
        },
    },
    {
        "type": "function",
        "name": "remember",
        "description": "Save something the user wants you to remember.",
        "parameters": {
            "type": "object",
            "properties": {"fact": {"type": "string"}},
            "required": ["fact"],
        },
    },
]

_memory = []

def dispatch_tool(name, args):
    if name == "get_weather":
        # In production: call a real weather API.
        return f"It's 68°F and partly cloudy in {args['city']}."
    if name == "remember":
        _memory.append(args["fact"])
        return f"Got it. I'll remember: {args['fact']}"
    return f"Unknown tool: {name}"

In production, replace the stubs with calls to a real weather API, your CRM, a database, or whatever your application actually does. The tool dispatcher is pure Python—anything you can do from a Python function, the voice agent can do.

Step 5: Run it

python agent.py

The agent greets you. Try:

• "What's the weather in San Francisco?"
• "Remember that my passport expires in March."
• "What did I just ask you to remember?"

The full flow: your speech → STT → LLM (with tools available) → tool call (if applicable) → tool result → LLM continues → TTS → speaker. End to end, that round trip runs in about a second, on one WebSocket.

Latency: keeping the round trip under a second

End-to-end latency on the Voice Agent API lands at roughly one second—from the moment you stop talking to the moment you hear the reply start. Because output audio streams as it's synthesized, the perceived latency (when the first word lands) is lower still, typically sub-second. A natural-feeling agent needs to stay in that range; best-in-class teams push toward 500ms. Here's where the milliseconds go:

The Voice Agent API streams audio output as it's generated, so the user hears the first word of the reply while the rest is still synthesizing. The biggest latency wins on the Python side:

• Don't buffer mic audio. Send 50ms chunks as they arrive—that's what the audio.py example does.
• Don't block in the tool dispatcher. If a tool call takes more than 500ms, the silence becomes audible. Cache hot data, set aggressive timeouts, and consider returning a placeholder ("Let me check on that") while the real call resolves.
• Play reply.audio chunks as they arrive. Never wait for the full response.

Handling interruptions

Real conversations include interruptions. The user changes their mind, asks a follow-up while the agent is still talking, or says "wait, no, the other one." The Voice Agent API handles this server-side, and barge-in is semantic: back-channels like "uh-huh" don't trigger an interruption, but "wait, stop" does.

When the user actually interrupts, the server sends reply.done with status: "interrupted" (and transcript.agent with interrupted: true and the trimmed text). Your client should flush any queued speaker audio and drop any pending tool results, exactly as shown in the receive() loop above.

This is the piece that's hardest to fake with a chained pipeline and simple VAD—and it's where teams tend to feel the difference.

Going to production

The agent above runs against your local microphone. To deploy it, swap the audio transport:

• Phone calls (PSTN)—Bridge through Twilio Media Streams. The Voice Agent API supports audio/pcmu (G.711 μ-law at 8 kHz) natively, so phone audio stays in μ-law end to end with no resampling. Prefer an orchestrator? Use the LiveKit voice agent guide.
• Web apps—Capture audio in the browser with AudioWorklet, then stream it to the Voice Agent API. See Browser integration for the temporary-token flow that keeps your API key off the client.
• Mobile—Same pattern. The native audio-capture APIs (iOS AVAudioEngine, Android AudioRecord) emit PCM you can forward through your server.

For all production deployments, add:

• Session resumption—save the session_id from session.ready and reconnect within 30 seconds without losing context
• Per-session structured logs (user transcript, agent transcript, tool calls, tool results)
• PII redaction on transcripts before they hit your warehouse
• A timeout-and-retry policy for tool calls so a slow backend doesn't kill the call

Using an orchestrator instead: LiveKit and Pipecat

If you already run LiveKit or Pipecat, you don't have to hand-roll the WebSocket loop from this tutorial. The Voice Agent API ships as a drop-in plugin for both, so you get the same accuracy, turn detection, and flat pricing while keeping your existing room management, telephony, and session infrastructure. The Python code above is the right choice when you want full control and minimal dependencies; the plugins are the right choice when you're already invested in an orchestrator.

The complete repository

Fork the runnable Python repo at github.com/kelsey-aai/python-voice-agent-api. It includes mic capture, speaker playback, the WebSocket loop, the tool dispatcher, and example tools you can swap for your own—around 200 lines of Python end to end.

Here's the thing most latency debates miss: the number that actually decides whether an agent feels human isn't the round-trip total, it's whether the agent hears you correctly on the first try. A 900ms reply to the right question beats a 400ms reply to the wrong one every time—because the second one costs you a "sorry, can you repeat that," and now you're three seconds deep. That's why we put the accuracy work into agent_context and dedicated STT before we optimized the last 100ms of TTS. Build the listener first.

Ready to build?

• Talk to a live agent—hear the latency, accuracy, and barge-in for yourself
• Get your free API key and run this tutorial in the next 10 minutes
• Read the full API reference for every event type and config option

Comparing options before you commit? Our guide to choosing an STT API for voice agents breaks down what matters, and pricing has the full flat-rate breakdown.

Frequently asked questions

How do I build a real-time voice agent in Python?

The fastest way to build a real-time voice agent in Python in 2026 is to open a WebSocket to the AssemblyAI Voice Agent API at wss://agents.assemblyai.com/v1/ws, stream microphone audio in as input.audio events, and play the reply.audio events you get back. The Voice Agent API handles streaming speech-to-text, the LLM, text-to-speech, turn detection, and tool calling on a single connection at $4.50/hour, so you don't need to wire up three separate providers. With PyAudio for microphone access and the websockets library, the entire agent fits in well under 100 lines of Python.

What's the difference between the Voice Agent API and chaining STT-LLM-TTS in Python?

The chained architecture uses three providers: a streaming STT API like AssemblyAI's Universal-3.5 Pro Realtime, an LLM, and a streaming TTS. You write the WebSocket bridge, turn-detection logic, and retry handling yourself, and you get one bill and one debugging surface per vendor. The Voice Agent API replaces all of that with a single WebSocket—one provider, one bill, one set of logs. Chained pipelines give you finer control over each layer and let you bring your own LLM; the Voice Agent API is faster to ship, cheaper to run, and easier to operate at scale. Most teams start with the Voice Agent API and only drop to the chained path when they need a specific model that isn't offered.

How does the Voice Agent API compare to the OpenAI Realtime API?

The OpenAI Realtime API is a single speech-to-speech model that does everything in one pass. The Voice Agent API uses dedicated best-in-class models for each step—Universal-3.5 Pro Realtime for listening, a frontier LLM for reasoning, purpose-built TTS for speaking—which gives it stronger accuracy on structured data like account numbers and emails, neural turn detection instead of basic VAD, and a flat $4.50/hour price that's roughly 4x cheaper than the OpenAI Realtime API's ~$18/hour. Pick speech-to-speech if you want one model to own the whole loop; pick the Voice Agent API if you want the best listener and a predictable bill.

How do I add tool calling to a Python voice agent?

Define tools as JSON schemas in the tools field of your session.update message—each needs "type": "function", a name, a description, and a parameter schema. When the LLM decides to call a tool, the Voice Agent API emits a tool.call event with the tool name, arguments (already parsed as a Python dict), and a call_id. Your Python dispatcher runs the actual function, then you send back a tool.result event with that call_id and the result. Send tool results inside your reply.done handler, not immediately on tool.call—the agent speaks a transition phrase while the tools run.

How low can latency go on a Python voice agent?

End to end, a well-tuned Python voice agent on the Voice Agent API lands at about one second from when you stop talking to when the reply starts, with perceived latency (first word out) typically in the 450–950ms range. The biggest wins: keep mic chunks small (~50ms) so end-of-turn detection fires fast, don't block in your tool dispatcher, and play reply.audio chunks as they arrive instead of buffering. Just as important as raw speed is accuracy on the first try—agent_context passes the agent's question to the STT model so short or mumbled replies resolve correctly and you avoid the "can you repeat that" round trip.

Can I use a different LLM with the Voice Agent API?

The Voice Agent API ships with frontier-quality LLMs selected for low-latency conversational performance. If you specifically need a model that isn't available through it, drop to a chained architecture: use Universal-3.5 Pro Realtime for the STT layer and bring your own LLM and TTS. Most teams find the built-in model selection meets their needs and prefer the simpler single-WebSocket architecture.

How do I handle interruptions in a Python voice agent?

The Voice Agent API detects barge-in semantically: back-channels like "uh-huh" don't interrupt, but "wait, stop" does. When the user actually interrupts, the server emits reply.done with status: "interrupted" and transcript.agent with interrupted: true. Your Python client should flush the speaker buffer (close and reopen the PyAudio output stream), drop any pending tool results, and continue listening for the user's new turn. Handling this well is what makes interruptions feel natural—the agent stops talking immediately instead of finishing the previous reply.

‍