Chapter 9: Production Telephony Patterns
Chapter 8 got your agent connected to the phone network. Calls come in, the agent answers, conversations happen. But there is a wide gap between “it works” and “it works reliably in production.” Six patterns close that gap: managing call lifecycle, pre-caching audio for zero-latency playback, handling silence and timeouts, recording calls, publishing real-time progress to dashboards, and managing phone number pools.
These patterns come from running a production voice agent that handles hundreds of calls. Each one addresses a specific failure mode we encountered and solved.
Call Lifecycle Management
Every phone call moves through a predictable sequence of states: ringing, connected, active, ending, ended. Your agent code needs to track where it is in that sequence and respond appropriately at each transition.
LiveKit models this through room events and participant state. When a SIP call comes in, the caller joins the LiveKit room as a participant. The agent joins separately. The call is “active” when both are present and audio is flowing.
from livekit.agents import AgentSession, JobContext
@server.rtc_session()
async def my_agent(ctx: JobContext):
session = AgentSession(
stt=deepgram.STT(),
llm=LLM(model="gpt-4.1-mini"),
tts=cartesia.TTS(),
vad=ctx.proc.userdata["vad"],
)
await session.start(agent=MyAgent(), room=ctx.room)
# Connect to the room -- this is when the agent becomes active
await ctx.connect()Handling Caller Disconnect
The most common mid-call event is the caller hanging up. When this happens, the SIP participant leaves the room. You can detect this and clean up:
from livekit import rtc
@server.rtc_session()
async def my_agent(ctx: JobContext):
session = AgentSession(...)
await session.start(agent=MyAgent(), room=ctx.room)
await ctx.connect()
@ctx.room.on("participant_disconnected")
def on_participant_left(participant: rtc.RemoteParticipant):
logger.info("Caller disconnected: %s", participant.identity)
# Cancel any pending operations
# Flush collected data to databaseGraceful Shutdown
When your agent process receives a shutdown signal (during deployment, for example), you do not want to drop active calls. LiveKit’s agent framework supports shutdown callbacks that let you complete pending work:
@server.rtc_session()
async def my_agent(ctx: JobContext):
lk_api = api.LiveKitAPI()
ctx.add_shutdown_callback(lk_api.aclose)
# The shutdown callback ensures the API client is properly closed
# even if the process is terminated mid-callFor longer cleanup operations (flushing data to a database, completing an API call), structure your tools so that critical operations happen before the response is sent back to the caller. Do not rely on post-call cleanup for anything essential.
Timeout Handling
LiveKit’s AgentSession has built-in support for detecting when the caller goes silent. The user_away_timeout parameter sets how long to wait before triggering a state change:
session = AgentSession(
stt=deepgram.STT(),
llm=LLM(model="gpt-4.1-mini"),
tts=cartesia.TTS(),
user_away_timeout=30.0, # seconds of silence before "away"
)This feeds directly into the silence handling pattern covered later in this chapter.
Hold Music and Cached TTS
One of the most impactful patterns for production voice agents is pre-synthesizing audio at startup. When your agent needs to play a hold message while running a slow tool (an API call, a database write, a website deployment), you cannot afford to wait for TTS synthesis. The caller is sitting in silence, wondering if the call dropped.
The solution is to synthesize the hold audio once, cache the frames, and play them instantly when needed.
Pre-synthesizing at Startup
The prewarm function runs once per worker process, before any calls are handled. This is the ideal place to synthesize and cache audio:
from livekit import rtc
from livekit.agents import AgentServer, JobProcess
HOLD_TEXT = (
"Alright, I'm setting up your website right now. "
"This will just take a moment."
)
server = AgentServer()
def prewarm(proc: JobProcess):
proc.userdata["vad"] = silero.VAD.load()
proc.userdata["hold_frames"] = None
server.setup_fnc = prewarmThen on the first job, synthesize and cache the frames for all subsequent calls:
@server.rtc_session()
async def my_agent(ctx: JobContext):
tts = cartesia.TTS()
if ctx.proc.userdata["hold_frames"] is None:
hold_frames_list: list[rtc.AudioFrame] = []
async for event in tts.synthesize(HOLD_TEXT):
hold_frames_list.append(event.frame)
ctx.proc.userdata["hold_frames"] = hold_frames_list
logger.info("Pre-synthesized hold audio (%d frames)", len(hold_frames_list))
hold_frames: list[rtc.AudioFrame] = ctx.proc.userdata["hold_frames"]The key detail: proc.userdata persists across all jobs handled by the same worker process. The first call pays the synthesis cost; every subsequent call plays from cache with zero latency.
Playing Cached Audio During Tool Execution
When a tool needs to run a slow operation, play the cached audio immediately, then run the operation concurrently:
from collections.abc import AsyncIterator
class MainAgent(Agent):
@function_tool()
async def trigger_onboarding_pipeline(self, ctx: RunContext) -> str:
"""Deploy the caller's website and assign a phone number."""
# Wrap cached frames as an async iterator
async def _cached_hold_audio() -> AsyncIterator[rtc.AudioFrame]:
for frame in self._hold_frames:
yield frame
# Play instantly from cache -- no TTS latency
hold_handle = ctx.session.say(
self._hold_text,
audio=_cached_hold_audio(),
add_to_chat_ctx=False,
)
# Run the slow operation while audio plays
result = await call_onboard_api(self._confirmed_profile)
# Wait for hold audio to finish if it's still playing
await hold_handle
return format_success_result(result)The add_to_chat_ctx=False parameter is critical. Without it, the hold message gets added to the conversation history, and the LLM sees it as part of the dialogue. This pollutes the context and can cause the agent to repeat hold messages or reference them in future responses. By setting it to False, the audio plays for the caller but the LLM never knows it happened.
Handling Caller Silence and Timeouts
Real callers get distracted. They put the phone down, get interrupted by someone at the door, or simply stop talking because they are thinking. Your agent needs to handle all of these gracefully.
Detecting Silence with user_away
The user_away_timeout parameter on AgentSession triggers a state change when the caller has been silent for the specified duration. You listen for this event and respond:
from src.prompts import USER_AWAY_INSTRUCTIONS
session = AgentSession(
stt=deepgram.STT(model="nova-3", language="en"),
llm=LLM(model="gpt-4.1-mini"),
tts=cartesia.TTS(),
user_away_timeout=30.0,
)
@session.on("user_state_changed")
def _on_user_state_changed(ev) -> None:
if ev.new_state == "away":
session.generate_reply(instructions=USER_AWAY_INSTRUCTIONS)The instructions tell the agent what to do when the caller goes silent. For a brief timeout, you might prompt:
USER_AWAY_INSTRUCTIONS = (
"The caller has been silent for 30 seconds. "
"Say a brief goodbye and use the end_call tool to hang up."
)You can also build a two-stage approach: first prompt “Are you still there?”, then end the call if silence continues. This requires tracking state:
@session.on("user_state_changed")
def _on_user_state_changed(ev) -> None:
if ev.new_state == "away":
if not hasattr(session, "_prompted_once"):
session._prompted_once = True
session.generate_reply(
instructions="The caller has been silent. Ask: 'Are you still there?'"
)
else:
session.generate_reply(
instructions="The caller is still silent after being prompted. "
"Say goodbye and end the call."
)
elif ev.new_state == "present":
# Reset the prompt flag when they speak again
session._prompted_once = FalseChoosing Timeout Duration
The right timeout depends on your use case. For a fast transactional call (booking an appointment), 15-20 seconds works well. For an onboarding flow where the caller might need to look up information, 30-45 seconds is more appropriate. We settled on 30 seconds after observing that shorter timeouts caused false positives when callers were looking up their business hours or email address.
Call Recording with Egress
Recording calls is essential for compliance in many industries and invaluable for debugging agent behavior. LiveKit provides this through its Egress service, which captures room audio and video and writes it to cloud storage.
Starting a Recording
Use the LiveKit server API to start an egress when the call begins:
from livekit import api
async def start_recording(lk_api: api.LiveKitAPI, room_name: str) -> str:
"""Start recording a room and return the egress ID."""
egress_info = await lk_api.egress.start_room_composite_egress(
api.RoomCompositeEgressRequest(
room_name=room_name,
audio_only=True,
file_outputs=[
api.EncodedFileOutput(
file_type=api.EncodedFileType.OGG,
filepath=f"recordings/{room_name}.ogg",
s3=api.S3Upload(
bucket="your-bucket",
region="us-east-1",
access_key="...",
secret="...",
),
)
],
)
)
return egress_info.egress_idStopping a Recording
Stop the egress when the call ends, either through a shutdown callback or when the participant disconnects:
async def stop_recording(lk_api: api.LiveKitAPI, egress_id: str) -> None:
"""Stop an active recording."""
await lk_api.egress.stop_egress(api.StopEgressRequest(egress_id=egress_id))Wire it into your agent lifecycle:
@server.rtc_session()
async def my_agent(ctx: JobContext):
lk_api = api.LiveKitAPI()
ctx.add_shutdown_callback(lk_api.aclose)
egress_id = await start_recording(lk_api, ctx.room.name)
async def _stop_on_shutdown():
await stop_recording(lk_api, egress_id)
ctx.add_shutdown_callback(_stop_on_shutdown)
# ... rest of agent setupThe Egress service runs separately from your agent. It connects to the LiveKit room as a hidden participant, captures the audio mix, encodes it, and uploads to your configured storage. Your agent code only needs to start and stop it.
Room Metadata for Dashboards
When you have an admin dashboard monitoring active calls, you need a way to push real-time progress updates. LiveKit room metadata is a lightweight mechanism for this: you write a JSON string to the room, and any client connected to that room (or polling via API) can read it.
The CallProgress Pattern
Create a dedicated class that owns the metadata lifecycle:
import asyncio
import json
from typing import Any
from livekit import api
class CallProgress:
"""Tracks and publishes onboarding progress to room metadata."""
def __init__(self, livekit_api: api.LiveKitAPI, room_name: str) -> None:
self._api = livekit_api
self._room_name = room_name
self._lock = asyncio.Lock()
self._data: dict[str, Any] = {
"step": "greeting",
"fields": {},
}
async def update(self, **kwargs: Any) -> None:
"""Merge kwargs into progress data and push to room metadata."""
async with self._lock:
if "fields" in kwargs:
existing_fields = self._data.get("fields", {})
existing_fields.update(kwargs.pop("fields"))
self._data["fields"] = existing_fields
self._data.update(kwargs)
try:
await self._api.room.update_room_metadata(
api.UpdateRoomMetadataRequest(
room=self._room_name,
metadata=json.dumps(self._data),
)
)
except Exception:
logger.warning(
"Failed to update room metadata for %s",
self._room_name,
exc_info=True,
)Using CallProgress in Your Agent
Pass the progress tracker to your agent and update it as the conversation moves through stages:
@server.rtc_session()
async def my_agent(ctx: JobContext):
lk_api = api.LiveKitAPI()
ctx.add_shutdown_callback(lk_api.aclose)
progress = CallProgress(lk_api, ctx.room.name)
session = AgentSession(...)
await session.start(
agent=MainAgent(progress=progress),
room=ctx.room,
)
await ctx.connect()Inside your tasks, update progress at each significant step:
class CollectBusinessIdentityTask(AgentTask):
@function_tool()
async def record_business_name_and_trade(
self, ctx: RunContext, business_name: str, trade: str
) -> str:
"""Record the business name and trade."""
await self._progress.update(
step="collecting_identity",
fields={"business_name": business_name, "trade": trade},
)
return f"Recorded: {business_name} ({trade})"Reading Metadata from the Dashboard
On the frontend, read room metadata by listing rooms or subscribing to room events:
// Dashboard polling (Next.js example)
async function fetchCallProgress(roomName: string) {
const room = await livekitApi.getRoom(roomName);
const metadata = JSON.parse(room.metadata || '{}');
return {
step: metadata.step, // "greeting" | "collecting_identity" | "deploying"
businessName: metadata.fields?.business_name,
siteUrl: metadata.fields?.site_url,
refCode: metadata.fields?.ref_code,
};
}The metadata updates are fast enough for real-time display. The lock in CallProgress prevents concurrent updates from corrupting the JSON. The try/except ensures that a metadata write failure never crashes the call. The conversation continues even if the dashboard misses an update.
Phone Number Management
A single phone number works for a demo. Production requires a pool of numbers, routing rules, and automated provisioning.
Number Pools
When you have multiple agents or multiple business lines, you need a pool of phone numbers that can be assigned dynamically. Telnyx (and similar providers) offer APIs for this:
import telnyx
async def provision_number(area_code: str = "416") -> str:
"""Search for and purchase a phone number in the given area code."""
# Search available numbers
available = telnyx.AvailablePhoneNumber.list(
filter={
"national_destination_code": area_code,
"features": ["sms", "voice"],
"limit": 1,
}
)
if not available.data:
raise ValueError(f"No numbers available in area code {area_code}")
# Purchase the number
phone_number = available.data[0]
order = telnyx.NumberOrder.create(
phone_numbers=[{"phone_number": phone_number.phone_number}],
)
return order.phone_numbers[0].phone_numberRouting Rules
Once you have multiple numbers, you need to route each one to the right agent or configuration. The SIP trunk configuration determines which LiveKit room a call is routed to, and room naming conventions let your agent code determine which business or workflow to load:
@server.rtc_session()
async def my_agent(ctx: JobContext):
room_name = ctx.room.name # e.g., "inbound-4165551234"
# Extract the dialed number from SIP headers or room name
dialed_number = extract_dialed_number(room_name)
# Look up which business this number belongs to
business = await db.get_business_by_phone(dialed_number)
if business:
agent = ReceptionistAgent(business=business)
else:
agent = OnboardingAgent()
session = AgentSession(...)
await session.start(agent=agent, room=ctx.room)
await ctx.connect()Number Assignment
After onboarding a new business, assign a number from the pool and configure its SIP routing:
async def assign_number_to_business(business_id: str) -> str:
"""Assign an available number from the pool to a business."""
# Get an unassigned number from our pool
number = await db.get_unassigned_number()
if not number:
# Pool is empty -- provision a new one
number = await provision_number()
# Configure SIP trunk to route this number to our LiveKit server
await configure_sip_routing(number, trunk_id="your-sip-trunk-id")
# Record the assignment
await db.assign_number(number, business_id)
return numberThe pool approach gives you flexibility: you can pre-provision numbers in bulk during off-hours, assign them instantly during calls, and reclaim them when businesses churn.
These six patterns (lifecycle management, cached TTS, silence handling, call recording, room metadata, and number management) form the operational foundation of a production voice agent. None of them affect what your agent says or how it reasons. They affect whether the experience feels professional, whether you can debug problems after the fact, and whether the system scales beyond a single number handling a handful of calls. Get these right early. Retrofitting them into a running system is significantly harder than building them in from the start.