PyPI - Wheel

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

AssemblyAI's Python SDK
Overview
Documentation
Migrating to 1.0
Quick Start
Advanced

Documentation

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

Quick Start

Installation

pip install -U assemblyai

Upgrading from 0.x? See the 1.0 migration guide for the breaking changes (LeMUR and the audio-capture extras were removed) and the recommended patterns going forward.

Examples

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

import assemblyai as aai

# set the API key
aai.settings.api_key = f"{ASSEMBLYAI_API_KEY}"

Choosing a transcriber

Class	Use it for
`assemblyai.prerecorded.v2.Transcriber`	Long-form audio, URLs, and the audio-intelligence features (speaker labels, chapters, sentiment, …), over the polled job API
`assemblyai.prerecorded.v2.AsyncTranscriber`	The same, from asyncio code
`assemblyai.sync.v1.SyncTranscriber`	Short clips (≤120s, ≤40MB) where you want the transcript back in one request, at the lowest latency — from a file you already have (`transcribe()`) or uploaded while it is still being recorded (`open_live()`, `transcribe_live()`)
`assemblyai.sync.v1.AsyncSyncTranscriber`	The same, from asyncio code
`assemblyai.dictation.v1.DictationTranscriber`	Dictation: short spoken notes uploaded as they are spoken, returned as one transcript with an optional LLM rewrite (`final_text`). Push audio from a callback (`open_live()`) or hand over an iterator, file object, bytes or path (`transcribe_live()`)
`assemblyai.dictation.v1.AsyncDictationTranscriber`	The same, from asyncio code
`assemblyai.streaming.v3.RealTimeTranscriber`	Live audio (microphone, telephony, voice agents), transcribed as it arrives over a websocket session
`assemblyai.streaming.v3.AsyncRealTimeTranscriber`	The same, from asyncio code

The versioned path is the preferred import for new code. The prerecorded, sync and dictation classes are also available as top-level shortcuts (aai.Transcriber, aai.SyncTranscriber, aai.DictationTranscriber, …) — see Migrating to 1.0 for the details.

Core Examples

import assemblyai as aai

aai.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)

if transcript.status == aai.TranscriptStatus.error:
    raise RuntimeError(f"Transcription failed: {transcript.error}")
print(f"\nFull Transcript:\n\n{transcript.text}")

import assemblyai as aai

aai.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)

if transcript.status == aai.TranscriptStatus.error:
    raise RuntimeError(f"Transcription failed: {transcript.error}")
print(f"\nFull Transcript:\n\n{transcript.text}")

import assemblyai as aai

aai.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)

import assemblyai as aai

aai.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)

if transcript.status == "error":
  raise RuntimeError(f"Transcription failed: {transcript.error}")

srt = transcript.export_subtitles_srt(
  # Optional: Customize the maximum number of characters per caption
  chars_per_caption=32
  )

with open(f"transcript_{transcript.id}.srt", "w") as srt_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)

import assemblyai as aai

aai.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)

if transcript.status == "error":
  raise RuntimeError(f"Transcription failed: {transcript.error}")

sentences = transcript.get_sentences()
for sentence in sentences:
  print(sentence.text)
  print()

paragraphs = transcript.get_paragraphs()
for paragraph in paragraphs:
  print(paragraph.text)
  print()

import assemblyai as aai

aai.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)

if transcript.status == "error":
  raise RuntimeError(f"Transcription failed: {transcript.error}")

# Set the words you want to search for
words = ["foo", "bar", "foo bar", "42"]

matches = transcript.word_search(words)

for match in matches:
  print(f"Found '{match.text}' {match.count} times in the transcript")

import assemblyai as aai

aai.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)

if transcript.status == "error":
  raise RuntimeError(f"Transcription failed: {transcript.error}")

print(transcript.text)

import assemblyai as aai

transcriber = aai.Transcriber()
upload_url = transcriber.upload_file(data)

import assemblyai as aai

aai.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)

if transcript.status == "error":
  raise RuntimeError(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)

This returns a page of transcripts you created.

import assemblyai as aai

transcriber = aai.Transcriber()

page = transcriber.list_transcripts()
print(page.page_details)  # Page details
print(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)
while page.page_details.before_id_of_prev_url is not None:
    params.before_id = page.page_details.before_id_of_prev_url
    page = 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.

