diff --git a/CLAUDE.md b/CLAUDE.md
index 515c7869..7550613c 100644
--- a/CLAUDE.md
+++ b/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
diff --git a/README.md b/README.md
index 073a1883..315349c2 100644
--- a/README.md
+++ b/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:
+
+ Transcribe live audio as it is recorded (`transcribe_live()`)
+
+`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 = ""
+
+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.
+
+
+
+
+ Transcribe live audio from asyncio
+
+`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 = ""
+
+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())
+```
+
+
+
Use it from asyncio (`AsyncSyncTranscriber`)
-`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
```
diff --git a/assemblyai/__version__.py b/assemblyai/__version__.py
index 67bc602a..3e8d9f94 100644
--- a/assemblyai/__version__.py
+++ b/assemblyai/__version__.py
@@ -1 +1 @@
-__version__ = "1.3.0"
+__version__ = "1.4.0"
diff --git a/assemblyai/sync/v1/__init__.py b/assemblyai/sync/v1/__init__.py
index 05936f41..4e3691aa 100644
--- a/assemblyai/sync/v1/__init__.py
+++ b/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",
diff --git a/assemblyai/sync/v1/_base.py b/assemblyai/sync/v1/_base.py
index 4e4b26b2..1d8ec3e1 100644
--- a/assemblyai/sync/v1/_base.py
+++ b/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,
+ )
diff --git a/assemblyai/sync/v1/_multipart.py b/assemblyai/sync/v1/_multipart.py
new file mode 100644
index 00000000..7ad83786
--- /dev/null
+++ b/assemblyai/sync/v1/_multipart.py
@@ -0,0 +1,183 @@
+"""Multipart encoding for streamed uploads.
+
+Kept apart from `_base.py` so the transport modules can use it: `_base.py`
+imports `api.py`, so anything `api.py` needs must not live there.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import re
+import secrets
+from typing import (
+ AsyncIterable,
+ AsyncIterator,
+ BinaryIO,
+ Iterable,
+ Iterator,
+ Optional,
+ Union,
+)
+
+# Audio for a streamed upload, delivered in whatever pieces the producer has
+# ready. A file object is read lazily. Audio the caller already holds whole
+# belongs in `transcribe()`, which is faster for it.
+AudioChunks = Union[Iterable[bytes], BinaryIO]
+AsyncAudioChunks = Union[AsyncIterable[bytes], Iterable[bytes], BinaryIO]
+
+# Read size for file objects handed to the streaming path.
+_STREAM_READ_SIZE = 32 * 1024
+
+# Escaping for quoted `Content-Disposition` parameters, matching what httpx's
+# own multipart encoder does (the HTML5 form-encoding rules): a double quote
+# becomes `%22`, a backslash is doubled, and control characters other than ESC
+# are percent-encoded so a name cannot break out of the header.
+_FORM_PARAM_REPLACEMENTS = {'"': "%22", "\\": "\\\\"}
+_FORM_PARAM_REPLACEMENTS.update(
+ {chr(c): "%{:02X}".format(c) for c in range(0x1F + 1) if c != 0x1B}
+)
+_FORM_PARAM_RE = re.compile("|".join(map(re.escape, _FORM_PARAM_REPLACEMENTS)))
+
+
+def _escape_form_param(value: str) -> str:
+ """Escapes `value` for use inside a quoted multipart header parameter."""
+
+ return _FORM_PARAM_RE.sub(lambda m: _FORM_PARAM_REPLACEMENTS[m.group(0)], value)
+
+
+class StreamingMultipartEncoder:
+ """
+ Builds a `multipart/form-data` body one part at a time.
+
+ The buffered path hands httpx a `files=` mapping and lets it encode the
+ whole body up front. A streamed upload cannot: the audio does not exist yet
+ when the request starts. This emits the framing itself so the `config` part
+ can go out immediately and audio can follow as it arrives.
+
+ **Part order is significant and is enforced here by construction.** The
+ server decodes audio as it lands, so it needs `sample_rate` and `channels`
+ before the first audio byte; `config` therefore precedes `audio`. Audio
+ first is rejected by the server rather than buffered, which is the whole
+ point of the endpoint.
+ """
+
+ def __init__(self, boundary: Optional[str] = None) -> None:
+ """
+ Args:
+ boundary: the multipart boundary. Generated when absent; pass one
+ only to make a test deterministic.
+ """
+ self.boundary = boundary or secrets.token_hex(16)
+
+ @property
+ def content_type(self) -> str:
+ """The `Content-Type` header the body must be sent with."""
+
+ return f"multipart/form-data; boundary={self.boundary}"
+
+ def config_part(self, config: Optional[dict]) -> bytes:
+ """
+ Encodes the JSON `config` part.
+
+ Always emitted: the streaming endpoint rejects a body whose audio is
+ not preceded by a `config` part, so when the caller set no options an
+ empty object is sent. The buffered route omits the part instead.
+ """
+
+ return (
+ (
+ f"--{self.boundary}\r\n"
+ 'Content-Disposition: form-data; name="config"\r\n'
+ "Content-Type: application/json\r\n\r\n"
+ ).encode("utf-8")
+ + json.dumps(config or {}).encode("utf-8")
+ + b"\r\n"
+ )
+
+ def audio_header(self, filename: str, content_type: str) -> bytes:
+ """
+ Encodes the `audio` part's headers, which precede its first byte.
+
+ `filename` is escaped so quotes and control characters in a file's
+ name cannot corrupt the framing.
+ """
+
+ return (
+ f"--{self.boundary}\r\n"
+ "Content-Disposition: form-data; "
+ f'name="audio"; filename="{_escape_form_param(filename)}"\r\n'
+ f"Content-Type: {content_type}\r\n\r\n"
+ ).encode("utf-8")
+
+ def closing(self) -> bytes:
+ """Encodes the terminating boundary."""
+
+ return f"\r\n--{self.boundary}--\r\n".encode("utf-8")
+
+
+def _as_bytes(chunk: object) -> bytes:
+ """
+ Returns `chunk` as `bytes`, or raises if it is not binary audio.
+
+ Names the likely cause when the producer yields text — a file opened
+ without `"b"`, or a generator of `str` — so the mistake does not surface
+ as an opaque failure deep inside the transport.
+ """
+
+ if isinstance(chunk, bytes):
+ return chunk
+ if isinstance(chunk, (bytearray, memoryview)):
+ return bytes(chunk)
+ if isinstance(chunk, str):
+ raise TypeError(
+ "transcribe_live() audio chunks must be bytes, not str. Open the "
+ "file in binary mode ('rb') or encode the producer's output."
+ )
+ raise TypeError(
+ f"transcribe_live() audio chunks must be bytes, not {type(chunk).__name__}"
+ )
+
+
+def iter_chunks(data: AudioChunks) -> Iterator[bytes]:
+ """
+ Yields audio pieces from a file object or any iterable of bytes.
+
+ Raises `TypeError` on the first piece that is not bytes-like.
+ """
+
+ if hasattr(data, "read"):
+ while True:
+ chunk = data.read(_STREAM_READ_SIZE)
+ if not chunk:
+ return
+ yield _as_bytes(chunk)
+ else:
+ for chunk in data: # type: ignore[union-attr]
+ yield _as_bytes(chunk)
+
+
+async def aiter_chunks(data: AsyncAudioChunks) -> AsyncIterator[bytes]:
+ """
+ Yields audio pieces from an async iterable, a file object, or a plain
+ iterable.
+
+ A file object is read in a worker thread, so a `read()` that blocks on a
+ pipe or socket does not stall the event loop. A plain iterable is accepted
+ so one producer can feed both transcribers, but it is consumed inline and
+ so must not block.
+
+ Raises `TypeError` on the first piece that is not bytes-like.
+ """
+ if hasattr(data, "__aiter__"):
+ async for chunk in data: # type: ignore[union-attr]
+ yield _as_bytes(chunk)
+ elif hasattr(data, "read"):
+ while True:
+ chunk = await asyncio.to_thread(data.read, _STREAM_READ_SIZE) # type: ignore[union-attr]
+ if not chunk:
+ return
+ yield _as_bytes(chunk)
+ else:
+ for chunk in data: # type: ignore[union-attr]
+ yield _as_bytes(chunk)
diff --git a/assemblyai/sync/v1/api.py b/assemblyai/sync/v1/api.py
index 7be3635c..559de78d 100644
--- a/assemblyai/sync/v1/api.py
+++ b/assemblyai/sync/v1/api.py
@@ -1,13 +1,15 @@
import json
-from typing import Dict, Optional, Tuple
+from typing import Dict, Iterator, Optional, Tuple
import httpx
from ... import types
+from ._multipart import AudioChunks, StreamingMultipartEncoder, iter_chunks
# Canonical paths since the sync API gained a /v1 prefix (#18103); the
# unprefixed routes remain served for SDK versions that predate it.
ENDPOINT_TRANSCRIBE = "/v1/transcribe"
+ENDPOINT_TRANSCRIBE_STREAM = "/v1/transcribe/stream"
ENDPOINT_WARM = "/v1/warm"
MODEL_HEADER = "X-AAI-Model"
@@ -109,3 +111,75 @@ def transcribe(
raise _error_from_response(response)
return types.SyncTranscriptResponse.parse_obj(response.json())
+
+
+def transcribe_live(
+ client: httpx.Client,
+ *,
+ base_url: str,
+ chunks: AudioChunks,
+ filename: str,
+ audio_content_type: str,
+ model: str,
+ config: Optional[dict],
+ timeout: float,
+) -> types.SyncTranscriptResponse:
+ """
+ Posts a transcription request whose audio is uploaded as it arrives.
+
+ Sends the body with chunked transfer encoding — httpx frames an unsized
+ iterator that way — so the request can start before the audio exists. The
+ server transcribes each speech segment as it lands, leaving only the final
+ segment's inference to wait on once the caller stops speaking.
+
+ Args:
+ client: the HTTP client (carries the `Authorization` header).
+ base_url: the sync API base URL, e.g. `https://sync.assemblyai.com`.
+ chunks: audio pieces (WAV container or S16LE PCM), in order.
+ filename: name for the audio multipart part.
+ audio_content_type: `audio/wav` or `audio/pcm`; selects the decoder.
+ model: sent as the `X-AAI-Model` routing header.
+ config: the JSON `config` part. None sends an empty object: the
+ streaming endpoint requires the part ahead of the audio.
+ timeout: per-operation timeout in seconds, as for every httpx request:
+ it bounds connecting, each socket write and each read while waiting
+ for the response, not the request end to end. Time blocked in the
+ caller's producer is not counted, so it need not cover the
+ recording.
+
+ Returns: the parsed transcript response.
+
+ Raises: `SyncTranscriptError` on any non-200 response — including one the
+ server sends while the upload is still in flight, which it may do for
+ auth, rate-limit and capacity failures. `TypeError` if the producer
+ yields something other than bytes. Any other exception the producer
+ raises propagates unchanged; the connection is dropped and no
+ transcript is returned.
+ """
+ encoder = StreamingMultipartEncoder()
+
+ def body() -> Iterator[bytes]:
+ # config first: the server needs sample_rate and channels before it can
+ # decode a single audio byte, and rejects audio that arrives first.
+ head = encoder.config_part(config) + encoder.audio_header(
+ filename, audio_content_type
+ )
+ yield head
+
+ for chunk in iter_chunks(chunks):
+ if chunk:
+ yield chunk
+
+ yield encoder.closing()
+
+ response = client.post(
+ base_url.rstrip("/") + ENDPOINT_TRANSCRIBE_STREAM,
+ content=body(),
+ headers={MODEL_HEADER: model, "Content-Type": encoder.content_type},
+ timeout=timeout,
+ )
+
+ if response.status_code != httpx.codes.OK:
+ raise _error_from_response(response)
+
+ return types.SyncTranscriptResponse.parse_obj(response.json())
diff --git a/assemblyai/sync/v1/async_api.py b/assemblyai/sync/v1/async_api.py
index 06012b47..b3a6d6b2 100644
--- a/assemblyai/sync/v1/async_api.py
+++ b/assemblyai/sync/v1/async_api.py
@@ -5,14 +5,20 @@
"""
import json
-from typing import Dict, Optional, Tuple
+from typing import AsyncIterator, Dict, Optional, Tuple
import httpx
from ... import types
-from .api import ENDPOINT_TRANSCRIBE, MODEL_HEADER, _error_from_response
+from ._multipart import AsyncAudioChunks, StreamingMultipartEncoder, aiter_chunks
+from .api import (
+ ENDPOINT_TRANSCRIBE,
+ ENDPOINT_TRANSCRIBE_STREAM,
+ MODEL_HEADER,
+ _error_from_response,
+)
-__all__ = ["transcribe"]
+__all__ = ["transcribe", "transcribe_live"]
async def transcribe(
@@ -66,3 +72,64 @@ async def transcribe(
raise _error_from_response(response)
return types.SyncTranscriptResponse.parse_obj(response.json())
+
+
+async def transcribe_live(
+ client: httpx.AsyncClient,
+ *,
+ base_url: str,
+ chunks: AsyncAudioChunks,
+ filename: str,
+ audio_content_type: str,
+ model: str,
+ config: Optional[dict],
+ timeout: float,
+) -> types.SyncTranscriptResponse:
+ """
+ Posts a transcription request whose audio is uploaded as it arrives.
+
+ The asyncio counterpart of `api.transcribe_live`; same endpoint, same
+ chunked framing, same errors.
+
+ Args:
+ client: the HTTP client (carries the `Authorization` header).
+ base_url: the sync API base URL, e.g. `https://sync.assemblyai.com`.
+ chunks: audio pieces (WAV container or S16LE PCM), in order.
+ filename: name for the audio multipart part.
+ audio_content_type: `audio/wav` or `audio/pcm`; selects the decoder.
+ model: sent as the `X-AAI-Model` routing header.
+ config: the JSON `config` part. None sends an empty object: the
+ streaming endpoint requires the part ahead of the audio.
+ timeout: per-operation timeout in seconds; see `api.transcribe_live`.
+
+ Returns: the parsed transcript response.
+
+ Raises: `SyncTranscriptError` on any non-200 response — including one the
+ server sends while the upload is still in flight.
+ """
+ encoder = StreamingMultipartEncoder()
+
+ async def body() -> AsyncIterator[bytes]:
+ # config first: the server needs sample_rate and channels before it can
+ # decode a single audio byte, and rejects audio that arrives first.
+ yield encoder.config_part(config) + encoder.audio_header(
+ filename, audio_content_type
+ )
+
+ async for chunk in aiter_chunks(chunks):
+ if chunk:
+ yield chunk
+
+ yield encoder.closing()
+
+ response = await client.post(
+ base_url.rstrip("/") + ENDPOINT_TRANSCRIBE_STREAM,
+ content=body(),
+ headers={MODEL_HEADER: model, "Content-Type": encoder.content_type},
+ timeout=timeout,
+ )
+
+ if response.status_code != httpx.codes.OK:
+ raise _error_from_response(response)
+
+ return types.SyncTranscriptResponse.parse_obj(response.json())
diff --git a/assemblyai/sync/v1/async_client.py b/assemblyai/sync/v1/async_client.py
index 6921c4bd..d87d9cc3 100644
--- a/assemblyai/sync/v1/async_client.py
+++ b/assemblyai/sync/v1/async_client.py
@@ -12,7 +12,15 @@
from ... import async_client as _async_client
from ... import types
from . import api, async_api
-from ._base import AudioInput, _config_to_json, _resolve_audio, check_config
+from ._base import (
+ AudioInput,
+ _config_to_json,
+ _resolve_audio,
+ check_chunks,
+ check_config,
+ stream_filename,
+)
+from ._multipart import AsyncAudioChunks
_T = TypeVar("_T")
@@ -142,6 +150,76 @@ async def transcribe(
timeout=self._client.settings.sync_http_timeout,
)
+ async def transcribe_live(
+ self,
+ data: AsyncAudioChunks,
+ config: Optional[types.SyncTranscriptionConfig] = None,
+ ) -> types.SyncTranscriptResponse:
+ """
+ Transcribes audio uploaded as it is produced.
+
+ The asyncio counterpart of `SyncTranscriber.transcribe_live`. Where
+ `transcribe()` needs the whole clip before it can send anything, this
+ starts the request immediately and uploads chunks as they arrive, so
+ authorization, the upload and every speech segment but the last resolve
+ while the caller is still recording.
+
+ That only pays off when the audio is genuinely still being produced —
+ a live microphone, an in-progress call. Streaming a file already on
+ disk is slower than `transcribe()`. The saving also needs enough audio
+ to have segments to release early: below roughly a minute, only the
+ elided upload counts.
+
+ The caller must keep producing: an upload that goes silent for long
+ enough is aborted server-side. Stop by ending the iterator, not by
+ pausing it.
+
+ Args:
+ data: An async iterable of audio chunks, a file object, or a plain
+ iterable. A file object is read in a worker thread, so a
+ blocking `read()` is fine; a plain iterable is consumed inline
+ and so must not block. Raw PCM also requires `sample_rate` and
+ `channels` on the config.
+ config: Options for this call. If `None`, the transcriber's default
+ configuration is used.
+
+ Raises:
+ TypeError: if `config` is not a `SyncTranscriptionConfig`; if
+ `data` is a path or a bytes buffer rather than a stream; or if
+ a chunk is not bytes (a file opened in text mode, say).
+ SyncTranscriptError: if the request fails. Auth, rate-limit and
+ capacity failures can surface part-way through the upload.
+ Exception: anything the producer raises mid-upload propagates
+ unchanged. The connection is dropped and no transcript is
+ returned.
+
+ Example:
+ ```python
+ async def mic_chunks():
+ while recording:
+ yield await stream.read(4096)
+
+ async with aai.AsyncSyncTranscriber() as transcriber:
+ result = await transcriber.transcribe_live(mic_chunks())
+ ```
+ """
+ check_config(type(self).__name__, config)
+
+ config = config or self.config
+ check_chunks(data, allow_async=True)
+ filename, content_type = stream_filename(data, config)
+
+ return await async_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,
+ )
+
async def warm(self) -> bool:
"""
Opens the connection to the sync API ahead of time.
diff --git a/assemblyai/sync/v1/client.py b/assemblyai/sync/v1/client.py
index a3293695..8c5fa74c 100644
--- a/assemblyai/sync/v1/client.py
+++ b/assemblyai/sync/v1/client.py
@@ -9,7 +9,7 @@
from ... import client as _client
from ... import types
from . import api
-from ._base import AudioInput, _SyncTranscriberImpl, check_config
+from ._base import AudioChunks, AudioInput, _SyncTranscriberImpl, check_config
class SyncTranscriber:
@@ -130,6 +130,63 @@ def transcribe_async(
config=config,
)
+ def transcribe_live(
+ self,
+ data: AudioChunks,
+ config: Optional[types.SyncTranscriptionConfig] = None,
+ ) -> types.SyncTranscriptResponse:
+ """
+ Transcribes audio uploaded as it is produced.
+
+ Where `transcribe()` needs the whole clip before it can send anything,
+ this starts the request immediately and uploads chunks as they arrive,
+ 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.
+
+ That only pays off when the audio is genuinely still being produced —
+ a live microphone, an in-progress call. Streaming a file that is
+ already on disk is slower than `transcribe()`, which uploads it in one
+ piece; the saving comes from overlapping the recording, not from the
+ chunking itself. The saving also needs enough audio to have segments to
+ release early: below roughly a minute, only the elided upload counts.
+
+ The caller must keep producing: an upload that goes silent for long
+ enough is aborted server-side. Stop by ending the iterator, not by
+ pausing it.
+
+ Args:
+ data: An iterable of audio chunks, or a binary file object read as
+ it fills. Raw PCM also requires `sample_rate` and `channels` on
+ the config. Audio you already hold whole belongs in
+ `transcribe()`.
+ config: Options for this call. If `None`, the transcriber's default
+ configuration is used.
+
+ Raises:
+ TypeError: if `config` is not a `SyncTranscriptionConfig`; if
+ `data` is a path, a bytes buffer or an async iterable rather
+ than a stream; or if a chunk is not bytes (a file opened in
+ text mode, say).
+ SyncTranscriptError: if the request fails. Auth, rate-limit and
+ capacity failures can surface part-way through the upload.
+ Exception: anything the producer raises mid-upload propagates
+ unchanged. The connection is dropped and no transcript is
+ returned.
+
+ Example:
+ ```python
+ def mic_chunks():
+ while recording:
+ yield stream.read(4096)
+
+ result = aai.SyncTranscriber().transcribe_live(mic_chunks())
+ ```
+ """
+ check_config(type(self).__name__, config)
+
+ return self._impl.transcribe_live(data=data, config=config)
+
def warm(self) -> bool:
"""
Opens the connection to the sync API ahead of time.
diff --git a/assemblyai/types.py b/assemblyai/types.py
index b0af7309..36ce1969 100644
--- a/assemblyai/types.py
+++ b/assemblyai/types.py
@@ -119,6 +119,9 @@ class Settings(BaseSettings):
sync_http_timeout: float = 60.0
"The HTTP timeout for synchronous transcription requests. Kept above the server's 30s deadline so the client doesn't race it."
+ sync_live_http_timeout: float = 180.0
+ "The HTTP timeout for streamed synchronous transcription requests (`transcribe_live`). Like every httpx timeout it bounds each socket operation, not the request as a whole: connecting, each write of a chunk, and each read while waiting for the response after the last byte. Time spent waiting on the caller's producer is not counted, so it need not cover the recording. The sync API caps audio at 120 seconds and independently aborts an upload that goes silent."
+
polling_interval: float = Field(default=3.0, gt=0.0)
"The default polling interval for long-running requests (e.g. polling the `Transcript`'s status)"
diff --git a/tests/unit/test_sync_live.py b/tests/unit/test_sync_live.py
new file mode 100644
index 00000000..ab33e31f
--- /dev/null
+++ b/tests/unit/test_sync_live.py
@@ -0,0 +1,475 @@
+"""Tests for the streamed-upload sync path (`transcribe_live`).
+
+Body-shape assertions go through `httpx.MockTransport`, which consumes the
+request stream the way a real transport does. `pytest_httpx` records the
+request without reading it on older httpx, so a body read after the call sees a
+spent generator or a closed file.
+"""
+
+import asyncio
+import email
+import io
+import threading
+from typing import List, Optional
+
+import httpx
+import pytest
+from pytest_httpx import HTTPXMock
+
+import assemblyai as aai
+from assemblyai.sync.v1 import api, async_api
+from assemblyai.sync.v1._multipart import _STREAM_READ_SIZE
+
+aai.settings.api_key = "test"
+
+STREAM_URL = f"{aai.settings.sync_base_url}/v1/transcribe/stream"
+
+_OK_RESPONSE = {
+ "text": "hello world",
+ "words": [{"text": "hello", "start": 0, "end": 200, "confidence": 0.9}],
+ "confidence": 0.92,
+ "audio_duration_ms": 400,
+ "session_id": "eb92c4ff-4bbb-429f-9b99-7279d7fe738f",
+ "request_time_ms": 243.7,
+}
+
+
+def _mock_ok(httpx_mock: HTTPXMock) -> None:
+ httpx_mock.add_response(
+ url=STREAM_URL,
+ method="POST",
+ status_code=httpx.codes.OK,
+ json=_OK_RESPONSE,
+ )
+
+
+def _chunks(*pieces: bytes):
+ for piece in pieces:
+ yield piece
+
+
+def _parts(request: httpx.Request):
+ """Parses a recorded request body into its multipart parts, in order."""
+
+ raw = b"Content-Type: " + request.headers["content-type"].encode() + b"\r\n\r\n"
+ message = email.message_from_bytes(raw + request.content)
+
+ return [
+ (part.get_param("name", header="content-disposition"), part)
+ for part in message.get_payload()
+ ]
+
+
+def _send(chunks, config: Optional[dict] = None, **kwargs) -> List[httpx.Request]:
+ """Runs `api.transcribe_live` against a transport that records the body."""
+
+ seen: List[httpx.Request] = []
+
+ def handler(request: httpx.Request) -> httpx.Response:
+ seen.append(request)
+ return httpx.Response(httpx.codes.OK, json=_OK_RESPONSE)
+
+ with httpx.Client(transport=httpx.MockTransport(handler)) as client:
+ api.transcribe_live(
+ client,
+ base_url=aai.settings.sync_base_url,
+ chunks=chunks,
+ filename=kwargs.get("filename", "audio.wav"),
+ audio_content_type=kwargs.get("audio_content_type", "audio/wav"),
+ model="u3-sync-pro",
+ config=config,
+ timeout=30.0,
+ )
+
+ return seen
+
+
+async def _asend(chunks, config: Optional[dict] = None) -> List[httpx.Request]:
+ """The asyncio counterpart of `_send`."""
+
+ seen: List[httpx.Request] = []
+
+ def handler(request: httpx.Request) -> httpx.Response:
+ seen.append(request)
+ return httpx.Response(httpx.codes.OK, json=_OK_RESPONSE)
+
+ async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
+ await async_api.transcribe_live(
+ client,
+ base_url=aai.settings.sync_base_url,
+ chunks=chunks,
+ filename="audio.wav",
+ audio_content_type="audio/wav",
+ model="u3-sync-pro",
+ config=config,
+ timeout=30.0,
+ )
+
+ return seen
+
+
+# --- body shape -------------------------------------------------------------
+
+
+def test_body_sends_config_before_audio():
+ # Given a config alongside the audio
+ # When the body is built
+ request = _send(_chunks(b"RIFF", b"fake"), config={"prompt": "a prompt"})[0]
+
+ # Then config precedes audio: the server decodes as bytes land and rejects
+ # audio that arrives first
+ assert [name for name, _ in _parts(request)] == ["config", "audio"]
+
+
+def test_body_carries_config_json_and_joined_audio():
+ # Given audio produced in several pieces
+ # When the body is built
+ request = _send(_chunks(b"RIFF", b"fake", b"wav"), config={"prompt": "a prompt"})[0]
+
+ # Then the pieces arrive as one audio part and the config part is JSON
+ parts = dict(_parts(request))
+ assert parts["audio"].get_payload(decode=True) == b"RIFFfakewav"
+ assert parts["config"].get_payload(decode=True) == b'{"prompt": "a prompt"}'
+
+
+def test_body_sends_an_empty_config_part_when_there_is_no_config():
+ # Given no config
+ # When the body is built
+ request = _send(_chunks(b"RIFF"))[0]
+
+ # Then a config part still precedes the audio, as an empty object: the
+ # streaming endpoint rejects audio that no config part came before
+ parts = _parts(request)
+ assert [name for name, _ in parts] == ["config", "audio"]
+ assert dict(parts)["config"].get_payload(decode=True) == b"{}"
+
+
+def test_body_skips_empty_chunks():
+ # Given a producer that yields an empty piece between real ones
+ # When the body is built
+ request = _send(_chunks(b"RIFF", b"", b"wav"))[0]
+
+ # Then it is dropped rather than ending the chunked body early
+ assert dict(_parts(request))["audio"].get_payload(decode=True) == b"RIFFwav"
+
+
+def test_body_is_uploaded_chunked():
+ # Given audio that does not exist yet when the request starts
+ # When the body is built
+ request = _send(_chunks(b"RIFF", b"more"))[0]
+
+ # Then it is framed without a length, which is what lets the request start
+ assert request.headers.get("transfer-encoding") == "chunked"
+ assert "content-length" not in request.headers
+
+
+class _SpyFile(io.BytesIO):
+ """A file object that records the size of every read asked of it."""
+
+ def __init__(self, payload: bytes) -> None:
+ super().__init__(payload)
+ self.reads: List[int] = []
+ self.threads: List[threading.Thread] = []
+
+ def read(self, size: int = -1) -> bytes:
+ self.reads.append(size)
+ self.threads.append(threading.current_thread())
+ return super().read(size)
+
+
+def test_body_reads_a_file_object_in_bounded_pieces():
+ # Given a file object holding more than two read sizes of audio
+ payload = b"RIFF" + bytes(2 * _STREAM_READ_SIZE + 100)
+ source = _SpyFile(payload)
+
+ # When it is streamed
+ request = _send(source, filename="call.wav")[0]
+
+ # Then it is read in fixed pieces until exhausted, never slurped whole
+ assert source.reads == [_STREAM_READ_SIZE] * 4
+ part = dict(_parts(request))["audio"]
+ assert part.get_payload(decode=True) == payload
+ assert part.get_filename() == "call.wav"
+
+
+def test_async_body_reads_a_file_object_off_the_event_loop():
+ # Given a file object handed to the async transport
+ source = _SpyFile(b"RIFF" + bytes(_STREAM_READ_SIZE + 1))
+
+ # When it is streamed
+ request = asyncio.run(_asend(source))[0]
+
+ # Then every read ran in a worker thread, so a blocking read cannot stall
+ # the loop, and the body is intact
+ assert source.reads == [_STREAM_READ_SIZE] * 3
+ assert all(t is not threading.main_thread() for t in source.threads)
+ assert dict(_parts(request))["audio"].get_payload(decode=True) == source.getvalue()
+
+
+def test_body_escapes_the_filename():
+ # Given a file whose name would otherwise break the part header
+ # When the body is built
+ request = _send(_chunks(b"RIFF"), filename='we"ird\r\nname.wav')[0]
+
+ # Then the name is escaped the way httpx escapes its own multipart parts
+ # and the audio part still parses
+ parts = dict(_parts(request))
+ assert parts["audio"].get_filename() == "we%22ird%0D%0Aname.wav"
+ assert parts["audio"].get_payload(decode=True) == b"RIFF"
+
+
+def test_body_accepts_bytearray_and_memoryview_chunks():
+ # Given a producer handing out buffer objects rather than bytes
+ # When the body is built
+ request = _send(iter([bytearray(b"RIFF"), memoryview(b"fake")]))[0]
+
+ # Then they are sent as bytes
+ assert dict(_parts(request))["audio"].get_payload(decode=True) == b"RIFFfake"
+
+
+def test_body_rejects_text_chunks():
+ # Given a producer yielding str
+ # When the body is built, then the mistake is named rather than failing
+ # inside the transport
+ with pytest.raises(TypeError, match="binary mode"):
+ _send(iter(["RIFF", "fake"]))
+
+
+def test_body_rejects_a_text_mode_file():
+ # Given a file object opened without "b"
+ # When the body is built, then the mistake is named
+ with pytest.raises(TypeError, match="binary mode"):
+ _send(io.StringIO("RIFFfake"))
+
+
+def test_body_hits_the_stream_endpoint():
+ # Given a streamed upload
+ # When it is sent
+ request = _send(_chunks(b"RIFF"))[0]
+
+ # Then the buffered route is left alone
+ assert str(request.url) == STREAM_URL
+ assert request.headers["X-AAI-Model"] == "u3-sync-pro"
+
+
+def test_async_body_matches_its_sync_twin():
+ # Given the same audio through the async transport
+ async def chunks():
+ for piece in (b"RIFF", b"fake", b"wav"):
+ yield piece
+
+ # When the body is built
+ request = asyncio.run(_asend(chunks(), config={"prompt": "a prompt"}))[0]
+
+ # Then it is framed identically
+ parts = dict(_parts(request))
+ assert [name for name, _ in _parts(request)] == ["config", "audio"]
+ assert parts["audio"].get_payload(decode=True) == b"RIFFfakewav"
+ assert request.headers.get("transfer-encoding") == "chunked"
+
+
+def test_async_body_accepts_a_sync_iterable():
+ # Given a plain iterable handed to the async transport
+ # When the body is built
+ request = asyncio.run(_asend(_chunks(b"RIFF", b"fake")))[0]
+
+ # Then one producer can feed both transports
+ assert dict(_parts(request))["audio"].get_payload(decode=True) == b"RIFFfake"
+
+
+# --- transcriber behaviour --------------------------------------------------
+
+
+def test_transcribe_live_parses_response(httpx_mock: HTTPXMock):
+ # Given a mocked streaming endpoint
+ _mock_ok(httpx_mock)
+
+ # When streaming audio chunks
+ result = aai.SyncTranscriber().transcribe_live(_chunks(b"RIFF", b"fake"))
+
+ # Then the response is parsed like the buffered path's
+ assert isinstance(result, aai.SyncTranscriptResponse)
+ assert result.text == "hello world"
+ assert result.session_id == _OK_RESPONSE["session_id"]
+
+
+def test_transcribe_live_sends_model_header(httpx_mock: HTTPXMock):
+ # Given a mocked streaming endpoint
+ _mock_ok(httpx_mock)
+
+ # When streaming with a model set
+ aai.SyncTranscriber().transcribe_live(
+ _chunks(b"RIFF"),
+ config=aai.SyncTranscriptionConfig(model="u3-sync-pro"),
+ )
+
+ # Then the routing header goes out, as on the buffered path
+ request = httpx_mock.get_requests()[0]
+ assert str(request.url) == STREAM_URL
+ assert request.headers["X-AAI-Model"] == "u3-sync-pro"
+
+
+def test_transcribe_live_uses_the_stream_timeout(httpx_mock: HTTPXMock):
+ # Given a mocked streaming endpoint
+ _mock_ok(httpx_mock)
+
+ # When streaming audio
+ aai.SyncTranscriber().transcribe_live(_chunks(b"RIFF"))
+
+ # Then the request gets the longer budget, which must cover the recording
+ # as well as the transcription
+ timeout = httpx_mock.get_requests()[0].extensions["timeout"]
+ assert timeout["read"] == aai.settings.sync_live_http_timeout
+
+
+def test_transcribe_live_names_a_file_object(monkeypatch):
+ # Given a file object carrying a name
+ captured = {}
+
+ def fake(client, **kwargs):
+ captured.update(kwargs)
+ return aai.SyncTranscriptResponse.parse_obj(_OK_RESPONSE)
+
+ monkeypatch.setattr(api, "transcribe_live", fake)
+
+ stream = io.BytesIO(b"RIFF")
+ stream.name = "/tmp/call.wav"
+
+ # When it is streamed
+ aai.SyncTranscriber().transcribe_live(stream)
+
+ # Then the audio part takes its name and type from the file
+ assert captured["filename"] == "call.wav"
+ assert captured["audio_content_type"] == "audio/wav"
+
+
+def test_transcribe_live_marks_pcm_from_config(monkeypatch):
+ # Given a config carrying the fields only raw PCM needs
+ captured = {}
+
+ def fake(client, **kwargs):
+ captured.update(kwargs)
+ return aai.SyncTranscriptResponse.parse_obj(_OK_RESPONSE)
+
+ monkeypatch.setattr(api, "transcribe_live", fake)
+
+ # When streaming
+ aai.SyncTranscriber().transcribe_live(
+ _chunks(b"\x00\x01"),
+ config=aai.SyncTranscriptionConfig(sample_rate=16000, channels=1),
+ )
+
+ # Then the audio part selects the PCM decoder
+ assert captured["audio_content_type"] == "audio/pcm"
+ assert captured["filename"] == "audio.pcm"
+
+
+def test_transcribe_live_requires_both_pcm_fields():
+ # Given a config with only half of what raw PCM needs
+ config = aai.SyncTranscriptionConfig(sample_rate=16000)
+
+ # When streaming, then it is rejected before any request is made
+ with pytest.raises(ValueError, match="sample_rate and channels"):
+ aai.SyncTranscriber().transcribe_live(_chunks(b"\x00"), config=config)
+
+
+def test_transcribe_live_rejects_bytes():
+ # Given audio the caller already holds whole
+ # When streaming it, then it is named as a mistake and sent to transcribe()
+ with pytest.raises(TypeError, match="transcribe\\(\\)"):
+ aai.SyncTranscriber().transcribe_live(b"RIFFfake-wav-bytes")
+
+
+def test_transcribe_live_rejects_a_path():
+ # Given a path rather than an open file
+ # When streaming it, then it is rejected rather than silently opened
+ with pytest.raises(TypeError, match="not a path"):
+ aai.SyncTranscriber().transcribe_live("./call.wav")
+
+
+def test_transcribe_live_rejects_an_async_iterable():
+ # Given an async producer handed to the synchronous transcriber
+ async def chunks():
+ yield b"RIFF"
+
+ # When streaming it, then it is turned away before any request is made
+ # and pointed at the transcriber that can drive it
+ with pytest.raises(TypeError, match="AsyncSyncTranscriber"):
+ aai.SyncTranscriber().transcribe_live(chunks())
+
+
+def test_transcribe_live_rejects_job_api_config():
+ # Given the job API's config type
+ # When streaming with it, then the mismatch is named
+ with pytest.raises(TypeError, match="SyncTranscriptionConfig"):
+ aai.SyncTranscriber().transcribe_live(
+ _chunks(b"RIFF"),
+ config=aai.TranscriptionConfig(),
+ )
+
+
+def test_transcribe_live_raises_on_error_response(httpx_mock: HTTPXMock):
+ # Given a rejection the server can send while the upload is still in flight
+ httpx_mock.add_response(
+ url=STREAM_URL,
+ method="POST",
+ status_code=httpx.codes.TOO_MANY_REQUESTS,
+ json={"status": 429, "title": "Rate Limited", "detail": "slow down"},
+ headers={"Retry-After": "3"},
+ )
+
+ # When streaming audio
+ with pytest.raises(aai.SyncTranscriptError) as exc:
+ aai.SyncTranscriber().transcribe_live(_chunks(b"RIFF"))
+
+ # Then it is surfaced with the same envelope parsing as the buffered path
+ assert exc.value.status_code == 429
+ assert exc.value.error_code == "rate_limited"
+ assert exc.value.retry_after == 3
+
+
+@pytest.mark.asyncio
+async def test_async_transcribe_live_parses_response(httpx_mock: HTTPXMock):
+ # Given a mocked streaming endpoint
+ _mock_ok(httpx_mock)
+
+ async def chunks():
+ yield b"RIFF"
+
+ # When streaming from an async producer
+ async with aai.AsyncSyncTranscriber() as transcriber:
+ result = await transcriber.transcribe_live(chunks())
+
+ # Then the response is parsed like its sync twin's
+ assert result.text == "hello world"
+ assert str(httpx_mock.get_requests()[0].url) == STREAM_URL
+
+
+@pytest.mark.asyncio
+async def test_async_transcribe_live_rejects_bytes():
+ # Given audio the caller already holds whole
+ # When streaming it, then the async path names the same mistake
+ async with aai.AsyncSyncTranscriber() as transcriber:
+ with pytest.raises(TypeError, match="transcribe\\(\\)"):
+ await transcriber.transcribe_live(b"RIFF")
+
+
+@pytest.mark.asyncio
+async def test_async_transcribe_live_raises_on_error_response(httpx_mock: HTTPXMock):
+ # Given a rejection
+ httpx_mock.add_response(
+ url=STREAM_URL,
+ method="POST",
+ status_code=httpx.codes.SERVICE_UNAVAILABLE,
+ json={"status": 503, "title": "Capacity Exceeded", "detail": "no capacity"},
+ )
+
+ async def chunks():
+ yield b"RIFF"
+
+ # When streaming audio, then the error surfaces
+ async with aai.AsyncSyncTranscriber() as transcriber:
+ with pytest.raises(aai.SyncTranscriptError) as exc:
+ await transcriber.transcribe_live(chunks())
+
+ assert exc.value.error_code == "capacity_exceeded"