CLAUDE.md

AssemblyAI Python SDK

Speech-to-text and audio intelligence SDK. Supports pre-recorded transcription, real-time streaming, audio analysis features, and LLM access through AssemblyAI's LLM Gateway.

Quick start

pip install -U assemblyai

import os
import assemblyai as aai

aai.settings.api_key = os.environ["ASSEMBLYAI_API_KEY"]

transcript = aai.Transcriber().transcribe(
    "https://assembly.ai/wildfires.mp3",
    config=aai.TranscriptionConfig(
        speech_models=["universal-3-5-pro", "universal-2"],
        speaker_labels=True,
    ),
)

print(transcript.text)
for utterance in transcript.utterances:
    print(f"Speaker {utterance.speaker}: {utterance.text}")

Auth

Set ASSEMBLYAI_API_KEY as an environment variable, or:

aai.settings.api_key = "your-key"

Key classes

• aai.Transcriber — Transcribe files, URLs, or streams. Methods: transcribe(), transcribe_async(), submit(), list_transcripts()
• aai.AsyncTranscriber — Asyncio counterpart of Transcriber. Same options, every API call a coroutine. Methods: transcribe(), submit(), transcribe_group(), get_by_id(), delete_by_id(), list_transcripts(), upload_file()
• aai.AsyncTranscript — What AsyncTranscriber returns. Same fields as Transcript, coroutine methods
• aai.AsyncClient — Shared httpx.AsyncClient pool for one or more AsyncTranscribers
• assemblyai.prerecorded.v2 — Canonical module for both transcribers, matching the /v2/transcript API. The top-level aai.* names re-export it
• aai.TranscriptionConfig — All transcription options: speech_models, speaker_labels, sentiment_analysis, entity_detection, auto_chapters, content_safety, language_detection, summarization, word_boost, disfluencies
• aai.Transcript — Result object with .text, .status, .utterances, .words, .chapters, .entities, .sentiment_analysis. Methods: get_sentences(), get_paragraphs(), export_subtitles_srt(), export_subtitles_vtt()
• aai.SyncTranscriber — Synchronous pre-recorded transcription: audio in, transcript out, one request (no polling). Methods: transcribe(), transcribe_live(), open_live(), transcribe_async(), warm()
• aai.AsyncSyncTranscriber — Asyncio counterpart of SyncTranscriber. Same input types, config, result, and errors; transcribe(), transcribe_live() and warm() are coroutines; open_live() returns an AsyncLiveSession. Owns an HTTP pool: use async with or await aclose(), or pass an aai.AsyncClient to share one
• aai.SyncTranscriptionConfig — Sync options: model (default universal-3-5-pro), prompt, keyterms_prompt, conversation_context, language_codes, timestamps, sample_rate, channels
• aai.SyncTranscriptResponse — Sync result: .text, .words (SyncWord with confidence always, start/end only when timestamps=True), .confidence, .audio_duration_ms, .session_id, .request_time_ms
• aai.DictationTranscriber — Dictation API: audio uploaded as it is spoken over one live request, transcript (and optional LLM pass) out. Methods: open_live() (push-style DictationLiveSession: write()/close()/result()/abort()), transcribe_live() (iterable, file object, bytes or path), warm(). There is no buffered transcribe(): every entry point posts to /v1/transcribe/live
• aai.AsyncDictationTranscriber — Asyncio counterpart of DictationTranscriber; open_live() returns an AsyncDictationLiveSession. Owns an HTTP pool: use async with or await aclose()
• aai.DictationConfig — Dictation options: sample_rate, channels, language_codes, keyterms_prompt, llm_instruction. Rejects unknown fields
• aai.DictationResponse — Dictation result: .text, .words (DictationWord with text + confidence), .confidence, .llm_response, .llm_error, .audio_duration_ms, .session_id, .request_time_ms, .sync_time_ms, and .final_text (the LLM rewrite, falling back to .text)
• assemblyai.streaming.v3.RealTimeTranscriber — Real-time streaming with event-based API (threaded)
• assemblyai.streaming.v3.AsyncRealTimeTranscriber — Asyncio-native counterpart; same options/events
• aai.LLMGateway — LLM Gateway client. Resources: models.list(), chat.completions.create() (incl. stream=True), understanding.create()/.validate()
• aai.AsyncLLMGateway — Asyncio counterpart of LLMGateway. Same resources, every API call a coroutine; owns an HTTP pool (async with or await aclose())
• assemblyai.llm_gateway.v1 — Canonical module for both gateways, matching the gateway's /v1 API. The top-level aai.* names re-export it
• aai.LLMGatewayMessageParam — TypedDict for a chat message; pass plain dicts ({"role": "user", "content": "..."})

