diff --git a/CLAUDE.md b/CLAUDE.md
index e017e0f..f2d835c 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -45,6 +45,7 @@ aai.settings.api_key = "your-key"
- `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.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.StreamingClient` — Real-time streaming with event-based API (threaded)
@@ -209,8 +210,42 @@ result = aai.SyncTranscriber().transcribe(raw_pcm_bytes, config=config)
```
**Concurrency**: `transcribe_async()` returns a `concurrent.futures.Future` (thread-based,
-not asyncio) for fanning out a handful of files. (An asyncio-native `AsyncSyncTranscriber`
-is a planned follow-up for high-concurrency servers and event-loop codebases.)
+not asyncio) for fanning out a handful of files. In asyncio code use
+`aai.AsyncSyncTranscriber` instead (see "Asyncio sync transcription" below).
+
+## Asyncio sync transcription (`AsyncSyncTranscriber`)
+
+`AsyncSyncTranscriber` is `SyncTranscriber` for the event loop: same input types
+(path/bytes/file object — no URLs), same `SyncTranscriptionConfig`, same
+`SyncTranscriptResponse` and `SyncTranscriptError`, with `transcribe()` and `warm()`
+as coroutines. Path and file-object reads run off the loop. Use it in FastAPI,
+aiohttp, and voice agents, where the threaded `transcribe()` would block the loop
+and `transcribe_async()`'s `concurrent.futures.Future` is not awaitable.
+
+```python
+import asyncio
+import assemblyai as aai
+
+aai.settings.api_key = os.environ["ASSEMBLYAI_API_KEY"]
+
+async def main():
+ async with aai.AsyncSyncTranscriber() as transcriber:
+ asyncio.create_task(transcriber.warm()) # optional: fire as recording starts
+ audio = await record_until_done()
+ result = await transcriber.transcribe(audio)
+ print(result.text)
+
+asyncio.run(main())
+```
+
+**Lifecycle**: like `AsyncTranscriber`, it owns an HTTP connection pool — use
+`async with`, or call `await transcriber.aclose()`. To share one pool, pass an
+`aai.AsyncClient`, which stays yours to close. There is no process-wide default
+async client (an `httpx.AsyncClient` pool is bound to the event loop that first
+used it).
+
+**Concurrency**: plain asyncio — `await asyncio.gather(transcriber.transcribe(a),
+transcriber.transcribe(b))`. There is no `transcribe_async()` and no thread pool.
**Errors**: failures raise `aai.SyncTranscriptError` with `.status_code`, a
machine-readable `.error_code` — the snake_cased problem-details `title` from the
@@ -380,7 +415,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. In asyncio code use `aai.AsyncTranscriber` (see "Asyncio transcription" above)
+- **`transcribe_async()` returns a `concurrent.futures.Future`**, not an asyncio coroutine. In asyncio code use `aai.AsyncTranscriber` (see "Asyncio transcription" above) — or `aai.AsyncSyncTranscriber` for the sync API
- **Timestamps are in milliseconds** throughout the SDK
- **Minimum Python**: 3.8+
diff --git a/README.md b/README.md
index 3f1fdc9..c06a7a5 100644
--- a/README.md
+++ b/README.md
@@ -499,6 +499,31 @@ with aai.SyncTranscriber() as transcriber:
+
+ 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.
+
+```python
+import asyncio
+import assemblyai as aai
+
+aai.settings.api_key = ""
+
+async def main():
+ async with aai.AsyncSyncTranscriber() as transcriber:
+ asyncio.create_task(transcriber.warm()) # optional: fire as recording starts
+ audio = await record_until_done()
+ result = await transcriber.transcribe(audio)
+ print(result.text)
+
+asyncio.run(main())
+```
+
+The transcriber owns an HTTP connection pool: use `async with`, or call `await transcriber.aclose()`. To share one pool between transcribers, pass an `aai.AsyncClient`, which stays yours to close. Concurrency is plain asyncio — `await asyncio.gather(transcriber.transcribe(a), transcriber.transcribe(b))`.
+
+
+
Handle errors
diff --git a/assemblyai/__init__.py b/assemblyai/__init__.py
index 7acffa6..d5fe3e9 100644
--- a/assemblyai/__init__.py
+++ b/assemblyai/__init__.py
@@ -3,7 +3,7 @@
from .async_client import AsyncClient
from .client import Client
from .prerecorded.v2 import AsyncTranscriber, AsyncTranscript
-from .sync import SyncTranscriber
+from .sync.v1 import AsyncSyncTranscriber, SyncTranscriber
from .transcriber import Transcriber, Transcript, TranscriptGroup
from .types import (
AssemblyAIError,
@@ -78,6 +78,7 @@
# types
"AssemblyAIError",
"AsyncClient",
+ "AsyncSyncTranscriber",
"AsyncTranscriber",
"AsyncTranscript",
"AutohighlightResponse",
diff --git a/assemblyai/__version__.py b/assemblyai/__version__.py
index b422ee5..70c5d8e 100644
--- a/assemblyai/__version__.py
+++ b/assemblyai/__version__.py
@@ -1 +1 @@
-__version__ = "0.66.00"
+__version__ = "0.67.00"
diff --git a/assemblyai/sync/v1/__init__.py b/assemblyai/sync/v1/__init__.py
index c6417ff..05936f4 100644
--- a/assemblyai/sync/v1/__init__.py
+++ b/assemblyai/sync/v1/__init__.py
@@ -6,9 +6,11 @@
SyncWord,
)
from ._base import AudioInput
+from .async_client import AsyncSyncTranscriber
from .client import SyncTranscriber
__all__ = [
+ "AsyncSyncTranscriber",
"AudioInput",
"SyncSpeechModel",
"SyncTranscriber",
diff --git a/assemblyai/sync/v1/async_api.py b/assemblyai/sync/v1/async_api.py
new file mode 100644
index 0000000..06012b4
--- /dev/null
+++ b/assemblyai/sync/v1/async_api.py
@@ -0,0 +1,68 @@
+"""The asyncio counterpart of `api.py`.
+
+Calls the same endpoint as its sync twin and raises the same
+`SyncTranscriptError` through `api._error_from_response`.
+"""
+
+import json
+from typing import Dict, Optional, Tuple
+
+import httpx
+
+from ... import types
+from .api import ENDPOINT_TRANSCRIBE, MODEL_HEADER, _error_from_response
+
+__all__ = ["transcribe"]
+
+
+async def transcribe(
+ client: httpx.AsyncClient,
+ *,
+ base_url: str,
+ audio: bytes,
+ filename: str,
+ audio_content_type: str,
+ model: str,
+ config: Optional[dict],
+ timeout: float,
+) -> types.SyncTranscriptResponse:
+ """
+ Posts a single synchronous transcription request.
+
+ Args:
+ client: the HTTP client (carries the `Authorization` header).
+ base_url: the sync API base URL, e.g. `https://sync.assemblyai.com`.
+ audio: raw audio bytes (WAV container or S16LE PCM).
+ 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, or None to omit it.
+ timeout: per-request timeout in seconds.
+
+ Returns: the parsed transcript response.
+
+ Raises: `SyncTranscriptError` on any non-200 response.
+ """
+ files: Dict[str, Tuple[Optional[str], bytes, str]] = {
+ "audio": (filename, audio, audio_content_type)
+ }
+ if config:
+ # httpx <0.23 rejects a `str` multipart part; encode to bytes so the
+ # config part works across the full supported httpx range (>=0.19).
+ files["config"] = (
+ None,
+ json.dumps(config).encode("utf-8"),
+ "application/json",
+ )
+
+ response = await client.post(
+ base_url.rstrip("/") + ENDPOINT_TRANSCRIBE,
+ files=files,
+ headers={MODEL_HEADER: model},
+ 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
new file mode 100644
index 0000000..6925065
--- /dev/null
+++ b/assemblyai/sync/v1/async_client.py
@@ -0,0 +1,190 @@
+"""The asyncio counterpart of `client.py`."""
+
+from __future__ import annotations
+
+import asyncio
+from types import TracebackType
+from typing import Any, Callable, Optional, Type, TypeVar
+
+import httpx
+from typing_extensions import Self
+
+from ... import async_client as _async_client
+from ... import types
+from . import api, async_api
+from ._base import AudioInput, _config_to_json, _resolve_audio
+
+_T = TypeVar("_T")
+
+
+async def _run_in_thread(func: Callable[..., _T], *args: Any) -> _T:
+ """Runs a blocking call on the default executor."""
+
+ loop = asyncio.get_event_loop()
+
+ return await loop.run_in_executor(None, func, *args)
+
+
+class AsyncSyncTranscriber:
+ """
+ The asyncio counterpart of `SyncTranscriber`: audio in, transcript out,
+ one request — without blocking the event loop.
+
+ Like `SyncTranscriber`, it posts the audio to the sync API and returns
+ the finished `SyncTranscriptResponse` directly; there is no job id or
+ status to poll. Accepts a local file path, raw bytes, or a binary file
+ object — but not a URL. Use it in asyncio code (FastAPI, aiohttp, voice
+ agents), where `SyncTranscriber.transcribe()` would block the loop and
+ `transcribe_async()`'s `concurrent.futures.Future` is not awaitable.
+
+ The transcriber owns an HTTP connection pool. Close it with `aclose()`,
+ or use the transcriber as an async context manager.
+
+ Example:
+ ```python
+ import asyncio
+ import assemblyai as aai
+
+ aai.settings.api_key = "your-key"
+
+ async def main():
+ async with aai.AsyncSyncTranscriber() as transcriber:
+ result = await transcriber.transcribe("./call.wav")
+ print(result.text)
+
+ asyncio.run(main())
+ ```
+
+ Transcribing several clips concurrently is plain asyncio:
+ ```python
+ async with aai.AsyncSyncTranscriber() as transcriber:
+ results = await asyncio.gather(
+ transcriber.transcribe("./one.wav"),
+ transcriber.transcribe("./two.wav"),
+ )
+ ```
+ """
+
+ def __init__(
+ self,
+ *,
+ client: Optional[_async_client.AsyncClient] = None,
+ config: Optional[types.SyncTranscriptionConfig] = None,
+ ) -> None:
+ """
+ Creates an `AsyncSyncTranscriber`.
+
+ Args:
+ client: The `AsyncClient` to use. If `None`, the transcriber
+ creates one from the global `settings` and closes it on
+ `aclose()`. Pass a client to share one pool between
+ transcribers.
+ config: Default transcription options. Per-call `config`
+ overrides it.
+ """
+ from ... import settings as default_settings
+
+ self._owns_client = client is None
+ self._client = client or _async_client.AsyncClient(settings=default_settings)
+ self.config = config or types.SyncTranscriptionConfig()
+
+ @property
+ def client(self) -> _async_client.AsyncClient:
+ """The `AsyncClient` this transcriber sends requests with."""
+
+ return self._client
+
+ async def transcribe(
+ self,
+ data: AudioInput,
+ config: Optional[types.SyncTranscriptionConfig] = None,
+ ) -> types.SyncTranscriptResponse:
+ """
+ Transcribes audio and returns the finished transcript.
+
+ Reads path and file-object input off the event loop.
+
+ Args:
+ data: A local file path, raw audio bytes, or a binary file object.
+ 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: `SyncTranscriptError` if the request fails.
+ """
+ config = config or self.config
+ audio, filename, content_type = await _run_in_thread(
+ _resolve_audio, data, config
+ )
+
+ return await async_api.transcribe(
+ self._client.http_client,
+ base_url=self._client.settings.sync_base_url,
+ audio=audio,
+ filename=filename,
+ audio_content_type=content_type,
+ model=config.model,
+ config=_config_to_json(config),
+ timeout=self._client.settings.sync_http_timeout,
+ )
+
+ async def warm(self) -> bool:
+ """
+ Opens the connection to the sync API ahead of time.
+
+ The sync API is a single request/response, so a `transcribe()` that
+ opens its connection on demand pays the full DNS + TCP + TLS handshake
+ on the critical path — one network round trip that, for a distant
+ client, can rival the transcription itself. Awaiting `warm()` as soon
+ as you know audio is coming — typically while the clip is still being
+ recorded, e.g. via `asyncio.create_task(transcriber.warm())` — spends
+ that setup concurrently: the next `transcribe()` reuses the
+ already-open connection.
+
+ The warmed connection is reused while it stays in the HTTP pool —
+ `settings.keepalive_expiry` seconds (httpx's 5s default unless raised).
+ Call `warm()` shortly before `transcribe()`, or raise
+ `keepalive_expiry` (e.g. to 120, the sync audio cap) so a single call
+ covers a whole in-progress recording. `warm()` is idempotent and cheap,
+ so calling it again to refresh the connection is fine.
+
+ Routing the same `config.model` as the eventual transcription ensures
+ the warmed connection lands on the right backend.
+
+ Returns:
+ True once the connection is open (any HTTP response — even a
+ non-200 — means the socket is established); False if the
+ connection could not be opened (transport error).
+ """
+ settings = self._client.settings
+ url = settings.sync_base_url.rstrip("/") + api.ENDPOINT_WARM
+ try:
+ await self._client.http_client.get(
+ url,
+ headers={api.MODEL_HEADER: self.config.model},
+ timeout=min(settings.sync_http_timeout, 10.0),
+ )
+ except httpx.HTTPError:
+ return False
+ return True
+
+ async def aclose(self) -> None:
+ """
+ Closes the HTTP connection pool.
+
+ Leaves a client that was passed in alone. Its creator closes it.
+ """
+
+ if self._owns_client:
+ await self._client.aclose()
+
+ async def __aenter__(self) -> Self:
+ return self
+
+ async def __aexit__(
+ self,
+ exc_type: Optional[Type[BaseException]],
+ exc_value: Optional[BaseException],
+ traceback: Optional[TracebackType],
+ ) -> None:
+ await self.aclose()
diff --git a/tests/unit/test_sync_async.py b/tests/unit/test_sync_async.py
new file mode 100644
index 0000000..f0fa8e9
--- /dev/null
+++ b/tests/unit/test_sync_async.py
@@ -0,0 +1,285 @@
+import asyncio
+
+import httpx
+import pytest
+from pytest_httpx import HTTPXMock
+
+import assemblyai as aai
+
+pytestmark = pytest.mark.asyncio
+
+aai.settings.api_key = "test"
+
+TRANSCRIBE_URL = f"{aai.settings.sync_base_url}/v1/transcribe"
+WARM_URL = f"{aai.settings.sync_base_url}/v1/warm"
+
+_OK_RESPONSE = {
+ "text": "hello world",
+ "words": [
+ {"text": "hello", "start": 0, "end": 200, "confidence": 0.9},
+ {"text": "world", "start": 220, "end": 400, "confidence": 0.95},
+ ],
+ "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=TRANSCRIBE_URL,
+ method="POST",
+ status_code=httpx.codes.OK,
+ json=_OK_RESPONSE,
+ )
+
+
+async def test_transcribe_bytes_parses_response(httpx_mock: HTTPXMock):
+ # Given a mocked sync endpoint
+ _mock_ok(httpx_mock)
+
+ # When transcribing raw audio bytes
+ async with aai.AsyncSyncTranscriber() as transcriber:
+ result = await transcriber.transcribe(b"RIFFfake-wav-bytes")
+
+ # Then the response is parsed into a SyncTranscriptResponse
+ assert isinstance(result, aai.SyncTranscriptResponse)
+ assert result.text == "hello world"
+ assert result.session_id == _OK_RESPONSE["session_id"]
+ assert result.words[0].start == 0
+ assert result.words[0].end == 200
+ assert result.words[1].text == "world"
+ assert result.request_time_ms == 243.7
+
+
+async def test_transcribe_sends_model_header_and_wav_part(httpx_mock: HTTPXMock):
+ # Given a mocked sync endpoint
+ _mock_ok(httpx_mock)
+
+ # When transcribing bytes with the default config
+ async with aai.AsyncSyncTranscriber() as transcriber:
+ await transcriber.transcribe(b"RIFFfake-wav-bytes")
+
+ # Then the request routes via X-AAI-Model and ships a WAV audio part
+ request = httpx_mock.get_requests()[0]
+ assert request.headers["X-AAI-Model"] == "universal-3-5-pro"
+ body = request.read()
+ assert b'name="audio"' in body
+ assert b"Content-Type: audio/wav" in body
+ # And no config part is sent when the config is empty
+ assert b'name="config"' not in body
+
+
+async def test_transcribe_sends_config_part(httpx_mock: HTTPXMock):
+ # Given a mocked sync endpoint
+ _mock_ok(httpx_mock)
+
+ # When transcribing with a prompt and keyterms_prompt
+ config = aai.SyncTranscriptionConfig(
+ prompt="Transcribe verbatim.",
+ keyterms_prompt=["AssemblyAI"],
+ )
+ async with aai.AsyncSyncTranscriber() as transcriber:
+ await transcriber.transcribe(b"RIFFfake-wav-bytes", config=config)
+
+ # Then a config JSON part carries the options
+ body = httpx_mock.get_requests()[0].read()
+ assert b'name="config"' in body
+ assert b"Transcribe verbatim." in body
+ assert b'"AssemblyAI"' in body
+ # And the routing model is never placed in the body
+ assert b'"model"' not in body
+
+
+async def test_transcribe_uses_default_config_and_per_call_override(
+ httpx_mock: HTTPXMock,
+):
+ # Given a transcriber with a default config
+ _mock_ok(httpx_mock)
+ _mock_ok(httpx_mock)
+ default = aai.SyncTranscriptionConfig(prompt="default prompt")
+
+ async with aai.AsyncSyncTranscriber(config=default) as transcriber:
+ # When transcribing without a per-call config
+ await transcriber.transcribe(b"RIFFfake-wav-bytes")
+ # And with a per-call override
+ override = aai.SyncTranscriptionConfig(prompt="override prompt")
+ await transcriber.transcribe(b"RIFFfake-wav-bytes", config=override)
+
+ # Then the default applies to the first call and the override to the second
+ first, second = (request.read() for request in httpx_mock.get_requests())
+ assert b"default prompt" in first
+ assert b"override prompt" in second
+
+
+async def test_transcribe_pcm_sends_pcm_part_and_rate(httpx_mock: HTTPXMock):
+ # Given a mocked sync endpoint
+ _mock_ok(httpx_mock)
+
+ # When transcribing bytes with sample_rate + channels (raw PCM)
+ config = aai.SyncTranscriptionConfig(sample_rate=16000, channels=1)
+ async with aai.AsyncSyncTranscriber() as transcriber:
+ await transcriber.transcribe(b"\x00\x01" * 100, config=config)
+
+ # Then the audio part is PCM and the config carries rate + channels
+ body = httpx_mock.get_requests()[0].read()
+ assert b"Content-Type: audio/pcm" in body
+ assert b'"sample_rate"' in body
+ assert b'"channels"' in body
+
+
+async def test_transcribe_pcm_without_rate_raises():
+ # Given a config with sample_rate but no channels (partial PCM intent)
+ config = aai.SyncTranscriptionConfig(sample_rate=16000)
+
+ # When transcribing, Then it fails locally before any request
+ async with aai.AsyncSyncTranscriber() as transcriber:
+ with pytest.raises(ValueError, match="sample_rate and channels"):
+ await transcriber.transcribe(b"\x00\x01" * 100, config=config)
+
+
+async def test_transcribe_rejects_url():
+ # Given an http URL as input
+ async with aai.AsyncSyncTranscriber() as transcriber:
+ # When transcribing, Then it is rejected with a pointer to Transcriber
+ with pytest.raises(ValueError, match="does not accept URLs"):
+ await transcriber.transcribe("https://example.com/audio.wav")
+
+
+async def test_transcribe_path_input(httpx_mock: HTTPXMock, tmp_path):
+ # Given a local WAV file
+ _mock_ok(httpx_mock)
+ audio_file = tmp_path / "call.wav"
+ audio_file.write_bytes(b"RIFFfake-wav-bytes")
+
+ # When transcribing the path
+ async with aai.AsyncSyncTranscriber() as transcriber:
+ result = await transcriber.transcribe(str(audio_file))
+
+ # Then it succeeds and ships the file under its own name
+ assert result.text == "hello world"
+ body = httpx_mock.get_requests()[0].read()
+ assert b'filename="call.wav"' in body
+
+
+async def test_transcribe_gather_runs_concurrently(httpx_mock: HTTPXMock):
+ # Given a mocked sync endpoint answering twice
+ _mock_ok(httpx_mock)
+ _mock_ok(httpx_mock)
+
+ # When fanning two clips out with asyncio.gather on one transcriber
+ async with aai.AsyncSyncTranscriber() as transcriber:
+ results = await asyncio.gather(
+ transcriber.transcribe(b"RIFFone"),
+ transcriber.transcribe(b"RIFFtwo"),
+ )
+
+ # Then both finish and parse
+ assert [result.text for result in results] == ["hello world", "hello world"]
+
+
+async def test_problem_details_envelope_maps_to_sync_transcript_error(
+ httpx_mock: HTTPXMock,
+):
+ # Given the server rejects oversized audio with a problem-details body
+ httpx_mock.add_response(
+ url=TRANSCRIBE_URL,
+ method="POST",
+ status_code=413,
+ json={"status": 413, "title": "Audio Too Large", "detail": "too long"},
+ )
+
+ # When transcribing, Then a SyncTranscriptError carries the snake_cased
+ # title as error_code, plus the status and detail
+ async with aai.AsyncSyncTranscriber() as transcriber:
+ with pytest.raises(aai.SyncTranscriptError) as exc_info:
+ await transcriber.transcribe(b"RIFFfake-wav-bytes")
+
+ error = exc_info.value
+ assert error.status_code == 413
+ assert error.error_code == "audio_too_large"
+ assert "too long" in str(error)
+
+
+async def test_rate_limit_surfaces_retry_after(httpx_mock: HTTPXMock):
+ # Given a rate-limit response with a Retry-After header
+ httpx_mock.add_response(
+ url=TRANSCRIBE_URL,
+ method="POST",
+ status_code=429,
+ json={
+ "status": 429,
+ "title": "Too Many Requests",
+ "detail": "Too many requests",
+ },
+ headers={"Retry-After": "5"},
+ )
+
+ # When transcribing, Then retry_after and the snake_cased title are parsed
+ async with aai.AsyncSyncTranscriber() as transcriber:
+ with pytest.raises(aai.SyncTranscriptError) as exc_info:
+ await transcriber.transcribe(b"RIFFfake-wav-bytes")
+
+ error = exc_info.value
+ assert error.status_code == 429
+ assert error.error_code == "too_many_requests"
+ assert error.retry_after == 5
+
+
+async def test_warm_opens_connection_with_model_header(httpx_mock: HTTPXMock):
+ # Given a mocked warm endpoint
+ httpx_mock.add_response(url=WARM_URL, method="GET", status_code=httpx.codes.OK)
+
+ # When warming the transcriber
+ async with aai.AsyncSyncTranscriber() as transcriber:
+ warmed = await transcriber.warm()
+
+ # Then it returns True and routes the probe via X-AAI-Model
+ assert warmed is True
+ request = httpx_mock.get_requests()[0]
+ assert request.url == WARM_URL
+ assert request.method == "GET"
+ assert request.headers["X-AAI-Model"] == "universal-3-5-pro"
+
+
+async def test_warm_returns_true_on_non_200(httpx_mock: HTTPXMock):
+ # Given a warm route that the load balancer answers with a 404
+ httpx_mock.add_response(url=WARM_URL, method="GET", status_code=404)
+
+ # When warming, Then the socket is still established, so warm() is True
+ async with aai.AsyncSyncTranscriber() as transcriber:
+ assert await transcriber.warm() is True
+
+
+async def test_warm_returns_false_on_transport_error(httpx_mock: HTTPXMock):
+ # Given the sync host is unreachable
+ httpx_mock.add_exception(httpx.ConnectError("connection refused"))
+
+ # When warming, Then the failure is swallowed and reported as False
+ async with aai.AsyncSyncTranscriber() as transcriber:
+ assert await transcriber.warm() is False
+
+
+async def test_context_manager_closes_owned_client():
+ # Given a transcriber that created its own client
+ async with aai.AsyncSyncTranscriber() as transcriber:
+ assert isinstance(transcriber, aai.AsyncSyncTranscriber)
+ assert not transcriber.client.http_client.is_closed
+
+ # Then leaving the block closes the owned connection pool
+ assert transcriber.client.http_client.is_closed
+
+
+async def test_aclose_leaves_shared_client_open():
+ # Given a transcriber built on a caller-owned client
+ async with aai.AsyncClient(settings=aai.settings) as client:
+ transcriber = aai.AsyncSyncTranscriber(client=client)
+
+ # When closing the transcriber
+ await transcriber.aclose()
+
+ # Then the shared pool stays open — its creator closes it
+ assert not client.http_client.is_closed
+
+ assert client.http_client.is_closed