Voice Agent API — legal intake agent with case-management write-back
# Voice Agent API: legal intake voice agent
import asyncio, json, websockets
API_KEY = "YOUR_API_KEY"
async def run_intake_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 legal intake assistant for Reeves & Partners. "
"Capture caller name, case type, key dates, and a one-line "
"summary. Never offer legal advice. Confirm captured fields "
"and route to the attorney on call via schedule_consultation."
),
"greeting": "Reeves & Partners, how can I help you today?",
"input": {"keyterms": ["deposition", "subpoena", "motion to compel", "discovery"]},
"output": {"voice": "ivy"},
"tools": [{
"type": "function",
"name": "schedule_consultation",
"description": "Book a consultation with the attorney on call.",
"parameters": {
"type": "object",
"properties": {
"client_name": {"type": "string"},
"case_type": {"type": "string"},
"summary": {"type": "string"},
},
"required": ["client_name", "case_type"],
},
}],
},
}))
async for msg in ws:
handle(json.loads(msg)) # transcript.user, reply.audio, tool.call, ...
Universal-3.5 Pro Realtime — live deposition transcript with PII redaction
# Universal-3.5 Pro Realtime: live deposition transcript
import asyncio, json, websockets
from urllib.parse import urlencode
API_KEY = "YOUR_API_KEY"
params = urlencode({
"sample_rate": 16000,
"speech_model": "u3-rt-pro",
"keyterms_prompt": json.dumps([
"Williams v. Meridian Corp", "Case No. 2026-CV-04471",
"Exhibit A", "deposition", "subpoena duces tecum",
"motion to compel",
]),
"format_turns": "true",
"speaker_labels": "true", # counsel vs. witness vs. judge
"redact_pii": "true", # mask privileged PII inline
"redact_pii_policies": json.dumps([
"us_social_security_number", "date_of_birth",
"phone_number", "email_address", "person_name",
"credit_card_number", "location",
]),
"redact_pii_sub": "entity_name", # e.g. [PERSON_NAME]
})
async def transcribe_deposition(audio_iter, export_to_case_mgmt):
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 with speaker_label + PII-redacted transcript
export_to_case_mgmt({
"speaker": evt.get("speaker_label"),
"transcript": evt["transcript"],
"words": evt.get("words", []), # word-level timestamps
})