Universal-3.6 Pro Realtime — live deposition transcript with PII redaction
# Universal-3.6 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": "universal-3-6-pro",
"keyterms_prompt": json.dumps([
"Williams v. Meridian Corp", "Case No. 2026-CV-04471",
"Exhibit A", "deposition", "subpoena duces tecum",
"motion to compel",
]),
"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
})