feat(transcriber): add AsyncTranscriber for asyncio callers

CLAUDE.md

@@ -38,6 +38,9 @@ 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 `AsyncTranscriber`s
 - `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_async()`
@@ -80,6 +83,62 @@ config.set_redact_pii(
 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.
+
+```python
+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:
+```python
+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:
+```python
+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.
+
+**LeMUR** is sync-only. An `AsyncTranscript` works as a `LemurSource`, but the LeMUR call
+blocks. Run it off the loop, for example with `asyncio.to_thread`.
+
 ## Sync transcription (pre-recorded, single request)
 
 `SyncTranscriber` posts a whole audio file and returns the finished transcript in one
@@ -323,7 +382,7 @@ async with AsyncStreamingClient(StreamingClientOptions(token=token_from_server))
 - **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.
 - **Microphone streaming needs extras**: `pip install "assemblyai[extras]"` for `pyaudio`
-- **`transcribe_async()` returns a `concurrent.futures.Future`**, not an asyncio coroutine
+- **`transcribe_async()` returns a `concurrent.futures.Future`**, not an asyncio coroutine. In asyncio code use `aai.AsyncTranscriber` (see "Asyncio transcription" above)
 - **Timestamps are in milliseconds** throughout the SDK
 - **Minimum Python**: 3.8+
 

README.md

@@ -49,6 +49,7 @@ See [Coding agent prompts](https://www.assemblyai.com/docs/coding-agent-prompts)
   - [Installation](#installation)
   - [Examples](#examples)
     - [**Core Examples**](#core-examples)
+    - [**Asyncio Examples**](#asyncio-examples)
     - [**Speech Understanding Examples**](#speech-understanding-examples)
     - [**Streaming Examples**](#streaming-examples)
     - [**Change the default settings**](#change-the-default-settings)
@@ -58,6 +59,7 @@ See [Coding agent prompts](https://www.assemblyai.com/docs/coding-agent-prompts)
     - [Defining Defaults](#defining-defaults)
     - [Overriding Defaults](#overriding-defaults)
   - [Synchronous vs Asynchronous](#synchronous-vs-asynchronous)
+  - [Asyncio](#asyncio)
   - [Getting the HTTP status code](#getting-the-http-status-code)
   - [Polling Intervals](#polling-intervals)
   - [Retrieving Existing Transcripts](#retrieving-existing-transcripts)
@@ -518,6 +520,146 @@ except aai.SyncTranscriptError as error:
 
 ---
 
+### **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`](https://docs.python.org/3/library/concurrent.futures.html),
+which is not awaitable.
+
+<details>
+  <summary>Transcribe a file with asyncio</summary>
+
+```python
+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()`.
+
+</details>
+
+<details>
+  <summary>Transcribe many files concurrently</summary>
+
+```python
+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.
+
+</details>
+
+<details>
+  <summary>Submit now, collect later</summary>
+
+```python
+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://example.org/audio.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())
+```
+
+</details>
+
+<details>
+  <summary>Share one connection pool across transcribers</summary>
+
+```python
+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")
+```
+
+</details>
+
+---
+
 ### **Speech Understanding Examples**
 
 <details>
@@ -1176,6 +1318,45 @@ The asynchronous approach allows the application to continue running while the t
 
 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`](#asyncio) 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:
+
+- `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)`.
+- LeMUR is sync-only. An `AsyncTranscript` works as a `LemurSource`, but the LeMUR call
+  blocks. Run it in a thread, for example with `asyncio.to_thread`.
+
+For real-time streaming, use `assemblyai.streaming.v3.AsyncStreamingClient`.
+
 ## Getting the HTTP status code
 
 There are two ways of accessing the HTTP status code:

assemblyai/__init__.py

@@ -1,5 +1,7 @@
 from . import extras
 from .__version__ import __version__
+from .async_client import AsyncClient
+from .async_transcriber import AsyncTranscriber, AsyncTranscript
 from .client import Client
 from .lemur import Lemur
 from .sync import SyncTranscriber
@@ -91,6 +93,9 @@
 __all__ = [
     # types
     "AssemblyAIError",
+    "AsyncClient",
+    "AsyncTranscriber",
+    "AsyncTranscript",
     "AutohighlightResponse",
     "AutohighlightResult",
     "Chapter",