import assemblyai as aai

aai.settings.api_key = "<YOUR_API_KEY>"

result = aai.SyncTranscriber().transcribe("./call.wav")

print(result.text)
for word in result.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.

Every entry point here opens the same connection: the audio is uploaded as a stream, and the service transcribes each speech segment as it lands. A clip you already have is simply a stream whose bytes are all ready at once, so transcribe() is the ergonomic shape rather than a different request. Audio must be WAV or raw PCM.

import assemblyai as aai

aai.settings.api_key = "<YOUR_API_KEY>"

config = aai.SyncTranscriptionConfig(
    prompt="Transcribe verbatim. Preserve disfluencies.",  # max 6000 chars
    keyterms_prompt=["AssemblyAI", "Universal", "U3-Pro"], # max 100 terms / 8000 chars
    conversation_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)

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.

import assemblyai as aai

aai.settings.api_key = "<YOUR_API_KEY>"

config = aai.SyncTranscriptionConfig(language_codes=["es"])  # or ["en", "es"] for multilingual
result = aai.SyncTranscriber().transcribe("./call.wav", config=config)
print(result.text)

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.

import assemblyai as aai

aai.settings.api_key = "<YOUR_API_KEY>"

config = aai.SyncTranscriptionConfig(timestamps=True)
result = aai.SyncTranscriber().transcribe("./call.wav", config=config)

for word in result.words:
    print(word.text, word.start, word.end)  # milliseconds

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.

import assemblyai as aai

aai.settings.api_key = "<YOUR_API_KEY>"

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

transcribe() needs the whole clip before it can send anything. A live session starts the request immediately and uploads audio as you produce it, so the upload and all but the last speech segment are done by the time the speaker stops. What remains is one final segment: on a paced 7 s clip that cut the wait after the last byte roughly in half.

Most audio sources hand you chunks in a callback. open_live() is built for that: write() from the callback, close() when the speaker stops, and result() for the transcript, which is the same SyncTranscriptResponse that transcribe() returns. The request runs on one of the transcriber's worker threads, so write() never blocks and is safe from any thread.

import assemblyai as aai
import sounddevice as sd  # pip install sounddevice

aai.settings.api_key = "<YOUR_API_KEY>"

RATE = 16000
config = aai.SyncTranscriptionConfig(sample_rate=RATE, channels=1)  # raw PCM needs both

with aai.SyncTranscriber() as transcriber:
    with transcriber.open_live(config) as session:
        microphone = sd.RawInputStream(
            samplerate=RATE, channels=1, dtype="int16",
            callback=lambda data, *_: session.write(bytes(data)),
        )
        with microphone:
            input("Recording, press Enter to stop... ")
    # leaving the block ends the audio; the final segment is all that is left
    print(session.result().text)

Leaving the with block calls close(). If the block raises, the session is aborted instead and result() raises. Call abort() yourself to drop a recording you no longer want transcribed.

This is not the streaming API. A live session returns one finished transcript when the audio ends; use aai.RealTimeTranscriber when you need words back while the speaker is still talking.

Use it only when the audio is genuinely still being produced. Streaming a file that is already on disk is slower than transcribe(), which sends it in one piece. Short clips gain less: below about a minute the saving is the elided upload only. The same 120 s cap applies, and the server aborts an upload that goes silent for long, so keep writing until you are done rather than pausing.

Errors that would normally arrive at the end (bad key, rate limit, capacity) can surface part-way through the upload; result() raises them as SyncTranscriptError. A bad key is reported at the first segment boundary, roughly 30 s in. warm() opens the connection early but does not validate the key.

When the audio already comes as something you can iterate, or as a file object that fills over time, hand it to transcribe_live() directly. It takes an iterable of bytes chunks or a binary file object, blocks until the audio ends, and returns the transcript. open_live() is this method with a queue in front of it.

import subprocess
import assemblyai as aai

aai.settings.api_key = "<YOUR_API_KEY>"
config = aai.SyncTranscriptionConfig(sample_rate=16000, channels=1)

# A file object that fills over time, e.g. the stdout of a recorder (sox):
recorder = subprocess.Popen(
    ["rec", "-q", "-t", "raw", "-r", "16000", "-c", "1", "-b", "16", "-e", "signed", "-"],
    stdout=subprocess.PIPE,
)
result = aai.SyncTranscriber().transcribe_live(recorder.stdout, config=config)

