Skip to content

Repository files navigation


CI PassingGitHub LicensePyPI versionPyPI Python VersionsPyPI - WheelAssemblyAI TwitterAssemblyAI YouTubeDiscord

AssemblyAI's Python SDK

Build with AI models that can transcribe and understand audio

With a single API call, get access to AI models built on the latest AI breakthroughs to transcribe and understand audio and speech data securely at large scale.

Using with AI coding agents

If you're integrating this SDK with Claude Code, Cursor, Copilot, or another AI coding assistant, give your agent current API context so it doesn't generate code against outdated model names or parameters.

The most effective option is project instructions. Add this to your CLAUDE.md, .cursorrules, AGENTS.md, or equivalent agent instructions file:

Always fetch https://assemblyai.com/docs/llms.txt before writing AssemblyAI code. The API has changed, do not rely on memorized parameter names.

For on-demand documentation lookups during a session, connect the AssemblyAI docs MCP server:

claude mcp add assemblyai-docs --transport http https://mcp.assemblyai.com/docs

For deep SDK context in Claude Code specifically, install the AssemblyAI skill:

claude install-skill https://github.com/AssemblyAI/assemblyai-skill

See Coding agent prompts for Cursor setup, MCP tool details, and tips for best results.

Overview

Documentation

Visit our AssemblyAI API Documentation to get an overview of our models!

Quick Start

Installation

pip install -U assemblyai

Examples

Before starting, you need to set the API key. If you don't have one yet, sign up for one!

importassemblyaiasaai# set the API keyaai.settings.api_key=f"{ASSEMBLYAI_API_KEY}"

Core Examples

Transcribe a local audio file
importassemblyaiasaaiaai.settings.base_url="https://api.assemblyai.com"aai.settings.api_key="YOUR_API_KEY"audio_file="./example.mp3"config=aai.TranscriptionConfig(
speech_models=["universal-3-5-pro", "universal-2"],
language_detection=True,
speaker_labels=True,
)
transcript=aai.Transcriber().transcribe(audio_file, config=config)
iftranscript.status==aai.TranscriptStatus.error:
raiseRuntimeError(f"Transcription failed: {transcript.error}")
print(f"\nFull Transcript:\n\n{transcript.text}")
Transcribe an URL
importassemblyaiasaaiaai.settings.base_url="https://api.assemblyai.com"aai.settings.api_key="YOUR_API_KEY"audio_file="https://assembly.ai/wildfires.mp3"config=aai.TranscriptionConfig(
speech_models=["universal-3-5-pro", "universal-2"],
language_detection=True,
speaker_labels=True,
)
transcript=aai.Transcriber().transcribe(audio_file, config=config)
iftranscript.status==aai.TranscriptStatus.error:
raiseRuntimeError(f"Transcription failed: {transcript.error}")
print(f"\nFull Transcript:\n\n{transcript.text}")
Transcribe binary data
importassemblyaiasaaiaai.settings.base_url="https://api.assemblyai.com"aai.settings.api_key="YOUR_API_KEY"transcriber=aai.Transcriber()
# Binary data is supported directly:transcript=transcriber.transcribe(data)
# Or: Upload data separately:upload_url=transcriber.upload_file(data)
transcript=transcriber.transcribe(upload_url)
Export subtitles of an audio file
importassemblyaiasaaiaai.settings.api_key="<YOUR_API_KEY>"# audio_file = "./local_file.mp3"audio_file="https://assembly.ai/wildfires.mp3"config=aai.TranscriptionConfig(
speech_models=["universal-3-5-pro", "universal-2"],
language_detection=True
)
transcript=aai.Transcriber(config=config).transcribe(audio_file)
iftranscript.status=="error":
raiseRuntimeError(f"Transcription failed: {transcript.error}")
srt=transcript.export_subtitles_srt(
# Optional: Customize the maximum number of characters per captionchars_per_caption=32
)
withopen(f"transcript_{transcript.id}.srt", "w") assrt_file:
srt_file.write(srt)
# vtt = transcript.export_subtitles_vtt()# with open(f"transcript_{transcript_id}.vtt", "w") as vtt_file:# vtt_file.write(vtt)
List all sentences and paragraphs
importassemblyaiasaaiaai.settings.api_key="<YOUR_API_KEY>"# audio_file = "./local_file.mp3"audio_file="https://assembly.ai/wildfires.mp3"config=aai.TranscriptionConfig(
speech_models=["universal-3-5-pro", "universal-2"],
language_detection=True
)
transcript=aai.Transcriber(config=config).transcribe(audio_file)
iftranscript.status=="error":
raiseRuntimeError(f"Transcription failed: {transcript.error}")
sentences=transcript.get_sentences()
forsentenceinsentences:
print(sentence.text)
print()
paragraphs=transcript.get_paragraphs()
forparagraphinparagraphs:
print(paragraph.text)
print()
Search for words in a transcript
importassemblyaiasaaiaai.settings.api_key="<YOUR_API_KEY>"# audio_file = "./local_file.mp3"audio_file="https://assembly.ai/wildfires.mp3"config=aai.TranscriptionConfig(
speech_models=["universal-3-5-pro", "universal-2"],
language_detection=True
)
transcript=aai.Transcriber(config=config).transcribe(audio_file)
iftranscript.status=="error":
raiseRuntimeError(f"Transcription failed: {transcript.error}")
# Set the words you want to search forwords= ["foo", "bar", "foo bar", "42"]
matches=transcript.word_search(words)
formatchinmatches:
print(f"Found '{match.text}' {match.count} times in the transcript")
Add custom spellings on a transcript
importassemblyaiasaaiaai.settings.api_key="<YOUR_API_KEY>"# audio_file = "./local_file.mp3"audio_file="https://assembly.ai/wildfires.mp3"config=aai.TranscriptionConfig(
speech_models=["universal-3-5-pro", "universal-2"],
language_detection=True
)
config.set_custom_spelling(
{
"Gettleman": ["gettleman"],
"SQL": ["Sequel"],
}
)
transcript=aai.Transcriber(config=config).transcribe(audio_file)
iftranscript.status=="error":
raiseRuntimeError(f"Transcription failed: {transcript.error}")
print(transcript.text)
Upload a file
importassemblyaiasaaitranscriber=aai.Transcriber()
upload_url=transcriber.upload_file(data)
Delete a transcript
importassemblyaiasaaiaai.settings.api_key="<YOUR_API_KEY>"# audio_file = "./local_file.mp3"audio_file="https://assembly.ai/wildfires.mp3"config=aai.TranscriptionConfig(
speech_models=["universal-3-5-pro", "universal-2"],
language_detection=True
)
transcript=aai.Transcriber(config=config).transcribe(audio_file)
iftranscript.status=="error":
raiseRuntimeError(f"Transcription failed: {transcript.error}")
print(transcript.text)
transcript.delete_by_id(transcript.id)
transcript=aai.Transcript.get_by_id(transcript.id)
print(transcript.text)
List transcripts

