feat(sync): add AsyncSyncTranscriber for asyncio callers

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+
 

README.md

@@ -499,6 +499,31 @@ with aai.SyncTranscriber() as transcriber:
 
 </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.
+
+```python
+import asyncio
+import assemblyai as aai
+
+aai.settings.api_key = "<YOUR_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))`.
+
+</details>
+
 <details>
   <summary>Handle errors</summary>
 

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",

assemblyai/__version__.py

@@ -1 +1 @@
-__version__ = "0.66.00"
+__version__ = "0.67.00"

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",

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())

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()