# Or any generator of chunks:
def from_call(call):
    while call.active:
        yield call.read_frame()

result = aai.SyncTranscriber().transcribe_live(from_call(call), config=config)

# Audio you already hold whole belongs in transcribe(), which is faster for it:
result = aai.SyncTranscriber().transcribe("./call.wav")

End the iterator to finish. Anything the producer raises propagates unchanged and drops the upload.

AsyncSyncTranscriber.open_live() returns an AsyncLiveSession: write() and close() are plain functions so a callback can call them, and result() and abort() are coroutines. The request runs as a task on the event loop. write() must be called on the loop's thread; from an audio library's capture thread, schedule it with loop.call_soon_threadsafe(session.write, chunk).

import asyncio
import assemblyai as aai

aai.settings.api_key = "<YOUR_API_KEY>"
config = aai.SyncTranscriptionConfig(sample_rate=16000, channels=1)


async def handle(websocket):
    """Transcribes the PCM frames a browser sends over one websocket."""
    async with aai.AsyncSyncTranscriber() as transcriber:
        async with transcriber.open_live(config) as session:
            async for frame in websocket:
                session.write(frame)
        result = await session.result()
        await websocket.send(result.text)

AsyncSyncTranscriber.transcribe_live() is the pull-style twin: it takes an async iterable of chunks as well as a file object or plain iterable. A file object is read in a worker thread, so a blocking read() is fine; a plain iterable is consumed inline and must not block the loop.

async def frames(websocket):
    async for frame in websocket:
        yield frame

result = await transcriber.transcribe_live(frames(websocket), config=config)

aai.AsyncSyncTranscriber is the asyncio counterpart of aai.SyncTranscriber — same input types, config, result, and errors, with transcribe(), transcribe_live() and warm() as coroutines and open_live() returning an AsyncLiveSession. Use it in asyncio code (FastAPI, aiohttp, voice agents), where the threaded transcribe() would block the event loop and transcribe_async()'s concurrent.futures.Future is not awaitable.

import asyncio
import assemblyai as aai

aai.settings.api_key = "<YOUR_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())

The transcriber owns an HTTP connection pool: use async with, or call await transcriber.aclose(). To share one pool between transcribers, pass an aai.AsyncClient, which stays yours to close. Concurrency is plain asyncio — await asyncio.gather(transcriber.transcribe(a), transcriber.transcribe(b)).

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.

import assemblyai as aai

aai.settings.api_key = "<YOUR_API_KEY>"

try:
    result = aai.SyncTranscriber().transcribe("./call.wav")
    print(result.text)
except aai.SyncTranscriptError as error:
    print(error.status_code, error.error_code, error.retry_after)

Dictation Examples

aai.DictationTranscriber is for dictation: a person speaks a short note and gets it back as text, optionally cleaned up or reformatted by an LLM. It has one connection, the live upload — audio is sent as it is spoken, the service transcribes each speech segment as it lands, and when the speaker stops what remains is the final segment and the LLM pass. There is no job id and no polling. Audio must be WAV or raw 16-bit PCM, up to 120 s per request.

The result is a DictationResponse: text is the raw transcript, llm_response the rewritten text when the LLM pass ran, and final_text is the one to show the user — the rewrite when there is one, the raw transcript otherwise.

Most audio sources deliver chunks in a callback. open_live() is built for that: the request starts the moment the session opens, write() from the callback (safe from any thread, never blocks), leave the with block when the speaker stops, and result() returns the transcript.

import assemblyai as aai
import sounddevice as sd  # pip install sounddevice

aai.settings.api_key = "<YOUR_API_KEY>"

RATE = 16000
config = aai.DictationConfig(sample_rate=RATE, channels=1)  # raw PCM needs both

with aai.DictationTranscriber() as transcriber:
    with transcriber.open_live(config) as session:
        microphone = sd.RawInputStream(
            samplerate=RATE, channels=1, dtype="int16",
            callback=lambda data, *_: session.write(bytes(data)),
        )
        with microphone:
            input("Dictating, press Enter to stop... ")
    # leaving the block ends the audio; the final segment is all that is left
    print(session.result().final_text)

