feat(sync): push-style live sessions via open_live()

CLAUDE.md

@@ -44,8 +44,8 @@ aai.settings.api_key = "your-key"
 - `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()`, `transcribe_async()`, `warm()`
-- `aai.AsyncSyncTranscriber` — Asyncio counterpart of `SyncTranscriber`. Same input types, config, result, and errors; `transcribe()`, `transcribe_live()` and `warm()` are coroutines. Owns an HTTP pool: use `async with` or `await aclose()`, or pass an `aai.AsyncClient` to share one
+- `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`
 - `assemblyai.streaming.v3.RealTimeTranscriber` — Real-time streaming with event-based API (threaded)
@@ -213,7 +213,7 @@ result = aai.SyncTranscriber().transcribe(raw_pcm_bytes, config=config)
 not asyncio) for fanning out a handful of files. In asyncio code use
 `aai.AsyncSyncTranscriber` instead (see "Asyncio sync transcription" below).
 
-**Live upload** (`transcribe_live()`): uploads audio as it is produced instead of
+**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
@@ -225,6 +225,17 @@ def mic_chunks():
 
 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:
+```python
+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

README.md

@@ -102,7 +102,7 @@ aai.settings.api_key = f"{ASSEMBLYAI_API_KEY}"
 | --- | --- |
 | `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 (`transcribe_live()`) |
+| `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.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 |
@@ -519,107 +519,116 @@ with aai.SyncTranscriber() as transcriber:
 </details>
 
 <details>
-  <summary>Transcribe live audio as it is recorded (`transcribe_live()`)</summary>
+  <summary>Transcribe live audio as it is recorded (`open_live()`)</summary>
 
-`transcribe()` needs the whole clip before it can send anything. `transcribe_live()` starts the request immediately and uploads audio as your code produces 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.
+`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.
 
-It takes an iterable of `bytes` chunks or a binary file object read as it fills, and returns the same `SyncTranscriptResponse` as `transcribe()`. Pass raw S16LE PCM with `sample_rate` and `channels` set; that is what microphones and telephony hand you, and it needs no container header.
+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.
 
 ```python
-import queue
-import threading
-
 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)
+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)
+```
 
-def microphone(stop: threading.Event):
-    """Yields PCM chunks from the default microphone until `stop` is set."""
-    chunks: queue.Queue[bytes] = queue.Queue()
-    with sd.RawInputStream(
-        samplerate=RATE, channels=1, dtype="int16",
-        callback=lambda data, *_: chunks.put(bytes(data)),
-    ):
-        while not stop.is_set():
-            try:
-                yield chunks.get(timeout=0.1)
-            except queue.Empty:
-                pass
+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.
 
-stop = threading.Event()
-threading.Thread(target=lambda: (input("Recording, press Enter to stop... "), stop.set()), daemon=True).start()
+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.
 
-result = aai.SyncTranscriber().transcribe_live(microphone(stop), config=config)
-print(result.text)
-```
+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.
 
-Any producer works, not only a microphone: a file object on a pipe or socket, chunks arriving over a websocket, a telephony media stream. End the iterator to finish. The server aborts an upload that goes silent for long, so keep sending until you are done rather than pausing.
+</details>
 
-This is **not** the streaming API. `transcribe_live()` returns one finished transcript when the audio ends; use `aai.RealTimeTranscriber` when you need words back while the speaker is still talking.
+<details>
+  <summary>Transcribe live audio from an iterator or file object (`transcribe_live()`)</summary>
 
-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.
+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.
 
 ```python
 import subprocess
+import assemblyai as aai
 
-# Audio you already hold whole: use transcribe(), which is faster for it.
-result = aai.SyncTranscriber().transcribe("./call.wav")
+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):
-proc = subprocess.Popen(["rec", "-q", "-t", "raw", "-r", "16000", "-c", "1", "-b", "16", "-e", "signed", "-"], stdout=subprocess.PIPE)
-result = aai.SyncTranscriber().transcribe_live(proc.stdout, config=config)
+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")
 ```
 
-Errors that would normally arrive at the end (bad key, rate limit, capacity) can surface part-way through the upload as a `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.
+End the iterator to finish. Anything the producer raises propagates unchanged and drops the upload.
 
 </details>
 
 <details>
   <summary>Transcribe live audio from asyncio</summary>
 
-`AsyncSyncTranscriber.transcribe_live()` 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.
+`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)`.
 
 ```python
 import asyncio
 import assemblyai as aai
 
 aai.settings.api_key = "<YOUR_API_KEY>"
-
 config = aai.SyncTranscriptionConfig(sample_rate=16000, channels=1)
 
 
-async def microphone(frames: asyncio.Queue):
-    """Yields PCM chunks pushed onto `frames` by a capture callback; None ends it."""
-    while (chunk := await frames.get()) is not None:
-        yield chunk
-
+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)
+```
 
-async def main():
-    frames: asyncio.Queue = asyncio.Queue()
-    start_capture(lambda pcm: frames.put_nowait(pcm))  # your audio callback
+`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 with aai.AsyncSyncTranscriber() as transcriber:
-        task = asyncio.create_task(transcriber.transcribe_live(microphone(frames), config=config))
-        await asyncio.sleep(5)          # ... or until the user stops speaking
-        frames.put_nowait(None)         # end the iterator; the upload finishes
-        result = await task
-        print(result.text)
+```python
+async def frames(websocket):
+    async for frame in websocket:
+        yield frame
 
-asyncio.run(main())
+result = await transcriber.transcribe_live(frames(websocket), config=config)
 ```
 
 </details>
 
 <details>
   <summary>Use it from asyncio (`AsyncSyncTranscriber`)</summary>
 
-`aai.AsyncSyncTranscriber` is the asyncio counterpart of `aai.SyncTranscriber` — same input types, config, result, and errors, with `transcribe()`, `transcribe_live()` and `warm()` as coroutines. 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.
+`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.
 
 ```python
 import asyncio

assemblyai/sync/v1/__init__.py

@@ -7,14 +7,16 @@
 )
 from ._base import AudioInput
 from ._multipart import AsyncAudioChunks, AudioChunks
-from .async_client import AsyncSyncTranscriber
-from .client import SyncTranscriber
+from .async_client import AsyncLiveSession, AsyncSyncTranscriber
+from .client import LiveSession, SyncTranscriber
 
 __all__ = [
     "AsyncAudioChunks",
+    "AsyncLiveSession",
     "AsyncSyncTranscriber",
     "AudioChunks",
     "AudioInput",
+    "LiveSession",
     "SyncSpeechModel",
     "SyncTranscriber",
     "SyncTranscriptError",

assemblyai/sync/v1/_base.py

@@ -122,6 +122,15 @@ def _config_to_json(config: types.SyncTranscriptionConfig) -> Optional[dict]:
     return data or None
 
 
+class _Aborted(Exception):
+    """
+    Raised inside a live session's producer when the session is aborted.
+
+    Propagates out of the transport, which drops the connection, and is
+    swallowed by `abort()`; it never reaches the caller.
+    """
+
+
 def check_chunks(data: object, *, allow_async: bool = False) -> None:
     """
     Raises unless `data` can be streamed.