You can build a real-time voice agent in Python that listens, thinks, calls your own functions, and talks back, in three short files. This tutorial builds one on AssemblyAI's speech stack: Universal-3.6 Pro Realtime for streaming speech-to-text, an LLM through LLM Gateway, and the AssemblyAI Text-to-Speech API for the voice. The Text-to-Speech API delivers frontier-level quality, runs faster and costs less than other production voice agent TTS models, and has unlimited concurrency. All three run on one AssemblyAI API key and one Python SDK.
The focus here is the part that turns a talking chatbot into an agent: tool calling. By the end you'll have a runnable agent that listens through your microphone, decides when to call Python functions, speaks the result, and stops talking the moment you interrupt it.
If you want the simplest possible loop first, start with our 5-minute voice agent quickstart. For a longer, step-by-step walkthrough of a chained Python agent, see our Python voice agent tutorial.
Why build a Python voice agent on AssemblyAI's speech stack
A voice agent has three jobs: hear the caller, decide what to do, and answer out loud. Doing each with a dedicated model, and owning the loop between them in your own code, gives you three properties that matter in production:
- Speech accuracy that survives real conversations. The listening layer is Universal-3.6 Pro Realtime. On AssemblyAI's English voice-agent benchmark, 12,460 scripted caller scenarios, it posts a 5.19% word error rate, against 7.78% for ElevenLabs Scribe v2, 8.64% for Deepgram Nova-3, and 13.50% for Deepgram Flux EN. That gap is the difference between an agent that hears "my account is 4-9-3-A" and one that doesn't.
- Tool calling in a standard LLM request. Tools are OpenAI-style JSON schemas sent with each LLM call. Pick any tool-capable model on LLM Gateway, or point the same code at your own provider.
- One key, one SDK. The assemblyai Python package covers streaming speech-to-text, LLM Gateway, and text-to-speech, so there's one credential to manage and one bill.
The part most STT APIs miss: agent context
Here's the capability that changes accuracy for voice agents. Universal-3.6 Pro Realtime accepts the agent's own last line as input, a parameter called 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 full of things that trip up 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 expects an email, and resolves "j dot smith at gmail" into an address instead of word soup.
On the structured values agents act on, Universal-3.6 Pro Realtime holds a 14.4% entity error rate on the same benchmark, against 30.1% for Deepgram Flux EN. Because you own the loop in this build, you pass agent_context yourself after every reply. It's one line of code in Step 5.
How this compares to a speech-to-speech model
The most common alternative is a single speech-to-speech model, such as OpenAI's Realtime API: audio goes in, audio comes out, and one model does everything in between. This build takes the cascaded approach instead, with a dedicated model for each step, and that shows up in three ways that matter for a production agent:
- Accuracy on structured data. A speech-to-speech model has no dedicated speech-to-text layer tuned for the account numbers, emails, and confirmation codes that voice agents depend on. Universal-3.6 Pro Realtime with agent_context is built for exactly that.
- A transcript you can see. Every user turn arrives as text before the LLM sees it, so you can log it, redact it, route on it, and evaluate it.
- Each layer is replaceable. Swap the LLM per use case, change the voice per brand, or move to a different orchestrator, without rebuilding the rest.
Get a free API key and run this tutorial: streaming speech-to-text, LLM Gateway, and text-to-speech on one key.
Architecture
Microphone
│ 16 kHz 16-bit mono PCM
▼
Universal-3.6 Pro Realtime (streaming speech-to-text)
│ Turn events: partials, then end_of_turn
▼
Your Python script ──► LLM Gateway (chat completions + tools)
│ │
│ tool_calls ◄─────────┘
├─► dispatch_tool() runs your Python function
│ result ─────────────► LLM Gateway again
▼
AssemblyAI Text-to-Speech (one session for the conversation)
│ 24 kHz 16-bit mono PCM, streamed
▼
Speaker
Your script owns the loop. Speech-to-text streams Turn events. When a turn ends, the script sends the conversation to the LLM with your tool definitions. If the model asks for a tool, the script runs the Python function, sends the result back, and repeats until the model answers in plain text. That answer goes to a text-to-speech session, which streams audio back as it's made. If the user talks over the agent, the script cancels the speech mid-sentence.
Before you start
You'll need:
- An AssemblyAI account and API key. LLM Gateway uses the same key; check that your plan includes it.
- Python 3.9 or later.
- A microphone and speakers. Use headphones: a laptop mic picks up the agent's own voice, and the agent will interrupt itself.
- PortAudio installed system-wide (brew install portaudio on macOS, apt install portaudio19-dev on Debian or Ubuntu).
Install the dependencies. Text-to-speech is in the AssemblyAI Python SDK from version 1.8.2, so upgrade with -U:
pip install -U assemblyai pyaudio python-dotenv
Put your API key in a .env file:
ASSEMBLYAI_API_KEY=your_key_here
Step 1: Capture microphone audio and play replies
PyAudio handles both directions. The two sides run at different rates: the microphone sends 16 kHz audio to speech-to-text, and text-to-speech returns 24 kHz audio, its native rate, as raw 16-bit mono PCM with no header.
# audio.py
import threading
from queue import Queue
import pyaudio
MIC_RATE = 16000 # what we send to speech-to-text
SPEAKER_RATE = 24000 # what AssemblyAI text-to-speech sends back
MIC_CHUNK = 800 # 50 ms of 16 kHz, 16-bit mono audio
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=MIC_RATE,
input=True, frames_per_buffer=MIC_CHUNK,
)
threading.Thread(target=self._loop, daemon=True).start()
def _loop(self):
while self._running:
self.queue.put(self._stream.read(MIC_CHUNK, exception_on_overflow=False))
def chunks(self):
# A generator the streaming client can consume directly
while self._running:
yield self.queue.get()
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._pa.open(
format=pyaudio.paInt16, channels=1, rate=SPEAKER_RATE, output=True,
)
def play(self, audio_bytes):
# Blocks for roughly the length of the chunk, so playback paces itself
self._stream.write(audio_bytes)
def close(self):
self._stream.stop_stream()
self._stream.close()
self._pa.terminate()
Step 2: Set up the three clients
agent.py starts by creating one client per job: a streaming transcriber for listening, an LLM Gateway client for thinking, and a text-to-speech session for speaking. The session stays open for the whole conversation and takes one request per reply. Set the voice explicitly; the example uses Audrey from the voice library.
# agent.py
import json
import os
import queue
import string
import threading
import time
import assemblyai as aai
from assemblyai.streaming.v3 import (
BeginEvent,
RealTimeError,
RealTimeEvents,
RealTimeParameters,
RealTimeSessionParameters,
RealTimeTranscriber,
RealTimeTranscriberOptions,
TurnEvent,
)
from dotenv import load_dotenv
from audio import MIC_RATE, Mic, Speaker
from tools import TOOLS, dispatch_tool
load_dotenv()
API_KEY = os.environ["ASSEMBLYAI_API_KEY"]
LLM_MODEL = "claude-haiku-4-5-20251001" # any tool-capable model on LLM Gateway
MAX_TOOL_ROUNDS = 3
GREETING = "Hi! I can check the weather or remember things for you. What do you need?"
SYSTEM_PROMPT = f"""You are a helpful voice assistant.
Your replies are spoken aloud: one or two short sentences, no lists or markdown.
Use the available tools when they help answer the question.
You have already greeted the user with: {GREETING}"""
# Listen: Universal-3.6 Pro Realtime streaming speech-to-text
stt = RealTimeTranscriber(RealTimeTranscriberOptions(), api_key=API_KEY)
# Think: an LLM through LLM Gateway, on the same API key
llm = aai.LLMGateway(api_key=API_KEY)
# Speak: one text-to-speech session for the whole conversation
tts = aai.TTSSession(
aai.TTSConfig(
voice_id="49c443a4-8a1d-5f7e-b70b-98bdbd9823c3", # Audrey
language="english",
),
api_key=API_KEY,
)
mic = Mic()
speaker = Speaker()
messages = [{"role": "system", "content": SYSTEM_PROMPT}]
turns = queue.Queue() # finished user turns, in order
speaking = threading.Event() # set while agent audio is playing
interrupted = threading.Event() # set when the user talks over the agent
last_spoke_at = 0.0
The model ID can be any tool-capable model on the available models list.
Step 3: Listen, and handle barge-in
Universal-3.6 Pro Realtime sends Turn events as the user speaks: partial transcripts first, then one with end_of_turn set when the model decides the user has finished. Turn detection is built in, so there's no silence timer to tune here; the defaults are what you want for a conversational agent.
The same events drive barge-in. While the agent is speaking, a transcript with real content means the user is talking over it, so on_turn sets the interrupted flag. Backchannels such as "mhm" or "yeah" are filtered out first, so the agent keeps talking through them. This is the filter AssemblyAI's voice agent guide recommends for direct WebSocket integrations.
# agent.py (continued): listening and barge-in
BACKCHANNELS = frozenset({
"mhm", "mm", "mmhm", "uh", "uhhuh", "huh", "um", "umm", "hmm", "hm",
"ah", "oh", "yeah", "yep", "yup", "okay", "ok", "right", "alright", "gotcha",
})
_PUNCT = str.maketrans("", "", string.punctuation)
def is_filler(text, min_words):
tokens = text.lower().translate(_PUNCT).split()
return len(tokens) < min_words or all(t in BACKCHANNELS for t in tokens)
def on_begin(client, event: BeginEvent):
print(f"Listening (session {event.id})")
def on_turn(client, event: TurnEvent):
text = event.transcript.strip()
if not text:
return
if speaking.is_set():
if is_filler(text, min_words=2):
return # "mhm", "yeah": let the agent keep talking
interrupted.set() # a real interruption: stop the reply
elif time.monotonic() - last_spoke_at < 1.0 and is_filler(text, min_words=1):
return # a trailing "mhm" right after the agent finished
if event.end_of_turn:
print(f"You: {text}")
turns.put(text)
def on_error(client, error: RealTimeError):
print(f"Speech-to-text error: {error}")
Step 4: Add tool calling
Tools are plain Python functions plus a JSON schema the LLM can read. The schemas use the OpenAI-compatible format that LLM Gateway tool calling accepts for every provider.
# tools.py
TOOLS = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "Get the current weather for a city.",
"parameters": {
"type": "object",
"properties": {"city": {"type": "string"}},
"required": ["city"],
},
},
},
{
"type": "function",
"function": {
"name": "remember",
"description": "Save something the user wants you to remember.",
"parameters": {
"type": "object",
"properties": {"fact": {"type": "string"}},
"required": ["fact"],
},
},
},
{
"type": "function",
"function": {
"name": "recall",
"description": "List everything the user has asked you to remember.",
"parameters": {"type": "object", "properties": {}},
},
},
]
_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"Saved: {args['fact']}"
if name == "recall":
return _memory or "Nothing saved yet."
return f"Unknown tool: {name}"
think() sends the conversation and the tool list to the LLM. If the response contains tool calls, it runs each one, appends the result as a tool message, and calls the LLM again. When the model answers with text and no tool calls, that text is the reply.
# agent.py (continued): the LLM and tool calls
def tool_call_message(text, calls):
return {
"role": "assistant",
"content": text or None,
"tool_calls": [
{
"id": call.id,
"type": "function",
"function": {"name": call.function.name, "arguments": call.function.arguments},
}
for call in calls
],
}
def think(user_text):
messages.append({"role": "user", "content": user_text})
for _ in range(MAX_TOOL_ROUNDS):
completion = 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 completion.choices if c.message.content).strip()
calls = [call for c in completion.choices for call in (c.message.tool_calls or [])]
if not calls:
messages.append({"role": "assistant", "content": text})
return text
messages.append(tool_call_message(text, calls))
for call in calls:
args = json.loads(call.function.arguments or "{}")
result = dispatch_tool(call.function.name, args)
print(f"Tool: {call.function.name}({args}) -> {result}")
messages.append({
"role": "tool",
"tool_call_id": call.id,
"content": json.dumps({"result": result}),
})
return "Sorry, I couldn't finish that. Could you ask me again?"
Two details are easy to miss:
- Read every choice. LLM Gateway returns tool calls in the OpenAI-compatible structure, but some providers put the model's text in one choice and the tool call in another. Collecting both across all choices keeps the loop provider-agnostic.
- Arguments arrive as a JSON string. Parse call.function.arguments with json.loads() before you dispatch, and echo the original call.id back as tool_call_id.
In production, swap the stubs for a real weather API, your CRM, or a database. Anything you can do from a Python function, the agent can do.
Step 5: Speak, and run it
speak() sends the reply to the text-to-speech session and plays audio chunks as they arrive. Between chunks it checks the interrupted flag; on a barge-in it calls cancel(), which stops the request in flight and drops the rest of the reply. After each reply it updates agent_context, so the next short answer is transcribed in context.
# agent.py (continued): speaking, and the main loop
def speak(text):
global last_spoke_at
if not text:
return
print(f"Agent: {text}")
interrupted.clear()
speaking.set()
try:
for chunk in tts.speak(text):
if interrupted.is_set():
tts.cancel() # barge-in: the server drops the rest of this reply
print(" (interrupted)")
break
speaker.play(chunk)
finally:
speaking.clear()
last_spoke_at = time.monotonic()
# Tell speech-to-text what the agent just said, so a short answer
# ("yes", an email, an order number) is transcribed in context
stt.set_params(RealTimeSessionParameters(agent_context=text[-1750:]))
def main():
stt.on(RealTimeEvents.Begin, on_begin)
stt.on(RealTimeEvents.Turn, on_turn)
stt.on(RealTimeEvents.Error, on_error)
stt.connect(
RealTimeParameters(
speech_model="universal-3-6-pro",
sample_rate=MIC_RATE,
agent_context=GREETING,
)
)
tts.connect()
mic.start()
threading.Thread(target=stt.stream, args=(mic.chunks(),), daemon=True).start()
speak(GREETING)
try:
while True:
try:
user_text = turns.get(timeout=30)
except queue.Empty:
tts.keep_alive() # the TTS session closes after 2 idle minutes
continue
speak(think(user_text))
except KeyboardInterrupt:
print("\nShutting down...")
finally:
mic.stop()
stt.disconnect(terminate=True)
tts.terminate()
speaker.close()
if __name__ == "__main__":
main()
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 ask you to remember?"
- Start a long question, then talk over the agent's answer to interrupt it.
The full flow: your speech, then speech-to-text, then the LLM with tools available, then a tool call if needed, then the tool result back to the LLM, then text-to-speech, then your speaker.
Latency: where the time goes
A natural-feeling agent starts answering quickly after the user stops talking. The time splits across the three services plus your tool calls:
- End of turn. Universal-3.6 Pro Realtime posts a 307 ms median time to final transcript on the Pipecat open benchmark. Send audio in small chunks, as audio.py does with 50 ms reads, so the model always has the latest audio.
- The LLM. Pick a fast model for conversational turns and keep max_tokens low; spoken replies are short. For the lowest time to first audio, stream the model's tokens into the text-to-speech session with generate() and flush() instead of waiting for the full reply; the server starts speaking once it has enough complete sentences. The text-to-speech quickstart shows the pattern.
- Text-to-speech. One session per conversation, as in this build, means no new connection per reply. Play chunks as they arrive rather than buffering the whole reply.
- Your tools. A slow tool call becomes audible silence. Cache hot data, set tight timeouts, and consider having the agent say a short "let me check" before a slow lookup.
Sign up free and run the full agent: Universal-3.6 Pro Realtime, LLM Gateway, and AssemblyAI Text-to-Speech on one key.
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." In this build, barge-in is two pieces of code working together:
- Detecting it. on_turn treats a non-filler transcript during agent speech as an interruption. Short backchannels ("uh-huh", "right") don't count, and neither does a trailing "mhm" right after the agent finishes. Edit the BACKCHANNELS set for your domain. "Yes" and "no" are deliberately left out, because in a booking flow a bare "yes" is a real answer.
- Stopping it. speak() checks the flag between audio chunks and calls tts.cancel(). The server drops the rest of the reply, and the user's new turn is already in the queue for think().
Because each audio chunk is checked before it plays, the agent stops within one chunk of the interruption instead of finishing its sentence.
Going to production
The agent above runs against your local microphone. To deploy it, swap the audio transport:
- Phone calls. Bridge through Twilio Media Streams, which carry 8 kHz μ-law audio. Both services accept that format directly, so phone audio needs no resampling in your code:
import assemblyai as aai
from assemblyai.streaming.v3 import Encoding, RealTimeParameters
stt_params = RealTimeParameters(
speech_model="universal-3-6-pro",
encoding=Encoding.pcm_mulaw,
sample_rate=8000,
)
tts_config = aai.TTSConfig(
voice_id="49c443a4-8a1d-5f7e-b70b-98bdbd9823c3", # Audrey
encoding="pcm_mulaw",
sample_rate=8000, # resampled on the server from the native 24 kHz
)
- Web apps. Capture audio in the browser, and keep your API key on the server. Mint a temporary token for any client that connects directly.
- Long calls. A text-to-speech session lasts up to an hour and closes after two idle minutes, which is why the main loop calls keep_alive(). See limits.
For every production deployment, add per-session structured logs (user turns, agent replies, tool calls, tool results), PII redaction on transcripts before they reach your warehouse, and a timeout-and-retry policy for tool calls so a slow backend doesn't stall the conversation.
Using an orchestrator instead: LiveKit and Pipecat
If you already run LiveKit or Pipecat, you don't have to hand-roll the loop from this tutorial. Both have AssemblyAI plugins for streaming speech-to-text and for text-to-speech (LiveKit, Pipecat), and both orchestrators handle tool calling in their own LLM layer. The plain Python in this tutorial 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 code
The three files above, audio.py, tools.py, and agent.py, are the whole agent: mic capture, speaker playback, streaming speech-to-text, the LLM tool loop, and text-to-speech with barge-in. Copy them into one folder next to your .env and run python agent.py.
Here's the thing most latency debates miss: the number that 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 slightly slower reply to the right question beats a fast reply to the wrong one, because the wrong one costs you a "sorry, can you repeat that," and now you're several seconds deep. That's why agent_context and a dedicated speech-to-text model come first in this build. Build the listener first.
Ready to build?
- Get your free API key and run this tutorial.
- Read the voice agent best practices for turn detection, prompting, and mid-session updates.
- Explore the Text-to-Speech API and its voice library.
Comparing options before you commit? Our guide to choosing an STT API for voice agents breaks down what matters, and pricing has current rates for every product.
Copy the code, swap in your own tools, and ship. Start with a free API key.
Frequently asked questions
How do I build a real-time voice agent in Python?
Connect three services in a loop. Stream microphone audio to Universal-3.6 Pro Realtime with the AssemblyAI Python SDK's RealTimeTranscriber, and wait for a Turn event with end_of_turn set. Send that transcript to an LLM through LLM Gateway, then speak the reply with an AssemblyAI TTSSession and play the raw PCM it returns. With PyAudio for the microphone and speaker, the whole agent fits in three short files and one API key.
How do I add tool calling to a Python voice agent?
Define each tool as an OpenAI-style JSON schema ({"type": "function", "function": {"name", "description", "parameters"}}) and pass the list as tools on every LLM Gateway chat completion. When the response contains tool_calls, parse each call's arguments JSON string, run your Python function, append the result as a tool message with the matching tool_call_id, and call the LLM again. Repeat until the model answers in plain text, then speak that text.
How does this compare to the OpenAI Realtime API?
The OpenAI Realtime API is a single speech-to-speech model that does everything in one pass. This build uses a dedicated model for each step: Universal-3.6 Pro Realtime for listening, an LLM of your choice for reasoning, and AssemblyAI Text-to-Speech for speaking. You get a dedicated speech-to-text layer for structured data like account numbers and emails, a text transcript of every turn, and the freedom to swap any layer independently.
How low can latency go on a Python voice agent?
The biggest levers are small audio chunks (about 50 ms) so end-of-turn detection fires on fresh audio, a fast LLM with a low max_tokens, one text-to-speech session reused for the whole conversation, and playing audio chunks as they arrive. To start speaking even sooner, stream LLM tokens into the session with generate() and flush(). Just as important is accuracy on the first try: agent_context helps short replies resolve correctly, so you avoid a "can you repeat that" round trip.
Can I use a different LLM?
Yes. Change LLM_MODEL to any tool-capable model on LLM Gateway's available models list, or swap the llm client for the openai client pointed at OpenAI directly. LLM Gateway is OpenAI-compatible, so the message and tool formats stay the same. The speech-to-text and text-to-speech code doesn't change.
How do I handle interruptions in a Python voice agent?
Watch Turn events while the agent is speaking. If a transcript has real content (two or more words that aren't all backchannels like "mhm" or "yeah"), set an interrupted flag. In your playback loop, check the flag before each audio chunk and call cancel() on the text-to-speech session when it's set. The server stops the reply, and the user's new turn goes to the LLM next.
What voice does the agent use, and can I change it?
The example sets voice_id to Audrey from AssemblyAI's voice library, which offers thousands of voices across six languages: English, Spanish, French, German, Italian, and Portuguese. Change voice_id, and set language to match your users. Browse voices in the dashboard or in the voices documentation.
