Building a Voice Assistant Backend with MCP (Model Context Protocol): Complete Guide
Voice assistants have evolved from simple command-and-response systems into sophisticated agents that can reason, retrieve context, and execute complex tasks. The Model Context Protocol (MCP)—an open standard introduced by Anthropic—provides a unified way to connect AI models to external data sources, tools, and services. In this tutorial, you'll learn how to build a production-ready voice assistant backend using MCP, covering everything from server setup to streaming audio-aware tool execution.
What Is the Model Context Protocol?
MCP is a JSON-RPC 2.0-based protocol that standardizes how AI applications communicate with external resources. Instead of writing bespoke integrations for every data source or API, you expose them through an MCP server that any MCP-compatible client (an LLM, a voice agent, an IDE) can consume. The protocol defines three core primitives:
- Resources — Read-only data sources (files, database rows, API responses) exposed via URIs.
- Tools — Executable functions the model can invoke, such as querying a calendar or controlling a smart home device.
- Prompts — Reusable prompt templates that can be parameterized and shared.
For a voice assistant, MCP is particularly powerful because it decouples the reasoning layer (the LLM) from the capability layer (tools and data). Your voice pipeline can focus on speech-to-text, intent routing, and text-to-speech, while MCP handles the messy business of calling APIs and retrieving context.
Why MCP Matters for Voice Assistants
Traditional voice assistant backends suffer from tight coupling between the NLU/LLM layer and integration code. Every new skill requires changes to the assistant's core logic. MCP solves this by providing a plug-and-play architecture:
- Modular skills: Add a new capability by registering a tool—no changes to the assistant's orchestration code.
- Transport flexibility: MCP supports stdio, HTTP, and Server-Sent Events (SSE), which is critical for streaming voice responses.
- Context awareness: Resources let your assistant pull in user-specific context (preferences, location, recent activity) without hardcoding it into prompts.
- Standardized security: MCP defines permission and consent flows, essential when a voice assistant acts on behalf of a user.
Architecture Overview
Our voice assistant backend consists of four layers:
- Audio Layer: Handles WebSocket connections for streaming audio in/out, speech-to-text (STT), and text-to-speech (TTS).
- Orchestration Layer: Manages conversation state, routes transcripts to the LLM, and coordinates tool calls.
- MCP Client: Connects to one or more MCP servers, discovers available tools, and relays tool-call requests from the LLM.
- MCP Servers: Expose domain-specific capabilities (calendar, smart home, knowledge base, etc.).
The key insight is that the MCP client sits between the LLM and the MCP servers, acting as a bridge. The LLM never talks to MCP servers directly—it emits tool calls, and the client translates those into MCP protocol messages.
Prerequisites and Project Setup
You'll need Python 3.11 or later. We'll use the official MCP Python SDK, FastAPI for the WebSocket audio layer, and the Anthropic SDK for the LLM. Create a new project:
mkdir voice-mcp-assistant && cd voice-mcp-assistant
python -m venv .venv
source .venv/bin/activate # On Windows: .venv\Scripts\activate
pip install mcp anthropic fastapi uvicorn websockets python-dotenv
Create a .env file with your API keys:
ANTHROPIC_API_KEY=sk-ant-...
ASSISTANT_NAME=Aria
Building Your First MCP Server
Let's start by building an MCP server that exposes tools a voice assistant would need: getting the current time, checking the weather, and managing a todo list. This server runs as a standalone process that the assistant connects to.
# servers/personal_tools_server.py
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
import json
import datetime
import asyncio
server = Server("personal-tools")
# In-memory todo store (replace with a database in production)
todos: list[dict] = []
@server.list_tools()
async def list_tools() -> list[Tool]:
return [
Tool(
name="get_current_time",
description="Get the current date and time in a human-friendly format. Use when the user asks what time it is.",
inputSchema={
"type": "object",
"properties": {
"timezone": {
"type": "string",
"description": "IANA timezone, e.g. 'America/New_York'. Defaults to UTC."
}
}
}
),
Tool(
name="get_weather",
description="Get the current weather for a given city. Use when the user asks about weather conditions.",
inputSchema={
"type": "object",
"properties": {
"city": {"type": "string", "description": "City name"}
},
"required": ["city"]
}
),
Tool(
name="add_todo",
description="Add an item to the user's todo list.",
inputSchema={
"type": "object",
"properties": {
"task": {"type": "string", "description": "The task description"}
},
"required": ["task"]
}
),
Tool(
name="list_todos",
description="List all items on the user's todo list.",
inputSchema={"type": "object", "properties": {}}
)
]
@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
if name == "get_current_time":
tz = arguments.get("timezone", "UTC")
try:
from zoneinfo import ZoneInfo
now = datetime.datetime.now(ZoneInfo(tz))
text = now.strftime("It's %I:%M %p on %A, %B %d.")
except Exception:
text = datetime.datetime.utcnow().strftime("It's %H:%M UTC on %A, %B %d.")
return [TextContent(type="text", text=text)]
elif name == "get_weather":
city = arguments["city"]
# Placeholder — integrate with a real weather API
text = f"The weather in {city} is currently 72°F and partly cloudy."
return [TextContent(type="text", text=text)]
elif name == "add_todo":
task = arguments["task"]
todos.append({"task": task, "done": False})
return [TextContent(type="text", text=f"Added '{task}' to your todo list.")]
elif name == "list_todos":
if not todos:
return [TextContent(type="text", text="Your todo list is empty.")]
lines = [f"{i+1}. {t['task']}" for i, t in enumerate(todos)]
return [TextContent(type="text", text="\n".join(lines))]
return [TextContent(type="text", text=f"Unknown tool: {name}")]
async def main():
async with stdio_server() as (read_stream, write_stream):
await server.run(read_stream, write_stream, server.create_initialization_options())
if __name__ == "__main__":
asyncio.run(main())
Notice how each tool includes a description that's written for an LLM audience. This is critical for voice assistants—the description is what the model uses to decide when to call the tool based on a spoken utterance. Write descriptions as if explaining to a colleague when this tool is appropriate.
Exposing Resources for Context
Tools are for actions; resources are for context. Let's add a resource that exposes user profile information the assistant can reference during conversations:
# Add to servers/personal_tools_server.py
from mcp.types import Resource
USER_PROFILE = {
"name": "Alex",
"preferred_name": "Alex",
"timezone": "America/New_York",
"location": "Brooklyn, NY",
"preferences": {
"temperature_unit": "fahrenheit",
"voice_speed": "normal"
}
}
@server.list_resources()
async def list_resources() -> list[Resource]:
return [
Resource(
uri="user://profile",
name="User Profile",
description="The current user's profile, preferences, and location.",
mimeType="application/json"
)
]
@server.read_resource()
async def read_resource(uri: str) -> str:
if uri == "user://profile":
return json.dumps(USER_PROFILE)
raise ValueError(f"Unknown resource: {uri}")
The LLM can now read user://profile to learn the user's name, timezone, and preferences before responding. This is far cleaner than injecting context into every system prompt manually.
Building the MCP Client Wrapper
Now let's build the client that connects to our MCP server and exposes its tools to the LLM. We'll use the MCP client SDK with stdio transport:
# assistant/mcp_client.py
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from contextlib import AsyncExitStack
import asyncio
import json
class MCPClientManager:
"""Manages connections to one or more MCP servers and exposes their tools."""
def __init__(self):
self.exit_stack = AsyncExitStack()
self.sessions: dict[str, ClientSession] = {}
self.available_tools: list[dict] = []
async def connect_server(self, name: str, command: str, args: list[str], env: dict | None = None):
"""Spawn and connect to an MCP server via stdio."""
server_params = StdioServerParameters(
command=command,
args=args,
env=env
)
stdio_transport = await self.exit_stack.enter_async_context(
stdio_client(server_params)
)
read_stream, write_stream = stdio_transport
session = await self.exit_stack.enter_async_context(
ClientSession(read_stream, write_stream)
)
await session.initialize()
# Discover tools
tools_response = await session.list_tools()
server_tools = [
{
"name": tool.name,
"description": tool.description,
"input_schema": tool.inputSchema,
"server": name
}
for tool in tools_response.tools
]
self.sessions[name] = session
self.available_tools.extend(server_tools)
print(f"[MCP] Connected to '{name}' with {len(server_tools)} tools")
def get_tools_for_llm(self) -> list[dict]:
"""Return tools in the format the Anthropic API expects."""
return [
{
"name": t["name"],
"description": t["description"],
"input_schema": t["input_schema"]
}
for t in self.available_tools
]
async def call_tool(self, tool_name: str, arguments: dict) -> str:
"""Call a tool on the appropriate MCP server."""
tool_info = next(t for t in self.available_tools if t["name"] == tool_name)
server_name = tool_info["server"]
session = self.sessions[server_name]
result = await session.call_tool(tool_name, arguments)
# Concatenate text content blocks
text_parts = [block.text for block in result.content if hasattr(block, "text")]
return "\n".join(text_parts)
async def read_resource(self, server_name: str, uri: str) -> str:
"""Read a resource from a specific MCP server."""
session = self.sessions[server_name]
result = await session.read_resource(uri)
return result.contents[0].text
async def cleanup(self):
await self.exit_stack.aclose()
This manager supports connecting to multiple MCP servers simultaneously. Each tool is tagged with its source server so the call_tool method can route correctly.
Building the Conversation Orchestrator
The orchestrator ties together the LLM and the MCP client. It handles the tool-use loop: the LLM may request tool calls, we execute them via MCP, feed results back, and repeat until the LLM produces a final text response.
# assistant/orchestrator.py
import anthropic
import json
from .mcp_client import MCPClientManager
SYSTEM_PROMPT = """You are Aria, a helpful voice assistant.
You communicate concisely since your responses will be spoken aloud.
Keep responses short and natural for speech. Avoid markdown, bullet points,
or special characters that don't read well aloud.
When you need information or need to take an action, use the available tools.
Always greet the user by their preferred name when available."""
class ConversationOrchestrator:
def __init__(self, mcp_manager: MCPClientManager):
self.client = anthropic.Anthropic()
self.mcp = mcp_manager
self.model = "claude-sonnet-4-20250514"
self.max_tool_rounds = 5
async def process_message(self, user_text: str, history: list[dict]) -> str:
"""Process a user message and return the assistant's text response."""
messages = history + [{"role": "user", "content": user_text}]
tools = self.mcp.get_tools_for_llm()
for round_num in range(self.max_tool_rounds):
response = self.client.messages.create(
model=self.model,
max_tokens=1024,
system=SYSTEM_PROMPT,
tools=tools if tools else None,
messages=messages
)
# If the model wants to call tools
if response.stop_reason == "tool_use":
tool_results = []
for block in response.content:
if block.type == "tool_use":
print(f"[Tool Call] {block.name}({json.dumps(block.input)})")
try:
result = await self.mcp.call_tool(block.name, block.input)
except Exception as e:
result = f"Error calling tool: {e}"
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": result
})
# Append the assistant's tool-use message and the results
messages.append({"role": "assistant", "content": response.content})
messages.append({"role": "user", "content": tool_results})
continue
# No tool calls — extract the final text response
text_parts = [b.text for b in response.content if b.type == "text"]
return "".join(text_parts)
return "I'm having trouble completing that request. Could you try again?"
async def load_user_context(self, server_name: str = "personal-tools"):
"""Preload user profile resource into context."""
try:
profile_json = await self.mcp.read_resource(server_name, "user://profile")
return f"User profile: {profile_json}"
except Exception:
return None
The max_tool_rounds limit prevents infinite loops where the model keeps calling tools. Five rounds is generous for most voice interactions—typically one or two tool calls suffice.
Adding the Audio Streaming Layer
A voice assistant needs to handle streaming audio over WebSockets. We'll build a FastAPI endpoint that accepts audio chunks, transcribes them, sends the transcript through the orchestrator, and streams back a TTS audio response. For this tutorial, we'll stub the STT and TTS components but show the full pipeline structure:
# assistant/audio_server.py
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
import asyncio
import json
from .mcp_client import MCPClientManager
from .orchestrator import ConversationOrchestrator
app = FastAPI(title="Voice MCP Assistant")
# Global instances (in production, use dependency injection)
mcp_manager = MCPClientManager()
orchestrator: ConversationOrchestrator = None
@app.on_event("startup")
async def startup():
global orchestrator
await mcp_manager.connect_server(
name="personal-tools",
command="python",
args=["servers/personal_tools_server.py"]
)
orchestrator = ConversationOrchestrator(mcp_manager)
print("[Startup] Assistant ready")
@app.on_event("shutdown")
async def shutdown():
await mcp_manager.cleanup()
@app.websocket("/voice")
async def voice_endpoint(websocket: WebSocket):
"""
WebSocket protocol:
- Client sends: {"type": "audio", "data": "<base64-audio-chunk>"}
- Client sends: {"type": "audio_end"} to signal end of utterance
- Server sends: {"type": "transcript", "text": "..."}
- Server sends: {"type": "response", "text": "..."}
- Server sends: {"type": "audio", "data": "<base64-tts-chunk>"}
- Server sends: {"type": "done"}
"""
await websocket.accept()
audio_buffer = bytearray()
history: list[dict] = []
# Preload user context
user_context = await orchestrator.load_user_context()
if user_context:
history.append({"role": "user", "content": user_context})
history.append({"role": "assistant", "content": "Understood. I have your profile loaded."})
try:
while True:
message = await websocket.receive_json()
if message["type"] == "audio":
# Accumulate audio chunks (decode base64 in production)
audio_buffer.extend(message["data"].encode())
elif message["type"] == "audio_end":
# Step 1: Transcribe audio (stub — replace with Whisper, Deepgram, etc.)
transcript = transcribe_audio(bytes(audio_buffer))
audio_buffer.clear()
await websocket.send_json({"type": "transcript", "text": transcript})
# Step 2: Process through orchestrator
response_text = await orchestrator.process_message(transcript, history)
# Update conversation history
history.append({"role": "user", "content": transcript})
history.append({"role": "assistant", "content": response_text})
await websocket.send_json({"type": "response", "text": response_text})
# Step 3: Synthesize speech (stub — replace with your TTS engine)
audio_chunks = synthesize_speech(response_text)
for chunk in audio_chunks:
await websocket.send_json({"type": "audio", "data": chunk})
await asyncio.sleep(0.02) # Simulate streaming cadence
await websocket.send_json({"type": "done"})
except WebSocketDisconnect:
print("[WebSocket] Client disconnected")
# --- Stub functions (replace with real implementations) ---
def transcribe_audio(audio_bytes: bytes) -> str:
"""Replace with your STT provider (Whisper, Deepgram, AssemblyAI, etc.)"""
# Example with OpenAI Whisper:
# from openai import OpenAI
# client = OpenAI()
# buffer = io.BytesIO(audio_bytes)
# buffer.name = "audio.wav"
# result = client.audio.transcriptions.create(model="whisper-1", file=buffer)
# return result.text
return "What time is it?"
def synthesize_speech(text: str) -> list[str]:
"""Replace with your TTS provider. Returns base64-encoded audio chunks."""
# Example with OpenAI TTS:
# response = client.audio.speech.create(model="tts-1", voice="alloy", input=text)
# Split response content into streaming chunks
return [text[:50]] # Placeholder
Run the server with:
uvicorn assistant.audio_server:app --host 0.0.0.0 --port 8000 --reload
Adding a Smart Home MCP Server
To demonstrate MCP's modularity, let's add a second server for smart home control. This shows how new capabilities are added without touching the orchestrator:
# servers/smart_home_server.py
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
import asyncio
server = Server("smart-home")
# Simulated device state
devices = {
"living_room_light": {"type": "light", "on": False, "brightness": 0},
"bedroom_light": {"type": "light", "on": False, "brightness": 0},
"thermostat": {"type": "thermostat", "temperature": 70, "mode": "auto"}
}
@server.list_tools()
async def list_tools() -> list[Tool]:
return [
Tool(
name="control_device",
description="Control a smart home device. Can turn devices on/off or set properties like brightness or temperature.",
inputSchema={
"type": "object",
"properties": {
"device": {
"type": "string",
"description": "Device ID: 'living_room_light', 'bedroom_light', or 'thermostat'",
"enum": list(devices.keys())
},
"action": {
"type": "string",
"description": "Action to perform: 'on', 'off', 'set'",
"enum": ["on", "off", "set"]
},
"value": {
"type": "number",
"description": "Value for 'set' action (brightness 0-100 or temperature in Fahrenheit)"
}
},
"required": ["device", "action"]
}
),
Tool(
name="get_device_status",
description="Get the current status of all smart home devices.",
inputSchema={"type": "object", "properties": {}}
)
]
@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
if name == "control_device":
device_id = arguments["device"]
action = arguments["action"]
device = devices.get(device_id)
if not device:
return [TextContent(type="text", text=f"Device '{device_id}' not found.")]
if action == "on":
device["on"] = True
if device["type"] == "light":
device["brightness"] = 100
return [TextContent(type="text", text=f"Turned on {device_id}.")]
elif action == "off":
device["on"] = False
if device["type"] == "light":
device["brightness"] = 0
return [TextContent(type="text", text=f"Turned off {device_id}.")]
elif action == "set":
value = arguments.get("value")
if value is None:
return [TextContent(type="text", text="A 'value' is required for the 'set' action.")]
if device["type"] == "light":
device["brightness"] = max(0, min(100, int(value)))
device["on"] = device["brightness"] > 0
return [TextContent(type="text", text=f"Set {device_id} brightness to {device['brightness']}%.")]
elif device["type"] == "thermostat":
device["temperature"] = int(value)
return [TextContent(type="text", text=f"Set thermostat to {device['temperature']}°F.")]
elif name == "get_device_status":
lines = []
for did, d in devices.items():
if d["type"] == "light":
state = f"{'on' if d['on'] else 'off'} at {d['brightness']}% brightness"
elif d["type"] == "thermostat":
state = f"{d['temperature']}°F, mode: {d['mode']}"
lines.append(f"- {did}: {state}")
return [TextContent(type="text", text="\n".join(lines))]
return [TextContent(type="text", text=f"Unknown tool: {name}")]
async def main():
async with stdio_server() as (read_stream, write_stream):
await server.run(read_stream, write_stream, server.create_initialization_options())
if __name__ == "__main__":
asyncio.run(main())
Register it in the startup handler:
await mcp_manager.connect_server(
name="smart-home",
command="python",
args=["servers/smart_home_server.py"]
)
Now the assistant can say "turn off the living room light" or "set the thermostat to 68" without any changes to the orchestrator or audio layer.
Using SSE Transport for Remote MCP Servers
While stdio is great for local development, production deployments often need remote MCP servers accessible over HTTP. The MCP SDK supports SSE transport. Here's how to run a server with SSE:
# servers/personal_tools_sse.py
from mcp.server import Server
from mcp.server.sse import SseServerTransport
from starlette.applications import Starlette
from starlette.routing import Mount, Route
import uvicorn
# Reuse the same tool definitions from personal_tools_server.py
from personal_tools_server import server as mcp_server
transport = SseServerTransport("/messages/")
async def handle_sse(request):
async with transport.connect_sse(request.scope, request.receive, request._send) as (read, write):
await mcp_server.run(read, write, mcp_server.create_initialization_options())
return Response()
sse_app = Starlette(
routes=[
Route("/sse", endpoint=handle_sse),
Mount("/messages/", app=transport.handle_post_message),
]
)
if __name__ == "__main__":
uvicorn.run(sse_app, host="0.0.0.0", port=3001)
On the client side, connect using the SSE client:
from mcp.client.sse import sse_client
async def connect_remote_server(url: str):
async with sse_client(url) as (read, write):
session = ClientSession(read, write)
await session.initialize()
return session
Best Practices
Write Tool Descriptions for Voice
Voice transcripts are often noisy and ambiguous. Write tool descriptions that account for paraphrasing. Instead of "Get weather", use "Get current weather conditions for a location. Use when the user asks about temperature, rain, snow, or what the weather is like." Include synonyms and common phrasings.
Keep Responses Concise
Voice is a linear medium—users can't skim. Instruct the LLM to keep responses under 2-3 sentences for simple queries. Use the system prompt to enforce this, and consider post-processing long responses into a summary for TTS while keeping the full text for display in a companion app.
Handle Tool Failures Gracefully
Tools will fail. Network timeouts, invalid inputs, and service outages are inevitable. Always wrap tool calls in try/except and return user-friendly error messages. The orchestrator already does this, but your MCP servers should also validate inputs and return clear error text rather than raising exceptions:
@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
try:
if name == "get_weather":
city = arguments.get("city", "").strip()
if not city:
return [TextContent(type="text", text="I need a city name to check the weather.")]
# ... fetch weather
except Exception as e:
return [TextContent(type="text", text=f"I couldn't complete that request. {str(e)}")]
Implement Conversation Memory
Voice interactions are multi-turn. Maintain conversation history per session and pass it to the LLM. For long conversations, implement a sliding window or summarization strategy to stay within context limits. Store history in Redis or a database keyed by session ID rather than in memory for production deployments.
Use Streaming Where Possible
Latency is the enemy of voice UX. Stream LLM responses token-by-token and feed chunks to TTS as they arrive. With the Anthropic API, use client.messages.stream() instead of create(). For tool-use rounds, you still need to wait for the complete response, but the final text response can be streamed:
async def process_message_streaming(self, user_text: str, history: list[dict]):
messages = history + [{"role": "user", "content": user_text}]
tools = self.mcp.get_tools_for_llm()
for _ in range(self.max_tool_rounds):
with self.client.messages.stream(
model=self.model,
max_tokens=1024,
system=SYSTEM_PROMPT,
tools=tools if tools else None,
messages=messages
) as stream:
for text in stream.text_stream:
yield text # Stream tokens to TTS as they arrive
response = stream.get_final_message()
if response.stop_reason == "tool_use":
# Handle tool calls as before, then continue the loop
tool_results = []
for block in response.content:
if block.type == "tool_use":
result = await self.mcp.call_tool(block.name, block.input)
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": result
})
messages.append({"role": "assistant", "content": response.content})
messages.append({"role": "user", "content": tool_results})
else:
return # Done
Secure Tool Execution
MCP servers can execute arbitrary code and access external APIs. In production: authenticate MCP connections, run servers in isolated containers, implement rate limiting, and require explicit user consent for sensitive tools. The MCP specification includes a consent flow—use it for actions like sending messages or making purchases.
Test Tool Discovery Dynamically
Don't hardcode tool lists in the orchestrator. Always call list_tools() at connection time and after reconnections. This ensures the assistant picks up new tools when MCP servers are updated without requiring a restart of the orchestrator.
Testing the Assistant
You can test the orchestrator without audio by writing a simple text-based client:
# test_text.py
import asyncio
from assistant.mcp_client import MCPClientManager
from assistant.orchestrator import ConversationOrchestrator
async def main():
manager = MCPClientManager()
await manager.connect_server("personal-tools", "python", ["servers/personal_tools_server.py"])
await manager.connect_server("smart-home", "python", ["servers/smart_home_server.py"])
orch = ConversationOrchestrator(manager)
history = []
# Preload context
ctx = await orch.load_user_context()
if ctx:
history.append({"role": "user", "content": ctx})
history.append({"role": "assistant", "content": "Got it."})
test_inputs = [
"What time is it?",
"Add 'buy groceries' to my todo list.",
"What's on my todo list?",
"Turn on the living room light.",
"What's the status of my devices?"
]
for user_input in test_inputs:
print(f"\nUser: {user_input}")
response = await orch.process_message(user_input, history)
print(f"Aria: {response}")
history.append({"role": "user", "content": user_input})
history.append({"role": "assistant", "content": response})
await manager.cleanup()
asyncio.run(main())
Run it to verify the full tool-use loop works end-to-end before wiring up audio.
Conclusion
The Model Context Protocol transforms voice assistant development from a monolithic integration challenge into a modular, composable architecture. By separating the reasoning layer (LLM), the capability layer (MCP servers), and the audio layer (WebSocket streaming), you can add new skills by simply spinning up another MCP server—no changes to the orchestrator required. This tutorial covered the full stack: building MCP servers with tools and resources, connecting them through a client manager, orchestrating multi-turn conversations with tool-use loops, and exposing everything through a streaming WebSocket endpoint. As you move to production, focus on streaming latency, conversation persistence, security boundaries for tool execution, and writing tool descriptions that account for the imprecision of spoken language. The MCP ecosystem is growing rapidly, and voice assistants built on this foundation will be able to adopt new capabilities with minimal effort as community-built MCP servers become available for everything from CRM access to home automation.