Leaving the with block calls close(). If the block raises, the session is aborted instead and result() raises. Call abort() yourself to drop a dictation you no longer want transcribed.

This is not the streaming API. A session returns one finished transcript when the audio ends; use aai.RealTimeTranscriber when you need words back while the speaker is still talking.

The server aborts an upload that goes silent for long, so keep writing until the speaker is done rather than pausing. Errors that would normally arrive at the end (bad key, rate limit, capacity, audio too large) can surface part-way through the upload; result() raises them as DictationError.

When the audio already comes as something you can iterate, or as a file object that fills over time, hand it to transcribe_live() directly. It blocks until the audio ends and returns the transcript. open_live() is this method with a queue in front of it.

Audio you already hold whole — raw bytes, or a local path — goes to transcribe_live() too. It travels the same connection as a single chunk; there is no separate buffered request.

import subprocess
import assemblyai as aai

aai.settings.api_key = "<YOUR_API_KEY>"
config = aai.DictationConfig(sample_rate=16000, channels=1)

# A file object that fills over time, e.g. the stdout of a recorder (sox):
recorder = subprocess.Popen(
    ["rec", "-q", "-t", "raw", "-r", "16000", "-c", "1", "-b", "16", "-e", "signed", "-"],
    stdout=subprocess.PIPE,
)
result = aai.DictationTranscriber().transcribe_live(recorder.stdout, config=config)

# Or any generator of chunks:
def from_call(call):
    while call.active:
        yield call.read_frame()

result = aai.DictationTranscriber().transcribe_live(from_call(call), config=config)

# Or a recording you already have (WAV carries its own rate and channels):
result = aai.DictationTranscriber().transcribe_live("./note.wav")
print(result.final_text)

End the iterator to finish. Anything the producer raises propagates unchanged and drops the upload. URLs are not accepted; use aai.Transcriber for URL ingestion.

Set llm_instruction to have the service run a follow-up pass over the transcript. The rewrite comes back in llm_response; the raw transcript stays in text. If the pass fails, llm_error says why and final_text falls back to text, so reading final_text is safe either way.

import assemblyai as aai

aai.settings.api_key = "<YOUR_API_KEY>"

config = aai.DictationConfig(llm_instruction="Format this as a SOAP note.")

result = aai.DictationTranscriber().transcribe_live("./visit.wav", config=config)

print(result.final_text)      # the SOAP note
print(result.text)            # what was actually said
if result.llm_error:
    print("LLM pass failed:", result.llm_error)

The instruction is capped at 2048 characters.

import assemblyai as aai

aai.settings.api_key = "<YOUR_API_KEY>"

config = aai.DictationConfig(
    language_codes=["en", "es"],                 # one code, or several for multilingual audio
    stt_prompt="A doctor dictating a visit note.", # what the audio is about; ≤6000 chars
    keyterms_prompt=["AssemblyAI", "Universal"], # bias the decoder; ≤100 terms / 8000 chars
    llm_instruction="Fix punctuation only.",     # optional LLM pass
)

result = aai.DictationTranscriber().transcribe_live("./note.wav", config=config)

stt_prompt describes the situation the audio was recorded in and steers the decoder as it writes the transcript; llm_instruction reshapes that transcript afterwards. sample_rate and channels are required only for raw PCM; WAV carries them in its own header. A config passed to transcribe_live() or open_live() overrides the transcriber's default config for that call. DictationConfig rejects unknown fields, so a typo or a sync-only option surfaces as a validation error instead of a setting that quietly does nothing.

A request that connects on demand pays the full DNS + TCP + TLS handshake before the first audio byte can leave. Call warm() as soon as you know audio is coming — when the user reaches for the record button — so the request that follows reuses the open connection.

import assemblyai as aai

aai.settings.api_key = "<YOUR_API_KEY>"

with aai.DictationTranscriber() as transcriber:
    transcriber.warm()                       # fire when the user is about to speak
    with transcriber.open_live(config) as session:
        record_into(session.write)
    print(session.result().final_text)

aai.AsyncDictationTranscriber is the asyncio counterpart — same audio sources, config, result and errors, with transcribe_live() and warm() as coroutines. open_live() returns an AsyncDictationLiveSession: write() and close() are plain functions so a callback can call them, and result() and abort() are coroutines. The request runs as a task on the event loop. write() must be called on the loop's thread; from an audio library's capture thread, schedule it with loop.call_soon_threadsafe(session.write, chunk).