This returns a page of transcripts you created.

importassemblyaiasaaitranscriber=aai.Transcriber()
page=transcriber.list_transcripts()
print(page.page_details) # Page detailsprint(page.transcripts) # List of transcripts

You can apply filter parameters:

params=aai.ListTranscriptParameters(
limit=3,
status=aai.TranscriptStatus.completed,
)
page=transcriber.list_transcripts(params)

You can also paginate over all pages by using the helper property before_id_of_prev_url.

The prev_url always points to a page with older transcripts. If you extract the before_id of the prev_url query parameters, you can paginate over all pages from newest to oldest.

transcriber=aai.Transcriber()
params=aai.ListTranscriptParameters()
page=transcriber.list_transcripts(params)
whilepage.page_details.before_id_of_prev_urlisnotNone:
params.before_id=page.page_details.before_id_of_prev_urlpage=transcriber.list_transcripts(params)

Sync STT Transcription Examples

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

Transcribe a local file synchronously
importassemblyaiasaaiaai.settings.api_key="<YOUR_API_KEY>"result=aai.SyncTranscriber().transcribe("./call.wav")
print(result.text)
forwordinresult.words:
print(word.text, word.confidence)

The input can be a local file path, raw bytes, or a binary file object — but not a URL. Pass a path/bytes, or use aai.Transcriber for URL ingestion.

Configure the transcription
importassemblyaiasaaiaai.settings.api_key="<YOUR_API_KEY>"config=aai.SyncTranscriptionConfig(
prompt="Transcribe verbatim. Preserve disfluencies.", # max 4096 charskeyterms_prompt=["AssemblyAI", "Lemur", "U3-Pro"], # max 2048 chars totalconversation_context=[
# prior turns from the same conversation, oldest first"I'd like to book a flight to Denver.",
"Sure, what date were you thinking?",
],
)
result=aai.SyncTranscriber().transcribe("./call.wav", config=config)
print(result.text)

