Introduction to Building a Voice Assistant Backend with Claude Code
Voice assistants have evolved from simple command-response systems into sophisticated conversational agents capable of understanding context, maintaining dialogue state, and performing complex tasks. With the release of Claude Code and the Anthropic API, developers now have access to a powerful reasoning engine that can be integrated into voice-first applications. This tutorial walks you through building a production-ready voice assistant backend that combines speech-to-text, Claude's language capabilities, and text-to-speech into a seamless pipeline.
What Is a Voice Assistant Backend?
A voice assistant backend is the server-side infrastructure that processes audio input, transcribes it into text, generates an intelligent response using a large language model, and converts that response back into audio for playback. Unlike chat-based assistants, voice assistants must handle additional concerns such as streaming latency, audio format conversion, interruption handling, and session management for ephemeral interactions.
Claude Code refers to Anthropic's family of models accessible via the Messages API, including the Claude 3.5 and Claude 3.7 series. These models excel at instruction following, tool use, and maintaining long conversational context — all critical for voice assistants that need to feel responsive and intelligent.
Why It Matters
Building a voice assistant backend matters for several reasons. First, voice is the most natural human interface, and demand for hands-free, eyes-free interactions continues to grow across automotive, healthcare, accessibility, and smart home domains. Second, Claude's strong reasoning and tool-use capabilities enable assistants that can actually do things rather than just retrieve information. Third, by building your own backend rather than relying on a closed platform, you retain full control over data privacy, customization, latency optimization, and integration with proprietary systems.
Key advantages of using Claude for voice assistants include:
- Low latency streaming: Claude supports streaming responses, which lets you begin text-to-speech synthesis before the full response is generated, dramatically reducing perceived latency.
- Tool use: Claude can call external functions to fetch data, control IoT devices, query databases, or trigger workflows.
- Extended context: Large context windows allow the assistant to remember details across a long conversation without aggressive summarization.
- System prompt customization: You can define personality, guardrails, and domain expertise through detailed system prompts.
Architecture Overview
Before writing code, it helps to understand the architecture. A typical voice assistant backend consists of the following components:
- Audio Ingestion Layer: Receives audio chunks from the client over WebSocket or HTTP.
- Speech-to-Text (STT) Engine: Converts audio to text. Common options include OpenAI Whisper, Deepgram, AssemblyAI, or a self-hosted Whisper instance.
- Conversation Manager: Maintains session state, message history, and orchestrates the flow between STT, LLM, and TTS.
- Claude Integration Layer: Sends the transcribed text along with conversation history to the Anthropic API and streams back the response.
- Text-to-Speech (TTS) Engine: Converts Claude's text response into audio. Options include ElevenLabs, OpenAI TTS, Amazon Polly, or Coqui.
- Audio Delivery Layer: Streams synthesized audio back to the client.
The diagram below illustrates the data flow conceptually: Audio Input → STT → Conversation Manager → Claude API → TTS → Audio Output. The critical optimization is that STT, Claude streaming, and TTS can all operate in a pipelined fashion so that audio begins playing while Claude is still generating.
Prerequisites and Project Setup
You will need the following before starting:
- Python 3.11 or later
- An Anthropic API key (obtainable from the Anthropic Console)
- A speech-to-text provider API key (this tutorial uses Deepgram for its streaming capabilities, but the code is adaptable)
- A text-to-speech provider API key (this tutorial uses ElevenLabs for natural-sounding voices)
uvorpipfor dependency management
Create a new project directory and set up a virtual environment:
mkdir voice-assistant-backend
cd voice-assistant-backend
python -m venv .venv
source .venv/bin/activate
Install the required dependencies:
pip install anthropic fastapi uvicorn websockets python-dotenv httpx pydantic
Create a .env file to store your API keys securely:
ANTHROPIC_API_KEY=sk-ant-xxxxxxxxxxxxxxxxxxxxxxxx
DEEPGRAM_API_KEY=your_deepgram_key_here
ELEVENLABS_API_KEY=your_elevenlabs_key_here
ASSISTANT_VOICE_ID=your_preferred_voice_id
Never commit this file to version control. Add it to your .gitignore.
Building the Conversation Manager
The conversation manager is the heart of the backend. It tracks sessions, stores message history, and coordinates the pipeline. Let us start by defining the data models and session storage.
Create a file named models.py:
from pydantic import BaseModel, Field
from enum import Enum
from typing import Optional
import time
class Role(str, Enum):
USER = "user"
ASSISTANT = "assistant"
SYSTEM = "system"
class Message(BaseModel):
role: Role
content: str
timestamp: float = Field(default_factory=time.time)
class Session(BaseModel):
session_id: str
messages: list[Message] = Field(default_factory=list)
created_at: float = Field(default_factory=time.time)
last_active: float = Field(default_factory=time.time)
def add_message(self, role: Role, content: str) -> Message:
msg = Message(role=role, content=content)
self.messages.append(msg)
self.last_active = time.time()
return msg
def to_claude_messages(self) -> list[dict]:
"""Convert session history to Anthropic API message format."""
return [
{"role": m.role.value, "content": m.content}
for m in self.messages
if m.role != Role.SYSTEM
]
Next, create a session store that manages active sessions in memory. For production, you would replace this with Redis or a database-backed store.
Create session_store.py:
import time
from collections import defaultdict
from models import Session, Role
class SessionStore:
def __init__(self, max_sessions: int = 1000, ttl_seconds: int = 1800):
self._sessions: dict[str, Session] = {}
self._max_sessions = max_sessions
self._ttl = ttl_seconds
def get_or_create(self, session_id: str) -> Session:
if session_id not in self._sessions:
self._evict_expired()
if len(self._sessions) >= self._max_sessions:
self._evict_oldest()
self._sessions[session_id] = Session(session_id=session_id)
return self._sessions[session_id]
def _evict_expired(self):
now = time.time()
expired = [
sid for sid, s in self._sessions.items()
if now - s.last_active > self._ttl
]
for sid in expired:
del self._sessions[sid]
def _evict_oldest(self):
if not self._sessions:
return
oldest = min(self._sessions.values(), key=lambda s: s.last_active)
del self._sessions[oldest.session_id]
def delete(self, session_id: str):
self._sessions.pop(session_id, None)
Integrating Claude with Streaming
The Claude integration layer is where the assistant's intelligence lives. We will use the Anthropic Python SDK with streaming enabled so that we can forward partial responses to the TTS engine as they arrive. This pipelining is the single most important optimization for voice assistants because it reduces time-to-first-audio from several seconds to under a second.
Create claude_service.py:
import os
from typing import AsyncGenerator
from anthropic import AsyncAnthropic
from models import Session, Role
class ClaudeService:
def __init__(self):
self.client = AsyncAnthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
self.model = "claude-3-5-sonnet-20241022"
self.system_prompt = (
"You are Aria, a helpful voice assistant. Keep responses concise "
"and conversational since they will be spoken aloud. Avoid markdown "
"formatting, bullet points, or special characters that do not "
"translate well to speech. If the user asks a follow-up question, "
"use conversation history for context. If you do not know something, "
"say so honestly rather than guessing."
)
async def stream_response(
self, session: Session, user_text: str
) -> AsyncGenerator[str, None]:
"""Stream Claude's response token by token."""
session.add_message(Role.USER, user_text)
messages = session.to_claude_messages()
full_response = ""
async with self.client.messages.stream(
model=self.model,
max_tokens=512,
system=self.system_prompt,
messages=messages,
) as stream:
async for text in stream.text_stream:
full_response += text
yield text
session.add_message(Role.ASSISTANT, full_response)
Notice that we cap max_tokens at 512. For voice interactions, shorter responses are almost always better because users are waiting in real time. You can adjust this based on your use case, but be mindful that long monologues make for poor voice experiences.
Adding Tool Use for Real Actions
A voice assistant that can only chat is limited. Claude's tool-use feature lets the assistant call functions to retrieve information or perform actions. Let us add a simple weather tool as an example.
Add this to claude_service.py or create a separate tools.py file:
import httpx
import os
async def get_weather(location: str) -> dict:
"""Fetch current weather for a given location."""
# Replace with a real weather API such as OpenWeatherMap
async with httpx.AsyncClient() as client:
response = await client.get(
"https://api.openweathermap.org/data/2.5/weather",
params={
"q": location,
"appid": os.getenv("OPENWEATHER_API_KEY", ""),
"units": "metric",
},
)
if response.status_code == 200:
data = response.json()
return {
"location": location,
"temperature": data["main"]["temp"],
"description": data["weather"][0]["description"],
}
return {"error": f"Could not fetch weather for {location}"}
WEATHER_TOOL = {
"name": "get_weather",
"description": "Get the current weather for a given city or location.",
"input_schema": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "The city name, e.g. 'San Francisco, CA'",
}
},
"required": ["location"],
},
}
Now update the stream_response method to include the tool and handle tool-use responses. This requires a two-step flow: first Claude may request a tool call, then you execute the tool and send the result back.
import json
class ClaudeService:
# ... previous __init__ code ...
def __init__(self):
self.client = AsyncAnthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
self.model = "claude-3-5-sonnet-20241022"
self.tools = [WEATHER_TOOL]
self.system_prompt = (
"You are Aria, a helpful voice assistant. Keep responses concise "
"and conversational since they will be spoken aloud. "
"Use the get_weather tool when users ask about weather conditions."
)
async def stream_response(
self, session: Session, user_text: str
) -> AsyncGenerator[str, None]:
session.add_message(Role.USER, user_text)
messages = session.to_claude_messages()
full_response = ""
async with self.client.messages.stream(
model=self.model,
max_tokens=512,
system=self.system_prompt,
tools=self.tools,
messages=messages,
) as stream:
async for text in stream.text_stream:
full_response += text
yield text
# Check if Claude wants to use a tool
response = await stream.get_final_message()
if response.stop_reason == "tool_use":
async for chunk in self._handle_tool_use(response, session):
yield chunk
else:
if full_response:
session.add_message(Role.ASSISTANT, full_response)
async def _handle_tool_use(
self, response, session: Session
) -> AsyncGenerator[str, None]:
"""Execute tool calls and stream the follow-up response."""
# Append the assistant's tool-use message
session.messages.append(Message(
role=Role.ASSISTANT,
content=str(response.content)
))
tool_results = []
for block in response.content:
if block.type == "tool_use":
if block.name == "get_weather":
result = await get_weather(**block.input)
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": json.dumps(result),
})
messages = session.to_claude_messages()
messages.append({"role": "assistant", "content": response.content})
messages.append({"role": "user", "content": tool_results})
full_response = ""
async with self.client.messages.stream(
model=self.model,
max_tokens=512,
system=self.system_prompt,
tools=self.tools,
messages=messages,
) as stream:
async for text in stream.text_stream:
full_response += text
yield text
if full_response:
session.add_message(Role.ASSISTANT, full_response)
You will need to add the Message import at the top of the file. This tool-use pattern can be extended to any number of tools — calendar lookups, smart home control, database queries, and more.
Integrating Speech-to-Text
For real-time voice assistants, streaming STT is essential. Deepgram provides a WebSocket-based streaming API that transcribes audio as it arrives. We will create a service that connects to Deepgram and forwards transcribed text to the conversation manager.
Create stt_service.py:
import os
import json
import asyncio
import websockets
from typing import AsyncGenerator
class STTService:
DEEPGRAM_URL = "wss://api.deepgram.com/v1/listen"
def __init__(self):
self.api_key = os.getenv("DEEPGRAM_API_KEY")
async def transcribe_stream(
self, audio_stream: AsyncGenerator[bytes, None]
) -> AsyncGenerator[str, None]:
"""Stream audio to Deepgram and yield transcribed text."""
params = "?encoding=linear16&sample_rate=16000&channels=1"
headers = {"Authorization": f"Token {self.api_key}"}
async with websockets.connect(
self.DEEPGRAM_URL + params, extra_headers=headers
) as ws:
# Forward audio to Deepgram
async def send_audio():
async for chunk in audio_stream:
await ws.send(chunk)
await ws.send(json.dumps({"type": "CloseStream"}))
# Receive transcriptions
async def receive_text():
async for message in ws:
data = json.loads(message)
if data.get("type") == "Results":
transcript = data["channel"]["alternatives"][0]["transcript"]
if transcript and data.get("is_final"):
yield transcript
send_task = asyncio.create_task(send_audio())
async for text in receive_text():
yield text
await send_task
Integrating Text-to-Speech
For TTS, we will use ElevenLabs' streaming API, which returns audio chunks as they are synthesized. This allows us to stream audio back to the client while Claude is still generating text.
Create tts_service.py:
import os
import httpx
from typing import AsyncGenerator
class TTSService:
BASE_URL = "https://api.elevenlabs.io/v1/text-to-speech"
def __init__(self):
self.api_key = os.getenv("ELEVENLABS_API_KEY")
self.voice_id = os.getenv("ASSISTANT_VOICE_ID", "21m00Tcm4TlvDq8ikWAM")
async def synthesize_stream(
self, text: str
) -> AsyncGenerator[bytes, None]:
"""Stream synthesized audio chunks from ElevenLabs."""
url = f"{self.BASE_URL}/{self.voice_id}/stream"
headers = {
"xi-api-key": self.api_key,
"Content-Type": "application/json",
}
payload = {
"text": text,
"model_id": "eleven_turbo_v2",
"voice_settings": {
"stability": 0.5,
"similarity_boost": 0.75,
},
}
async with httpx.AsyncClient() as client:
async with client.stream(
"POST", url, headers=headers, json=payload, timeout=30.0
) as response:
response.raise_for_status()
async for chunk in response.aiter_bytes():
yield chunk
Building the WebSocket Server
Now we tie everything together with a FastAPI WebSocket endpoint. The server accepts audio from the client, transcribes it, sends the text to Claude, and streams synthesized audio back. This is the orchestration layer that makes the pipeline work.
Create main.py:
import os
import uuid
import json
import asyncio
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from dotenv import load_dotenv
from session_store import SessionStore
from claude_service import ClaudeService
from stt_service import STTService
from tts_service import TTSService
load_dotenv()
app = FastAPI(title="Voice Assistant Backend")
session_store = SessionStore()
claude_service = ClaudeService()
stt_service = STTService()
tts_service = TTSService()
@app.websocket("/voice/{session_id}")
async def voice_endpoint(websocket: WebSocket, session_id: str):
await websocket.accept()
session = session_store.get_or_create(session_id)
audio_queue: asyncio.Queue[bytes] = asyncio.Queue()
text_queue: asyncio.Queue[str] = asyncio.Queue()
async def receive_audio():
"""Receive audio chunks from the client."""
try:
while True:
data = await websocket.receive_bytes()
await audio_queue.put(data)
except WebSocketDisconnect:
await audio_queue.put(None)
async def transcribe():
"""Run STT on incoming audio and forward text."""
async def audio_stream():
while True:
chunk = await audio_queue.get()
if chunk is None:
break
yield chunk
async for text in stt_service.transcribe_stream(audio_stream()):
await text_queue.put(text)
await text_queue.put(None)
async def generate_responses():
"""Send transcribed text to Claude and stream back audio."""
while True:
user_text = await text_queue.get()
if user_text is None:
break
# Accumulate text for TTS in sentence-sized chunks
buffer = ""
async for token in claude_service.stream_response(session, user_text):
buffer += token
# Flush on sentence boundaries for lower latency
if any(p in buffer for p in [".", "!", "?", "\n"]):
if buffer.strip():
async for audio_chunk in tts_service.synthesize_stream(buffer.strip()):
await websocket.send_bytes(audio_chunk)
buffer = ""
# Flush any remaining text
if buffer.strip():
async for audio_chunk in tts_service.synthesize_stream(buffer.strip()):
await websocket.send_bytes(audio_chunk)
# Send end-of-response marker
await websocket.send_text(json.dumps({"type": "response_complete"}))
try:
await asyncio.gather(
receive_audio(),
transcribe(),
generate_responses(),
)
except WebSocketDisconnect:
pass
finally:
session_store.delete(session_id)
@app.get("/health")
async def health():
return {"status": "healthy"}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
Run the server with:
python main.py
The server will start on port 8000. Clients can connect to ws://localhost:8000/voice/{session_id}, send raw PCM audio (16-bit, 16kHz, mono), and receive audio chunks back.
Adding Interruption Handling
Real voice assistants must handle interruptions gracefully. If a user starts speaking while the assistant is responding, the assistant should stop generating and start listening. This is known as barge-in. Implementing full barge-in requires voice activity detection on the incoming audio stream while TTS is playing. Here is a simplified approach:
class BargeInManager:
def __init__(self):
self._cancelled = False
def cancel(self):
self._cancelled = True
def reset(self):
self._cancelled = False
@property
def is_cancelled(self) -> bool:
return self._cancelled
In the generate_responses function, check the cancellation flag before sending each audio chunk:
async def generate_responses():
barge_in = BargeInManager()
while True:
user_text = await text_queue.get()
if user_text is None:
break
barge_in.reset()
buffer = ""
async for token in claude_service.stream_response(session, user_text):
if barge_in.is_cancelled:
break
buffer += token
if any(p in buffer for p in [".", "!", "?", "\n"]):
if buffer.strip():
async for audio_chunk in tts_service.synthesize_stream(buffer.strip()):
if barge_in.is_cancelled:
break
await websocket.send_bytes(audio_chunk)
buffer = ""
await websocket.send_text(json.dumps({"type": "response_complete"}))
To trigger barge-in, you would run a lightweight VAD (voice activity detection) model like Silero VAD on the incoming audio stream and call barge_in.cancel() when speech is detected during response playback.
Best Practices
Optimize for Time-to-First-Audio
The most important metric for voice assistants is the delay between when a user finishes speaking and when the assistant begins responding. Aim for under 800 milliseconds. Achieve this by streaming at every stage: stream STT results, stream Claude's response, and stream TTS synthesis. Flush TTS buffers at sentence boundaries rather than waiting for the full response.
Keep Responses Concise
Instruct Claude in the system prompt to keep responses short and conversational. Long responses frustrate voice users. A good rule of thumb is 1-3 sentences for most interactions. Use max_tokens as a hard cap, but rely on the system prompt for soft guidance.
Manage Context Windows Carefully
Voice sessions can run long. Implement a sliding window or summarization strategy to keep the message history within Claude's context limits. A simple approach is to keep the last 20 messages and summarize older context periodically:
def trim_history(session: Session, max_messages: int = 20):
if len(session.messages) <= max_messages:
return
# Keep the system message and the most recent messages
kept = session.messages[-max_messages:]
session.messages = kept
Handle Errors Gracefully
Network failures, API rate limits, and malformed audio will all occur in production. Wrap each pipeline stage in error handling and send user-friendly error messages via TTS when something goes wrong:
async def generate_responses():
while True:
user_text = await text_queue.get()
if user_text is None:
break
try:
# ... streaming logic ...
except Exception as e:
error_msg = "I'm sorry, I encountered an error. Please try again."
async for audio_chunk in tts_service.synthesize_stream(error_msg):
await websocket.send_bytes(audio_chunk)
await websocket.send_text(json.dumps({
"type": "error",
"message": str(e)
}))
Secure Your Endpoints
Add authentication to your WebSocket endpoint. A simple approach is to pass a bearer token as a query parameter and validate it on connection:
from fastapi import Query
@app.websocket("/voice/{session_id}")
async def voice_endpoint(
websocket: WebSocket,
session_id: str,
token: str = Query(...)
):
if token != os.getenv("WEBSOCKET_AUTH_TOKEN"):
await websocket.close(code=4001, reason="Unauthorized")
return
await websocket.accept()
# ... rest of the handler ...
Use the Right Model for the Task
Claude 3.5 Haiku is significantly faster and cheaper than Sonnet, making it ideal for simple voice interactions. Use Sonnet or Opus for complex reasoning tasks, tool use, or when response quality is more important than latency. You can even route dynamically based on query complexity.
Log and Monitor Everything
Instrument your pipeline with timing metrics for each stage. Log the time from audio receipt to first STT result, from STT to first Claude token, and from first token to first audio chunk sent. These metrics will guide your optimization efforts.
import time
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("voice-assistant")
async def generate_responses():
while True:
user_text = await text_queue.get()
if user_text is None:
break
t0 = time.time()
first_token_time = None
first_audio_time = None
async for token in claude_service.stream_response(session, user_text):
if first_token_time is None:
first_token_time = time.time()
logger.info(f"First token: {first_token_time - t0:.3f}s")
# ... TTS logic ...
if first_audio_time is None:
first_audio_time = time.time()
logger.info(f"First audio: {first_audio_time - t0:.3f}s")
logger.info(f"Total response time: {time.time() - t0:.3f}s")
Testing the Backend
You can test the backend with a simple Python client that sends a pre-recorded audio file and receives the response. Here is a minimal test client:
import asyncio
import websockets
import json
async def test_client():
uri = "ws://localhost:8000/voice/test-session-001"
async with websockets.connect(uri) as ws:
# Send a pre-recorded WAV file (strip header, send PCM)
with open("test_audio.wav", "rb") as f:
f.read(44) # Skip WAV header
while True:
chunk = f.read(3200) # 100ms at 16kHz, 16-bit
if not chunk:
break
await ws.send(chunk)
await asyncio.sleep(0.1)
# Signal end of audio
await ws.send(json.dumps({"type": "audio_end"}))
# Receive response audio
with open("response.wav", "wb") as out:
while True:
msg = await ws.recv()
if isinstance(msg, bytes):
out.write(msg)
elif isinstance(msg, str):
data = json.loads(msg)
if data.get("type") == "response_complete":
print("Response complete")
break
asyncio.run(test_client())
Deployment Considerations
When deploying to production, consider the following. Run the server behind a reverse proxy like Nginx that supports WebSocket proxying. Use multiple worker processes with uvicorn --workers 4 or a process manager like Gunicorn with Uvicorn workers. If you need horizontal scaling, move session storage to Redis so that any worker can handle any session. Use HTTPS and WSS in production — never expose unencrypted WebSocket endpoints. Set appropriate rate limits per client to prevent abuse. Monitor API usage and set billing alerts on your Anthropic, Deepgram, and ElevenLabs accounts to avoid unexpected costs.
Conclusion
Building a voice assistant backend with Claude Code is a rewarding project that combines real-time audio processing, large language model orchestration, and thoughtful UX design into a single system. By leveraging Claude's streaming API, tool-use capabilities, and strong conversational reasoning, you can create an assistant that feels genuinely intelligent and responsive. The key to a great voice experience lies in pipeline optimization — streaming at every stage to minimize time-to-first-audio — combined with concise response generation, robust error handling, and graceful interruption support. Start with the architecture described here, iterate on the system prompt to shape your assistant's personality, expand the tool set to give it real capabilities, and continuously monitor latency metrics as you scale. With these foundations in place, you are well equipped to ship a voice assistant that users will actually enjoy talking to.