Voice Agent API — hands-free agent with FSM write-back
# Voice Agent API: hands-free field service voice 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 hands-free assistant for an HVAC field technician. "
"Capture part numbers, asset IDs, and work status. Always "
"confirm captured fields back to the tech before calling "
"update_work_order. Keep responses under 2 sentences."
),
"greeting": "Ready when you are — what's the update?",
"input": {"keyterms": ["Carrier 50XC", "Trane XR", "scroll assembly", "compressor"]},
"output": {"voice": "ivy"},
"tools": [{
"type": "function",
"name": "update_work_order",
"description": "Push captured fields to the FSM platform.",
"parameters": {
"type": "object",
"properties": {
"wo_id": {"type": "string"},
"part_number": {"type": "string"},
"status": {"type": "string"},
},
"required": ["wo_id", "status"],
},
}],
},
}))
async for msg in ws:
handle(json.loads(msg)) # transcript.user, reply.audio, tool.call, ...
Universal-3.5 Pro Realtime — voice notes to structured fields
# Universal-3.5 Pro Realtime: voice notes → structured work order
import asyncio, json, websockets
from urllib.parse import urlencode
API_KEY = "YOUR_API_KEY"
params = urlencode({
"sample_rate": 16000,
"speech_model": "u3-rt-pro",
"language_detection": "true", # tag each turn with detected language
"keyterms_prompt": json.dumps([
"Carrier 50XC", "Trane XR", "scroll assembly",
"compressor seized", "refrigerant leak",
"3BAK-0601", "FS-20260518",
]),
"format_turns": "true",
"speaker_labels": "true", # tech vs. customer on-site
})
async def stream_field_notes(audio_iter, send_to_fsm):
url = f"wss://streaming.assemblyai.com/v3/ws?{params}"
async with websockets.connect(
url, additional_headers={"Authorization": API_KEY},
) as ws:
async def send_audio():
async for chunk in audio_iter:
await ws.send(chunk)
asyncio.create_task(send_audio())
async for raw in ws:
evt = json.loads(raw)
if evt.get("type") == "Turn" and evt.get("end_of_turn"):
# finalized turn → LLM Gateway extracts {wo_id, part, status}
fields = extract_work_order_fields(evt["transcript"])
send_to_fsm(fields)