Raw S16LE PCM audio needs sample_rate and channels; WAV reads them from its header.

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

language_codes steers the model toward one or more languages — a single-element list for monolingual audio, or several codes for multilingual audio. It is mutually exclusive with prompt; pass one or the other, not both.

importassemblyaiasaaiaai.settings.api_key="<YOUR_API_KEY>"config=aai.SyncTranscriptionConfig(language_codes=["es"]) # or ["en", "es"] for multilingualresult=aai.SyncTranscriber().transcribe("./call.wav", config=config)
print(result.text)
Get word timestamps

Word timestamps are opt-in. By default each word carries text and confidence only — start/end are None. Set timestamps=True to compute accurate per-word timings at a small latency cost.

importassemblyaiasaaiaai.settings.api_key="<YOUR_API_KEY>"config=aai.SyncTranscriptionConfig(timestamps=True)
result=aai.SyncTranscriber().transcribe("./call.wav", config=config)
forwordinresult.words:
print(word.text, word.start, word.end) # milliseconds
Pre-warm the connection

The sync API is a single request/response, so a transcribe() that connects on demand pays the full DNS + TCP + TLS handshake on the critical path. Call warm() as soon as you know audio is coming — for example while it is still being recorded — so the next transcribe() reuses the open connection.

importassemblyaiasaaiaai.settings.api_key="<YOUR_API_KEY>"withaai.SyncTranscriber() astranscriber:
transcriber.warm() # fire as recording startsaudio=record_until_done()
result=transcriber.transcribe(audio) # reuses the hot connectionprint(result.text)
Handle errors

Failures raise aai.SyncTranscriptError with the HTTP status_code, a machine-readable error_code (bad_audio, audio_too_short, audio_too_large, capacity_exceeded, …), and retry_after (seconds) on 429/503 responses.

importassemblyaiasaaiaai.settings.api_key="<YOUR_API_KEY>"try:
result=aai.SyncTranscriber().transcribe("./call.wav")
print(result.text)
exceptaai.SyncTranscriptErroraserror:
print(error.status_code, error.error_code, error.retry_after)

Speech Understanding Examples

PII Redact a transcript
importassemblyaiasaaiaai.settings.api_key="<YOUR_API_KEY>"# audio_file = "./local_file.mp3"audio_file="https://assembly.ai/wildfires.mp3"config=aai.TranscriptionConfig(
speech_models=["universal-3-5-pro", "universal-2"],
language_detection=True,
).set_redact_pii(
policies=[
aai.PIIRedactionPolicy.person_name,
aai.PIIRedactionPolicy.organization,
aai.PIIRedactionPolicy.occupation,
],
substitution=aai.PIISubstitutionPolicy.hash,
)
transcript=aai.Transcriber().transcribe(audio_file, config)
print(f"Transcript ID:", transcript.id)
print(transcript.text)

To request a copy of the original audio file with the redacted information "beeped" out, set redact_pii_audio=True in the config. Once the Transcript object is returned, you can access the URL of the redacted audio file with get_redacted_audio_url, or save the redacted audio directly to disk with save_redacted_audio.

importassemblyaiasaaiaai.settings.api_key="<YOUR_API_KEY>"# audio_file = "./local_file.mp3"audio_file="https://assembly.ai/wildfires.mp3"config=aai.TranscriptionConfig(
speech_models=["universal-3-5-pro", "universal-2"],
language_detection=True,
).set_redact_pii(
policies=[
aai.PIIRedactionPolicy.person_name,
aai.PIIRedactionPolicy.organization,
aai.PIIRedactionPolicy.occupation,
],
substitution=aai.PIISubstitutionPolicy.hash,
redact_audio=True
)
transcript=aai.Transcriber().transcribe(audio_file, config)
print(f"Transcript ID:", transcript.id)
print(transcript.text)
print(transcript.get_redacted_audio_url())

Read more about PII redaction here.