Common patterns

Transcribe a local file:

transcript = aai.Transcriber().transcribe("./recording.mp3")

With multiple features:

config = aai.TranscriptionConfig(
    speech_models=["universal-3-5-pro", "universal-2"],
    speaker_labels=True,
    sentiment_analysis=True,
    entity_detection=True,
    auto_chapters=True,
    language_detection=True,
)
transcript = aai.Transcriber().transcribe(audio_url, config=config)

PII redaction (uses setter, not constructor):

config = aai.TranscriptionConfig()
config.set_redact_pii(
    policies=[aai.PIIRedactionPolicy.email_address, aai.PIIRedactionPolicy.phone_number],
    substitution=aai.PIISubstitutionPolicy.hash,
)

Retrieve existing transcript:

transcript = aai.Transcript.get_by_id("transcript-id")

Asyncio transcription (AsyncTranscriber)

AsyncTranscriber is Transcriber with coroutines instead of a thread pool. Use it in asyncio code (FastAPI, aiohttp, voice agents). Do not use transcribe_async() there: its concurrent.futures.Future is not awaitable, and .result() blocks the event loop. Same TranscriptionConfig, same fields on the result.

import asyncio
import assemblyai as aai

aai.settings.api_key = os.environ["ASSEMBLYAI_API_KEY"]

async def main():
    async with aai.AsyncTranscriber() as transcriber:
        transcript = await transcriber.transcribe("./audio.mp3")
        if transcript.status == aai.TranscriptStatus.error:
            raise RuntimeError(transcript.error)
        print(transcript.text)

        sentences = await transcript.get_sentences()   # follow-ups are coroutines too
        srt = await transcript.export_subtitles_srt()

asyncio.run(main())

Lifecycle: the transcriber owns an HTTP connection pool. Use async with, or call await transcriber.aclose(). To share one pool, pass a client, which stays yours to close:

async with aai.AsyncClient(settings=aai.settings) as client:
    transcriber = aai.AsyncTranscriber(client=client)

There is no process-wide default async client, unlike sync Client. An httpx.AsyncClient pool is bound to the event loop that first used it, so a global one breaks under a second asyncio.run(...).

Concurrency: use asyncio.gather(...), or the group helpers to cap in-flight work. Results come back in input order:

transcripts = await transcriber.transcribe_group(files, max_concurrency=8)
transcripts, errors = await transcriber.transcribe_group(files, return_failures=True)

Unlike the sync transcribe_group, failures are never dropped. The first error raises after the batch settles, or you collect them with return_failures=True.

Method mapping (sync → async): Transcript.get_by_id(id) → await transcriber.get_by_id(id); Transcript.delete_by_id(id) → await transcriber.delete_by_id(id). Both moved onto the transcriber, which owns the connection pool.

Uploads: local paths and file objects stream to the upload endpoint in chunks read on a worker thread. The request sets Content-Length when the size is known. A large file never blocks the loop and never loads fully into memory.

Sync transcription (pre-recorded, single request)

SyncTranscriber posts a whole audio file and returns the finished transcript in one round trip — no job id, no polling, no status enum. It targets the sync API (sync.assemblyai.com), distinct from Transcriber's async job API. Use it for short clips where you want the answer inline; use Transcriber for long-form audio, URLs, or the rich audio-intelligence features (speaker labels, chapters, sentiment, …) the sync API doesn't expose.

import assemblyai as aai

aai.settings.api_key = os.environ["ASSEMBLYAI_API_KEY"]

result = aai.SyncTranscriber().transcribe("./call.wav")
print(result.text, result.session_id)
for w in result.words:
    print(w.text, w.confidence)  # w.start/w.end need timestamps=True (see below)

Input: a local file path, raw bytes, or a binary file object. Not a URL — pass a path/bytes or use Transcriber for URL ingestion.

Config (all optional):

