How to build a voice agent with AssemblyAI: streaming STT, LLM, and TTS

Building a voice agent used to mean wiring together three or four vendors: a speech-to-text provider, an LLM, a text-to-speech engine, and some glue for turn detection, each with its own key, protocol, and bill. AssemblyAI now covers the whole loop. Universal-3.6 Pro Realtime listens, LLM Gateway gives you an LLM with tool calling, and the AssemblyAI Text-to-Speech API speaks, all on one API key.

This is a hands-on build guide. It explains how each piece works and why, with protocol-level code over plain WebSockets that assembles into one working agent: it listens, thinks, calls your own tools, talks, and stops when it's interrupted.

What is AssemblyAI's voice agent stack?

A voice agent built on AssemblyAI is a cascade of three services, with your code in the middle:

1. Listen. Universal-3.6 Pro Realtime, AssemblyAI's streaming speech-to-text model, at wss://streaming.assemblyai.com/v3/ws. You stream microphone audio in and get Turn events back, with turn detection built in.
2. Think. An LLM through LLM Gateway, an OpenAI-compatible endpoint at https://llm-gateway.assemblyai.com/v1 that reaches models from several providers on the same AssemblyAI key. Tool calling lives here, in the LLM request.
3. Speak. AssemblyAI Text-to-Speech at wss://tts.us.assemblyai.com/v1/ws: one session per conversation, one request per reply, audio streamed back as it's made. It delivers frontier-level quality, runs faster and costs less than other production voice agent TTS models, and has unlimited concurrency.

AssemblyAI is infrastructure here. We're the Voice AI layer your agent runs on, and you own the loop: every protocol is standard JSON over WebSocket or HTTP, there's no proprietary framework you have to adopt, and it works cleanly with coding agents like Claude Code, Cursor, and Windsurf.

Build the loop yourself, or use an orchestrator

There are two ways to assemble the stack, and it's worth being explicit about when to choose which:

• Your own loop (this guide). Connect the three services directly and write the turn-taking logic yourself. Most control, fewest dependencies, and you see every event.
• An orchestrator. LiveKit and Pipecat handle transport, turn-taking, and interruptions, with AssemblyAI plugins for speech-to-text and text-to-speech. Our Agora and Daily.co guides show transport-specific builds.

This guide covers the first path; the plugins come at the end.

Why build on AssemblyAI's speech stack?

Two reasons stand out: accuracy and a single bill.

Accuracy, on real conversations. On AssemblyAI's English voice-agent benchmark, 12,460 scripted voice-agent scenarios, Universal-3.6 Pro Realtime posts a 5.19% word error rate. From the Universal-3.6 Pro Realtime launch results:

The entity error rate column is the one to watch. When a caller says a phone number, an email, or a name, that's usually the whole point of the call, and it's exactly where weaker models fall apart. A 14.4% entity error rate against Deepgram Flux EN's 30.1% is the difference between an agent that books the appointment and one that mishears the callback number. See the full methodology on our benchmarks page.

One key, one bill. Speech-to-text, LLM Gateway, and text-to-speech all run on the same AssemblyAI API key. Each is billed on its own usage: streaming speech-to-text at $0.45/hr, LLM Gateway per token for the model you pick, and text-to-speech at $15 per 1M characters you send. Current rates for each are on the pricing page.

Connect to streaming speech-to-text

The listening side is one WebSocket. Put the model and sample rate in the query string, authenticate with your API key in the Authorization header, and send microphone audio as binary frames of raw 16-bit PCM. The server answers with a Begin message, then Turn messages as the user speaks.

# agent.py: a voice agent on AssemblyAI's speech stack, over raw WebSockets
# pip install -U websockets sounddevice openai
import asyncio
import base64
import json
import os
import string
from urllib.parse import urlencode

import sounddevice as sd
import websockets
from openai import OpenAI

API_KEY = os.environ["ASSEMBLYAI_API_KEY"]
HEADERS = {"Authorization": API_KEY}  # the raw key, no "Bearer" prefix
MIC_RATE, SPEAKER_RATE = 16000, 24000