Summarize the content of a transcript over time
importassemblyaiasaaiaai.settings.api_key="<YOUR_API_KEY>"# audio_file = "./local_file.mp3"audio_file="https://assembly.ai/wildfires.mp3"config=aai.TranscriptionConfig(
speech_models=["universal-3-5-pro", "universal-2"],
language_detection=True,
auto_chapters=True
)
transcript=aai.Transcriber().transcribe(audio_file, config)
print(f"Transcript ID:", transcript.id)
forchapterintranscript.chapters:
print(f"{chapter.start}-{chapter.end}: {chapter.headline}")

Read more about auto chapters here.

Summarize the content of a transcript
importassemblyaiasaaiaai.settings.api_key="<YOUR_API_KEY>"# audio_file = "./local_file.mp3"audio_file="https://assembly.ai/wildfires.mp3"config=aai.TranscriptionConfig(
speech_models=["universal-3-5-pro", "universal-2"],
language_detection=True,
summarization=True,
summary_model=aai.SummarizationModel.informative,
summary_type=aai.SummarizationType.bullets
)
transcript=aai.Transcriber().transcribe(audio_file, config)
print(f"Transcript ID: ", transcript.id)
print(transcript.summary)

By default, the summarization model will be informative and the summarization type will be bullets. Read more about summarization models and types here.

To change the model and/or type, pass additional parameters to the TranscriptionConfig:

config=aai.TranscriptionConfig(
summarization=True,
summary_model=aai.SummarizationModel.catchy,
summary_type=aai.SummarizationType.headline
)
Detect sensitive content in a transcript
importassemblyaiasaaiaai.settings.api_key="<YOUR_API_KEY>"# audio_file = "./local_file.mp3"audio_file="https://assembly.ai/wildfires.mp3"config=aai.TranscriptionConfig(
speech_models=["universal-3-5-pro", "universal-2"],
language_detection=True,
content_safety=True
)
transcript=aai.Transcriber().transcribe(audio_file, config)
print(f"Transcript ID:", transcript.id)
forresultintranscript.content_safety.results:
print(result.text)
print(f"Timestamp: {result.timestamp.start} - {result.timestamp.end}")
# Get category, confidence, and severity.forlabelinresult.labels:
print(f"{label.label} - {label.confidence} - {label.severity}") # content safety category# Get the confidence of the most common labels in relation to the entire audio file.forlabel, confidenceintranscript.content_safety.summary.items():
print(f"{confidence*100}% confident that the audio contains {label}")
# Get the overall severity of the most common labels in relation to the entire audio file.forlabel, severity_confidenceintranscript.content_safety.severity_score_summary.items():
print(f"{severity_confidence.low*100}% confident that the audio contains low-severity {label}")
print(f"{severity_confidence.medium*100}% confident that the audio contains medium-severity {label}")
print(f"{severity_confidence.high*100}% confident that the audio contains high-severity {label}")

Read more about the content safety categories.

By default, the content safety model will only include labels with a confidence greater than 0.5 (50%). To change this, pass content_safety_confidence (as an integer percentage between 25 and 100, inclusive) to the TranscriptionConfig:

config=aai.TranscriptionConfig(
content_safety=True,
content_safety_confidence=80, # only include labels with a confidence greater than 80%
)
Analyze the sentiment of sentences in a transcript
importassemblyaiasaaiaai.settings.api_key="<YOUR_API_KEY>"# audio_file = "./local_file.mp3"audio_file="https://assembly.ai/wildfires.mp3"config=aai.TranscriptionConfig(
speech_models=["universal-3-5-pro", "universal-2"],
language_detection=True,
sentiment_analysis=True
)
transcript=aai.Transcriber().transcribe(audio_file, config)
print(f"Transcript ID:", transcript.id)
forsentiment_resultintranscript.sentiment_analysis:
print(sentiment_result.text)
print(sentiment_result.sentiment) # POSITIVE, NEUTRAL, or NEGATIVEprint(sentiment_result.confidence)
print(f"Timestamp: {sentiment_result.start} - {sentiment_result.end}")

If speaker_labels is also enabled, then each sentiment analysis result will also include a speaker field.

# ...config=aai.TranscriptionConfig(sentiment_analysis=True, speaker_labels=True)
# ...forsentiment_resultintranscript.sentiment_analysis:
print(sentiment_result.speaker)

Read more about sentiment analysis here.