config = aai.SyncTranscriptionConfig(
    prompt="Transcribe verbatim. Preserve disfluencies.",  # max 4096 chars
    keyterms_prompt=["AssemblyAI", "Universal", "U3-Pro"], # max 2048 chars total
    language_codes=["es"],                                 # or e.g. ["en", "es"] for multilingual; defaults to English
)
result = aai.SyncTranscriber().transcribe("./call.wav", config=config)

Conversation context: conversation_context carries prior turns from the same conversation so the model keeps continuity and proper-noun spelling across a multi-turn exchange. List them oldest-first (most recent last); a single prior turn can be a bare string. Capped at 100 turns / 4096 chars total — over-cap context is trimmed (oldest turns first), not rejected; oldest turns are likewise dropped first when over the model token budget.

config = aai.SyncTranscriptionConfig(
    conversation_context=[
        "I'd like to book a flight to Denver.",
        "Sure, what date were you thinking?",
    ],
)
result = aai.SyncTranscriber().transcribe("./reply.wav", config=config)

Language: language_codes takes a list of ISO 639-1 codes — a single-element list for monolingual audio or several codes for multilingual audio — and steers the default prompt toward those languages; ignored when you pass a custom prompt. Supported: en, es, de, fr, it, pt, tr, nl, sv, no, da, fi, hi, vi, ar, he, ja, ur, zh.

Word timestamps are opt-in. By default words carry text and confidence only — start/end are None. timestamps=True computes accurate per-word timings for a small latency cost:

config = aai.SyncTranscriptionConfig(timestamps=True)
result = aai.SyncTranscriber().transcribe("./call.wav", config=config)
for w in result.words:
    print(w.text, w.start, w.end)  # milliseconds

Raw PCM (S16LE) needs sample_rate + channels; WAV reads them from its header. Setting either field routes the audio as audio/pcm, and both must be present:

config = aai.SyncTranscriptionConfig(sample_rate=16000, channels=1)
result = aai.SyncTranscriber().transcribe(raw_pcm_bytes, config=config)

Concurrency: transcribe_async() returns a concurrent.futures.Future (thread-based, not asyncio) for fanning out a handful of files. In asyncio code use aai.AsyncSyncTranscriber instead (see "Asyncio sync transcription" below).

Live upload (open_live() / transcribe_live()): uploads audio as it is produced instead of waiting for the whole clip. The request starts immediately, so authorization, the upload and every speech segment but the last resolve while the caller is still recording; what is left to wait for once they stop is the final segment. Takes an iterable of chunks or a file object read as it fills — not a path, and not a bytes buffer:

def mic_chunks():
    while recording:
        yield stream.read(4096)

result = aai.SyncTranscriber().transcribe_live(mic_chunks())

For callback-driven sources (most microphone libraries, WebRTC, telephony) use the push-style session instead; write() is thread-safe and never blocks:

with transcriber.open_live(config) as session:
    start_capture(on_audio=session.write)   # e.g. sounddevice callback
    wait_until_done()
result = session.result()                    # raises SyncTranscriptError on failure

Leaving the block calls close(); an exception in the block calls abort() instead. The asyncio twin is AsyncSyncTranscriber.open_live() → AsyncLiveSession (write()/close() plain functions, result()/abort() coroutines; call write() on the loop thread). It pays off only when the audio is genuinely still being produced. Streaming a file already on disk is slower than transcribe(), which sends it in one piece — the saving comes from overlapping the recording, not from chunking. It also needs enough audio to have segments to release early: below roughly a minute only the elided upload counts, so short clips gain little. Keep producing until done — an upload that goes silent for long enough is aborted server-side; end the iterator to finish, don't pause it. Auth, rate-limit and capacity failures can surface part-way through the upload rather than at the end. settings.sync_live_http_timeout (180s) bounds each socket operation, as every httpx timeout does, not the request end to end: time spent waiting on the producer is not counted, so it does not need to cover the recording.

Asyncio sync transcription (AsyncSyncTranscriber)

AsyncSyncTranscriber is SyncTranscriber for the event loop: same input types (path/bytes/file object — no URLs), same SyncTranscriptionConfig, same SyncTranscriptResponse and SyncTranscriptError, with transcribe() and warm() as coroutines. Path and file-object reads run off the loop. Use it in FastAPI, aiohttp, and voice agents, where the threaded transcribe() would block the loop and transcribe_async()'s concurrent.futures.Future is not awaitable.