# Listen: Universal-3.6 Pro Realtime streaming speech-to-text
STT_URL = "wss://streaming.assemblyai.com/v3/ws?" + urlencode(
    {"speech_model": "universal-3-6-pro", "sample_rate": MIC_RATE}
)
# Speak: AssemblyAI Text-to-Speech, one session for the whole conversation
TTS_URL = "wss://tts.us.assemblyai.com/v1/ws?" + urlencode({
    "voice_id": "49c443a4-8a1d-5f7e-b70b-98bdbd9823c3",  # Audrey
    "language": "english",
    "sample_rate": SPEAKER_RATE,
    "encoding": "pcm_s16le",
})


async def stream_mic(stt_ws):
    loop = asyncio.get_running_loop()
    chunks = asyncio.Queue()

    def on_audio(indata, frames, time_info, status):
        loop.call_soon_threadsafe(chunks.put_nowait, bytes(indata))

    # 50 ms blocks of 16-bit mono PCM, sent as binary WebSocket frames
    with sd.RawInputStream(samplerate=MIC_RATE, channels=1, dtype="int16",
                           blocksize=800, callback=on_audio):
        while True:
            await stt_ws.send(await chunks.get())

A few details that trip people up:

• Auth. The Authorization header takes the raw API key, with no Bearer prefix. Browser clients should never hold your key: mint a temporary token on your server instead.
• Audio format. Send 16 kHz, 16-bit mono PCM as binary frames, in chunks of about 50 ms. Higher sample rates don't improve accuracy.
• Text-to-speech is 24 kHz. The speaking side returns 24 kHz audio, its native rate, so the speaker stream runs at a different rate from the microphone.

How turn detection works

This is the part most stacks get wrong, and it's where the speech model earns its keep.

Traditional voice pipelines detect end-of-turn with a voice activity detector, essentially a silence timer. Someone stops making sound for N milliseconds, and the agent assumes they're done. The problem shows up the first time a caller pauses mid-sentence to think: the agent barges in, and the conversation falls apart.

Universal-3.6 Pro Realtime makes the turn decision inside the speech service. It sends partial Turn messages as the user speaks. After a short pause it checks whether the transcript reads as a finished sentence, and if it does, sends a Turn with end_of_turn: true; a longer maximum silence ends the turn regardless. A short mid-sentence pause doesn't end the turn, because the sentence isn't finished yet. On Pipecat's open STT benchmark, it posts a 307 ms median time to final transcript.

For most agents, the right configuration is the default: send no silence parameters, and act on end_of_turn. What you will want to tune is context, mid-conversation, with an UpdateConfiguration message.

Give the model context mid-conversation

Universal-3.6 Pro Realtime accepts updates while the session is open. The most useful one for a voice agent is agent_context: what your agent just said. A caller's reply to "What's the email on file?" is short and full of spelled-out characters; a model that knows the question expects an email and transcribes one. Send it after every agent reply, along with the rest of the listening logic:

# agent.py (continued): turns, barge-in, and agent context
BACKCHANNELS = {"mhm", "mm", "uh", "um", "hmm", "yeah", "yep", "okay", "ok", "right", "gotcha"}


def is_filler(text):
    words = text.lower().translate(str.maketrans("", "", string.punctuation)).split()
    return len(words) < 2 or all(w in BACKCHANNELS for w in words)


async def listen(stt_ws, turns, speaking, interrupted):
    async for raw in stt_ws:
        msg = json.loads(raw)
        if msg["type"] == "Begin":
            print(f"Listening (session {msg['id']})")
        elif msg["type"] == "Turn":
            text = msg["transcript"].strip()
            if not text:
                continue
            if speaking.is_set():
                if is_filler(text):
                    continue          # "mhm", "yeah": keep talking
                interrupted.set()     # a real interruption
            if msg["end_of_turn"]:
                print(f"You:   {text}")
                await turns.put(text)