Identify entities in a transcript
importassemblyaiasaaiaai.settings.api_key="<YOUR_API_KEY>"# audio_file = "./local_file.mp3"audio_file="https://assembly.ai/wildfires.mp3"config=aai.TranscriptionConfig(
speech_models=["universal-3-5-pro", "universal-2"],
language_detection=True,
entity_detection=True
)
transcript=aai.Transcriber().transcribe(audio_file, config)
print(f"Transcript ID:", transcript.id)
forentityintranscript.entities:
print(entity.text)
print(entity.entity_type)
print(f"Timestamp: {entity.start} - {entity.end}\n")

Read more about entity detection here.

Detect topics in a transcript (IAB Classification)
importassemblyaiasaaiaai.settings.api_key="<YOUR_API_KEY>"# audio_file = "./local_file.mp3"audio_file="https://assembly.ai/wildfires.mp3"config=aai.TranscriptionConfig(
speech_models=["universal-3-5-pro", "universal-2"],
language_detection=True,
iab_categories=True
)
transcript=aai.Transcriber().transcribe(audio_file, config)
print(f"Transcript ID:", transcript.id)
# Get the parts of the transcript that were tagged with topicsforresultintranscript.iab_categories.results:
print(result.text)
print(f"Timestamp: {result.timestamp.start} - {result.timestamp.end}")
forlabelinresult.labels:
print(f"{label.label} ({label.relevance})")
# Get a summary of all topics in the transcriptfortopic, relevanceintranscript.iab_categories.summary.items():
print(f"Audio is {relevance*100}% relevant to {topic}")

Read more about IAB classification here.

Identify important words and phrases in a transcript
importassemblyaiasaaiaai.settings.api_key="<YOUR_API_KEY>"# audio_file = "./local_file.mp3"audio_file="https://assembly.ai/wildfires.mp3"config=aai.TranscriptionConfig(
speech_models=["universal-3-5-pro", "universal-2"],
language_detection=True,
auto_highlights=True
)
transcript=aai.Transcriber().transcribe(audio_file, config)
print(f"Transcript ID:", transcript.id)
forresultintranscript.auto_highlights.results:
print(f"Highlight: {result.text}, Count: {result.count}, Rank: {result.rank}, Timestamps: {result.timestamps}")

Read more about auto highlights here.


Streaming Examples

Real-time speech-to-text via WebSocket against the universal-3-5-pro model. The SDK ships two clients with identical option/event/handler surfaces — StreamingClient (threaded) and AsyncStreamingClient (asyncio). Pick whichever fits your codebase.

Handler contract: every handler is called as handler(client, event). Plain functions and async def functions both work; AsyncStreamingClient awaits async handlers inline on the read task, so don't block — use asyncio.create_task(...) if you need concurrent work.

Read more about the streaming service.

Stream a local file (sync)
importassemblyaiasaaifromassemblyai.streaming.v3import (
BeginEvent, StreamingClient, StreamingClientOptions, StreamingError,
StreamingEvents, StreamingParameters, TerminationEvent, TurnEvent,
)
defon_begin(client, event: BeginEvent):
print(f"Session started: {event.id}")
defon_turn(client, event: TurnEvent):
print(f"{event.transcript} (end_of_turn={event.end_of_turn})")
defon_terminated(client, event: TerminationEvent):
print(f"Done: {event.audio_duration_seconds}s of audio processed")
defon_error(client, error: StreamingError):
print(f"Error: {error} (code={error.code})")
client=StreamingClient(StreamingClientOptions(api_key="<YOUR_API_KEY>"))
client.on(StreamingEvents.Begin, on_begin)
client.on(StreamingEvents.Turn, on_turn)
client.on(StreamingEvents.Termination, on_terminated)
client.on(StreamingEvents.Error, on_error)
client.connect(StreamingParameters(
sample_rate=16000, speech_model="universal-3-5-pro",
))
try:
client.stream(aai.extras.stream_file(filepath="audio.wav", sample_rate=16000))
finally:
client.disconnect(terminate=True)
Stream your microphone (sync)

MicrophoneStream requires PyAudio:

pip install -U "assemblyai[extras]"
importassemblyaiasaaifromassemblyai.streaming.v3import (
StreamingClient, StreamingClientOptions, StreamingEvents, StreamingParameters,
)
defon_turn(client, event):
print(f"{event.transcript} (end_of_turn={event.end_of_turn})")
client=StreamingClient(StreamingClientOptions(api_key="<YOUR_API_KEY>"))
client.on(StreamingEvents.Turn, on_turn)
client.connect(StreamingParameters(sample_rate=16000, speech_model="universal-3-5-pro"))
try:
client.stream(aai.extras.MicrophoneStream(sample_rate=16000))
finally:
client.disconnect(terminate=True)
Dual-channel: mic + system audio in one session

