Voice Agent API — recommended
# Voice Agent API: sales lead qualification agent
import asyncio, json, websockets
API_KEY = "YOUR_API_KEY"
async def run_agent():
async with websockets.connect(
"wss://agents.assemblyai.com/v1/ws",
additional_headers={"Authorization": f"Bearer {API_KEY}"},
) as ws:
await ws.send(json.dumps({
"type": "session.update",
"session": {
"system_prompt": (
"You are a sales qualification agent for Acme Corp. "
"Ask about budget, timeline, and decision-maker."
),
"greeting": "Hi, thanks for your interest in Acme — how can I help?",
"input": {"keyterms": ["Acme Pro", "Enterprise Plan", "tier-2"]},
"output": {"voice": "ivy"},
},
}))
# Stream audio in, get audio + transcript back
async for msg in ws:
handle(json.loads(msg)) # transcript.user, reply.audio, tool.call, ...
Universal-3.5 Pro Realtime + LiveKit — BYO stack
# LiveKit + AssemblyAI STT in a cascading sales agent pipeline
from livekit.agents import Agent, AgentSession, TurnHandlingOptions
from livekit.plugins import assemblyai, cartesia, openai, silero
class SalesAgent(Agent):
def __init__(self):
super().__init__(
instructions=(
"You are a sales qualification agent for Acme Corp. "
"Be concise. Qualify on budget, timeline, and authority."
),
)
async def entrypoint(ctx):
session = AgentSession(
stt=assemblyai.STT(
model="u3-rt-pro",
min_turn_silence=100,
max_turn_silence=1000, # let buyers finish thoughts on email/phone numbers
vad_threshold=0.3,
keyterms_prompt=["Acme Pro", "Enterprise Plan", "tier-2"],
),
llm=openai.LLM(model="gpt-4o"),
tts=cartesia.TTS(),
vad=silero.VAD.load(activation_threshold=0.3),
turn_handling=TurnHandlingOptions(
turn_detection="stt",
endpointing={"min_delay": 0}, # avoid additive latency in STT-driven turns
),
)
await session.start(room=ctx.room, agent=SalesAgent())