async def set_agent_context(stt_ws, agent_text):
    # Tell the model what the agent just said, so a short answer
    # ("yes", an email, an order number) is transcribed in context
    await stt_ws.send(json.dumps({
        "type": "UpdateConfiguration",
        "agent_context": agent_text[-1750:],
    }))

UpdateConfiguration also takes prompt and keyterms_prompt, so you can steer recognition by conversation stage. From AssemblyAI's voice agent best practices:

How interruption handling works

Interruptions need two decisions: did the user really interrupt, and how do you stop talking?

Detecting it. While the agent is speaking, users produce backchannels ("uh-huh," "right," "mm-hm") that aren't interruptions. The listen() function above handles this the way AssemblyAI's direct WebSocket guide recommends: during agent speech, a Turn with fewer than two words, or made entirely of backchannels, is ignored; anything else sets the interrupted flag. Edit the backchannel set for your domain, and leave "yes" and "no" out, since they're often real answers.

Stopping it. On the speaking side, send a Cancel frame. The server stops synthesis, discards the rest of the reply, and answers with Cancelled. A short tail of audio that was already in transit can arrive before Cancelled, so drop every Audio frame between the two:

# agent.py (continued): speaking, with barge-in
async def speak(tts_ws, speaker, text, interrupted):
    await tts_ws.send(json.dumps({"type": "Generate", "text": text}))
    await tts_ws.send(json.dumps({"type": "Flush"}))
    cancelled = False
    async for raw in tts_ws:
        msg = json.loads(raw)
        if interrupted.is_set() and not cancelled:
            await tts_ws.send(json.dumps({"type": "Cancel"}))
            cancelled = True
        if cancelled:
            if msg["type"] == "Cancelled":
                return            # the server dropped the rest of the reply
            continue              # discard audio that was already in transit
        if msg["type"] == "Audio":
            pcm = base64.b64decode(msg["audio"])  # raw 24 kHz 16-bit PCM
            await asyncio.to_thread(speaker.write, pcm)
        elif msg["type"] == "FlushDone":
            return
        elif msg["type"] == "Error":
            raise RuntimeError(msg["error"])

The session stays open after a cancel, so the next reply goes out on the same connection. See message frames for the full protocol.

Add tool calling to your agent

Tools live in the LLM request. Define each one as an OpenAI-style function schema, pass the list as tools, and when the response contains tool_calls, run the function and send the result back as a tool message. Because LLM Gateway is OpenAI-compatible, the standard openai client works by pointing base_url at the gateway and using your AssemblyAI key:

# agent.py (continued): the LLM, with tool calling
llm = OpenAI(base_url="https://llm-gateway.assemblyai.com/v1", api_key=API_KEY)
LLM_MODEL = "claude-haiku-4-5-20251001"  # any tool-capable model on LLM Gateway
SYSTEM_PROMPT = ("You are a friendly support agent for an online store. Your replies are "
                 "spoken aloud: one or two short sentences, no lists or markdown.")

TOOLS = [{
    "type": "function",
    "function": {
        "name": "lookup_order",
        "description": "Look up the status of an order by its order number.",
        "parameters": {
            "type": "object",
            "properties": {"order_id": {"type": "string"}},
            "required": ["order_id"],
        },
    },
}]


def run_tool(name, args):
    if name == "lookup_order":
        # In production: query your order system
        return {"order_id": args["order_id"], "status": "shipped", "eta": "Thursday"}
    return {"error": f"unknown tool {name}"}


def think(messages, user_text):
    messages.append({"role": "user", "content": user_text})
    for _ in range(3):
        resp = llm.chat.completions.create(
            model=LLM_MODEL, messages=messages, tools=TOOLS, max_tokens=300,
        )
        # Text and tool calls can arrive in separate choices, so read them all
        text = " ".join(c.message.content for c in resp.choices if c.message.content).strip()
        calls = [tc for c in resp.choices for tc in (c.message.tool_calls or [])]
        if not calls:
            messages.append({"role": "assistant", "content": text})
            return text
        messages.append({"role": "assistant", "content": text or None,
                         "tool_calls": [tc.model_dump() for tc in calls]})
        for tc in calls:
            result = run_tool(tc.function.name, json.loads(tc.function.arguments or "{}"))
            messages.append({"role": "tool", "tool_call_id": tc.id,
                             "content": json.dumps(result)})
    return "Sorry, I couldn't finish that. Could you ask me again?"