import asyncio
import assemblyai as aai

aai.settings.api_key = os.environ["ASSEMBLYAI_API_KEY"]

async def main():
    async with aai.AsyncSyncTranscriber() as transcriber:
        asyncio.create_task(transcriber.warm())   # optional: fire as recording starts
        audio = await record_until_done()
        result = await transcriber.transcribe(audio)
        print(result.text)

asyncio.run(main())

Lifecycle: like AsyncTranscriber, it owns an HTTP connection pool — use async with, or call await transcriber.aclose(). To share one pool, pass an aai.AsyncClient, which stays yours to close. There is no process-wide default async client (an httpx.AsyncClient pool is bound to the event loop that first used it).

Concurrency: plain asyncio — await asyncio.gather(transcriber.transcribe(a), transcriber.transcribe(b)). There is no transcribe_async() and no thread pool.

Errors: failures raise aai.SyncTranscriptError with .status_code, a machine-readable .error_code — the snake_cased problem-details title from the server (bad_audio, audio_too_short, audio_too_large, capacity_exceeded, inference_timeout, …) — and .retry_after (seconds) on 429/503. Audio limits: 80 ms–120 s, ≤40 MB, 16-bit, mono/stereo, sample rate ∈ {8000, 16000, 22050, 24000, 32000, 44100, 48000}.

Model: config.model defaults to "universal-3-5-pro" and is sent as the X-AAI-Model routing header (aai.SyncSpeechModel.universal_3_5_pro).

Pre-warming the connection: the sync API is one request/response, so a transcribe() that connects on demand pays the full DNS + TCP + TLS handshake on the critical path — a network round trip that, for a distant client, can rival the transcription. Call warm() as soon as you know audio is coming (e.g. while it is still being recorded) to spend that setup concurrently; the next transcribe() reuses the open connection.

aai.settings.keepalive_expiry = 120          # optional: hold idle connections 120s

with aai.SyncTranscriber() as transcriber:
    transcriber.warm()                       # fire as recording starts
    audio = record_until_done()
    result = transcriber.transcribe(audio)   # reuses the hot connection

warm() returns True once the socket is open, False on a transport failure. The warmed connection is reused while it stays in the pool — settings.keepalive_expiry seconds, which defaults to httpx's 5s. For "warm at the start of a recording, send when it ends," raise keepalive_expiry (e.g. to 120, the sync audio cap) so one warm() covers the whole clip; otherwise call warm() shortly before transcribe() (it is idempotent, so re-warming to refresh is fine). It only helps the first request: sync calls share the client's connection pool, so later transcribe() calls already reuse the open connection. Note keepalive_expiry applies to the whole client (sync and async).

Streaming (real-time)

Use universal-3-5-pro as the streaming model — it's the flagship and what every example below targets. Two clients with identical option/event/handler surfaces: RealTimeTranscriber (threaded) and AsyncRealTimeTranscriber (asyncio).

The former Streaming* names (StreamingClient, AsyncStreamingClient, StreamingClientOptions, StreamingParameters, StreamingSessionParameters, StreamingEvents, StreamingError, StreamingErrorCodes) are aliases bound to the same objects, so existing code keeps working; prefer the RealTime* names in new code.

Handler contract — important: every handler is called as handler(client, event). Two positional args. Plain functions and async def functions both work for the async client; async handlers are awaited inline on the read task, so don't block — use asyncio.create_task(...) if you need to fan out work. Exceptions inside handlers are logged and swallowed.

Event payloads (what the second arg to each handler is):