For note-taker apps that capture two live sources (microphone and system/speaker output) but want them handled as one streaming session — while still knowing which source each word came from — wrap the client in a ChannelStreamer.

You declare named channels and feed each channel's PCM separately. The SDK runs per-channel energy VAD, mixes the channels into a single mono stream over one websocket, and — for handlers registered on the coordinator — delivers an enriched DualChannelTurnEvent whose words/turn carry their originating channel (turn.channel and per-word word.channel). The base Word / TurnEvent stay unchanged, so single-stream payloads aren't affected. Attribution is fully client-side and model-agnostic, so it composes with speaker_labels, multilingual, and universal-3-5-pro. It is a separate dimension from diarizationword.channel (physical source) is independent of word.speaker (voice): two people on the same system channel get distinct speaker labels, while one person heard on two channels keeps a single speaker label.

Unlike a browser sample, the SDK does not capture audio — you supply 16-bit PCM for each channel (from sounddevice, pyaudio, a loopback device, files, …).

fromassemblyai.streaming.v3import (
ChannelStreamer, StreamingClient, StreamingClientOptions,
StreamingEvents, StreamingParameters,
)
defon_turn(client, event): # event is a DualChannelTurnEventprint(f"[{event.channel}] {event.transcript}")
forwinevent.words:
print(f" {w.text!r} -> channel={w.channel} speaker={w.speaker}")
client=StreamingClient(StreamingClientOptions(api_key="<YOUR_API_KEY>"))
# Declare the channels and the session sample rate (must be pcm_s16le).mixer=ChannelStreamer(client, channels=["mic", "system"], sample_rate=16000)
# Register handlers on the mixer: Turn handlers receive the enriched event,# other events (Begin/Error/…) are forwarded to the client.mixer.on(StreamingEvents.Turn, on_turn)
client.connect(StreamingParameters(
sample_rate=16000, speech_model="universal-3-5-pro", speaker_labels=True,
))
# Feed each source separately — e.g. from two capture callbacks. Send# continuous PCM for every channel (silence as zeros), at the same rate.mixer.stream("mic", mic_pcm)
mixer.stream("system", system_pcm)
mixer.flush() # push trailing buffered audioclient.disconnect(terminate=True)

AsyncChannelStreamer is the asyncio-native equivalent (await mixer.stream(...) / await mixer.close_channel(...) / await mixer.flush()); register handlers the same way with mixer.on(...).

