feat: stream the sync upload with transcribe_stream()

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_async()`
-- `aai.AsyncSyncTranscriber` — Asyncio counterpart of `SyncTranscriber`. Same input types, config, result, and errors; `transcribe()` 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()`, `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.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,6 +213,29 @@ 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
+waiting for the whole clip. The request starts immediately, so authorization, the upload
+and every speech segment but the last resolve while the caller is still recording; what
+is left to wait for once they stop is the final segment. Takes an iterable of chunks or a
+file object read as it fills — not a path, and not a bytes buffer:
+```python
+def mic_chunks():
+    while recording:
+        yield stream.read(4096)
+
+result = aai.SyncTranscriber().transcribe_live(mic_chunks())
+```
+It pays off only when the audio is genuinely still being produced. Streaming a file
+already on disk is *slower* than `transcribe()`, which sends it in one piece — the saving
+comes from overlapping the recording, not from chunking. It also needs enough audio to
+have segments to release early: below roughly a minute only the elided upload counts, so
+short clips gain little. Keep producing until done — an upload that goes silent for long
+enough is aborted server-side; end the iterator to finish, don't pause it. Auth,
+rate-limit and capacity failures can surface part-way through the upload rather than at
+the end. `settings.sync_live_http_timeout` (180s) bounds each socket operation, as every
+httpx timeout does, not the request end to end: time spent waiting on the producer is not
+counted, so it does not need to cover the recording.
+
 ## Asyncio sync transcription (`AsyncSyncTranscriber`)
 
 `AsyncSyncTranscriber` is `SyncTranscriber` for the event loop: same input types

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 |
+| `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.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 |
@@ -518,10 +518,108 @@ with aai.SyncTranscriber() as transcriber:
 
 </details>
 