• Begin → BeginEvent(id: str, expires_at: datetime)
• Turn → TurnEvent(transcript: str, end_of_turn: bool, turn_is_formatted: bool, end_of_turn_confidence: float, words: list[Word], language_code: str | None, language_confidence: float | None, turn_order: int)
• Termination → TerminationEvent(audio_duration_seconds: int | None, session_duration_seconds: int | None)
• SpeechStarted → SpeechStartedEvent(timestamp: int) (ms)
• Warning → WarningEvent(warning_code: int, warning: str)
• Error → RealTimeError (an Exception subclass) with .code: int | None; str(error) is the message. Server-side errors come through on_error rather than being raised, and the payload is a RealTimeError, not the wire ErrorEvent class.
• LLMGatewayResponse → LLMGatewayResponseEvent(turn_order: int, transcript: str, data: Any)
• SpeakerRevision → SpeakerRevisionEvent(revisions: list[SpeakerRevisionItem]) — diarization-only. Sent once per offline-recluster resolve. Each SpeakerRevisionItem(turn_order: int, speaker_label: str | None, words: list[Word]) is an earlier Turn whose labels changed (unchanged turns are omitted). For each item, match by turn_order against the original Turn and replace its per-word speaker (and the turn-level speaker_label) with the revision's values. Text and word timestamps are unchanged. By default only the end-of-stream revision is sent; set speaker_labels_revision_interval_ms on RealTimeParameters (ms of audio time; values below 120_000 are raised to 120_000 server-side, larger values honored, 300_000 recommended) to also get them mid-stream — the first can only arrive after ~120 s of speech, and one is sent only when a label actually changed.

Sync streaming:

from assemblyai.streaming.v3 import (
    RealTimeTranscriber, RealTimeTranscriberOptions, RealTimeEvents,
    RealTimeParameters, TurnEvent,
)

def on_turn(client, event: TurnEvent):
    print(event.transcript, "end_of_turn=", event.end_of_turn)

def on_error(client, error):
    print(f"Error {error.code}: {error}")

client = RealTimeTranscriber(RealTimeTranscriberOptions(api_key=os.environ["ASSEMBLYAI_API_KEY"]))
client.on(RealTimeEvents.Turn, on_turn)
client.on(RealTimeEvents.Error, on_error)
client.connect(RealTimeParameters(
    sample_rate=16000, speech_model="universal-3-5-pro",
))
try:
    client.stream(audio_bytes_or_iterable)
finally:
    client.disconnect(terminate=True)

Async streaming (preferred for voice agents and asyncio-native codebases):

import asyncio
from assemblyai.streaming.v3 import (
    AsyncRealTimeTranscriber, RealTimeTranscriberOptions, RealTimeEvents,
    RealTimeParameters,
)

async def on_turn(client, event):
    print(event.transcript)

async def main():
    async with AsyncRealTimeTranscriber(
        RealTimeTranscriberOptions(api_key=os.environ["ASSEMBLYAI_API_KEY"])
    ) as client:
        client.on(RealTimeEvents.Turn, on_turn)
        await client.connect(RealTimeParameters(
            sample_rate=16000, speech_model="universal-3-5-pro",
        ))
        await client.stream(audio_async_generator)  # async iterable of bytes

asyncio.run(main())

Microphone streaming: the SDK does not capture audio — supply 16-bit PCM chunks yourself (e.g. via pyaudio or sounddevice) and pass the iterable to client.stream(...).

Voice-agent tuning (knobs that matter most when building a voice agent):

RealTimeParameters(
    sample_rate=8000, speech_model="universal-3-5-pro", encoding="pcm_mulaw",  # telephony
    prompt="Transcribe verbatim with filler words and full punctuation.",
    agent_context="What is your account number?",  # agent's last reply; biases U3Pro context (U3Pro only)
    keyterms_prompt=["yes", "no", "uh-huh", "mm-hmm"],  # bias short acknowledgements
    min_turn_silence=200,         # ms; lower = snappier, may split phrases
    max_turn_silence=1000,        # ms; hard cap before forcing end-of-turn
    end_of_turn_confidence_threshold=0.7,
)

Use pcm_s16le (default) + sample_rate=16000 for microphone capture.

agent_context (like prompt) can also be updated mid-stream after each agent turn: client.set_params(RealTimeSessionParameters(agent_context="...")).

Error codes: RealTimeErrorCodes is a dict[int, str] mapping wire codes to human messages. Use .get(...) for lookup — not enum-style attribute access:

from assemblyai.streaming.v3 import RealTimeErrorCodes

def on_error(client, error):
    message = RealTimeErrorCodes.get(error.code, str(error))
    print(f"Streaming error {error.code}: {message}")

Common codes worth branching on: 4001 Not Authorized, 4002 Insufficient Funds, 4029 Client sent audio too fast, 4031 Session idle for too long.

Events enum: RealTimeEvents.{Begin, Turn, Termination, SpeechStarted, Error, Warning, LLMGatewayResponse, SpeakerRevision}. Register a handler for each you care about; the call is the same shape: client.on(RealTimeEvents.Begin, on_begin), client.on(RealTimeEvents.Error, on_error), etc.