Sources that end mid-session. Mixing keeps channels aligned by consuming the shortest buffer, so it assumes every channel keeps delivering PCM (send silence as zeros, don't omit it). When a source genuinely ends (file EOF, screen share stopped, device removed), call mixer.close_channel(name) so the session degrades to the surviving channel(s) instead of stalling — the ended channel is then padded with silence.

Swappable VAD. The default detector is the built-in energy-based EnergyVad. Supply your own (e.g. a DNN VAD such as Silero) via ChannelAttributionOptions.create_vad, which is called once per channel with the channel name; subclass VadDetector (process(frame) -> VadResult, reset()). Pass on_vad=callback to observe raw per-frame activity (e.g. a live "who's talking" meter). Tune the default with EnergyVad(threshold_ratio=3.0, noise_floor_alpha=0.05, hangover_frames=10)threshold_ratio below ~2 is too sensitive, above ~6 misses quiet onsets/offsets.

Resolving unknown channels. A word is "unknown" when no channel was clearly dominant in its window — silence, or two channels too close to call (the top must beat the runner-up by dominance_ratio, default 4). ChannelAttributionOptions.resolve_unknown_channels_method back-fills these:

  • "window" (default) — from the dominant non-"unknown" channel among ±resolution_window_words neighbor words.
  • "speaker-history" — from the speaker's session-wide channel evidence (requires speaker_labels).
  • "none" — leave "unknown" as-is.

Back-filled words are flagged word.channel_resolved = True; confident per-word decisions are never overwritten. The method is validated at construction, so a typo raises immediately rather than silently disabling resolution.

Caveats.

  • Requires 16-bit PCM (pcm_s16le, the default) — linear mixing is invalid for pcm_mulaw.
  • Capturing the system/speaker output is platform-specific: macOS needs a loopback driver (e.g. BlackHole); Windows uses WASAPI loopback; Linux a PulseAudio/PipeWire monitor source.
  • If the mic physically picks up the speakers, that bleed can pull attribution toward mic. Apply acoustic echo cancellation at capture (getUserMedia({ audio: { echoCancellation: true } }) in browser front-ends, or an AEC-capable native path) — the SDK only receives already-captured PCM, so it can't apply AEC itself. Transcription quality is unaffected; only the channel field.

See examples/streaming_dual_channel.py for a complete runnable demo.

Stream a local file (async)

AsyncStreamingClient mirrors StreamingClient with async methods. It's safe to use as an async context manager — disconnect() runs on block exit even if user code raises. Don't pass extras.stream_file directly (it uses blocking time.sleep); pace from an async generator instead.

importasynciofromassemblyai.streaming.v3import (
AsyncStreamingClient, StreamingClientOptions, StreamingEvents, StreamingParameters,
)
asyncdefstream_file_async(path: str, sample_rate: int, chunk_duration: float=0.3):
bytes_per_chunk=int(sample_rate*chunk_duration) *2withopen(path, "rb") asf:
whilechunk:=f.read(bytes_per_chunk):
yieldchunkawaitasyncio.sleep(chunk_duration)
asyncdefon_turn(client, event):
print(f"{event.transcript} (end_of_turn={event.end_of_turn})")
asyncdefmain():
asyncwithAsyncStreamingClient(StreamingClientOptions(api_key="<YOUR_API_KEY>")) asclient:
client.on(StreamingEvents.Turn, on_turn)
awaitclient.connect(StreamingParameters(
sample_rate=16000, speech_model="universal-3-5-pro",
))
awaitclient.stream(stream_file_async("audio.wav", 16000))
asyncio.run(main())
Handle errors

Server-side errors arrive on the Error event rather than being raised. The handler receives a StreamingError (an Exception subclass) with .code: int | Nonenot the wire ErrorEvent class.

StreamingErrorCodes is a dict[int, str] mapping wire codes to human-readable messages. Use .get(...) for lookup:

fromassemblyai.streaming.v3importStreamingErrorCodesdefon_error(client, error):
message=StreamingErrorCodes.get(error.code, str(error))
print(f"Streaming error {error.code}: {message}")

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

Change settings mid-session

set_params updates an active session. Typical use: enable turn formatting (punctuation, casing) only on confirmed end-of-turn so partial transcripts stay raw:

fromassemblyai.streaming.v3importStreamingSessionParametersdefon_turn(client, event):
ifevent.end_of_turnandnotevent.turn_is_formatted:
client.set_params(StreamingSessionParameters(format_turns=True))

For voice agents, force_endpoint() flushes the current turn — useful when an external signal (UI button, barge-in detection) determines the user has stopped speaking before VAD does:

client.force_endpoint() # ends the current turn immediately
Temporary tokens for browser / edge clients

Don't ship your API key to browsers. Mint a short-lived token server-side and pass it to the client.

Sync server (Flask / WSGI / scripts):

client=StreamingClient(StreamingClientOptions(api_key="<YOUR_API_KEY>"))
token=client.create_temporary_token(expires_in_seconds=60)
# Send `token` to the browser, which connects with options(token=token).

Async server (FastAPI / asyncio): always wrap in async with even though you don't call connect()create_temporary_token lazily opens an httpx.AsyncClient pool. The context manager closes it on exit; without it you leak a pool every request.

fromfastapiimportFastAPIfromassemblyai.streaming.v3importAsyncStreamingClient, StreamingClientOptionsapp=FastAPI()
MASTER_KEY="<YOUR_API_KEY>"@app.get("/streaming-token")asyncdefstreaming_token():
asyncwithAsyncStreamingClient(StreamingClientOptions(api_key=MASTER_KEY)) asclient:
return {"token": awaitclient.create_temporary_token(expires_in_seconds=60)}

Browser / edge client: pass the token via StreamingClientOptions(token=...):

client=StreamingClient(StreamingClientOptions(token="<TOKEN_FROM_SERVER>"))
client.connect(StreamingParameters(sample_rate=16000, speech_model="universal-3-5-pro"))

Change the default settings

You'll find the Settings class with all default values in types.py.

Change the default timeout and polling interval
importassemblyaiasaaiaai.settings.base_url="https://api.assemblyai.com"aai.settings.api_key="YOUR_API_KEY"# The HTTP timeout in seconds for general requests, default is 30.0aai.settings.http_timeout=60.0# The polling interval in seconds for long-running requests, default is 3.0aai.settings.polling_interval=10.0

Playground

Visit our Playground to try our all of our Speech AI models and LeMUR for free:

Advanced

How the SDK handles Default Configurations

Defining Defaults

When no TranscriptionConfig is being passed to the Transcriber or its methods, it will use a default instance of a TranscriptionConfig.

If you would like to re-use the same TranscriptionConfig for all your transcriptions, you can set it on the Transcriber directly:

config=aai.TranscriptionConfig(punctuate=False, format_text=False)
transcriber=aai.Transcriber(config=config)
# will use the same config for all `.transcribe*(...)` operationstranscriber.transcribe("https://example.org/audio.wav")

Overriding Defaults

You can override the default configuration later via the .config property of the Transcriber:

transcriber=aai.Transcriber()
# override the `Transcriber`'s config with a new configtranscriber.config=aai.TranscriptionConfig(punctuate=False, format_text=False)

In case you want to override the Transcriber's configuration for a specific operation with a different one, you can do so via the config parameter of a .transcribe*(...) method:

config=aai.TranscriptionConfig(punctuate=False, format_text=False)
# set a default configurationtranscriber=aai.Transcriber(config=config)
transcriber.transcribe(
"https://example.com/audio.mp3",
# overrides the above configuration on the `Transcriber` with the followingconfig=aai.TranscriptionConfig(speech_models=["universal-3-5-pro", "universal-2"], multichannel=True, disfluencies=True)
)

Synchronous vs Asynchronous

Currently, the SDK provides two ways to transcribe audio files.

The synchronous approach halts the application's flow until the transcription has been completed.

The asynchronous approach allows the application to continue running while the transcription is being processed. The caller receives a concurrent.futures.Future object which can be used to check the status of the transcription at a later time.

You can identify those two approaches by the _async suffix in the Transcriber's method name (e.g. transcribe vs transcribe_async).

Getting the HTTP status code

There are two ways of accessing the HTTP status code:

  • All custom AssemblyAI Error classes have a status_code attribute.
  • The latest HTTP response is stored in aai.Client.get_default().latest_response after every API call. This approach works also if no Exception is thrown.
transcriber=aai.Transcriber()
# Option 1: Catch the errortry:
transcript=transcriber.submit("./example.mp3")
exceptaai.AssemblyAIErrorase:
print(e.status_code)
# Option 2: Access the latest response through the clientclient=aai.Client.get_default()
try:
transcript=transcriber.submit("./example.mp3")
except:
print(client.last_response)
print(client.last_response.status_code)

Polling Intervals

By default we poll the Transcript's status each 3s. In case you would like to adjust that interval:

importassemblyaiasaaiaai.settings.base_url="https://api.assemblyai.com"aai.settings.api_key="YOUR_API_KEY"aai.settings.polling_interval=1.0

Retrieving Existing Transcripts

Retrieving a Single Transcript

If you previously created a transcript, you can use its ID to retrieve it later.

importassemblyaiasaaiaai.settings.base_url="https://api.assemblyai.com"aai.settings.api_key="YOUR_API_KEY"transcript=aai.Transcript.get_by_id("<TRANSCRIPT_ID>")
print(transcript.id)
print(transcript.text)

Retrieving Multiple Transcripts as a Group

You can also retrieve multiple existing transcripts and combine them into a single TranscriptGroup object. This allows you to perform operations on the transcript group as a single unit.

importassemblyaiasaaiaai.settings.base_url="https://api.assemblyai.com"aai.settings.api_key="YOUR_API_KEY"transcript_group=aai.TranscriptGroup.get_by_ids(["<TRANSCRIPT_ID_1>", "<TRANSCRIPT_ID_2>"])

Retrieving Transcripts Asynchronously

Both Transcript.get_by_id and TranscriptGroup.get_by_ids have asynchronous counterparts, Transcript.get_by_id_async and TranscriptGroup.get_by_ids_async, respectively. These functions immediately return a Future object, rather than blocking until the transcript(s) are retrieved.

See the above section on Synchronous vs Asynchronous for more information.

About

AssemblyAI's Official Python SDK

Resources

Stars

207 stars

Watchers

4 watching

Forks

Releases

Used by

Contributors

Languages