+<details>
+  <summary>Transcribe live audio as it is recorded (`transcribe_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.
+
+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.
+
+```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)
+
+
+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
+
+
+stop = threading.Event()
+threading.Thread(target=lambda: (input("Recording, press Enter to stop... "), stop.set()), daemon=True).start()
+
+result = aai.SyncTranscriber().transcribe_live(microphone(stop), config=config)
+print(result.text)
+```
+
+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.
+
+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.
+
+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.
+
+```python
+import subprocess
+
+# Audio you already hold whole: use transcribe(), which is faster for it.
+result = aai.SyncTranscriber().transcribe("./call.wav")
+
+# 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)
+```
+
+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.
+
+</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.
+
+```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 main():
+    frames: asyncio.Queue = asyncio.Queue()
+    start_capture(lambda pcm: frames.put_nowait(pcm))  # your audio callback
+
+    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)
+
+asyncio.run(main())
+```
+
+</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()` 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. 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
@@ -1273,6 +1371,13 @@ 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
 ```
 
 </details>

assemblyai/__version__.py

@@ -1 +1 @@
-__version__ = "1.3.0"
+__version__ = "1.4.0"

assemblyai/sync/v1/__init__.py

@@ -6,11 +6,14 @@
     SyncWord,
 )
 from ._base import AudioInput
+from ._multipart import AsyncAudioChunks, AudioChunks
 from .async_client import AsyncSyncTranscriber
 from .client import SyncTranscriber
 
 __all__ = [
+    "AsyncAudioChunks",
     "AsyncSyncTranscriber",
+    "AudioChunks",
     "AudioInput",
     "SyncSpeechModel",
     "SyncTranscriber",

assemblyai/sync/v1/_base.py

@@ -7,6 +7,7 @@
 from ... import client as _client
 from ... import types
 from . import api
+from ._multipart import AudioChunks
 
 AudioInput = Union[str, bytes, bytearray, "os.PathLike[str]", BinaryIO]
 
@@ -73,6 +74,32 @@ def _resolve_audio(
     else:
         raise TypeError(f"unsupported audio input type: {type(data).__name__}")
 
+    resolved_filename, content_type = resolve_format(config, suffix, filename)
+
+    return audio, resolved_filename, content_type
+
+
+def resolve_format(
+    config: types.SyncTranscriptionConfig,
+    suffix: str = "",
+    filename: Optional[str] = None,
+) -> Tuple[str, str]:
+    """
+    Decides the multipart filename and Content-Type for the audio part.
+
+    PCM is selected when `suffix` is a PCM extension or when
+    `sample_rate`/`channels` are set on the config — the fields the sync API
+    requires only for raw PCM — and both must then be present. Everything else
+    is treated as a WAV container. Needs no audio bytes, so it serves the
+    streamed path as well as the buffered one.
+
+    Args:
+        config: the transcription options.
+        suffix: the source's lowercased file extension, if any.
+        filename: the name for the multipart part; defaulted when absent.
+
+    Returns: `(filename, content_type)`.
+    """
     wants_pcm = config.sample_rate is not None or config.channels is not None
     is_pcm = suffix in _PCM_SUFFIXES or wants_pcm
     if is_pcm and (config.sample_rate is None or config.channels is None):
@@ -85,7 +112,7 @@ def _resolve_audio(
     if not filename:
         filename = "audio.pcm" if is_pcm else "audio.wav"
 
-    return audio, filename, content_type
+    return filename, content_type
 
 
 def _config_to_json(config: types.SyncTranscriptionConfig) -> Optional[dict]:
@@ -95,6 +122,61 @@ def _config_to_json(config: types.SyncTranscriptionConfig) -> Optional[dict]:
     return data or None
 
 
+def check_chunks(data: object, *, allow_async: bool = False) -> None:
+    """
+    Raises unless `data` can be streamed.
+
+    Names the mistakes worth catching early — a whole audio buffer, which
+    belongs in `transcribe()`; a path, which the streaming path cannot open on
+    the caller's behalf without deciding when to read it; and an async
+    iterable handed to the synchronous transcriber, which cannot drive it.
+
+    Args:
+        data: the candidate audio source.
+        allow_async: whether an object exposing only `__aiter__` is acceptable,
+            i.e. whether the caller is `AsyncSyncTranscriber`.
+    """
+    if isinstance(data, (bytes, bytearray)):
+        raise TypeError(
+            "transcribe_live() expects an iterable of audio chunks or a file "
+            "object, not audio bytes. Audio you already hold whole should go to "
+            "transcribe(), which is faster for it."
+        )
+
+    if isinstance(data, (str, os.PathLike)):
+        raise TypeError(
+            "transcribe_live() expects an iterable of audio chunks or a file "
+            "object, not a path. Open the file and pass the file object, or use "
+            "transcribe() to let the SDK read it."
+        )
+
+    if hasattr(data, "read") or hasattr(data, "__iter__"):
+        return
+
+    if hasattr(data, "__aiter__"):
+        if allow_async:
+            return
+        raise TypeError(
+            "SyncTranscriber.transcribe_live() cannot consume an async "
+            "iterable. Use AsyncSyncTranscriber.transcribe_live(), or hand it "
+            "a plain iterable or file object."
+        )
+
+    raise TypeError(f"unsupported audio stream type: {type(data).__name__}")
+
+
+def stream_filename(
+    data: object, config: types.SyncTranscriptionConfig
+) -> Tuple[str, str]:
+    """Resolves the audio part's filename and Content-Type for a stream."""
+
+    name = getattr(data, "name", None)
+    filename = os.path.basename(name) if isinstance(name, str) and name else None
+    suffix = os.path.splitext(filename)[1].lower() if filename else ""
+
+    return resolve_format(config, suffix, filename)
+
+
 class _SyncTranscriberImpl:
     def __init__(
         self,
@@ -123,3 +205,24 @@ def transcribe(
             config=_config_to_json(config),
             timeout=self._client.settings.sync_http_timeout,
         )
+
+    def transcribe_live(
+        self,
+        *,
+        data: AudioChunks,
+        config: Optional[types.SyncTranscriptionConfig],
+    ) -> types.SyncTranscriptResponse:
+        config = config or self.config
+        check_chunks(data)
+        filename, content_type = stream_filename(data, config)
+
+        return api.transcribe_live(
+            self._client.http_client,
+            base_url=self._client.settings.sync_base_url,
+            chunks=data,
+            filename=filename,
+            audio_content_type=content_type,
+            model=config.model,
+            config=_config_to_json(config),
+            timeout=self._client.settings.sync_live_http_timeout,
+        )