Slow async handlers — fan-out pattern: async handlers are awaited inline on the read task. If on_turn calls a slow LLM/TTS, ingestion stalls. Fan out and drain on shutdown:

pending: set[asyncio.Task] = set()

async def on_turn(client, event):
    if not event.end_of_turn:
        return
    t = asyncio.create_task(slow_responder(event.transcript))
    pending.add(t); t.add_done_callback(pending.discard)

# After client.stream(...) returns, before leaving the `async with`:
if pending:
    await asyncio.gather(*pending, return_exceptions=True)

Mint a temporary token without streaming (typical FastAPI/server use):

async with AsyncRealTimeTranscriber(RealTimeTranscriberOptions(api_key=MASTER_KEY)) as client:
    return await client.create_temporary_token(expires_in_seconds=60)

Use async with even when you never call connect() — create_temporary_token lazily opens an httpx.AsyncClient and __aexit__ closes it. Without the context manager you leak the HTTP pool per request. The sync RealTimeTranscriber.create_temporary_token doesn't need this (no pool to close).

Pass the token to the streaming client via RealTimeTranscriberOptions(token=...) — same surface for RealTimeTranscriber and AsyncRealTimeTranscriber:

async with AsyncRealTimeTranscriber(RealTimeTranscriberOptions(token=token_from_server)) as client:
    await client.connect(RealTimeParameters(sample_rate=16000, speech_model="universal-3-5-pro"))
    await client.stream(audio)

Other gotchas:

• Don't feed AsyncRealTimeTranscriber.stream() from a sync generator that blocks (e.g. time.sleep pacing) — it starves the read task. Use an async def generator with await asyncio.sleep(...) instead.
• format_turns=True enables punctuation/casing on confirmed end-of-turns. Toggle mid-session via client.set_params(RealTimeSessionParameters(format_turns=True)).
• AsyncRealTimeTranscriber used as async with calls disconnect(terminate=True) on normal block exit and disconnect(terminate=False) on exception — no explicit disconnect() needed inside the block.

LLM Gateway

aai.LLMGateway is a client for AssemblyAI's LLM Gateway: model listing, OpenAI-shaped chat completions routed to OpenAI/Claude/Gemini/Bedrock models, and speech understanding over an existing transcript. It targets llm-gateway.assemblyai.com and rides on the same Client/AsyncClient as the rest of the SDK.

import assemblyai as aai

aai.settings.api_key = os.environ["ASSEMBLYAI_API_KEY"]

gateway = aai.LLMGateway()

for model in gateway.models.list().data:
    print(model.id, model.context_length, model.pricing.global_.prompt)

completion = gateway.chat.completions.create(
    model="claude-sonnet-5",
    messages=[{"role": "user", "content": "Summarize this call."}],
)
print(completion.choices[0].message.content)

Messages are plain dicts (aai.LLMGatewayMessageParam) — no message object to build. role is "user", "assistant", "system", or "tool" (a tool result, paired with tool_call_id); content is a string or a list of content-part dicts.

Request params: everything past model/messages/stream is a keyword argument forwarded verbatim into the request body — max_tokens, temperature, tools, tool_choice, response_format, plus the AssemblyAI extensions (fallbacks, fallback_config, zero_data_retention, transcript_id, reasoning, model_region, fail_fast, max_timeout, …). fallbacks is a list of dicts in the API's own shape: [{"model": "gpt-5-mini", "messages": [...]}].

completion = gateway.chat.completions.create(
    model="claude-sonnet-5",
    messages=[{"role": "user", "content": "..."}],
    max_tokens=512,
    fallbacks=[{"model": "gpt-5-mini", "messages": [{"role": "user", "content": "..."}]}],
    zero_data_retention=True,
    transcript_id="transcript_abc123",
)

Streaming: stream=True returns an iterator of LLMGatewayCompletionChunk:

stream = gateway.chat.completions.create(
    model="claude-sonnet-5",
    messages=[{"role": "user", "content": "..."}],
    stream=True,
)
for chunk in stream:
    if chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="")

Chunk parsing is verified for OpenAI-routed models only. Chunks can carry tool-call fragments (chunk.choices[0].delta.tool_calls), but this SDK does not reassemble them or accumulate the streamed text for you — collect chunks yourself, or use a non-streaming create().