The gotchas worth memorizing:

• Read every choice. LLM Gateway keeps tool calls in the OpenAI-compatible structure, but some providers return the model's text and its tool call in separate choices. Collect both across all choices.
• Arguments are a JSON string. Parse function.arguments before you dispatch, and echo the original id as tool_call_id.
• Pick a tool-capable model. The available models list shows which models support tools. See tool calling for the full reference.

Put it together

The main loop opens both WebSockets, starts the microphone and listener tasks, and then runs one turn at a time: take a finished user turn, think, speak, and update agent_context. The text-to-speech session sends Begin first; a rejected key or parameter arrives as Error instead. A KeepAlive every 30 idle seconds holds the session open, since it closes after two idle minutes.

# agent.py (continued): wiring it together
async def main():
    turns = asyncio.Queue()
    speaking, interrupted = asyncio.Event(), asyncio.Event()
    messages = [{"role": "system", "content": SYSTEM_PROMPT}]

    async with websockets.connect(STT_URL, additional_headers=HEADERS) as stt_ws, \
            websockets.connect(TTS_URL, additional_headers=HEADERS) as tts_ws:
        begin = json.loads(await tts_ws.recv())  # the TTS server speaks first
        if begin["type"] == "Error":
            raise SystemExit(begin["error"])

        with sd.RawOutputStream(samplerate=SPEAKER_RATE, channels=1, dtype="int16") as speaker:
            tasks = [
                asyncio.create_task(stream_mic(stt_ws)),
                asyncio.create_task(listen(stt_ws, turns, speaking, interrupted)),
            ]
            try:
                while True:
                    try:
                        user_text = await asyncio.wait_for(turns.get(), timeout=30)
                    except asyncio.TimeoutError:
                        # The TTS session closes after 2 idle minutes
                        await tts_ws.send(json.dumps({"type": "KeepAlive"}))
                        continue
                    reply = await asyncio.to_thread(think, messages, user_text)
                    print(f"Agent: {reply}")
                    interrupted.clear()
                    speaking.set()
                    try:
                        await speak(tts_ws, speaker, reply, interrupted)
                    finally:
                        speaking.clear()
                    await set_agent_context(stt_ws, reply)
            finally:
                for task in tasks:
                    task.cancel()
                await stt_ws.send(json.dumps({"type": "Terminate"}))
                await tts_ws.send(json.dumps({"type": "Terminate"}))


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        pass

Install the dependencies and run it, with headphones on so the agent doesn't hear itself:

pip install -U websockets sounddevice openai
export ASSEMBLYAI_API_KEY=your_key_here
python agent.py

Ask "Where is order 4 9 3 A?" and the agent calls lookup_order, then speaks the result. Talk over a long answer and it stops.

Go multilingual

Both ends of the stack handle more than English. Universal-3.6 Pro Realtime transcribes 32 languages. AssemblyAI Text-to-Speech speaks six: English, Spanish, French, German, Italian, and Portuguese, with thousands of voices in its voice library.

On the speaking side, set the language query parameter to the language's full English name, such as language=spanish. The text-to-speech API doesn't detect language from the text, so set it per session to match your users. The model, assemblyai-tts-1, speaks every voice in all six languages, so you can keep one voice across languages.

Drop-in plugins: LiveKit and Pipecat

If you're already building on an orchestration framework, you don't have to hand-wire anything. LiveKit and Pipecat are partners, and AssemblyAI ships plugins for both, so you can slot in the whole speech stack with a few lines of config. In LiveKit, the session looks like this (from the LiveKit text-to-speech guide):

import os