import asyncio
import assemblyai as aai

aai.settings.api_key = "<YOUR_API_KEY>"
config = aai.DictationConfig(sample_rate=16000, channels=1)


async def handle(websocket):
    """Transcribes the PCM frames a browser sends over one websocket."""
    async with aai.AsyncDictationTranscriber() as transcriber:
        async with transcriber.open_live(config) as session:
            async for frame in websocket:
                session.write(frame)
        result = await session.result()
        await websocket.send(result.final_text)

transcribe_live() also takes an async iterable of chunks, as well as a file object, plain iterable, bytes or path. A file object or path is read in a worker thread, so a blocking read() is fine; a plain iterable is consumed inline and must not block the loop.

async def frames(websocket):
    async for frame in websocket:
        yield frame

result = await transcriber.transcribe_live(frames(websocket), config=config)

The transcriber owns an HTTP connection pool: use async with, or call await transcriber.aclose(). To share one pool between transcribers, pass an aai.AsyncClient, which stays yours to close.

Failures raise aai.DictationError with the HTTP status_code, a machine-readable error_code (bad_audio, audio_too_large, capacity_exceeded, inference_timeout, …), and retry_after (seconds) on 429/503 responses. It is a separate class from aai.SyncTranscriptError.

import assemblyai as aai

aai.settings.api_key = "<YOUR_API_KEY>"

try:
    result = aai.DictationTranscriber().transcribe_live("./note.wav")
    print(result.final_text)
except aai.DictationError as error:
    print(error.status_code, error.error_code, error.retry_after)

Asyncio Examples

aai.AsyncTranscriber is the asyncio counterpart of aai.Transcriber. Every method that calls the API is a coroutine. Hundreds of transcriptions run concurrently on one thread, with no thread pool.

Use it in asyncio code (FastAPI, aiohttp, voice agents). Do not use transcribe_async there: it returns a concurrent.futures.Future, which is not awaitable.

import asyncio

import assemblyai as aai

aai.settings.api_key = "<YOUR_API_KEY>"


async def main():
    config = aai.TranscriptionConfig(
        speech_models=["universal-3-5-pro", "universal-2"],
        speaker_labels=True,
    )

    async with aai.AsyncTranscriber(config=config) as transcriber:
        transcript = await transcriber.transcribe("./example.mp3")

        if transcript.status == aai.TranscriptStatus.error:
            raise RuntimeError(f"Transcription failed: {transcript.error}")

        print(transcript.text)


asyncio.run(main())

The transcriber owns an HTTP connection pool. Close it with an async context manager, or call await transcriber.aclose().

import asyncio

import assemblyai as aai

aai.settings.api_key = "<YOUR_API_KEY>"


async def main():
    async with aai.AsyncTranscriber() as transcriber:
        # Plain asyncio - one coroutine per file, all in flight at once.
        transcripts = await asyncio.gather(
            transcriber.transcribe("./one.mp3"),
            transcriber.transcribe("./two.mp3"),
        )

        # Or let the transcriber cap how many run at a time. Results come back
        # in the order of the input.
        transcripts = await transcriber.transcribe_group(
            ["./one.mp3", "./two.mp3", "./three.mp3"],
            max_concurrency=2,
        )

        for transcript in transcripts:
            print(transcript.text)


asyncio.run(main())

Pass return_failures=True to get (transcripts, errors) instead of a raised error.

import asyncio

import assemblyai as aai

aai.settings.api_key = "<YOUR_API_KEY>"


async def main():
    async with aai.AsyncTranscriber() as transcriber:
        # Returns as soon as the job is queued - no polling.
        transcript = await transcriber.submit("https://assembly.ai/wildfires.mp3")
        print(transcript.id, transcript.status)

        # Later, in the same or another process:
        transcript = await transcriber.get_by_id(transcript.id)
        print(transcript.text)

        # Follow-up operations are coroutines too.
        sentences = await transcript.get_sentences()
        srt = await transcript.export_subtitles_srt()


asyncio.run(main())

import assemblyai as aai

aai.settings.api_key = "<YOUR_API_KEY>"