There is no built-in tool-calling loop: tools/tool_choice are forwarded like any other request param (see "Request params" above), and replaying a tool result is a plain {"role": "tool", "tool_call_id": ..., "content": ...} message per https://www.assemblyai.com/docs/llm-gateway/tool-calling — drive the loop yourself around create().

Understanding: speaker identification, translation, and custom formatting over an existing transcript — transcript_id is required, there is no raw-text path.

result = gateway.understanding.create(
    transcript_id="transcript_abc123",
    request={"speaker_identification": {"speaker_type": "name", "speakers": [{"name": "Ana"}, {"name": "Peter"}]}},
)
print(result.speech_understanding["response"], result.utterances)

gateway.understanding.validate(request={"translation": {"target_languages": ["es"]}})

request is a plain dict — the server has no fixed schema for it. validate() returns None when the request is valid and raises aai.LLMGatewayError with .errors populated otherwise; it costs the same auth/balance as a real call, so it is not a free dry run.

Errors: every failure raises aai.LLMGatewayError (an aai.AssemblyAIError) with .status_code, .request_id, and .errors (validation messages, when the server sent any). For a stream, only a failure before the first byte raises — a mid-stream failure truncates the iterator.

Unknown server fields survive: response models allow extras, so a field the gateway adds is readable as an attribute without an SDK upgrade, and collections the server sends as null are read as empty (completion.choices is always a list).

Settings: aai.settings.llm_gateway_base_url (default https://llm-gateway.assemblyai.com) and aai.settings.llm_gateway_http_timeout (default 30.0), both overridable via ASSEMBLYAI_LLM_GATEWAY_BASE_URL / ASSEMBLYAI_LLM_GATEWAY_HTTP_TIMEOUT.

Asyncio LLM Gateway (AsyncLLMGateway)

aai.AsyncLLMGateway mirrors aai.LLMGateway with coroutines — same resources, same request params, same response models. Every API call is awaited; stream=True resolves to an async iterator of LLMGatewayCompletionChunk you async for over.

import asyncio
import assemblyai as aai

aai.settings.api_key = os.environ["ASSEMBLYAI_API_KEY"]

async def main():
    async with aai.AsyncLLMGateway() as gateway:
        completion = await gateway.chat.completions.create(
            model="claude-sonnet-5",
            messages=[{"role": "user", "content": "Summarize this call."}],
        )
        print(completion.choices[0].message.content)

        stream = await gateway.chat.completions.create(
            model="claude-sonnet-5",
            messages=[{"role": "user", "content": "..."}],
            stream=True,
        )
        async for chunk in stream:
            print(chunk.choices[0].delta.content or "", end="")

asyncio.run(main())

Lifecycle: the gateway owns an HTTP connection pool — use async with, or call await gateway.aclose(). Pass client=aai.AsyncClient(settings=aai.settings) to share one pool; a client you pass in stays yours to close. Sync LLMGateway needs no such handling: it borrows the shared default Client and has no close().

Per-gateway key: both gateways take api_key=... instead of a client, and build their own client for it. Given alongside client, the key wins and the passed client is left untouched.

Important gotchas

• Always check status: if transcript.status == aai.TranscriptStatus.error — accessing .text on a failed transcript returns None, not an exception
• speech_models takes a list with fallback ordering: ["universal-3-5-pro", "universal-2"]
• PII redaction uses set_redact_pii(), not a constructor parameter
• Streaming v3 lives in its own module: assemblyai.streaming.v3 (there is no other streaming API in this SDK). See the "Streaming (real-time)" section above.
• The SDK does not capture microphone audio: bring your own capture (e.g. pyaudio, sounddevice) and stream the PCM chunks
• transcribe_async() returns a concurrent.futures.Future, not an asyncio coroutine. In asyncio code use aai.AsyncTranscriber (see "Asyncio transcription" above) — or aai.AsyncSyncTranscriber for the sync API
• LLM Gateway request params are **kwargs: everything past model/messages/stream is forwarded verbatim, so there is no client-side check on the name — spell params carefully
• Timestamps are in milliseconds throughout the SDK
• Minimum Python: 3.8+

Dependencies

httpx, pydantic, typing-extensions, websockets.

Docs

• Full documentation
• API reference
• LLM Gateway docs
• llms-full.txt (Python-filtered docs for LLMs)