diff --git a/CLAUDE.md b/CLAUDE.md
index 508b60e..a5c99eb 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -52,6 +52,11 @@ aai.settings.api_key = "your-key"
- `aai.AsyncDictationTranscriber` — Asyncio counterpart of `DictationTranscriber`; `open_live()` returns an `AsyncDictationLiveSession`. Owns an HTTP pool: use `async with` or `await aclose()`
- `aai.DictationConfig` — Dictation options: `sample_rate`, `channels`, `language_codes`, `keyterms_prompt`, `llm_instruction`, `rewrite_mode` (`aai.DictationRewriteMode.fast` default | `smart`). Rejects unknown fields
- `aai.DictationResponse` — Dictation result: `.text`, `.words` (`DictationWord` with `text` + `confidence`), `.confidence`, `.llm_response`, `.llm_error`, `.audio_duration_ms`, `.session_id`, `.request_time_ms`, `.sync_time_ms`, and `.final_text` (the LLM rewrite, falling back to `.text`)
+- `aai.TextToSpeech` — Text-to-Speech HTTP API: one text per request, audio out. Methods: `synthesize()` (complete audio as a `TTSResponse`), `stream()` (a `TTSStream` context manager yielding bytes as the API generates them), `close()` / `with` for a client built from `api_key=`. Host from `aai.settings.tts_base_url` (`https://tts.us.assemblyai.com`; set `https://tts.eu.assemblyai.com` for the EU zone)
+- `aai.AsyncTextToSpeech` — Asyncio counterpart of `TextToSpeech`; `synthesize()` is a coroutine and `stream()` returns an `AsyncTTSStream`. Owns an HTTP pool: use `async with` or `await aclose()`
+- `aai.TTSConfig` — Text-to-Speech options: `voice_id` (a library voice such as `aai-jane`, or a cloned voice by UUID or slug), `model_id` (`assemblyai-tts-1`), `language`, `sample_rate`, `encoding` (`aai.TTSEncoding`: `pcm_s16le` default | `pcm_mulaw` | `pcm_alaw` | `mp3`), `container` (`aai.TTSContainer`: `raw` default | `wav`), `bit_rate`, `extend_biometric_consent` (owner-only consent renewal of a cloned voice). Rejects unknown fields
+- `aai.TTSResponse` — Text-to-Speech result: `.audio` (bytes) plus the response headers: `.request_id`, `.region`, `.model`, `.voice_id`, `.language`, `.encoding`, `.sample_rate`, `.container`, `.bit_rate`, `.content_type`, `.voice_expires_at`, `.billed_characters`, `.audio_duration_ms`. `TTSStream.info` carries the same fields without `audio`; `audio_duration_ms` is `None` on a stream with audio (`0` when the text had nothing to speak)
+- `aai.TTSError` — Raised on any Text-to-Speech refusal: `.status_code`, `.error_code` (stable API code; `biometric_terms_not_accepted` 403 until the account accepts the biometric data terms in the dashboard, `auth_unavailable` 503 retryable), `.param`, `.request_id`, `.retryable`, `.retry_after`, `.errors`, `.body` (the whole envelope)
- `assemblyai.streaming.v3.RealTimeTranscriber` — Real-time streaming with event-based API (threaded)
- `assemblyai.streaming.v3.AsyncRealTimeTranscriber` — Asyncio-native counterpart; same options/events
- `aai.LLMGateway` — LLM Gateway client. Resources: `models.list()`, `chat.completions.create()` (incl. `stream=True`), `understanding.create()`/`.validate()`
@@ -452,6 +457,30 @@ async with AsyncRealTimeTranscriber(RealTimeTranscriberOptions(token=token_from_
- `format_turns=True` enables punctuation/casing on confirmed end-of-turns. Toggle mid-session via `client.set_params(RealTimeSessionParameters(format_turns=True))`.
- `AsyncRealTimeTranscriber` used as `async with` calls `disconnect(terminate=True)` on normal block exit and `disconnect(terminate=False)` on exception — no explicit `disconnect()` needed inside the block.
+## Text-to-Speech
+
+One text per HTTP request. `synthesize()` returns the whole audio; `stream()` yields it as it is generated. An API refusal is always `aai.TTSError`, and so is a stream body cut after the first audio (`retryable=True`, from the loop: retry the request). Connect and first-byte failures are httpx exceptions, as in the rest of the SDK. Settings are copied when a client is built: set `tts_base_url` before creating `TextToSpeech`. `synthesize()` waits for the whole clip under `settings.tts_synthesize_http_timeout` (610 s, over the API's 600 s request cap); `stream()` uses `settings.tts_http_timeout` (30 s) per socket operation.
+
+```python
+tts = aai.TextToSpeech(config=aai.TTSConfig(voice_id="aai-jane"))
+
+# A per-call config replaces the default whole: repeat the voice_id in it
+wav = tts.synthesize("Hello.", aai.TTSConfig(voice_id="aai-jane", container=aai.TTSContainer.wav)).audio
+
+with tts.stream("Hello.") as speech:
+ for chunk in speech: # raw 24 kHz s16le mono by default
+ play(chunk)
+
+# Telephony: 8 kHz mu-law (chunk sizes vary; re-frame to 160 bytes for 20 ms frames)
+telephony = aai.TTSConfig(voice_id="aai-jane", encoding=aai.TTSEncoding.pcm_mulaw, sample_rate=8000)
+with tts.stream("Hello.", telephony) as speech:
+ for chunk in speech:
+ send_to_call(chunk)
+
+# A cloned or library voice by id
+tts.synthesize("Hello.", aai.TTSConfig(voice_id="aai-alba"))
+```
+
## LLM Gateway
`aai.LLMGateway` is a client for AssemblyAI's LLM Gateway: model listing, OpenAI-shaped chat
diff --git a/README.md b/README.md
index 466736a..f6168a3 100644
--- a/README.md
+++ b/README.md
@@ -890,6 +890,144 @@ except aai.DictationError as error:
---
+### **Text-to-Speech Examples**
+
+`aai.TextToSpeech` speaks one text per HTTP request on the Text-to-Speech API. `synthesize()` returns the complete audio; `stream()` returns it as the API generates it, so playback can start long before the sentence ends. A text is 1 to 5,000 characters. Options go in a `TTSConfig`: the `voice_id` (a voice of the AssemblyAI voice library by its slug: `aai-jane` by default, `aai-alba`, `aai-anna`, `aai-charles`, `aai-eve`, `aai-george`, `aai-jean`, `aai-mary`, `aai-michael`, `aai-paul`, `aai-vera` and hundreds more; or a voice cloned by your account, by its UUID or slug), the `model_id` (`assemblyai-tts-1`), the `language` (`english`, `spanish`, `german`, `french`, `italian`, `portuguese`), the `sample_rate` (8000, 16000, 22050, 24000 by default, 44100, 48000), the `encoding` (`pcm_s16le` by default, `pcm_mulaw`/`pcm_alaw` for telephony, `mp3`) and the `container` (`raw` or `wav`). A per-call `TTSConfig` replaces the client's default config whole. Voices are created and managed in the AssemblyAI dashboard.
+
+Requests go to `aai.settings.tts_base_url`, `https://tts.us.assemblyai.com` by default; set it to `https://tts.eu.assemblyai.com` to keep them in the EU data zone.
+
+
+ Synthesize a WAV file
+
+```python
+import assemblyai as aai
+
+aai.settings.api_key = ""
+
+tts = aai.TextToSpeech()
+
+response = tts.synthesize(
+ "Hello from AssemblyAI.",
+ aai.TTSConfig(voice_id="aai-jane", container=aai.TTSContainer.wav),
+)
+# A config given per call replaces the client's default config whole.
+
+with open("hello.wav", "wb") as out:
+ out.write(response.audio)
+
+print(response.model, response.voice_id, response.audio_duration_ms, "ms")
+```
+
+`synthesize()` waits for the whole audio (`aai.settings.tts_synthesize_http_timeout`, 610 s by default, just over the API's 600-second limit on a request), so an API refusal is always a `TTSError`, never a cut response, and the WAV header carries real sizes. Network failures raise httpx exceptions, as everywhere in the SDK. Python's `wave` module reads PCM WAV only; a mu-law or A-law WAV needs another reader, or use `container="raw"` for telephony codecs. The response headers are on the result: `request_id`, `region`, `model`, `voice_id`, `language`, `encoding`, `sample_rate`, `container`, `bit_rate`, `billed_characters` and `audio_duration_ms`.
+
+
+
+
+ Stream audio as it is generated
+
+```python
+import assemblyai as aai
+
+aai.settings.api_key = ""
+
+tts = aai.TextToSpeech(config=aai.TTSConfig(voice_id="aai-alba"))
+
+with tts.stream("Streaming straight to the speakers.") as speech:
+ print("speaking with", speech.info.model, "at", speech.info.sample_rate, "Hz")
+ for chunk in speech:
+ play(chunk) # raw 24 kHz 16-bit mono PCM
+```
+
+`stream()` itself sends nothing: entering the `with` block (or the first read) sends the request and returns once the API has sent the first audio, usually well under a second. Chunk sizes vary and are not aligned to samples or frames: for 16-bit PCM carry an odd trailing byte over to the next chunk, and re-frame into fixed-size frames (20 ms of 8 kHz mu-law is 160 bytes) if your sink needs them. Breaking out of the loop early (a barge-in) drops that connection, so the next request pays a new TLS handshake, about 100 ms more to its first audio. The API commits to `200` with the first audio, so a failure after that cuts the body instead of changing the status: the loop then raises a `TTSError` with `retryable=True`, and the whole request should be retried.
+
+
+
+
+ Telephony audio and a cloned voice
+
+```python
+import assemblyai as aai
+
+aai.settings.api_key = ""
+
+telephony = aai.TTSConfig(
+ voice_id="01a0da61-138c-77c7-ab98-421431ed74e8", # a voice cloned in the dashboard
+ encoding=aai.TTSEncoding.pcm_mulaw,
+ sample_rate=8000,
+)
+
+with aai.TextToSpeech(config=telephony).stream("Your appointment is confirmed.") as speech:
+ for chunk in speech:
+ send_to_call(chunk)
+```
+
+
+
+
+ Handle errors and retries
+
+```python
+import time
+import assemblyai as aai
+
+aai.settings.api_key = "" # or the ASSEMBLYAI_API_KEY environment variable
+
+tts = aai.TextToSpeech()
+
+try:
+ response = tts.synthesize("Hello.", aai.TTSConfig(voice_id="aai-nobody"))
+except aai.TTSError as e:
+ print(e.status_code, e.error_code, e.param, e.request_id) # 422 voice_not_found voice_id ...
+ if e.retryable:
+ time.sleep(e.retry_after or 1)
+```
+
+`error_code` is the API's stable code (`invalid_parameter`, `invalid_request` for a voice the model does not speak, `voice_not_found`, `biometric_terms_not_accepted` (403: accept the biometric data terms in the dashboard, then retry), `rate_limited`, `at_capacity`, `auth_unavailable` (503, retryable), ...); branch on it rather than on the message. `body` holds the whole error envelope for the fields some codes add. `retryable` and `retry_after` (seconds) say whether and when to send the same request again.
+
+
+
+
+ Asyncio
+
+```python
+import asyncio
+import assemblyai as aai
+
+aai.settings.api_key = ""
+
+
+async def main():
+ async with aai.AsyncTextToSpeech() as tts:
+ response = await tts.synthesize("Hello from AssemblyAI.")
+ print(len(response.audio), "bytes")
+
+ async with tts.stream("And streamed.") as speech:
+ async for chunk in speech:
+ await play(chunk) # your own coroutine
+
+
+asyncio.run(main())
+```
+
+`AsyncTextToSpeech` owns an HTTP pool: use `async with` or `await aclose()`, or pass an `aai.AsyncClient` to share one.
+
+
+
+
+ Renew a cloned voice's consent
+
+A voice cloned by your account expires 60 days after its biometric consent was last given. Its owner renews it by sending `extend_biometric_consent=True` with a request that uses the voice; `voice_expires_at` on the response shows the new expiry. The account must have accepted the biometric data terms in the dashboard, or the request is refused with `biometric_terms_not_accepted`.
+
+```python
+response = tts.synthesize(
+ "Consent renewed.",
+ aai.TTSConfig(voice_id="", extend_biometric_consent=True),
+)
+print(response.voice_expires_at) # Unix seconds
+```
+
+
+
### **Asyncio Examples**
`aai.AsyncTranscriber` is the asyncio counterpart of `aai.Transcriber`. Every method
@@ -1837,8 +1975,15 @@ aai.settings.sync_live_http_timeout = 180.0
# to outlast the final segment's inference plus the LLM pass.
aai.settings.dictation_base_url = "https://dictation.assemblyai.com"
aai.settings.dictation_http_timeout = 300.0
+
+# Text-to-Speech: the host (US by default, EU below) and the timeouts
+aai.settings.tts_base_url = "https://tts.eu.assemblyai.com"
+aai.settings.tts_http_timeout = 30.0 # stream(): each socket operation
+aai.settings.tts_synthesize_http_timeout = 610.0 # synthesize(): the wait for the whole clip
```
+Set these before creating a client: `aai.TextToSpeech()` (like every client) copies the settings when it is built, so a later change applies to the next instance, not to an existing one. Every setting can also come from the environment with the `ASSEMBLYAI_` prefix, such as `ASSEMBLYAI_TTS_BASE_URL`.
+
---
@@ -1957,6 +2102,7 @@ All custom SDK exceptions inherit from `aai.AssemblyAIError`, except the streami
| `aai.AssemblyAIError` | Base exception for all AssemblyAI errors. | `status_code: Optional[int]` |
| `aai.TranscriptError` | Raised when a prerecorded transcript API call fails at the HTTP level or polling times out. | `status_code: Optional[int]` |
| `aai.SyncTranscriptError` | Raised when synchronous transcription fails (`aai.SyncTranscriber`). | `status_code: Optional[int]`, `error_code: Optional[str]`, `retry_after: Optional[int]` |
+| `aai.TTSError` | Raised when a Text-to-Speech request fails (`aai.TextToSpeech`). | `status_code: Optional[int]`, `error_code: Optional[str]`, `param: Optional[str]`, `request_id: Optional[str]`, `retryable: Optional[bool]`, `retry_after: Optional[int]`, `errors: Optional[List[Dict[str, Optional[str]]]]`, `body: Optional[dict]` |
| `assemblyai.types.RedactedAudioIncompleteError` | Raised when requesting redacted audio before processing completes. | `status_code: Optional[int]` |
| `assemblyai.types.RedactedAudioExpiredError` | Raised when requesting redacted audio that has expired and is no longer available. | `status_code: Optional[int]` |
| `assemblyai.types.RedactedAudioUnavailableError` | Raised when requesting redacted audio that is unavailable at the given URL. | `status_code: Optional[int]` |
diff --git a/assemblyai/__init__.py b/assemblyai/__init__.py
index b765b61..f625c97 100644
--- a/assemblyai/__init__.py
+++ b/assemblyai/__init__.py
@@ -29,6 +29,18 @@
from .prerecorded.v2 import AsyncTranscriber, AsyncTranscript
from .sync.v1 import AsyncSyncTranscriber, SyncTranscriber
from .transcriber import Transcriber, Transcript, TranscriptGroup
+from .tts.v1 import (
+ AsyncTextToSpeech,
+ AsyncTTSStream,
+ TextToSpeech,
+ TTSConfig,
+ TTSContainer,
+ TTSEncoding,
+ TTSError,
+ TTSResponse,
+ TTSResponseInfo,
+ TTSStream,
+)
from .types import (
AssemblyAIError,
AutohighlightResponse,
@@ -128,6 +140,8 @@ def __getattr__(name: str):
"AsyncDictationTranscriber",
"AsyncLLMGateway",
"AsyncSyncTranscriber",
+ "AsyncTTSStream",
+ "AsyncTextToSpeech",
"AsyncTranscriber",
"AsyncTranscript",
"AutohighlightResponse",
@@ -197,6 +211,14 @@ def __getattr__(name: str):
"SyncTranscriptResponse",
"SyncWord",
"Timestamp",
+ "TTSConfig",
+ "TTSContainer",
+ "TTSEncoding",
+ "TTSError",
+ "TTSResponse",
+ "TTSResponseInfo",
+ "TTSStream",
+ "TextToSpeech",
"Transcriber",
"TranscriptionConfig",
"Transcript",
diff --git a/assemblyai/__version__.py b/assemblyai/__version__.py
index 81162ee..14d9d2f 100644
--- a/assemblyai/__version__.py
+++ b/assemblyai/__version__.py
@@ -1 +1 @@
-__version__ = "1.6.5"
+__version__ = "1.7.0"
diff --git a/assemblyai/tts/__init__.py b/assemblyai/tts/__init__.py
new file mode 100644
index 0000000..4666a76
--- /dev/null
+++ b/assemblyai/tts/__init__.py
@@ -0,0 +1,32 @@
+"""The Text-to-Speech API: text in, audio out, one text per HTTP request.
+
+The implementation lives in ``tts/v1/``, mirroring the ``dictation/v1/`` and
+``sync/v1/`` layout. This ``__init__`` re-exports the public surface so
+``from assemblyai.tts import TextToSpeech`` works alongside the versioned path.
+"""
+
+from .v1 import (
+ AsyncTextToSpeech,
+ AsyncTTSStream,
+ TextToSpeech,
+ TTSConfig,
+ TTSContainer,
+ TTSEncoding,
+ TTSError,
+ TTSResponse,
+ TTSResponseInfo,
+ TTSStream,
+)
+
+__all__ = [
+ "AsyncTTSStream",
+ "AsyncTextToSpeech",
+ "TTSConfig",
+ "TTSContainer",
+ "TTSEncoding",
+ "TTSError",
+ "TTSResponse",
+ "TTSResponseInfo",
+ "TTSStream",
+ "TextToSpeech",
+]
diff --git a/assemblyai/tts/v1/__init__.py b/assemblyai/tts/v1/__init__.py
new file mode 100644
index 0000000..fd8224e
--- /dev/null
+++ b/assemblyai/tts/v1/__init__.py
@@ -0,0 +1,25 @@
+from .api import TTSStream
+from .async_api import AsyncTTSStream
+from .async_client import AsyncTextToSpeech
+from .client import TextToSpeech
+from .models import (
+ TTSConfig,
+ TTSContainer,
+ TTSEncoding,
+ TTSError,
+ TTSResponse,
+ TTSResponseInfo,
+)
+
+__all__ = [
+ "AsyncTTSStream",
+ "AsyncTextToSpeech",
+ "TTSConfig",
+ "TTSContainer",
+ "TTSEncoding",
+ "TTSError",
+ "TTSResponse",
+ "TTSResponseInfo",
+ "TTSStream",
+ "TextToSpeech",
+]
diff --git a/assemblyai/tts/v1/_base.py b/assemblyai/tts/v1/_base.py
new file mode 100644
index 0000000..e7fcd9f
--- /dev/null
+++ b/assemblyai/tts/v1/_base.py
@@ -0,0 +1,56 @@
+from __future__ import annotations
+
+from typing import Any, Dict, Optional
+
+import httpx
+
+from ... import _audio
+from ...types import Settings
+from .models import TTSConfig
+
+__all__ = ["check_config", "check_text", "request_body", "synthesize_timeout"]
+
+
+def synthesize_timeout(settings: Settings) -> httpx.Timeout:
+ """
+ The httpx timeout of a buffered request (`synthesize()`).
+
+ Its one read waits for the whole audio, so the read timeout is
+ `tts_synthesize_http_timeout`; connecting and sending use
+ `tts_http_timeout`, as a streamed request does.
+ """
+
+ return httpx.Timeout(
+ settings.tts_http_timeout, read=settings.tts_synthesize_http_timeout
+ )
+
+
+def check_config(owner: str, config: Optional[TTSConfig]) -> None:
+ """
+ Raises unless `config` is a `TTSConfig` or `None`.
+
+ Args:
+ owner: the class to name in the message, e.g. `TextToSpeech`.
+ config: the configuration to check.
+ """
+
+ _audio.check_config(owner, config, TTSConfig)
+
+
+def check_text(text: object) -> None:
+ """Raises unless `text` is a `str`; names the swapped-argument mistake."""
+
+ if isinstance(text, TTSConfig):
+ raise TypeError(
+ "text comes first: synthesize(text, config) / stream(text, config)"
+ )
+ if not isinstance(text, str):
+ raise TypeError(f"text must be a str, not {type(text).__name__}")
+
+
+def request_body(text: str, config: TTSConfig) -> Dict[str, Any]:
+ """The JSON body of a speech request: `text` plus every option that is set."""
+
+ body: Dict[str, Any] = _audio._config_to_json(config) or {}
+ body["text"] = text
+ return body
diff --git a/assemblyai/tts/v1/api.py b/assemblyai/tts/v1/api.py
new file mode 100644
index 0000000..ec2f7fc
--- /dev/null
+++ b/assemblyai/tts/v1/api.py
@@ -0,0 +1,312 @@
+import math
+from typing import Any, ContextManager, Dict, Iterator, Optional, Union
+
+import httpx
+
+from .models import TTSError, TTSResponse, TTSResponseInfo
+
+# The one route this client posts to. Streamed and buffered responses differ
+# only in the body's `stream` field.
+ENDPOINT_SPEECH = "/v1/speech"
+
+# Statuses an answer made before the request reached the API can carry; those
+# answers have no JSON body, so `retryable` is decided from the status alone.
+_RETRYABLE_STATUSES = frozenset({408, 429, 500, 502, 503, 504})
+
+# What httpx raises from a streamed body that ends before the audio is
+# complete: the API closing the connection or resetting the stream after a
+# failure, a dropped connection, or no audio for a whole read timeout.
+_CUT_ERRORS = (httpx.RemoteProtocolError, httpx.ReadError, httpx.ReadTimeout)
+
+Timeout = Union[float, httpx.Timeout]
+
+
+def _seconds(value: Any) -> Optional[int]:
+ """A retry hint as whole seconds, rounded up; `None` for anything else."""
+
+ if isinstance(value, bool):
+ return None
+ if isinstance(value, (int, float)):
+ return int(math.ceil(value))
+ if isinstance(value, str) and value.strip().isdigit():
+ return int(value)
+ return None
+
+
+def _int_header(headers: httpx.Headers, name: str) -> Optional[int]:
+ return _seconds(headers.get(name))
+
+
+def _voice_id(headers: httpx.Headers, requested: Optional[str]) -> Optional[str]:
+ """
+ The voice that spoke, from `X-TTS-Voice-Id`.
+
+ Until the API sends that header, it is derived from `X-TTS-Voice`: the
+ request's own `voice_id` for `custom`, else the built-in voice's slug.
+ """
+
+ voice_id = headers.get("x-tts-voice-id")
+ if voice_id is not None:
+ return voice_id
+ voice = headers.get("x-tts-voice")
+ if voice == "custom":
+ return requested
+ return f"aai-{voice}" if voice else None
+
+
+def _header_fields(
+ headers: httpx.Headers, requested_voice_id: Optional[str]
+) -> Dict[str, Any]:
+ """The `TTSResponseInfo` fields read from the `X-TTS-*` headers of a `200`."""
+
+ return {
+ "request_id": headers.get("x-request-id"),
+ "region": headers.get("x-tts-region"),
+ "model": headers.get("x-tts-model"),
+ "voice_id": _voice_id(headers, requested_voice_id),
+ "language": headers.get("x-tts-language"),
+ "encoding": headers.get("x-tts-encoding"),
+ "sample_rate": _int_header(headers, "x-tts-sample-rate"),
+ "container": headers.get("x-tts-container"),
+ "bit_rate": _int_header(headers, "x-tts-bit-rate"),
+ "content_type": headers.get("content-type"),
+ "voice_expires_at": _int_header(headers, "x-tts-voice-expires-at"),
+ "billed_characters": _int_header(headers, "x-tts-billed-characters"),
+ "audio_duration_ms": _int_header(headers, "x-tts-audio-duration-ms"),
+ }
+
+
+def _error_from_response(response: httpx.Response) -> TTSError:
+ """
+ Builds a `TTSError` from a non-200 response.
+
+ The API answers every refusal with one JSON envelope (`code`, `message`,
+ `param`, `request_id`, `errors`, `retryable`, `retry_after_seconds`).
+ An answer made before the request reached the API (a load balancer's
+ `502`, a plain-text `404`) has no envelope: its status and text are
+ reported, and `retryable` follows the status.
+ """
+
+ status = response.status_code
+ retry_after = _int_header(response.headers, "retry-after")
+ request_id = response.headers.get("x-request-id")
+
+ body: Any = None
+ if response.headers.get("content-type", "").startswith("application/json"):
+ try:
+ body = response.json()
+ except ValueError:
+ body = None
+
+ if isinstance(body, dict):
+ message = body.get("message") or body.get("error")
+ if not message:
+ message = f"text-to-speech request failed with status {status}"
+ if body.get("retry_after_seconds") is not None:
+ retry_after = _seconds(body["retry_after_seconds"])
+ retryable = body.get("retryable")
+ if retryable is None:
+ retryable = status in _RETRYABLE_STATUSES
+ return TTSError(
+ message,
+ status_code=status,
+ error_code=body.get("code"),
+ param=body.get("param"),
+ request_id=body.get("request_id") or request_id,
+ retryable=retryable,
+ retry_after=retry_after,
+ errors=body.get("errors"),
+ body=body,
+ )
+
+ text = response.text.strip()
+ message = f"text-to-speech request failed with status {status}"
+ if text:
+ message = f"{message}: {text[:200]}"
+ return TTSError(
+ message,
+ status_code=status,
+ request_id=request_id,
+ retryable=status in _RETRYABLE_STATUSES,
+ retry_after=retry_after,
+ )
+
+
+def _cut_error(info: Optional[TTSResponseInfo], error: Exception) -> TTSError:
+ """The error for a body cut after the first audio: retry the whole request."""
+
+ return TTSError(
+ "The response was cut before the audio was complete "
+ f"({type(error).__name__}). Retry the request.",
+ status_code=200,
+ request_id=info.request_id if info else None,
+ retryable=True,
+ )
+
+
+class TTSStream:
+ """
+ The audio of one request, read as the API generates it.
+
+ Returned by `TextToSpeech.stream()`. The request is sent when the stream
+ is entered as a context manager or first read, and `TTSError` is raised
+ there if the API refuses it. Use `with`, or call `close()`, so the
+ connection is released: a fully read stream's connection returns to the
+ pool, a partially read one is dropped. `info` holds what the response
+ headers say (model, voice id, encoding, sample rate, request id) once the
+ request has been sent; `iter_bytes()` yields the audio as it arrives and
+ `read()` collects all of it.
+
+ The API commits to `200` with the first audio, so a failure after that
+ cuts the body instead of changing the status: the iteration then raises
+ `TTSError` with `retryable=True`. Retry the whole request. Chunks are
+ not aligned to samples: for 16-bit PCM, carry an odd trailing byte over
+ to the next chunk.
+ """
+
+ def __init__(
+ self,
+ client: httpx.Client,
+ *,
+ url: str,
+ body: Dict[str, Any],
+ timeout: Timeout,
+ ) -> None:
+ self._client = client
+ self._url = url
+ self._body = body
+ self._timeout = timeout
+ self._cm: Optional[ContextManager[httpx.Response]] = None
+ self._response: Optional[httpx.Response] = None
+ self._info: Optional[TTSResponseInfo] = None
+
+ @property
+ def info(self) -> TTSResponseInfo:
+ """The response headers, once the stream is entered or first read."""
+
+ if self._info is None:
+ raise RuntimeError("enter or read the stream before `info`")
+ return self._info
+
+ def _open(self) -> httpx.Response:
+ if self._response is not None:
+ return self._response
+
+ # `client.stream()` rather than `build_request(timeout=)`, which the
+ # SDK's httpx floor lacks.
+ cm = self._client.stream(
+ "POST", self._url, json=self._body, timeout=self._timeout
+ )
+ response = cm.__enter__()
+ if response.status_code != httpx.codes.OK:
+ try:
+ response.read()
+ raise _error_from_response(response)
+ finally:
+ cm.__exit__(None, None, None)
+
+ self._cm = cm
+ self._response = response
+ self._info = TTSResponseInfo(
+ **_header_fields(response.headers, self._body.get("voice_id"))
+ )
+ return response
+
+ def __enter__(self) -> "TTSStream":
+ self._open()
+ return self
+
+ def __exit__(self, *exc: object) -> None:
+ self.close()
+
+ def __iter__(self) -> Iterator[bytes]:
+ return self.iter_bytes()
+
+ def iter_bytes(self) -> Iterator[bytes]:
+ """
+ Yields the audio bytes as they arrive.
+
+ Raises:
+ TTSError: from the first read if the API refuses the request, or
+ later if the body is cut before the audio is complete.
+ """
+
+ response = self._open()
+ try:
+ for chunk in response.iter_bytes():
+ if chunk:
+ yield chunk
+ except _CUT_ERRORS as error:
+ raise _cut_error(self._info, error) from error
+
+ def read(self) -> bytes:
+ """Reads the rest of the audio and returns it."""
+
+ return b"".join(self.iter_bytes())
+
+ def close(self) -> None:
+ """Releases the connection. Audio not yet read is discarded."""
+
+ cm, self._cm = self._cm, None
+ if cm is not None:
+ cm.__exit__(None, None, None)
+
+
+def synthesize(
+ client: httpx.Client,
+ *,
+ base_url: str,
+ body: Dict[str, Any],
+ timeout: Timeout,
+) -> TTSResponse:
+ """
+ Posts a buffered speech request and returns the complete audio.
+
+ Asks for `"stream": false`, so nothing arrives until the audio is
+ complete and an API failure is always an error status, never a cut body.
+
+ Args:
+ client: the HTTP client (carries the `Authorization` header).
+ base_url: the Text-to-Speech API base URL, e.g. `https://tts.us.assemblyai.com`.
+ body: the JSON body: `text` and the options of a `TTSConfig`.
+ timeout: the httpx timeout; its read timeout is the wait for the
+ whole audio.
+
+ Raises: `TTSError` on any non-200 response.
+ """
+
+ response = client.post(
+ base_url.rstrip("/") + ENDPOINT_SPEECH,
+ json={**body, "stream": False},
+ timeout=timeout,
+ )
+
+ if response.status_code != httpx.codes.OK:
+ raise _error_from_response(response)
+
+ return TTSResponse(
+ audio=response.content,
+ **_header_fields(response.headers, body.get("voice_id")),
+ )
+
+
+def stream_speech(
+ client: httpx.Client,
+ *,
+ base_url: str,
+ body: Dict[str, Any],
+ timeout: Timeout,
+) -> TTSStream:
+ """
+ Prepares a streamed speech request and returns it as a `TTSStream`.
+
+ Nothing is sent here: the stream sends the request when it is entered or
+ first read, and raises `TTSError` then if the API refuses it.
+ """
+
+ return TTSStream(
+ client,
+ url=base_url.rstrip("/") + ENDPOINT_SPEECH,
+ body={**body, "stream": True},
+ timeout=timeout,
+ )
diff --git a/assemblyai/tts/v1/async_api.py b/assemblyai/tts/v1/async_api.py
new file mode 100644
index 0000000..8802ce3
--- /dev/null
+++ b/assemblyai/tts/v1/async_api.py
@@ -0,0 +1,168 @@
+from typing import Any, AsyncContextManager, AsyncIterator, Dict, Optional
+
+import httpx
+
+from .api import (
+ _CUT_ERRORS,
+ ENDPOINT_SPEECH,
+ Timeout,
+ _cut_error,
+ _error_from_response,
+ _header_fields,
+)
+from .models import TTSResponse, TTSResponseInfo
+
+
+class AsyncTTSStream:
+ """
+ The audio of one request, read as the API generates it.
+
+ The asyncio counterpart of `TTSStream`, returned by
+ `AsyncTextToSpeech.stream()`. The request is sent on `async with` or on
+ the first read, and `TTSError` is raised there if the API refuses it.
+ Use `async with`, or `await aclose()`, so the connection is released (a
+ fully read stream's connection returns to the pool, a partially read one
+ is dropped). `info` holds what the response headers say once the request
+ has been sent; `aiter_bytes()` yields the audio as it arrives and
+ `aread()` collects all of it. A failure after the first audio cuts the
+ body: the iteration then raises `TTSError` with `retryable=True`.
+ """
+
+ def __init__(
+ self,
+ client: httpx.AsyncClient,
+ *,
+ url: str,
+ body: Dict[str, Any],
+ timeout: Timeout,
+ ) -> None:
+ self._client = client
+ self._url = url
+ self._body = body
+ self._timeout = timeout
+ self._cm: Optional[AsyncContextManager[httpx.Response]] = None
+ self._response: Optional[httpx.Response] = None
+ self._info: Optional[TTSResponseInfo] = None
+
+ @property
+ def info(self) -> TTSResponseInfo:
+ """The response headers, once the stream is entered or first read."""
+
+ if self._info is None:
+ raise RuntimeError("enter or read the stream before `info`")
+ return self._info
+
+ async def _open(self) -> httpx.Response:
+ if self._response is not None:
+ return self._response
+
+ cm = self._client.stream(
+ "POST", self._url, json=self._body, timeout=self._timeout
+ )
+ response = await cm.__aenter__()
+ if response.status_code != httpx.codes.OK:
+ try:
+ await response.aread()
+ raise _error_from_response(response)
+ finally:
+ await cm.__aexit__(None, None, None)
+
+ self._cm = cm
+ self._response = response
+ self._info = TTSResponseInfo(
+ **_header_fields(response.headers, self._body.get("voice_id"))
+ )
+ return response
+
+ async def __aenter__(self) -> "AsyncTTSStream":
+ await self._open()
+ return self
+
+ async def __aexit__(self, *exc: object) -> None:
+ await self.aclose()
+
+ def __aiter__(self) -> AsyncIterator[bytes]:
+ return self.aiter_bytes()
+
+ async def aiter_bytes(self) -> AsyncIterator[bytes]:
+ """
+ Yields the audio bytes as they arrive.
+
+ Raises:
+ TTSError: from the first read if the API refuses the request, or
+ later if the body is cut before the audio is complete.
+ """
+
+ response = await self._open()
+ try:
+ async for chunk in response.aiter_bytes():
+ if chunk:
+ yield chunk
+ except _CUT_ERRORS as error:
+ raise _cut_error(self._info, error) from error
+
+ async def aread(self) -> bytes:
+ """Reads the rest of the audio and returns it."""
+
+ chunks = []
+ async for chunk in self.aiter_bytes():
+ chunks.append(chunk)
+ return b"".join(chunks)
+
+ async def aclose(self) -> None:
+ """Releases the connection. Audio not yet read is discarded."""
+
+ cm, self._cm = self._cm, None
+ if cm is not None:
+ await cm.__aexit__(None, None, None)
+
+
+async def synthesize(
+ client: httpx.AsyncClient,
+ *,
+ base_url: str,
+ body: Dict[str, Any],
+ timeout: Timeout,
+) -> TTSResponse:
+ """
+ Posts a buffered speech request and returns the complete audio.
+
+ The asyncio counterpart of `api.synthesize`.
+
+ Raises: `TTSError` on any non-200 response.
+ """
+
+ response = await client.post(
+ base_url.rstrip("/") + ENDPOINT_SPEECH,
+ json={**body, "stream": False},
+ timeout=timeout,
+ )
+
+ if response.status_code != httpx.codes.OK:
+ raise _error_from_response(response)
+
+ return TTSResponse(
+ audio=response.content,
+ **_header_fields(response.headers, body.get("voice_id")),
+ )
+
+
+def stream_speech(
+ client: httpx.AsyncClient,
+ *,
+ base_url: str,
+ body: Dict[str, Any],
+ timeout: Timeout,
+) -> AsyncTTSStream:
+ """
+ Prepares a streamed speech request and returns it as an `AsyncTTSStream`.
+
+ The asyncio counterpart of `api.stream_speech`: nothing is sent here.
+ """
+
+ return AsyncTTSStream(
+ client,
+ url=base_url.rstrip("/") + ENDPOINT_SPEECH,
+ body={**body, "stream": True},
+ timeout=timeout,
+ )
diff --git a/assemblyai/tts/v1/async_client.py b/assemblyai/tts/v1/async_client.py
new file mode 100644
index 0000000..514d636
--- /dev/null
+++ b/assemblyai/tts/v1/async_client.py
@@ -0,0 +1,148 @@
+from typing import Optional
+
+from ... import async_client as _async_client
+from . import async_api
+from ._base import check_config, check_text, request_body, synthesize_timeout
+from .async_api import AsyncTTSStream
+from .models import TTSConfig, TTSResponse
+
+__all__ = ["AsyncTextToSpeech"]
+
+
+class AsyncTextToSpeech:
+ """
+ Text in, speech out, for asyncio: the counterpart of `TextToSpeech`.
+
+ Same options, result and errors; `synthesize()` is a coroutine and
+ `stream()` returns an `AsyncTTSStream`. Owns an HTTP pool: use
+ `async with` or `await aclose()`, or pass an `aai.AsyncClient` to share
+ one.
+
+ Example:
+ ```python
+ import asyncio
+ import assemblyai as aai
+
+ aai.settings.api_key = "your-key"
+
+ async def main():
+ async with aai.AsyncTextToSpeech() as tts:
+ async with tts.stream("Hello from AssemblyAI.") as speech:
+ async for chunk in speech:
+ await play(chunk)
+
+ asyncio.run(main())
+ ```
+ """
+
+ def __init__(
+ self,
+ *,
+ client: Optional[_async_client.AsyncClient] = None,
+ config: Optional[TTSConfig] = None,
+ api_key: Optional[str] = None,
+ ) -> None:
+ """
+ Creates an `AsyncTextToSpeech`.
+
+ Args:
+ client: The `AsyncClient` to use. If `None`, one is created from
+ the global `settings` and closed on `aclose()`. Pass a client
+ to share one pool between instances.
+ config: Default options. A per-call `config` replaces it whole,
+ so repeat the voice_id in a per-call config.
+ api_key: The API key to authenticate with. Builds and owns an
+ `AsyncClient` from it. Given alongside `client`, it takes
+ precedence: a client is built from a copy of that client's
+ settings with the key replaced, and the given client is left
+ untouched and stays the caller's to close.
+
+ Raises:
+ TypeError: if `config` is not a `TTSConfig`.
+ """
+ check_config(type(self).__name__, config)
+
+ self._owns_client = client is None or api_key is not None
+ self._client = _async_client._resolve_client(client, api_key)
+ self.config = config or TTSConfig()
+
+ @property
+ def config(self) -> TTSConfig:
+ """The default configuration of the `AsyncTextToSpeech`."""
+ return self._config
+
+ @config.setter
+ def config(self, config: TTSConfig) -> None:
+ check_config(type(self).__name__, config)
+
+ self._config = config or TTSConfig()
+
+ @property
+ def client(self) -> _async_client.AsyncClient:
+ """The `AsyncClient` this instance sends requests with."""
+
+ return self._client
+
+ async def synthesize(
+ self,
+ text: str,
+ config: Optional[TTSConfig] = None,
+ ) -> TTSResponse:
+ """
+ Speaks `text` and returns the complete audio.
+
+ The asyncio counterpart of `TextToSpeech.synthesize`.
+
+ Raises:
+ TTSError: on any API refusal.
+ """
+ check_text(text)
+ check_config(type(self).__name__, config)
+ body = request_body(text, config or self._config)
+
+ return await async_api.synthesize(
+ self._client.http_client,
+ base_url=self._client.settings.tts_base_url,
+ body=body,
+ timeout=synthesize_timeout(self._client.settings),
+ )
+
+ def stream(
+ self,
+ text: str,
+ config: Optional[TTSConfig] = None,
+ ) -> AsyncTTSStream:
+ """
+ Speaks `text` and returns the audio as it is generated.
+
+ The asyncio counterpart of `TextToSpeech.stream`. Nothing is sent
+ until the returned `AsyncTTSStream` is entered with `async with` or
+ first read, which is where a refusal raises. Use `async with`, or
+ `await aclose()` it, so the connection is released.
+
+ Raises:
+ TTSError: from the `async with` or the first read, if the API
+ refuses the request before any audio.
+ """
+ check_text(text)
+ check_config(type(self).__name__, config)
+ body = request_body(text, config or self._config)
+
+ return async_api.stream_speech(
+ self._client.http_client,
+ base_url=self._client.settings.tts_base_url,
+ body=body,
+ timeout=self._client.settings.tts_http_timeout,
+ )
+
+ async def aclose(self) -> None:
+ """Closes the HTTP pool this instance owns. A shared `client` is left open."""
+
+ if self._owns_client:
+ await self._client.aclose()
+
+ async def __aenter__(self) -> "AsyncTextToSpeech":
+ return self
+
+ async def __aexit__(self, *exc: object) -> None:
+ await self.aclose()
diff --git a/assemblyai/tts/v1/client.py b/assemblyai/tts/v1/client.py
new file mode 100644
index 0000000..f1c9de9
--- /dev/null
+++ b/assemblyai/tts/v1/client.py
@@ -0,0 +1,166 @@
+from typing import Optional
+
+from ... import client as _client
+from . import api
+from ._base import check_config, check_text, request_body, synthesize_timeout
+from .api import TTSStream
+from .models import TTSConfig, TTSResponse
+
+__all__ = ["TextToSpeech"]
+
+
+class TextToSpeech:
+ """
+ Text in, speech out: the Text-to-Speech HTTP API, one text per request.
+
+ `synthesize()` returns the complete audio; `stream()` returns it as the
+ API generates it, so playback can start long before the sentence ends.
+ Voices, models and options are those of the streaming WebSocket API,
+ without a session to manage: use this client for one text at a time (a
+ sentence of a voice agent, a file, a preview).
+
+ Requests go to `settings.tts_base_url`, `https://tts.us.assemblyai.com`
+ by default; set `https://tts.eu.assemblyai.com` to keep them in the EU.
+
+ Example:
+ ```python
+ import assemblyai as aai
+
+ aai.settings.api_key = "your-key"
+
+ tts = aai.TextToSpeech(config=aai.TTSConfig(voice_id="aai-alba"))
+
+ with open("hello.wav", "wb") as out:
+ wav = aai.TTSConfig(voice_id="aai-alba", container=aai.TTSContainer.wav)
+ out.write(tts.synthesize("Hello from AssemblyAI.", wav).audio)
+
+ with tts.stream("Streaming straight to the speakers.") as speech:
+ for chunk in speech:
+ play(chunk)
+ ```
+ """
+
+ def __init__(
+ self,
+ *,
+ client: Optional[_client.Client] = None,
+ config: Optional[TTSConfig] = None,
+ api_key: Optional[str] = None,
+ ) -> None:
+ """
+ Creates a `TextToSpeech`.
+
+ Args:
+ client: The HTTP client to use. Defaults to the shared default client.
+ config: Default options. A per-call `config` replaces it whole,
+ so repeat the voice_id in a per-call config.
+ api_key: The API key to authenticate with. Builds a `Client` for
+ this instance. Given alongside `client`, it takes precedence:
+ a client is built from a copy of that client's settings with
+ the key replaced, and the given client is left untouched.
+
+ Raises:
+ TypeError: if `config` is not a `TTSConfig`.
+ """
+ check_config(type(self).__name__, config)
+
+ # Only a client built from `api_key` is owned; `client=None` is the
+ # shared default client, which stays open.
+ self._owns_client = api_key is not None
+ self._client = _client._resolve_client(client, api_key)
+ self.config = config or TTSConfig()
+
+ @property
+ def config(self) -> TTSConfig:
+ """The default configuration of the `TextToSpeech`."""
+ return self._config
+
+ @config.setter
+ def config(self, config: TTSConfig) -> None:
+ check_config(type(self).__name__, config)
+
+ self._config = config or TTSConfig()
+
+ @property
+ def client(self) -> _client.Client:
+ """The `Client` this instance sends requests with."""
+
+ return self._client
+
+ def synthesize(
+ self,
+ text: str,
+ config: Optional[TTSConfig] = None,
+ ) -> TTSResponse:
+ """
+ Speaks `text` and returns the complete audio.
+
+ Nothing is sent back until the audio is complete, so a long text
+ takes several seconds and an API failure is always a `TTSError`,
+ never a cut response. Use `stream()` when the first audio matters.
+
+ Args:
+ text: 1 to 5,000 characters.
+ config: Options for this request. Replaces (does not merge with)
+ the instance's default `config`.
+
+ Returns: a `TTSResponse` with the `audio` bytes and the response headers.
+
+ Raises:
+ TTSError: on any API refusal.
+ """
+ check_text(text)
+ check_config(type(self).__name__, config)
+ body = request_body(text, config or self._config)
+
+ return api.synthesize(
+ self._client.http_client,
+ base_url=self._client.settings.tts_base_url,
+ body=body,
+ timeout=synthesize_timeout(self._client.settings),
+ )
+
+ def stream(
+ self,
+ text: str,
+ config: Optional[TTSConfig] = None,
+ ) -> TTSStream:
+ """
+ Speaks `text` and returns the audio as it is generated.
+
+ The request is sent when the returned `TTSStream` is entered as a
+ context manager or first read; entering it returns once the API has
+ sent the first audio. Iterate the stream for the bytes, and use
+ `with` (or call `close()`) so the connection is released.
+
+ Args:
+ text: 1 to 5,000 characters.
+ config: Options for this request. Replaces (does not merge with)
+ the instance's default `config`.
+
+ Raises:
+ TTSError: from the `with` or the first read, if the API refuses
+ the request before any audio.
+ """
+ check_text(text)
+ check_config(type(self).__name__, config)
+ body = request_body(text, config or self._config)
+
+ return api.stream_speech(
+ self._client.http_client,
+ base_url=self._client.settings.tts_base_url,
+ body=body,
+ timeout=self._client.settings.tts_http_timeout,
+ )
+
+ def close(self) -> None:
+ """Closes the HTTP pool this instance owns. A shared `client` is left open."""
+
+ if self._owns_client:
+ self._client.http_client.close()
+
+ def __enter__(self) -> "TextToSpeech":
+ return self
+
+ def __exit__(self, *exc: object) -> None:
+ self.close()
diff --git a/assemblyai/tts/v1/models.py b/assemblyai/tts/v1/models.py
new file mode 100644
index 0000000..bdebdbd
--- /dev/null
+++ b/assemblyai/tts/v1/models.py
@@ -0,0 +1,265 @@
+"""Models for the Text-to-Speech API.
+
+The request options, the audio result and the error of `/v1/speech` on
+`tts.us.assemblyai.com` (or `tts.eu.assemblyai.com`), alongside the client
+that uses them.
+"""
+
+from enum import Enum
+from typing import Any, Dict, List, Optional, Union
+
+try:
+ # pydantic v2 import (`model_validator` exists only in v2; pydantic 1.10
+ # also exports `ConfigDict`, so it is not the probe)
+ from pydantic import BaseModel, ConfigDict, Field, model_validator
+
+ pydantic_v2 = True
+except ImportError:
+ # pydantic v1 import (fallback for Python < 3.14)
+ from pydantic import BaseModel, Field, root_validator
+
+ pydantic_v2 = False
+
+from ...types import AssemblyAIError
+
+__all__ = [
+ "TTSConfig",
+ "TTSContainer",
+ "TTSEncoding",
+ "TTSError",
+ "TTSResponse",
+ "TTSResponseInfo",
+]
+
+
+class TTSError(AssemblyAIError):
+ """
+ Error raised when a Text-to-Speech request fails.
+
+ `error_code` is the API's stable `code` (`invalid_parameter`,
+ `voice_not_found`, `biometric_terms_not_accepted`, `rate_limited`,
+ `at_capacity`, `auth_unavailable`, ...): branch on it, not on the message,
+ which may be reworded. `retryable` says whether the same request can
+ succeed later and `retry_after` (seconds) when to send it; `param` names
+ the field at fault and `errors` lists every failing field
+ (`{"message", "param"}`) of an `invalid_parameter` refusal. `request_id`
+ is the value to quote to support.
+
+ `body` is the parsed error envelope as the API sent it, for fields some
+ codes add (`rate_limited` adds `limit` and `limit_window`; a text or
+ body over its limit adds `limit`).
+
+ `biometric_terms_not_accepted` (`403`, not retryable) refuses one of the
+ account's own cloned voices, or `extend_biometric_consent`, until the
+ account accepts the biometric data terms in the AssemblyAI dashboard.
+ `auth_unavailable` (`503`, retryable) means the credential, or those
+ terms, could not be checked just now.
+
+ A streamed response cut after its first audio raises a `TTSError` with
+ `status_code` `200`, no `error_code`, `retryable` `True` and the
+ request's `request_id`; the httpx error is its `__cause__`. Retry the
+ whole request.
+ """
+
+ def __init__(
+ self,
+ message: str,
+ status_code: Optional[int] = None,
+ error_code: Optional[str] = None,
+ param: Optional[str] = None,
+ request_id: Optional[str] = None,
+ retryable: Optional[bool] = None,
+ retry_after: Optional[int] = None,
+ errors: Optional[List[Dict[str, Optional[str]]]] = None,
+ body: Optional[Dict[str, Any]] = None,
+ ):
+ super().__init__(message, status_code)
+ self.error_code = error_code
+ self.param = param
+ self.request_id = request_id
+ self.retryable = retryable
+ self.retry_after = retry_after
+ self.errors = errors
+ self.body = body
+
+
+class TTSEncoding(str, Enum):
+ """Codec of the audio the API returns."""
+
+ pcm_s16le = "pcm_s16le"
+ "The default: mono signed 16-bit little-endian PCM."
+
+ pcm_mulaw = "pcm_mulaw"
+ "8-bit G.711 mu-law, for telephony."
+
+ pcm_alaw = "pcm_alaw"
+ "8-bit G.711 A-law, for telephony."
+
+ mp3 = "mp3"
+ "Constant bit rate MP3; see `TTSConfig.bit_rate`."
+
+ def __str__(self):
+ return self.value
+
+
+class TTSContainer(str, Enum):
+ """Container of a PCM or G.711 response."""
+
+ raw = "raw"
+ "The default: the codec bytes with no header."
+
+ wav = "wav"
+ "A WAV file. On a streamed response its size fields are unknown (all ones)."
+
+ def __str__(self):
+ return self.value
+
+
+def _refuse_voice_field(data: Any) -> Any:
+ """A `voice` key is refused with a message that points to `voice_id`."""
+
+ if isinstance(data, dict) and "voice" in data:
+ raise ValueError(
+ "TTSConfig has no field `voice`: name the voice with `voice_id`, "
+ 'such as voice_id="aai-jane"'
+ )
+ return data
+
+
+class TTSConfig(BaseModel):
+ """
+ Options for a Text-to-Speech request.
+
+ Every field is optional: the API speaks English with its default voice
+ (`aai-jane`) at 24 kHz as raw 16-bit PCM when nothing is set. Every voice
+ is named by `voice_id`. Unknown fields are rejected rather than silently
+ dropped, so a typo surfaces as a validation error. `speed` (only 1 is
+ supported) and `word_boundaries` (served by `/v1/speech/events`) are
+ left out on purpose.
+ """
+
+ voice_id: Optional[str] = None
+ """The voice: a voice of the AssemblyAI voice library by its slug, such as
+ `aai-jane` (the default), `aai-alba` or `aai-michael`; a voice's ID (a
+ UUID); or the slug of one of the account's own cloned voices. The
+ dashboard shows the value that names each voice. A voice belongs to the
+ data zone it was created in: on the other zone's host it is
+ `voice_not_found`."""
+
+ model_id: Optional[str] = None
+ """The model to run the request on, such as `assemblyai-tts-1`. Defaults to
+ the region's default model of the request's language. A cloned voice must
+ be paired with it."""
+
+ language: Optional[str] = None
+ """The language, by its English name: `english`, `spanish`, `german`,
+ `french`, `italian` or `portuguese`. Case-insensitive. Defaults to the
+ voice's own language. The eleven built-in voices (`aai-jane`, `aai-alba`,
+ `aai-anna`, `aai-charles`, `aai-eve`, `aai-george`, `aai-jean`,
+ `aai-mary`, `aai-michael`, `aai-paul`, `aai-vera`) speak all six; another
+ voice speaks a language other than its own only where its model can."""
+
+ sample_rate: Optional[int] = None
+ "Output sample rate in Hz: 8000, 16000, 22050, 24000 (the default), 44100 or 48000."
+
+ encoding: Optional[Union[TTSEncoding, str]] = None
+ """Output codec: a `TTSEncoding`, or its name as a string. The API also
+ takes the aliases `mulaw` and `alaw`. Defaults to `pcm_s16le`."""
+
+ container: Optional[Union[TTSContainer, str]] = None
+ """`raw` (the default) or `wav`, with a PCM or G.711 encoding only: a
+ `TTSContainer`, or its name as a string."""
+
+ bit_rate: Optional[int] = None
+ """MP3 bit rate in bits per second: 32000, 48000, 64000 (the default),
+ 96000 or 128000; at most 64000 at 8 kHz. With `encoding=mp3` only."""
+
+ extend_biometric_consent: Optional[bool] = None
+ """`True` renews the biometric consent of the cloned voice in `voice_id`,
+ which only the voice's owner may do; `voice_expires_at` on the response
+ shows the new expiry. A cloned voice expires 60 days after its consent
+ was last given. Refused `not_owner` on a voice the account does not own
+ and `403 biometric_terms_not_accepted` until the account has accepted the
+ biometric data terms in the dashboard."""
+
+ if pydantic_v2:
+ # `model_id` is the API's field name: `protected_namespaces=()` keeps
+ # pydantic 2.0 to 2.9 from warning about its `model_` prefix on import.
+ model_config = ConfigDict(extra="forbid", protected_namespaces=())
+
+ @model_validator(mode="before")
+ @classmethod
+ def _refuse_voice(cls, data: Any) -> Any:
+ return _refuse_voice_field(data)
+
+ else:
+
+ class Config:
+ extra = "forbid"
+
+ @root_validator(pre=True, allow_reuse=True)
+ def _refuse_voice(cls, values: Any) -> Any:
+ return _refuse_voice_field(values)
+
+
+class TTSResponseInfo(BaseModel):
+ """
+ What the response headers say about the audio of a request.
+
+ Every field is `None` when the API did not send its header.
+ """
+
+ request_id: Optional[str] = None
+ "The request's id, from `X-Request-Id`. Quote it in support requests."
+
+ region: Optional[str] = None
+ "The region that served the request, such as `us-east-1`."
+
+ model: Optional[str] = None
+ "The model that spoke, such as `assemblyai-tts-1`."
+
+ voice_id: Optional[str] = None
+ """The voice that spoke: the request's `voice_id`, or the default voice's
+ slug (`aai-jane`) when the request named none."""
+
+ language: Optional[str] = None
+ "The language used, lowercase."
+
+ encoding: Optional[str] = None
+ "The codec used, in canonical form (`pcm_mulaw`, never `mulaw`)."
+
+ sample_rate: Optional[int] = None
+ "The sample rate used, in Hz."
+
+ container: Optional[str] = None
+ "`raw` or `wav`; `None` for MP3."
+
+ bit_rate: Optional[int] = None
+ "The MP3 bit rate in bits per second; `None` otherwise."
+
+ content_type: Optional[str] = None
+ "The response's `Content-Type`, such as `audio/pcm;rate=24000;channels=1`."
+
+ voice_expires_at: Optional[int] = None
+ """For a cloned voice of the account that expires: its expiry as a Unix
+ time in seconds, after the request's consent renewal if it asked for one.
+ `None` for a library voice. On a stream, the value as the stream started."""
+
+ billed_characters: Optional[int] = None
+ """The characters the request bills. On a stream, the value as the stream
+ started: a body cut by a server failure bills nothing."""
+
+ audio_duration_ms: Optional[int] = None
+ """The duration of the audio in milliseconds, on a complete response:
+ `synthesize()`, and a stream whose text had nothing to speak (`0`). A
+ stream with audio does not know it up front: `None`."""
+
+
+class TTSResponse(TTSResponseInfo):
+ """The result of `TextToSpeech.synthesize()`: the whole audio, with its headers."""
+
+ audio: bytes = Field(..., repr=False)
+ """The audio in the requested encoding and container, complete. Its
+ duration is `audio_duration_ms`. Empty (or a header-only WAV) for a text
+ with nothing to speak, such as one made only of emoji, which bills `0`.
+ Left out of `repr()`, so a logged response does not carry the clip."""
diff --git a/assemblyai/types.py b/assemblyai/types.py
index a27768b..f722658 100644
--- a/assemblyai/types.py
+++ b/assemblyai/types.py
@@ -162,6 +162,15 @@ class Settings(BaseSettings):
llm_gateway_http_timeout: float = 30.0
"The HTTP timeout for LLM Gateway requests"
+ tts_base_url: str = "https://tts.us.assemblyai.com"
+ "The base URL for the Text-to-Speech API (used by `TextToSpeech`/`AsyncTextToSpeech`). Set `https://tts.eu.assemblyai.com` to keep requests in the EU data zone."
+
+ tts_http_timeout: float = 30.0
+ "The HTTP timeout for streamed Text-to-Speech requests (`stream()`). Like every httpx timeout it bounds each socket operation (connecting, the wait for the first audio, each read), not the response as a whole. Buffered requests (`synthesize()`) use it to connect and send."
+
+ tts_synthesize_http_timeout: float = 610.0
+ "The read timeout of buffered Text-to-Speech requests (`synthesize()`), whose one read waits for the complete audio. A little over the API's own 600-second limit on a request, so a request that reaches it gets the API's `504 session_expired` as a `TTSError`, not a client timeout."
+
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_tts.py b/tests/unit/test_tts.py
new file mode 100644
index 0000000..0c985cd
--- /dev/null
+++ b/tests/unit/test_tts.py
@@ -0,0 +1,639 @@
+"""Tests for the Text-to-Speech HTTP client.
+
+Every entry point posts one JSON body to `/v1/speech` on the TTS host and
+reads audio back; the response headers carry the metadata.
+"""
+
+import json
+import warnings
+from typing import Iterator
+
+import httpx
+import pytest
+from pytest_httpx import HTTPXMock
+
+import assemblyai as aai
+from assemblyai.tts.v1 import api
+
+aai.settings.api_key = "test"
+
+SPEECH_URL = f"{aai.settings.tts_base_url}/v1/speech"
+
+_AUDIO = b"\x00\x01" * 480 # 20 ms of 24 kHz s16le
+
+_OK_HEADERS = {
+ "content-type": "audio/pcm;rate=24000;channels=1",
+ "x-request-id": "0199c6f1-2d0b-7c3e-9a41-5b6f0e1d2c3a",
+ "x-tts-region": "us-east-1",
+ "x-tts-model": "assemblyai-tts-1",
+ "x-tts-voice-id": "aai-jane",
+ "x-tts-language": "english",
+ "x-tts-encoding": "pcm_s16le",
+ "x-tts-sample-rate": "24000",
+ "x-tts-container": "raw",
+ "x-tts-billed-characters": "12",
+}
+
+_ERROR_BODY = {
+ "code": "invalid_parameter",
+ "message": "sample_rate: sample_rate 12345 not in supported set",
+ "param": "sample_rate",
+ "request_id": "req-1",
+ "errors": [{"message": "not in supported set", "param": "sample_rate"}],
+ "retryable": False,
+ "retry_after_seconds": None,
+ "error": "sample_rate: sample_rate 12345 not in supported set",
+}
+
+
+def _mock_ok(httpx_mock: HTTPXMock, **headers: str) -> None:
+ httpx_mock.add_response(
+ url=SPEECH_URL,
+ method="POST",
+ status_code=httpx.codes.OK,
+ content=_AUDIO,
+ headers={**_OK_HEADERS, **headers},
+ )
+
+
+def _mock_error(httpx_mock: HTTPXMock, status: int, body: dict, **headers: str) -> None:
+ httpx_mock.add_response(
+ url=SPEECH_URL, method="POST", status_code=status, json=body, headers=headers
+ )
+
+
+def _sent_body(httpx_mock: HTTPXMock) -> dict:
+ requests = httpx_mock.get_requests()
+ assert len(requests) == 1
+ return json.loads(requests[0].content)
+
+
+def test_synthesize_posts_a_buffered_request_to_the_tts_host(httpx_mock: HTTPXMock):
+ # GIVEN a TTS host that answers audio
+ _mock_ok(httpx_mock, **{"x-tts-audio-duration-ms": "20"})
+
+ # WHEN a text is synthesized with no options
+ response = aai.TextToSpeech().synthesize("Hello there.")
+
+ # THEN one buffered request went to the TTS host with the raw key
+ request = httpx_mock.get_requests()[0]
+ assert str(request.url) == SPEECH_URL
+ assert request.headers["authorization"] == "test"
+ assert _sent_body(httpx_mock) == {"text": "Hello there.", "stream": False}
+
+ # AND the audio and the headers are on the response
+ assert response.audio == _AUDIO
+ assert response.request_id == "0199c6f1-2d0b-7c3e-9a41-5b6f0e1d2c3a"
+ assert response.region == "us-east-1"
+ assert response.model == "assemblyai-tts-1"
+ assert response.voice_id == "aai-jane"
+ assert response.language == "english"
+ assert response.encoding == "pcm_s16le"
+ assert response.sample_rate == 24000
+ assert response.container == "raw"
+ assert response.bit_rate is None
+ assert response.content_type == "audio/pcm;rate=24000;channels=1"
+ assert response.billed_characters == 12
+ assert response.audio_duration_ms == 20
+
+
+def test_config_fields_go_in_the_body_and_unset_ones_stay_out(httpx_mock: HTTPXMock):
+ # GIVEN a default config on the client and a per-call config
+ _mock_ok(httpx_mock)
+ tts = aai.TextToSpeech(
+ config=aai.TTSConfig(voice_id="aai-alba", language="english")
+ )
+
+ # WHEN a per-call config is given
+ tts.synthesize(
+ "Hola.",
+ aai.TTSConfig(
+ voice_id="aai-juan",
+ model_id="assemblyai-tts-1",
+ sample_rate=8000,
+ encoding=aai.TTSEncoding.pcm_mulaw,
+ container=aai.TTSContainer.wav,
+ ),
+ )
+
+ # THEN it replaces the default, enums serialize to their values
+ assert _sent_body(httpx_mock) == {
+ "text": "Hola.",
+ "voice_id": "aai-juan",
+ "model_id": "assemblyai-tts-1",
+ "sample_rate": 8000,
+ "encoding": "pcm_mulaw",
+ "container": "wav",
+ "stream": False,
+ }
+
+
+def test_default_config_is_used_when_no_per_call_config(httpx_mock: HTTPXMock):
+ _mock_ok(httpx_mock)
+ aai.TextToSpeech(config=aai.TTSConfig(voice_id="aai-alba")).synthesize("Hi.")
+
+ assert _sent_body(httpx_mock) == {
+ "text": "Hi.",
+ "voice_id": "aai-alba",
+ "stream": False,
+ }
+
+
+def test_voice_id_falls_back_to_x_tts_voice_until_the_api_sends_the_id(
+ httpx_mock: HTTPXMock,
+):
+ # GIVEN a server that still sends X-TTS-Voice only
+ legacy = {k: v for k, v in _OK_HEADERS.items() if k != "x-tts-voice-id"}
+ httpx_mock.add_response(
+ url=SPEECH_URL,
+ method="POST",
+ content=_AUDIO,
+ headers={**legacy, "x-tts-voice": "jane"},
+ )
+ httpx_mock.add_response(
+ url=SPEECH_URL,
+ method="POST",
+ content=_AUDIO,
+ headers={**legacy, "x-tts-voice": "custom"},
+ )
+ tts = aai.TextToSpeech()
+
+ # THEN a built-in voice's name becomes its slug, and `custom` the request's voice_id
+ assert tts.synthesize("Hi.").voice_id == "aai-jane"
+ cloned = tts.synthesize("Hi.", aai.TTSConfig(voice_id="aai-isaiah-2"))
+ assert cloned.voice_id == "aai-isaiah-2"
+
+
+def test_stream_yields_audio_as_it_arrives_and_reports_the_headers(
+ httpx_mock: HTTPXMock,
+):
+ # GIVEN a streamed answer
+ _mock_ok(httpx_mock)
+
+ # WHEN a text is streamed
+ with aai.TextToSpeech().stream("Hello there.") as speech:
+ chunks = list(speech)
+
+ # THEN the request asked for a streamed response and the bytes came back
+ assert _sent_body(httpx_mock) == {"text": "Hello there.", "stream": True}
+ assert b"".join(chunks) == _AUDIO
+ assert speech.info.model == "assemblyai-tts-1"
+ assert speech.info.voice_id == "aai-jane"
+ assert speech.info.sample_rate == 24000
+ assert speech.info.audio_duration_ms is None
+
+
+def test_stream_sends_nothing_until_entered_and_raises_there(httpx_mock: HTTPXMock):
+ _mock_error(
+ httpx_mock, 422, {**_ERROR_BODY, "code": "voice_not_found", "param": "voice_id"}
+ )
+
+ speech = aai.TextToSpeech().stream("Hi.", aai.TTSConfig(voice_id="aai-x"))
+ assert httpx_mock.get_requests() == []
+ with pytest.raises(RuntimeError, match="enter or read"):
+ speech.info
+
+ with pytest.raises(aai.TTSError) as exc_info:
+ with speech:
+ pass
+ assert exc_info.value.error_code == "voice_not_found"
+ speech.close() # a never-opened stream closes quietly
+
+
+def test_stream_can_be_iterated_without_a_context_manager(httpx_mock: HTTPXMock):
+ _mock_ok(httpx_mock)
+
+ speech = aai.TextToSpeech().stream("Hi.")
+ chunks = list(speech)
+ speech.close()
+
+ assert b"".join(chunks) == _AUDIO
+ assert speech.info.voice_id == "aai-jane"
+
+
+def test_stream_read_collects_the_whole_audio(httpx_mock: HTTPXMock):
+ _mock_ok(httpx_mock)
+
+ speech = aai.TextToSpeech().stream("Hello there.")
+ try:
+ assert speech.read() == _AUDIO
+ finally:
+ speech.close()
+
+
+class _CutStream(httpx.SyncByteStream):
+ def __init__(self, error: Exception) -> None:
+ self._error = error
+
+ def __iter__(self) -> Iterator[bytes]:
+ yield _AUDIO[:100]
+ raise self._error
+
+
+@pytest.mark.parametrize(
+ "error",
+ [
+ httpx.RemoteProtocolError("peer closed connection"),
+ httpx.ReadError("connection reset"),
+ httpx.ReadTimeout("no audio for 30 s"),
+ ],
+)
+def test_a_body_cut_after_the_first_audio_is_a_retryable_tts_error(error):
+ transport = httpx.MockTransport(
+ lambda request: httpx.Response(
+ 200, headers=_OK_HEADERS, stream=_CutStream(error)
+ )
+ )
+ with httpx.Client(transport=transport) as http:
+ speech = api.stream_speech(
+ http, base_url="https://tts.test", body={"text": "Hi."}, timeout=5.0
+ )
+ with speech, pytest.raises(aai.TTSError) as exc_info:
+ list(speech)
+
+ err = exc_info.value
+ assert err.retryable is True
+ assert err.status_code == 200
+ assert err.error_code is None
+ assert err.request_id == _OK_HEADERS["x-request-id"]
+ assert err.__cause__ is error
+
+
+class _RecordingStream(httpx.SyncByteStream):
+ closed = False
+
+ def __iter__(self) -> Iterator[bytes]:
+ yield _AUDIO[:100]
+ yield _AUDIO[100:]
+
+ def close(self) -> None:
+ self.closed = True
+
+
+def _transport(body: httpx.SyncByteStream, status: int = 200) -> httpx.MockTransport:
+ headers = _OK_HEADERS if status == 200 else {"content-type": "application/json"}
+ return httpx.MockTransport(
+ lambda request: httpx.Response(status, headers=headers, stream=body)
+ )
+
+
+def test_leaving_the_with_block_early_releases_the_connection():
+ body = _RecordingStream()
+ with httpx.Client(transport=_transport(body)) as http:
+ with api.stream_speech(
+ http, base_url="https://tts.test", body={"text": "Hi."}, timeout=5.0
+ ) as speech:
+ next(iter(speech)) # a barge-in after the first chunk
+ assert body.closed
+
+
+def test_close_releases_a_stream_read_outside_a_with_block():
+ body = _RecordingStream()
+ with httpx.Client(transport=_transport(body)) as http:
+ speech = api.stream_speech(
+ http, base_url="https://tts.test", body={"text": "Hi."}, timeout=5.0
+ )
+ speech.read()
+ assert body.closed # a fully read stream is released by httpx itself
+ speech.close() # and a second close is harmless
+
+
+class _JsonBody(httpx.SyncByteStream):
+ closed = False
+
+ def __iter__(self) -> Iterator[bytes]:
+ yield json.dumps({"code": "at_capacity", "message": "No room"}).encode()
+
+ def close(self) -> None:
+ self.closed = True
+
+
+def test_a_refused_stream_releases_its_connection():
+ body = _JsonBody()
+ with httpx.Client(transport=_transport(body, status=503)) as http:
+ speech = api.stream_speech(
+ http, base_url="https://tts.test", body={"text": "Hi."}, timeout=5.0
+ )
+ with pytest.raises(aai.TTSError) as exc_info:
+ speech.read()
+ assert exc_info.value.error_code == "at_capacity"
+ assert body.closed
+
+
+def test_a_client_built_from_api_key_is_closed_with_the_instance():
+ with aai.TextToSpeech(api_key="k") as tts:
+ http = tts.client.http_client
+ assert not http.is_closed
+ assert http.is_closed
+
+ # the shared default client is never closed by an instance
+ shared = aai.TextToSpeech()
+ shared.close()
+ assert not shared.client.http_client.is_closed
+
+
+def test_error_body_is_kept_for_the_fields_some_codes_add(httpx_mock: HTTPXMock):
+ _mock_error(
+ httpx_mock,
+ 429,
+ {
+ "code": "rate_limited",
+ "message": "slow down",
+ "retryable": True,
+ "limit": 100,
+ "limit_window": "minute",
+ },
+ )
+
+ with pytest.raises(aai.TTSError) as exc_info:
+ aai.TextToSpeech().synthesize("Hi.")
+
+ assert exc_info.value.body is not None
+ assert exc_info.value.body["limit"] == 100
+ assert exc_info.value.body["limit_window"] == "minute"
+
+
+def test_config_is_a_native_model_of_the_installed_pydantic():
+ import pydantic
+
+ # On pydantic 2 without pydantic-settings the SDK's shared models run in
+ # v1-compat mode; the TTS models must still be native v2 models.
+ assert issubclass(aai.TTSConfig, pydantic.BaseModel)
+ assert issubclass(aai.TTSResponse, pydantic.BaseModel)
+
+
+def test_defining_the_models_raises_no_pydantic_warnings():
+ import importlib
+
+ from assemblyai.tts.v1 import models
+
+ with warnings.catch_warnings():
+ warnings.simplefilter("error")
+ importlib.reload(models)
+
+
+def test_response_repr_leaves_the_audio_out(httpx_mock: HTTPXMock):
+ _mock_ok(httpx_mock)
+
+ response = aai.TextToSpeech().synthesize("Hi.")
+
+ assert "audio=" not in repr(response)
+ assert "voice_id='aai-jane'" in repr(response)
+ assert response.audio == _AUDIO
+
+
+def test_error_envelope_becomes_a_tts_error(httpx_mock: HTTPXMock):
+ # GIVEN a refusal with the API's JSON envelope
+ _mock_error(httpx_mock, 422, _ERROR_BODY, **{"x-request-id": "req-1"})
+
+ # WHEN the request is made
+ with pytest.raises(aai.TTSError) as exc_info:
+ aai.TextToSpeech().synthesize("Hi.", aai.TTSConfig(sample_rate=12345))
+
+ # THEN every field of the envelope is on the error
+ err = exc_info.value
+ assert isinstance(err, aai.AssemblyAIError)
+ assert str(err) == _ERROR_BODY["message"]
+ assert err.status_code == 422
+ assert err.error_code == "invalid_parameter"
+ assert err.param == "sample_rate"
+ assert err.request_id == "req-1"
+ assert err.retryable is False
+ assert err.retry_after is None
+ assert err.errors == _ERROR_BODY["errors"]
+
+
+def test_stream_raises_the_same_error_before_any_audio(httpx_mock: HTTPXMock):
+ _mock_error(
+ httpx_mock,
+ 429,
+ {
+ **_ERROR_BODY,
+ "code": "rate_limited",
+ "message": "Too many requests",
+ "param": None,
+ "errors": None,
+ "retryable": True,
+ "retry_after_seconds": 17,
+ },
+ **{"retry-after": "17"},
+ )
+
+ with pytest.raises(aai.TTSError) as exc_info:
+ with aai.TextToSpeech().stream("Hi."):
+ pass
+
+ assert exc_info.value.error_code == "rate_limited"
+ assert exc_info.value.retryable is True
+ assert exc_info.value.retry_after == 17
+
+
+def test_biometric_terms_refusal_and_auth_unavailable(httpx_mock: HTTPXMock):
+ _mock_error(
+ httpx_mock,
+ 403,
+ {
+ "code": "biometric_terms_not_accepted",
+ "message": "Accept the biometric data terms in the dashboard.",
+ "param": "voice_id",
+ "retryable": False,
+ },
+ )
+ _mock_error(
+ httpx_mock,
+ 503,
+ {"code": "auth_unavailable", "message": "Try again", "retryable": True},
+ **{"retry-after": "1"},
+ )
+ tts = aai.TextToSpeech()
+
+ with pytest.raises(aai.TTSError) as exc_info:
+ tts.synthesize(
+ "Hi.", aai.TTSConfig(voice_id="aai-x", extend_biometric_consent=True)
+ )
+ assert exc_info.value.status_code == 403
+ assert exc_info.value.error_code == "biometric_terms_not_accepted"
+ assert exc_info.value.retryable is False
+
+ with pytest.raises(aai.TTSError) as exc_info:
+ tts.synthesize("Hi.")
+ assert exc_info.value.error_code == "auth_unavailable"
+ assert exc_info.value.retryable is True
+ assert exc_info.value.retry_after == 1
+
+
+def test_retry_after_header_is_read_when_the_body_has_no_hint(httpx_mock: HTTPXMock):
+ _mock_error(
+ httpx_mock,
+ 503,
+ {"code": "at_capacity", "message": "No room", "retryable": True},
+ **{"retry-after": "2"},
+ )
+
+ with pytest.raises(aai.TTSError) as exc_info:
+ aai.TextToSpeech().synthesize("Hi.")
+
+ assert exc_info.value.status_code == 503
+ assert exc_info.value.error_code == "at_capacity"
+ assert exc_info.value.retry_after == 2
+
+
+def test_fractional_retry_after_is_rounded_up(httpx_mock: HTTPXMock):
+ _mock_error(
+ httpx_mock,
+ 429,
+ {"code": "rate_limited", "message": "slow down", "retry_after_seconds": 1.2},
+ )
+
+ with pytest.raises(aai.TTSError) as exc_info:
+ aai.TextToSpeech().synthesize("Hi.")
+
+ assert exc_info.value.retry_after == 2
+
+
+def test_json_error_without_retryable_falls_back_to_the_status(httpx_mock: HTTPXMock):
+ _mock_error(httpx_mock, 503, {"message": "upstream unavailable"})
+
+ with pytest.raises(aai.TTSError) as exc_info:
+ aai.TextToSpeech().synthesize("Hi.")
+
+ assert exc_info.value.retryable is True
+ assert exc_info.value.error_code is None
+
+
+def test_non_json_error_reports_the_status_and_text(httpx_mock: HTTPXMock):
+ # GIVEN an answer made before the request reached the API
+ httpx_mock.add_response(
+ url=SPEECH_URL,
+ method="POST",
+ status_code=httpx.codes.BAD_GATEWAY,
+ content=b"502 Bad Gateway",
+ headers={"content-type": "text/html"},
+ )
+
+ with pytest.raises(aai.TTSError) as exc_info:
+ aai.TextToSpeech().synthesize("Hi.")
+
+ err = exc_info.value
+ assert err.status_code == 502
+ assert err.error_code is None
+ assert err.retryable is True
+ assert "502" in str(err)
+ assert "Bad Gateway" in str(err)
+
+
+def test_tts_base_url_setting_picks_the_host(httpx_mock: HTTPXMock):
+ # GIVEN the EU host in the settings
+ eu_url = "https://tts.eu.assemblyai.com/v1/speech"
+ httpx_mock.add_response(
+ url=eu_url, method="POST", status_code=200, content=_AUDIO, headers=_OK_HEADERS
+ )
+ client = aai.Client(api_key="test")
+ client.settings.tts_base_url = "https://tts.eu.assemblyai.com"
+
+ # WHEN a request is made with that client
+ aai.TextToSpeech(client=client).synthesize("Hallo.")
+
+ # THEN it went to the EU host
+ assert str(httpx_mock.get_requests()[0].url) == eu_url
+
+
+def test_synthesize_and_stream_use_their_own_timeouts(httpx_mock: HTTPXMock):
+ _mock_ok(httpx_mock)
+ _mock_ok(httpx_mock)
+ client = aai.Client(api_key="test")
+ client.settings.tts_http_timeout = 7.0
+ client.settings.tts_synthesize_http_timeout = 99.0
+ tts = aai.TextToSpeech(client=client)
+
+ tts.synthesize("Hi.")
+ with tts.stream("Hi.") as speech:
+ speech.read()
+
+ timeouts = [r.extensions["timeout"] for r in httpx_mock.get_requests()]
+ # synthesize: the read waits for the whole clip, connecting does not
+ assert (timeouts[0]["read"], timeouts[0]["connect"]) == (99.0, 7.0)
+ assert timeouts[1]["read"] == 7.0
+
+
+def test_the_synthesize_read_timeout_outlasts_the_apis_request_limit():
+ assert aai.Settings().tts_synthesize_http_timeout > 600
+
+
+def test_setting_config_to_none_restores_the_defaults(httpx_mock: HTTPXMock):
+ _mock_ok(httpx_mock)
+ tts = aai.TextToSpeech(config=aai.TTSConfig(voice_id="aai-alba"))
+ tts.config = None # type: ignore[assignment]
+
+ tts.synthesize("Hi.")
+
+ assert _sent_body(httpx_mock) == {"text": "Hi.", "stream": False}
+
+
+def test_voice_expiry_header_is_reported(httpx_mock: HTTPXMock):
+ _mock_ok(
+ httpx_mock,
+ **{"x-tts-voice-id": "aai-x", "x-tts-voice-expires-at": "1791480600"},
+ )
+
+ response = aai.TextToSpeech().synthesize("Hi.", aai.TTSConfig(voice_id="aai-x"))
+
+ assert response.voice_id == "aai-x"
+ assert response.voice_expires_at == 1791480600
+
+
+def test_encoding_and_container_take_the_enum_or_a_string(httpx_mock: HTTPXMock):
+ _mock_ok(httpx_mock)
+ _mock_ok(httpx_mock)
+ tts = aai.TextToSpeech()
+
+ tts.synthesize("Hi.", aai.TTSConfig(encoding="mulaw", container="wav"))
+ tts.synthesize(
+ "Hi.",
+ aai.TTSConfig(
+ encoding=aai.TTSEncoding.pcm_alaw, container=aai.TTSContainer.raw
+ ),
+ )
+
+ bodies = [json.loads(r.content) for r in httpx_mock.get_requests()]
+ assert (bodies[0]["encoding"], bodies[0]["container"]) == ("mulaw", "wav")
+ assert (bodies[1]["encoding"], bodies[1]["container"]) == ("pcm_alaw", "raw")
+
+
+def test_config_rejects_unknown_fields():
+ with pytest.raises(ValueError):
+ aai.TTSConfig(voice_id="aai-jane", speed=2) # type: ignore[call-arg]
+
+
+def test_voice_is_not_a_config_field_and_the_error_names_voice_id():
+ with pytest.raises(ValueError, match="voice_id"):
+ aai.TTSConfig(voice="jane") # type: ignore[call-arg]
+
+
+def test_model_id_needs_no_protected_namespace_warning():
+ with warnings.catch_warnings():
+ warnings.simplefilter("error")
+ aai.TTSConfig(model_id="assemblyai-tts-1")
+
+
+def test_wrong_config_type_is_a_type_error():
+ with pytest.raises(TypeError):
+ aai.TextToSpeech(config=aai.DictationConfig()) # type: ignore[arg-type]
+
+ with pytest.raises(TypeError):
+ aai.TextToSpeech().synthesize("Hi.", aai.SyncTranscriptionConfig()) # type: ignore[arg-type]
+
+
+def test_text_must_be_a_str():
+ with pytest.raises(TypeError):
+ aai.TextToSpeech().synthesize(b"Hi.") # type: ignore[arg-type]
+
+
+def test_swapped_text_and_config_is_named():
+ with pytest.raises(TypeError, match="text comes first"):
+ aai.TextToSpeech().stream(aai.TTSConfig(voice_id="aai-jane"), "Hi.") # type: ignore[arg-type]
+
+
+def test_a_dict_config_is_a_type_error_not_an_attribute_error():
+ with pytest.raises(TypeError, match="TTSConfig"):
+ aai.TextToSpeech().synthesize("Hi.", {"voice_id": "aai-jane"}) # type: ignore[arg-type]
diff --git a/tests/unit/test_tts_async.py b/tests/unit/test_tts_async.py
new file mode 100644
index 0000000..23b3f1e
--- /dev/null
+++ b/tests/unit/test_tts_async.py
@@ -0,0 +1,281 @@
+"""Tests for the asyncio Text-to-Speech client.
+
+The body shape and the error parsing are covered by `test_tts.py`; these pin
+the asyncio twin's behaviour: same route, same audio and headers, same errors,
+plus pool ownership.
+"""
+
+import json
+from typing import AsyncIterator
+
+import httpx
+import pytest
+from pytest_httpx import HTTPXMock
+
+import assemblyai as aai
+from assemblyai.tts.v1 import async_api
+
+pytestmark = pytest.mark.asyncio
+
+aai.settings.api_key = "test"
+
+SPEECH_URL = f"{aai.settings.tts_base_url}/v1/speech"
+
+_AUDIO = b"\x00\x01" * 480
+
+_OK_HEADERS = {
+ "content-type": "audio/pcm;rate=24000;channels=1",
+ "x-request-id": "req-async",
+ "x-tts-region": "us-east-1",
+ "x-tts-model": "assemblyai-tts-1",
+ "x-tts-voice-id": "aai-jane",
+ "x-tts-language": "english",
+ "x-tts-encoding": "pcm_s16le",
+ "x-tts-sample-rate": "24000",
+ "x-tts-container": "raw",
+ "x-tts-billed-characters": "12",
+}
+
+
+def _mock_ok(httpx_mock: HTTPXMock, **headers: str) -> None:
+ httpx_mock.add_response(
+ url=SPEECH_URL,
+ method="POST",
+ status_code=httpx.codes.OK,
+ content=_AUDIO,
+ headers={**_OK_HEADERS, **headers},
+ )
+
+
+def _sent_body(httpx_mock: HTTPXMock) -> dict:
+ requests = httpx_mock.get_requests()
+ assert len(requests) == 1
+ return json.loads(requests[0].content)
+
+
+async def test_synthesize_matches_the_threaded_twin(httpx_mock: HTTPXMock):
+ _mock_ok(httpx_mock, **{"x-tts-audio-duration-ms": "20"})
+
+ async with aai.AsyncTextToSpeech(config=aai.TTSConfig(voice_id="aai-alba")) as tts:
+ response = await tts.synthesize("Hello there.")
+
+ request = httpx_mock.get_requests()[0]
+ assert str(request.url) == SPEECH_URL
+ assert request.headers["authorization"] == "test"
+ assert _sent_body(httpx_mock) == {
+ "text": "Hello there.",
+ "voice_id": "aai-alba",
+ "stream": False,
+ }
+ assert response.audio == _AUDIO
+ assert response.request_id == "req-async"
+ assert response.region == "us-east-1"
+ assert response.voice_id == "aai-jane"
+ assert response.sample_rate == 24000
+ assert response.audio_duration_ms == 20
+
+
+async def test_stream_yields_audio_and_reports_the_headers(httpx_mock: HTTPXMock):
+ _mock_ok(httpx_mock)
+
+ async with aai.AsyncTextToSpeech() as tts:
+ speech = tts.stream("Hello there.")
+ assert httpx_mock.get_requests() == []
+ async with speech:
+ chunks = [chunk async for chunk in speech]
+
+ assert _sent_body(httpx_mock) == {"text": "Hello there.", "stream": True}
+ assert b"".join(chunks) == _AUDIO
+ assert speech.info.model == "assemblyai-tts-1"
+
+
+async def test_stream_aread_collects_the_whole_audio(httpx_mock: HTTPXMock):
+ _mock_ok(httpx_mock)
+
+ async with aai.AsyncTextToSpeech() as tts:
+ speech = tts.stream("Hello there.")
+ try:
+ assert await speech.aread() == _AUDIO
+ finally:
+ await speech.aclose()
+
+
+class _CutStream(httpx.AsyncByteStream):
+ def __init__(self, error: Exception) -> None:
+ self._error = error
+
+ async def __aiter__(self) -> AsyncIterator[bytes]:
+ yield _AUDIO[:100]
+ raise self._error
+
+
+@pytest.mark.parametrize(
+ "error",
+ [
+ httpx.RemoteProtocolError("peer closed connection"),
+ httpx.ReadError("connection reset"),
+ httpx.ReadTimeout("no audio for 30 s"),
+ ],
+)
+async def test_a_body_cut_after_the_first_audio_is_a_retryable_tts_error(error):
+ transport = httpx.MockTransport(
+ lambda request: httpx.Response(
+ 200, headers=_OK_HEADERS, stream=_CutStream(error)
+ )
+ )
+ async with httpx.AsyncClient(transport=transport) as http:
+ speech = async_api.stream_speech(
+ http, base_url="https://tts.test", body={"text": "Hi."}, timeout=5.0
+ )
+ async with speech:
+ with pytest.raises(aai.TTSError) as exc_info:
+ await speech.aread()
+
+ assert exc_info.value.retryable is True
+ assert exc_info.value.status_code == 200
+ assert exc_info.value.request_id == "req-async"
+ assert exc_info.value.__cause__ is error
+
+
+class _RecordingStream(httpx.AsyncByteStream):
+ closed = False
+
+ async def __aiter__(self) -> AsyncIterator[bytes]:
+ yield _AUDIO[:100]
+ yield _AUDIO[100:]
+
+ async def aclose(self) -> None:
+ self.closed = True
+
+
+def _transport(body: httpx.AsyncByteStream, status: int = 200) -> httpx.MockTransport:
+ headers = _OK_HEADERS if status == 200 else {"content-type": "application/json"}
+ return httpx.MockTransport(
+ lambda request: httpx.Response(status, headers=headers, stream=body)
+ )
+
+
+async def test_leaving_the_async_with_block_early_releases_the_connection():
+ body = _RecordingStream()
+ async with httpx.AsyncClient(transport=_transport(body)) as http:
+ async with async_api.stream_speech(
+ http, base_url="https://tts.test", body={"text": "Hi."}, timeout=5.0
+ ) as speech:
+ await speech.aiter_bytes().__anext__() # a barge-in after the first chunk
+ assert body.closed
+
+
+class _JsonBody(httpx.AsyncByteStream):
+ closed = False
+
+ async def __aiter__(self) -> AsyncIterator[bytes]:
+ yield json.dumps({"code": "at_capacity", "message": "No room"}).encode()
+
+ async def aclose(self) -> None:
+ self.closed = True
+
+
+async def test_a_refused_stream_releases_its_connection():
+ body = _JsonBody()
+ async with httpx.AsyncClient(transport=_transport(body, status=503)) as http:
+ speech = async_api.stream_speech(
+ http, base_url="https://tts.test", body={"text": "Hi."}, timeout=5.0
+ )
+ with pytest.raises(aai.TTSError) as exc_info:
+ await speech.aread()
+ assert exc_info.value.error_code == "at_capacity"
+ assert body.closed
+
+
+async def test_an_owned_pool_is_closed_on_aclose():
+ tts = aai.AsyncTextToSpeech(api_key="k")
+ http = tts.client.http_client
+ await tts.aclose()
+ assert http.is_closed
+
+
+async def test_errors_are_tts_errors(httpx_mock: HTTPXMock):
+ httpx_mock.add_response(
+ url=SPEECH_URL,
+ method="POST",
+ status_code=httpx.codes.UNPROCESSABLE_ENTITY,
+ json={
+ "code": "voice_not_found",
+ "message": "Voice not found",
+ "param": "voice_id",
+ "request_id": "req-2",
+ "errors": None,
+ "retryable": False,
+ "retry_after_seconds": None,
+ },
+ )
+ httpx_mock.add_response(
+ url=SPEECH_URL,
+ method="POST",
+ status_code=httpx.codes.SERVICE_UNAVAILABLE,
+ json={"code": "at_capacity", "message": "No room", "retryable": True},
+ headers={"retry-after": "3"},
+ )
+ httpx_mock.add_response(
+ url=SPEECH_URL,
+ method="POST",
+ status_code=httpx.codes.FORBIDDEN,
+ json={"code": "biometric_terms_not_accepted", "message": "Accept the terms"},
+ )
+
+ async with aai.AsyncTextToSpeech() as tts:
+ with pytest.raises(aai.TTSError) as exc_info:
+ await tts.synthesize("Hi.", aai.TTSConfig(voice_id="nope"))
+ assert exc_info.value.error_code == "voice_not_found"
+ assert exc_info.value.param == "voice_id"
+ assert exc_info.value.request_id == "req-2"
+
+ with pytest.raises(aai.TTSError) as exc_info:
+ async with tts.stream("Hi."):
+ pass
+ assert exc_info.value.error_code == "at_capacity"
+ assert exc_info.value.retry_after == 3
+
+ with pytest.raises(aai.TTSError) as exc_info:
+ await tts.synthesize("Hi.", aai.TTSConfig(voice_id="aai-x"))
+ assert exc_info.value.status_code == 403
+ assert exc_info.value.error_code == "biometric_terms_not_accepted"
+ assert exc_info.value.retryable is False
+
+
+async def test_synthesize_and_stream_use_their_own_timeouts(httpx_mock: HTTPXMock):
+ _mock_ok(httpx_mock)
+ _mock_ok(httpx_mock)
+ client = aai.AsyncClient(api_key="test")
+ client.settings.tts_http_timeout = 7.0
+ client.settings.tts_synthesize_http_timeout = 99.0
+
+ async with aai.AsyncTextToSpeech(client=client) as tts:
+ await tts.synthesize("Hi.")
+ async with tts.stream("Hi.") as speech:
+ await speech.aread()
+ await client.aclose()
+
+ timeouts = [r.extensions["timeout"] for r in httpx_mock.get_requests()]
+ assert (timeouts[0]["read"], timeouts[0]["connect"]) == (99.0, 7.0)
+ assert timeouts[1]["read"] == 7.0
+
+
+async def test_a_shared_client_is_left_open_on_aclose(httpx_mock: HTTPXMock):
+ _mock_ok(httpx_mock)
+ _mock_ok(httpx_mock)
+
+ client = aai.AsyncClient(api_key="test")
+ async with aai.AsyncTextToSpeech(client=client) as tts:
+ await tts.synthesize("One.")
+
+ # The pool is still usable after the instance closed
+ await aai.AsyncTextToSpeech(client=client).synthesize("Two.")
+ await client.aclose()
+
+ assert len(httpx_mock.get_requests()) == 2
+
+
+async def test_wrong_config_type_is_a_type_error():
+ with pytest.raises(TypeError):
+ aai.AsyncTextToSpeech(config=aai.DictationConfig()) # type: ignore[arg-type]