from livekit.agents import AgentSession, TurnHandlingOptions
from livekit.plugins import assemblyai, openai

tts = assemblyai.TTS(
    model="assemblyai-tts-1",
    voice="49c443a4-8a1d-5f7e-b70b-98bdbd9823c3",  # Audrey
)

session = AgentSession(
    stt=assemblyai.STT(model="universal-3-6-pro", mode="balanced"),
    # The LLM Gateway, through its OpenAI-compatible API
    llm=openai.LLM(
        model="gemma-4-31b",
        base_url="https://llm-gateway.assemblyai.com/v1",
        api_key=os.environ["ASSEMBLYAI_API_KEY"],
    ),
    tts=tts,
    turn_handling=TurnHandlingOptions(
        turn_detection="stt",          # AssemblyAI decides when the user has finished
        endpointing={"min_delay": 0},  # so LiveKit adds no delay on top
    ),
)

The full LiveKit example adds Silero VAD for barge-in; see Universal-3.6 Pro on LiveKit for speech-to-text options. Pipecat works the same way, with AssemblyAI as the speech-to-text and text-to-speech services in your pipeline; see the Pipecat text-to-speech guide. Both frameworks handle transport and orchestration; AssemblyAI handles the listening and the voice.

Tune your agent, then ship it

Here's the thing about voice agents: the model quality compounds. Every point of WER you shave off means fewer misheard names, fewer wrong tool calls, and fewer moments where the caller has to repeat themselves. That's the difference between a demo and a system people trust with a phone number.

AssemblyAI's speech stack gives you that accuracy floor, built-in turn detection, and a voice built for conversation, on one API key, without asking you to become a distributed-systems expert first. Wire up two WebSockets and an LLM call, and the hard parts are handled.

Start with the voice agent best practices, then grab your API key and build your own.

Frequently asked questions

How do I build a voice agent with AssemblyAI?

Connect three services in a loop. Stream microphone audio to Universal-3.6 Pro Realtime at wss://streaming.assemblyai.com/v3/ws and wait for a Turn message with end_of_turn: true. Send that transcript to an LLM through LLM Gateway, which is OpenAI-compatible and supports tool calling. Then send the reply to AssemblyAI Text-to-Speech at wss://tts.us.assemblyai.com/v1/ws and play the audio it streams back. All three use the same API key.

What does each part of the stack handle?

Universal-3.6 Pro Realtime handles listening: streaming transcription, end-of-turn detection, and SpeechStarted events. LLM Gateway handles reasoning and tool calling across models from several providers. AssemblyAI Text-to-Speech handles the voice: streamed audio, a session that stays open for the conversation, and Cancel for barge-in. Your code owns the loop between them.

Which speech-to-text model powers it, and how accurate is it?

Universal-3.6 Pro Realtime, AssemblyAI's streaming flagship. On AssemblyAI's English voice-agent benchmark it posts a 5.19% word error rate and a 14.4% entity error rate, versus 30.1% entity error for Deepgram Flux EN, so names and phone numbers survive the call.

How do I authenticate to the WebSockets?

Send your AssemblyAI API key in the Authorization header when you open each WebSocket. Speech-to-text takes the raw key with no Bearer prefix. Browser and mobile clients should use a short-lived temporary token minted on your server rather than shipping the key.

How does the agent decide when a caller has finished speaking?

Universal-3.6 Pro Realtime makes the end-of-turn decision in the speech service, based on whether the transcript reads as a finished sentence, and marks it with end_of_turn: true on a Turn message. On Pipecat's open STT benchmark it posts a 307 ms median time to final transcript. For most agents the defaults are right. To improve recognition mid-call, send UpdateConfiguration with context such as agent_context or a prompt.

Do I need a proprietary SDK, or can I use my existing framework?

No SDK is required: both WebSockets speak standard JSON, and LLM Gateway works with the standard openai client. The AssemblyAI Python SDK wraps all three services if you prefer, and LiveKit and Pipecat have AssemblyAI plugins for speech-to-text and text-to-speech.