async def main():
    async with aai.AsyncClient(settings=aai.settings) as client:
        verbatim = aai.AsyncTranscriber(
            client=client,
            config=aai.TranscriptionConfig(disfluencies=True),
        )
        clean = aai.AsyncTranscriber(client=client)

        # A client that was passed in is not closed by the transcribers - the
        # `async with` above owns it.
        await verbatim.transcribe("./interview.mp3")
        await clean.transcribe("./interview.mp3")

Speech Understanding Examples

import assemblyai as aai

aai.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.

import assemblyai as aai

aai.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())

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 — RealTimeTranscriber (threaded) and AsyncRealTimeTranscriber (asyncio). Pick whichever fits your codebase.

The former Streaming* names (StreamingClient, AsyncStreamingClient, StreamingClientOptions, StreamingParameters, StreamingSessionParameters, StreamingEvents, StreamingError, StreamingErrorCodes) remain available as aliases of the RealTime* names — same objects, so existing code keeps working unchanged.

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

Change the default settings

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

import assemblyai as aai

aai.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.0
aai.settings.http_timeout = 60.0

# The polling interval in seconds for long-running requests, default is 3.0
aai.settings.polling_interval = 10.0

# Per-operation timeouts for the sync API: transcribe() (default 60.0) and
# transcribe_live() (default 180.0). Like every httpx timeout these bound each
# socket operation, not the request end to end, so the live value does not
# need to cover the length of the recording.
aai.settings.sync_http_timeout = 60.0
aai.settings.sync_live_http_timeout = 180.0

# The Dictation API host and its per-operation timeout (default 300.0), sized
# to outlast the final segment's inference plus the LLM pass.
aai.settings.dictation_base_url = "https://dictation.assemblyai.com"
aai.settings.dictation_http_timeout = 300.0

Playground

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

Playground

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*(...)` operations
transcriber.transcribe("https://assembly.ai/wildfires.mp3")

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 config
transcriber.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 configuration
transcriber = aai.Transcriber(config=config)

transcriber.transcribe(
    "https://assembly.ai/wildfires.mp3",
    # overrides the above configuration on the `Transcriber` with the following
    config=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).

A concurrent.futures.Future is not awaitable, and its .result() blocks the event loop. In an asyncio application, use AsyncTranscriber instead.

Asyncio

aai.AsyncTranscriber mirrors aai.Transcriber with coroutines instead of threads:

`Transcriber` (threads)	`AsyncTranscriber` (asyncio)
`transcribe(...)`	`await transcribe(...)`
`transcribe_async(...)` -> `Future`	`await transcribe(...)` (or `asyncio.gather`)
`submit(...)`	`await submit(...)`
`transcribe_group(...)`	`await transcribe_group(...)`
`Transcript.get_by_id(id)`	`await transcriber.get_by_id(id)`
`Transcript.delete_by_id(id)`	`await transcriber.delete_by_id(id)`
`transcript.get_sentences()`	`await transcript.get_sentences()`
`list_transcripts(...)`	`await list_transcripts(...)`

Notes:

Both transcribers live in assemblyai.prerecorded.v2, whose version matches the /v2/transcript API. The top-level aai.* names re-export them, so aai.AsyncTranscriber is all most callers need.
AsyncTranscriber owns an HTTP connection pool. Close it with an async context manager, or call await transcriber.aclose().
Pass client=aai.AsyncClient(settings=aai.settings) to share one pool between transcribers. A client you pass in stays yours to close.
There is no process-wide default async client. An httpx.AsyncClient pool belongs to the event loop that first used it, so a global pool fails on a second asyncio.run().
AsyncTranscript carries the same fields as Transcript. Only the methods that call the API became coroutines.
Uploads stream local files and file objects through a thread, so a large upload never blocks the loop.
transcribe_group and submit_group return results in input order. Both cap in-flight work at max_concurrency, which defaults to 8.
Neither group method drops a failure. Either the first error is raised, or you pass return_failures=True and get (transcripts, errors).

For real-time streaming, use assemblyai.streaming.v3.AsyncRealTimeTranscriber.

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 error
try:
    transcript = transcriber.submit("./example.mp3")
except aai.AssemblyAIError as e:
    print(e.status_code)

# Option 2: Access the latest response through the client
client = 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:

import assemblyai as aai

aai.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.

import assemblyai as aai

aai.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.

import assemblyai as aai

aai.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.

assemblyai