From 04e004b067af3cd5a205a83ea32537c178582d4e Mon Sep 17 00:00:00 2001 From: Aleksandar Mitov <140423361+aleks-mitov@users.noreply.github.com> Date: Sat, 24 Oct 2026 23:53:35 +0000 Subject: [PATCH 1/7] feat(tts): share one TLS context across session connects Every TTSSession and AsyncTTSSession connect built a new default TLS context, which loads and parses the CA certificates (about 15 ms with a CA file, as on Debian and Ubuntu system Pythons, and more on a busy host) before the first byte. Sessions now take one process-wide context (_shared_tls_context(), the streaming client's write-first context built on first use), so only the first connect pays for it. The streaming STT client is unchanged. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01KcyY6MAEUVjqBt5pMpetvD --- assemblyai/streaming/v3/_tls.py | 33 +++++++ assemblyai/tts/v1/async_session.py | 20 ++++- assemblyai/tts/v1/session.py | 18 +++- tests/unit/test_streaming_tls.py | 129 +++++++++++++++++++++++++++ tests/unit/test_tts_session.py | 58 +++++++++++- tests/unit/test_tts_session_async.py | 57 +++++++++++- 6 files changed, 305 insertions(+), 10 deletions(-) diff --git a/assemblyai/streaming/v3/_tls.py b/assemblyai/streaming/v3/_tls.py index 252fcdf..b7825a4 100644 --- a/assemblyai/streaming/v3/_tls.py +++ b/assemblyai/streaming/v3/_tls.py @@ -101,3 +101,36 @@ def _create_tls_context() -> ssl.SSLContext: if issubclass(_WriteFirstSSLSocket, context.sslsocket_class): context.sslsocket_class = _WriteFirstSSLSocket return context + + +# The context `_shared_tls_context()` returns once it has been built. +_shared_context: Optional[ssl.SSLContext] = None +_SHARED_CONTEXT_LOCK = threading.Lock() + + +def _shared_tls_context() -> ssl.SSLContext: + """Return one ``_create_tls_context()`` for the whole process. + + Building a context loads and parses the trusted CA certificates. Where + Python reads them from a CA file, as the system Pythons of Debian and + Ubuntu do, that takes about 15 ms on an idle machine and 20 to 50 ms on a + busy one, and a connect that builds its own context pays it before its + first byte. Connects that share this context pay it once per process. + + One context serves any number of connections, threads and event loops; + each connection keeps its own TLS state. Sync connects get + ``_WriteFirstSSLSocket`` sockets from it. asyncio wraps its sockets with + ``wrap_bio()``, which does not use the socket class. + + The context is built on the first call, so the CA certificates and the + ``ssl`` module it reflects (``SSL_CERT_FILE``, ``SSL_CERT_DIR``, a gevent + patch) are those in place at that time. + """ + global _shared_context + context = _shared_context + if context is None: + with _SHARED_CONTEXT_LOCK: + context = _shared_context + if context is None: + context = _shared_context = _create_tls_context() + return context diff --git a/assemblyai/tts/v1/async_session.py b/assemblyai/tts/v1/async_session.py index 094502a..d0b399f 100644 --- a/assemblyai/tts/v1/async_session.py +++ b/assemblyai/tts/v1/async_session.py @@ -2,6 +2,7 @@ import json import logging from typing import Any, AsyncIterator, Callable, Dict, List, Optional, Union +from urllib.parse import urlsplit from ... import _audio from ... import async_client as _async_client @@ -37,12 +38,23 @@ def websocket_connect_async( uri: str, additional_headers: Dict[str, Optional[str]] ) -> Any: - """Opens an asyncio websocket; the streaming client's seam, imported - lazily so `import assemblyai` does not load websockets. Tests patch this.""" + """Opens an asyncio websocket with whichever websockets client the + streaming client uses, importing it lazily so `import assemblyai` does + not load websockets. Tests patch this. + + `wss://` connects use the process's shared TLS context + (`_shared_tls_context()`), built by the first connect. Given no context, + asyncio would build a new one, loading the CA certificates, for every + connection. + """ - from ...streaming.v3.async_client import websocket_connect_async as _connect + from ...streaming.v3._tls import _shared_tls_context + from ...streaming.v3.async_client import _WS_HEADER_KW, _ws_connect - return _connect(uri, additional_headers=additional_headers) + kwargs: Dict[str, Any] = {_WS_HEADER_KW: additional_headers} + if urlsplit(uri).scheme == "wss": + kwargs["ssl"] = _shared_tls_context() + return _ws_connect(uri, **kwargs) class AsyncTTSSession: diff --git a/assemblyai/tts/v1/session.py b/assemblyai/tts/v1/session.py index 0251ddb..03adb54 100644 --- a/assemblyai/tts/v1/session.py +++ b/assemblyai/tts/v1/session.py @@ -2,6 +2,7 @@ import logging import threading from typing import Any, Callable, Dict, Iterator, List, Optional, Union +from urllib.parse import urlsplit from ... import _audio from ... import client as _client @@ -35,20 +36,29 @@ def websocket_connect(uri: str, **kwargs: Any) -> Any: - """Opens a sync websocket; the streaming client's seam, imported lazily so - `import assemblyai` does not load websockets. Tests patch this name.""" + """Opens a sync websocket, importing websockets lazily so `import + assemblyai` does not load it. Tests patch this name. + + `wss://` connects use the process's shared TLS context + (`_shared_tls_context()`): the streaming client's write-first context, + built by the first connect, so later connects skip loading the CA + certificates. + """ import inspect from websockets.sync.client import connect as _ws_connect - from ...streaming.v3.client import websocket_connect as _connect + from ...streaming.v3._tls import _shared_tls_context + from ...streaming.v3.client import _WS_SSL_KW # websockets 17 deprecates a bare connect() unless `legacy=True` (older # releases have no such parameter). if "legacy" in inspect.signature(_ws_connect).parameters: kwargs["legacy"] = True - return _connect(uri, **kwargs) + if urlsplit(uri).scheme == "wss": + kwargs[_WS_SSL_KW] = _shared_tls_context() + return _ws_connect(uri, **kwargs) class TTSSession: diff --git a/tests/unit/test_streaming_tls.py b/tests/unit/test_streaming_tls.py index 6a5b452..48f956c 100644 --- a/tests/unit/test_streaming_tls.py +++ b/tests/unit/test_streaming_tls.py @@ -1,3 +1,4 @@ +import asyncio import contextlib import inspect import shutil @@ -5,6 +6,7 @@ import ssl import subprocess import threading +import time import warnings from typing import Iterator, List, Optional, Tuple @@ -21,6 +23,10 @@ _tls, ) from assemblyai.streaming.v3.client import websocket_connect +from assemblyai.tts.v1.async_session import ( + websocket_connect_async as tts_websocket_connect_async, +) +from assemblyai.tts.v1.session import websocket_connect as tts_websocket_connect # websockets 13 renamed the sync ``connect()`` / ``serve()`` TLS parameter # from ``ssl_context`` to ``ssl``. @@ -341,3 +347,126 @@ def test_client_connect_passes_no_tls_context_for_ws(mocker: MockFixture): [(uri, kwargs)] = calls assert uri.startswith("ws://localhost:8080/v3/ws?") assert not {"ssl", "ssl_context"} & kwargs.keys() + + +def test_shared_tls_context_is_built_once(monkeypatch: pytest.MonkeyPatch): + # Given: no shared context built yet + monkeypatch.setattr(_tls, "_shared_context", None) + built: List[ssl.SSLContext] = [] + create_tls_context = _tls._create_tls_context + + def counting_create_tls_context() -> ssl.SSLContext: + context = create_tls_context() + built.append(context) + return context + + monkeypatch.setattr(_tls, "_create_tls_context", counting_create_tls_context) + + # When: asking for it twice + first = _tls._shared_tls_context() + second = _tls._shared_tls_context() + + # Then: one context was built and both calls return it, with the SDK + # context's socket class and verification + assert built == [first] + assert second is first + assert first.sslsocket_class is _tls._WriteFirstSSLSocket + assert first.verify_mode == ssl.CERT_REQUIRED + assert first.check_hostname is True + + +def test_shared_tls_context_is_built_once_by_concurrent_first_callers( + monkeypatch: pytest.MonkeyPatch, +): + # Given: no shared context yet, and a build slow enough for every thread + # to ask before it finishes + monkeypatch.setattr(_tls, "_shared_context", None) + built: List[ssl.SSLContext] = [] + create_tls_context = _tls._create_tls_context + + def slow_create_tls_context() -> ssl.SSLContext: + time.sleep(0.05) + context = create_tls_context() + built.append(context) + return context + + monkeypatch.setattr(_tls, "_create_tls_context", slow_create_tls_context) + callers = 8 + start = threading.Barrier(callers) + got: List[ssl.SSLContext] = [] + + def call() -> None: + start.wait() + got.append(_tls._shared_tls_context()) + + # When: several threads ask for it at once + threads = [threading.Thread(target=call) for _ in range(callers)] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=10) + + # Then: one context was built, and every thread got it + assert len(built) == 1 + assert len(got) == callers + assert all(context is built[0] for context in got) + + +@pytest.fixture +def trusting_shared_context( + monkeypatch: pytest.MonkeyPatch, client_context: ssl.SSLContext +) -> ssl.SSLContext: + """The SDK context trusting the test certificate, as the process's shared + context.""" + monkeypatch.setattr(_tls, "_shared_context", client_context) + return client_context + + +def test_tts_sessions_share_one_context_over_wss( + server_context: ssl.SSLContext, trusting_shared_context: ssl.SSLContext +): + # Given: a TLS 1.3 websockets echo server + with _websocket_server(server_context) as port: + for _ in range(2): + # When: a Text-to-Speech session connects over wss:// + connection = tts_websocket_connect( + f"wss://{_HOST}:{port}/v1/ws", open_timeout=10 + ) + + # Then: the handshake completes over a write-first socket of the + # shared context, on every connect + with connection: + assert isinstance(connection.socket, _tls._WriteFirstSSLSocket) + assert connection.socket.context is trusting_shared_context + assert connection.socket.version() == "TLSv1.3" + connection.send("ping") + assert connection.recv(timeout=10) == "ping" + + +def test_async_tts_sessions_share_one_context_over_wss( + server_context: ssl.SSLContext, trusting_shared_context: ssl.SSLContext +): + # Given: a TLS 1.3 websockets echo server + async def connect_twice(port: int) -> List[Tuple[object, Optional[str], str]]: + seen = [] + for _ in range(2): + # When: an asyncio Text-to-Speech session connects over wss:// + connection = await asyncio.wait_for( + tts_websocket_connect_async(f"wss://{_HOST}:{port}/v1/ws", {}), + timeout=10, + ) + try: + ssl_object = connection.transport.get_extra_info("ssl_object") + await connection.send("ping") + echoed = await asyncio.wait_for(connection.recv(), timeout=10) + seen.append((ssl_object.context, ssl_object.version(), echoed)) + finally: + await connection.close() + return seen + + with _websocket_server(server_context) as port: + seen = asyncio.run(connect_twice(port)) + + # Then: both handshakes completed with the shared context, which asyncio + # uses through wrap_bio() whatever its socket class + assert seen == [(trusting_shared_context, "TLSv1.3", "ping")] * 2 diff --git a/tests/unit/test_tts_session.py b/tests/unit/test_tts_session.py index 5d9985c..bbb804a 100644 --- a/tests/unit/test_tts_session.py +++ b/tests/unit/test_tts_session.py @@ -8,10 +8,11 @@ import base64 import json import queue +import ssl import threading import time from types import SimpleNamespace -from typing import Any, Iterable, List, Optional +from typing import Any, Iterable, List, Optional, Tuple from urllib.parse import parse_qs, urlsplit import pytest @@ -19,6 +20,8 @@ from websockets.frames import Close import assemblyai as aai +from assemblyai.streaming.v3 import _tls +from assemblyai.streaming.v3.client import _WS_SSL_KW aai.settings.api_key = "test" @@ -1017,3 +1020,56 @@ def test_cancel_ends_a_reader_following_the_request_being_written(server): def test_the_first_request_is_not_dropped_by_the_initial_watermark(server): with aai.TTSSession() as session: assert b"".join(session.speak("Zero.")) == b"Zero." + + +def _record_websockets_connect(mocker) -> List[Tuple[str, dict]]: + """Replaces websockets' sync connect, under the session's seam, with a + recorder that answers every connect with a new `_FakeServer`.""" + + calls: List[Tuple[str, dict]] = [] + + def connect(uri: str, **kwargs: Any) -> _FakeServer: + calls.append((uri, kwargs)) + return _FakeServer() + + mocker.patch("websockets.sync.client.connect", new=connect) + return calls + + +def test_sessions_over_wss_share_the_process_tls_context(mocker): + # Given: websockets' connect recorded, and the default https:// host + calls = _record_websockets_connect(mocker) + + # When: two sessions connect one after the other + for text in ("One.", "Two."): + with aai.TTSSession() as session: + assert b"".join(session.speak(text)) == text.encode() + + # Then: both connected over wss:// with the one shared context, passed + # under the installed websockets' TLS parameter name + assert len(calls) == 2 + shared = _tls._shared_tls_context() + for uri, kwargs in calls: + assert uri.startswith("wss://tts.us.assemblyai.com/v1/ws?") + assert {"ssl", "ssl_context"} & kwargs.keys() == {_WS_SSL_KW} + assert kwargs[_WS_SSL_KW] is shared + assert kwargs["additional_headers"]["Authorization"] == "test" + assert isinstance(shared, ssl.SSLContext) + assert shared.sslsocket_class is _tls._WriteFirstSSLSocket + assert shared.verify_mode == ssl.CERT_REQUIRED + assert shared.check_hostname is True + + +def test_a_session_over_ws_passes_no_tls_context(mocker, monkeypatch): + # Given: a plain http:// host, which a session reaches over ws:// + monkeypatch.setattr(aai.settings, "tts_base_url", "http://localhost:8080") + calls = _record_websockets_connect(mocker) + + # When: a session connects + with aai.TTSSession(): + pass + + # Then: no TLS argument is passed, which websockets rejects for ws:// + [(uri, kwargs)] = calls + assert uri.startswith("ws://localhost:8080/v1/ws?") + assert not {"ssl", "ssl_context"} & kwargs.keys() diff --git a/tests/unit/test_tts_session_async.py b/tests/unit/test_tts_session_async.py index 40d9638..e2da409 100644 --- a/tests/unit/test_tts_session_async.py +++ b/tests/unit/test_tts_session_async.py @@ -3,8 +3,9 @@ import asyncio import base64 import json +import ssl from types import SimpleNamespace -from typing import Any, AsyncIterable, List, Optional +from typing import Any, AsyncIterable, List, Optional, Tuple from urllib.parse import parse_qs, urlsplit import pytest @@ -13,6 +14,8 @@ from websockets.frames import Close import assemblyai as aai +from assemblyai.streaming.v3 import _tls +from assemblyai.streaming.v3.async_client import _WS_HEADER_KW pytestmark = pytest.mark.asyncio @@ -699,3 +702,55 @@ async def test_cancel_ends_a_reader_following_the_request_being_written(server): async def _collect(iterator): return [chunk async for chunk in iterator] + + +def _record_websockets_connect(mocker) -> List[Tuple[str, dict]]: + """Replaces the websockets asyncio connect, under the session's seam, with + a recorder that answers every connect with a new `_FakeAsyncServer`.""" + + calls: List[Tuple[str, dict]] = [] + + async def connect(uri: str, **kwargs: Any) -> _FakeAsyncServer: + calls.append((uri, kwargs)) + return _FakeAsyncServer() + + mocker.patch("assemblyai.streaming.v3.async_client._ws_connect", new=connect) + return calls + + +async def test_sessions_over_wss_share_the_process_tls_context(mocker): + # Given: websockets' connect recorded, and the default https:// host + calls = _record_websockets_connect(mocker) + + # When: two sessions connect one after the other + for text in ("One.", "Two."): + async with aai.AsyncTTSSession() as session: + got = b"".join([c async for c in session.speak(text)]) + assert got == text.encode() + + # Then: both connected over wss:// with the one shared context, which + # spares asyncio building a new default context per connection + assert len(calls) == 2 + shared = _tls._shared_tls_context() + for uri, kwargs in calls: + assert uri.startswith("wss://tts.us.assemblyai.com/v1/ws?") + assert kwargs["ssl"] is shared + assert kwargs[_WS_HEADER_KW]["Authorization"] == "test" + assert isinstance(shared, ssl.SSLContext) + assert shared.verify_mode == ssl.CERT_REQUIRED + assert shared.check_hostname is True + + +async def test_a_session_over_ws_passes_no_tls_context(mocker, monkeypatch): + # Given: a plain http:// host, which a session reaches over ws:// + monkeypatch.setattr(aai.settings, "tts_base_url", "http://localhost:8080") + calls = _record_websockets_connect(mocker) + + # When: a session connects + async with aai.AsyncTTSSession(): + pass + + # Then: no TLS argument is passed, which websockets rejects for ws:// + [(uri, kwargs)] = calls + assert uri.startswith("ws://localhost:8080/v1/ws?") + assert "ssl" not in kwargs From 7a5ba6a423ff1ecfbbe75cb8528837bd6c013bbb Mon Sep 17 00:00:00 2001 From: Aleksandar Mitov <140423361+aleks-mitov@users.noreply.github.com> Date: Sat, 24 Oct 2026 23:53:35 +0000 Subject: [PATCH 2/7] feat(tts): keep idle Text-to-Speech HTTP connections for 120 s TextToSpeech and AsyncTextToSpeech send their requests through a pool of their own (Client.tts_http_client / AsyncClient.tts_http_client), built on first use, whose idle connections stay settings.tts_keepalive_expiry seconds (120 by default, HTTP/1.1). A voice agent's next sentence a few seconds after the previous one reuses its connection instead of paying the TCP and TLS handshakes again under httpx's 5 s default. The pool of the other APIs and settings.keepalive_expiry are unchanged, and no request is retried. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01KcyY6MAEUVjqBt5pMpetvD --- CLAUDE.md | 2 +- README.md | 3 +- assemblyai/__version__.py | 2 +- assemblyai/async_client.py | 55 ++++++++- assemblyai/client.py | 66 ++++++++++- assemblyai/tts/v1/async_client.py | 4 +- assemblyai/tts/v1/client.py | 15 ++- assemblyai/types.py | 3 + tests/unit/test_tts.py | 178 +++++++++++++++++++++++++++++- tests/unit/test_tts_async.py | 52 +++++++++ 10 files changed, 364 insertions(+), 16 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index ede6f2a..946d4cc 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -52,7 +52,7 @@ 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.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). Requests use the client's own TTS pool (`client.tts_http_client`), which keeps idle connections `aai.settings.tts_keepalive_expiry` seconds (120), so consecutive sentences reuse one connection - `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.TTSSession` — Text-to-Speech WebSocket session: one connection, text in as it is written, audio out. `with` connects and waits for `Begin` (`session.begin`); `speak(text)` yields one request's audio; `generate(text)` / `flush()` -> flush_id / `audio(flush_id)` for streamed-in text and pipelining (`audio()` with no id follows the request being written, from another thread); `cancel()` for barge-in (ends the iteration at once); `keep_alive()`; `on(TTSEvents.X, handler(event))` for `Begin`, `Audio` (switches audio from pull to push), `FlushDone`, `WordBoundaries`, `Cancelled`, `Termination`, `Error`. Exit sends `Terminate` and sets `session.termination` (`None` after a failure). Refusals and failures raise `TTSError` (`status_code` None, `request_id` = session id). Raw PCM/G.711 only; 2,000 chars per generate unless `Begin` reports another `limits.max_generate_text_length`; ~100 chars/s sustained. `token=` for temporary tokens - `aai.AsyncTTSSession` — Asyncio counterpart: `async with`, coroutine sends, `async for chunk in session.speak(text)` diff --git a/README.md b/README.md index c90f9cf..ada0066 100644 --- a/README.md +++ b/README.md @@ -894,7 +894,7 @@ except aai.DictationError as error: `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. +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. They share a connection pool of their own that keeps an idle connection for `aai.settings.tts_keepalive_expiry` seconds (120 by default), so a request that follows the previous one within that time, such as a voice agent's next sentence, reuses its connection instead of opening a new one.
Synthesize a WAV file @@ -2042,6 +2042,7 @@ aai.settings.dictation_http_timeout = 300.0 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 +aai.settings.tts_keepalive_expiry = 120.0 # seconds an idle connection stays in the TTS pool ``` 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`. diff --git a/assemblyai/__version__.py b/assemblyai/__version__.py index 320141d..0a0a43a 100644 --- a/assemblyai/__version__.py +++ b/assemblyai/__version__.py @@ -1 +1 @@ -__version__ = "1.8.2" +__version__ = "1.9.0" diff --git a/assemblyai/async_client.py b/assemblyai/async_client.py index a0acf4e..cd7c3e6 100644 --- a/assemblyai/async_client.py +++ b/assemblyai/async_client.py @@ -5,7 +5,13 @@ from typing_extensions import Self from . import types -from .client import _MISSING_API_KEY_ERROR, _build_headers, _build_limits +from .client import ( + _CLOSED_CLIENT_ERROR, + _MISSING_API_KEY_ERROR, + _build_headers, + _build_limits, + _build_tts_limits, +) class AsyncClient: @@ -62,14 +68,20 @@ def __init__( async def _store_response(response: httpx.Response) -> None: self._last_response = response + self._event_hooks = {"response": [_store_response]} self._http_client = httpx.AsyncClient( base_url=self._settings.base_url, headers=_build_headers(self._settings), timeout=self._settings.http_timeout, limits=_build_limits(self._settings), - event_hooks={"response": [_store_response]}, + event_hooks=self._event_hooks, ) + # Built by `tts_http_client` on first use, so a client that never + # speaks pays nothing for it. + self._tts_http_client: Optional[httpx.AsyncClient] = None + self._closed = False + @property def last_response(self) -> Optional[httpx.Response]: """ @@ -102,10 +114,43 @@ def http_client(self) -> httpx.AsyncClient: return self._http_client - async def aclose(self) -> None: - """Closes the underlying HTTP connection pool.""" + @property + def tts_http_client(self) -> httpx.AsyncClient: + """ + The HTTP client of the Text-to-Speech API, built on first use. + + A connection pool of its own, whose idle connections stay + `settings.tts_keepalive_expiry` seconds (120 by default); see + `Client.tts_http_client`. `aclose()` closes it with `http_client`. - await self._http_client.aclose() + Raises: + RuntimeError: if the client was closed before the pool was ever + built. + """ + + client = self._tts_http_client + if client is None: + if self._closed: + raise RuntimeError(_CLOSED_CLIENT_ERROR) + client = self._tts_http_client = httpx.AsyncClient( + base_url=self._settings.tts_base_url, + headers=_build_headers(self._settings), + timeout=self._settings.tts_http_timeout, + limits=_build_tts_limits(self._settings), + event_hooks=self._event_hooks, + ) + + return client + + async def aclose(self) -> None: + """Closes the underlying HTTP connection pools.""" + + self._closed = True + try: + await self._http_client.aclose() + finally: + if self._tts_http_client is not None: + await self._tts_http_client.aclose() async def __aenter__(self) -> Self: return self diff --git a/assemblyai/client.py b/assemblyai/client.py index 2d180d8..169bf1f 100644 --- a/assemblyai/client.py +++ b/assemblyai/client.py @@ -36,6 +36,21 @@ def _build_limits(settings: types.Settings) -> httpx.Limits: ) +def _build_tts_limits(settings: types.Settings) -> httpx.Limits: + """ + Builds the limits of the Text-to-Speech pool: no cap on connections, as + from `_build_limits`, and idle connections kept + `settings.tts_keepalive_expiry` seconds. Shared with `AsyncClient`. + """ + + return httpx.Limits(keepalive_expiry=settings.tts_keepalive_expiry) + + +# What httpx raises from a request on a closed client, raised as well when a +# client closed before it built its Text-to-Speech pool is asked for that pool. +_CLOSED_CLIENT_ERROR = "Cannot send a request, as the client has been closed." + + _MISSING_API_KEY_ERROR = ( "Please provide an API key: set the ASSEMBLYAI_API_KEY environment variable, " "set aai.settings.api_key, or pass api_key= to the transcriber or client." @@ -79,14 +94,21 @@ def __init__( def _store_response(response): self._last_response = response + self._event_hooks = {"response": [_store_response]} self._http_client = httpx.Client( base_url=self.settings.base_url, headers=_build_headers(self._settings), timeout=self.settings.http_timeout, limits=_build_limits(self._settings), - event_hooks={"response": [_store_response]}, + event_hooks=self._event_hooks, ) + # Built by `tts_http_client` on first use, so a client that never + # speaks pays nothing for it. + self._tts_http_client: Optional[httpx.Client] = None + self._tts_http_client_closed = False + self._tts_http_client_lock = threading.Lock() + @property def last_response(self) -> Optional[httpx.Response]: """ @@ -119,6 +141,48 @@ def http_client(self) -> httpx.Client: return self._http_client + @property + def tts_http_client(self) -> httpx.Client: + """ + The HTTP client of the Text-to-Speech API, built on first use. + + A connection pool of its own, whose idle connections stay + `settings.tts_keepalive_expiry` seconds (120 by default): a voice + agent's next sentence reuses the connection of the previous one + instead of opening a new one. `http_client` and its + `settings.keepalive_expiry` are unaffected. Responses update + `last_response` as those of `http_client` do. + + Raises: + RuntimeError: if the pool was closed before it was ever built. + """ + + client = self._tts_http_client + if client is None: + with self._tts_http_client_lock: + client = self._tts_http_client + if client is None: + if self._tts_http_client_closed: + raise RuntimeError(_CLOSED_CLIENT_ERROR) + client = self._tts_http_client = httpx.Client( + base_url=self._settings.tts_base_url, + headers=_build_headers(self._settings), + timeout=self._settings.tts_http_timeout, + limits=_build_tts_limits(self._settings), + event_hooks=self._event_hooks, + ) + + return client + + def _close_tts_http_client(self) -> None: + """Closes the Text-to-Speech pool, and keeps it from being built later.""" + + with self._tts_http_client_lock: + self._tts_http_client_closed = True + client = self._tts_http_client + if client is not None: + client.close() + @classmethod def get_default(cls, api_key_required: bool = True): """ diff --git a/assemblyai/tts/v1/async_client.py b/assemblyai/tts/v1/async_client.py index 514d636..4d5f303 100644 --- a/assemblyai/tts/v1/async_client.py +++ b/assemblyai/tts/v1/async_client.py @@ -101,7 +101,7 @@ async def synthesize( body = request_body(text, config or self._config) return await async_api.synthesize( - self._client.http_client, + self._client.tts_http_client, base_url=self._client.settings.tts_base_url, body=body, timeout=synthesize_timeout(self._client.settings), @@ -129,7 +129,7 @@ def stream( body = request_body(text, config or self._config) return async_api.stream_speech( - self._client.http_client, + self._client.tts_http_client, base_url=self._client.settings.tts_base_url, body=body, timeout=self._client.settings.tts_http_timeout, diff --git a/assemblyai/tts/v1/client.py b/assemblyai/tts/v1/client.py index f1c9de9..b848ed0 100644 --- a/assemblyai/tts/v1/client.py +++ b/assemblyai/tts/v1/client.py @@ -21,6 +21,10 @@ class TextToSpeech: 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. + They share the client's Text-to-Speech pool (`Client.tts_http_client`), + which keeps an idle connection `settings.tts_keepalive_expiry` seconds + (120 by default), so a request within that time of the previous one + reuses its connection. Example: ```python @@ -114,7 +118,7 @@ def synthesize( body = request_body(text, config or self._config) return api.synthesize( - self._client.http_client, + self._client.tts_http_client, base_url=self._client.settings.tts_base_url, body=body, timeout=synthesize_timeout(self._client.settings), @@ -147,17 +151,20 @@ def stream( body = request_body(text, config or self._config) return api.stream_speech( - self._client.http_client, + self._client.tts_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.""" + """Closes the HTTP pools this instance owns. A shared `client` is left open.""" if self._owns_client: - self._client.http_client.close() + try: + self._client.http_client.close() + finally: + self._client._close_tts_http_client() def __enter__(self) -> "TextToSpeech": return self diff --git a/assemblyai/types.py b/assemblyai/types.py index f722658..acd7859 100644 --- a/assemblyai/types.py +++ b/assemblyai/types.py @@ -171,6 +171,9 @@ class Settings(BaseSettings): 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." + tts_keepalive_expiry: float = Field(default=120.0, ge=0.0) + "Seconds an idle connection stays in the Text-to-Speech HTTP pool (`TextToSpeech`/`AsyncTextToSpeech`), a pool of its own, so `keepalive_expiry` and the other APIs are unaffected. A voice agent's turns are often more than httpx's default 5 seconds apart; a request within this time of the previous one reuses its connection and skips the TCP and TLS handshakes (a few milliseconds in the API's region, two network round trips from farther away). Well under the hour the API keeps an idle connection open, so the client closes an idle connection before the API does." + 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 index 57bd943..424b113 100644 --- a/tests/unit/test_tts.py +++ b/tests/unit/test_tts.py @@ -4,9 +4,12 @@ reads audio back; the response headers carry the metadata. """ +import http.server import json +import threading +import time import warnings -from typing import Iterator +from typing import Iterator, List import httpx import pytest @@ -662,3 +665,176 @@ def test_swapped_text_and_config_is_named(): 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] + + +def _keepalive_expiry(http: httpx.Client) -> float: + """The idle time of `http`'s connection pool (httpcore's own field).""" + + return http._transport._pool._keepalive_expiry # type: ignore[attr-defined] + + +def test_the_tts_pool_keeps_idle_connections_120_s_by_default(): + assert aai.Settings().tts_keepalive_expiry == 120.0 + + client = aai.Client(settings=aai.Settings(api_key="test")) + pool = client.tts_http_client + + # A pool of its own, speaking HTTP/1.1, holding idle connections 120 s + assert pool is not client.http_client + assert pool is client.tts_http_client + assert _keepalive_expiry(pool) == 120.0 + assert pool._transport._pool._http2 is False # type: ignore[attr-defined] + # while the pool of the other APIs keeps httpx's default + assert _keepalive_expiry(client.http_client) == 5.0 + + +def test_the_tts_keepalive_expiry_is_a_setting(): + settings = aai.Settings(api_key="test", tts_keepalive_expiry=30.0) + assert _keepalive_expiry(aai.Client(settings=settings).tts_http_client) == 30.0 + + with pytest.raises(ValueError): + aai.Settings(tts_keepalive_expiry=-1.0) + + +def test_the_tts_pool_is_built_on_first_use_only(httpx_mock: HTTPXMock): + _mock_ok(httpx_mock) + client = aai.Client(api_key="test") + tts = aai.TextToSpeech(client=client) + assert client._tts_http_client is None # a client that never speaks + + tts.synthesize("Hi.") + + assert client._tts_http_client is not None + + +def test_requests_go_through_the_tts_pool(httpx_mock: HTTPXMock): + _mock_ok(httpx_mock) + _mock_ok(httpx_mock) + client = aai.Client(api_key="test") + tts = aai.TextToSpeech(client=client) + # The pool of the other APIs is not needed + client.http_client.close() + + tts.synthesize("Hi.") + with tts.stream("Hi.") as speech: + speech.read() + + assert len(httpx_mock.get_requests()) == 2 + # and responses still update last_response + assert client.last_response is not None + assert str(client.last_response.request.url) == SPEECH_URL + + +def test_concurrent_first_requests_build_one_tts_pool(): + client = aai.Client(api_key="test") + callers = 8 + start = threading.Barrier(callers) + pools: List[httpx.Client] = [] + + def take() -> None: + start.wait() + pools.append(client.tts_http_client) + + threads = [threading.Thread(target=take) for _ in range(callers)] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=10) + + assert len(pools) == callers + assert all(pool is pools[0] for pool in pools) + + +def test_closing_an_owned_client_closes_its_tts_pool(httpx_mock: HTTPXMock): + _mock_ok(httpx_mock) + with aai.TextToSpeech(api_key="k") as tts: + tts.synthesize("Hi.") + pool = tts.client.tts_http_client + assert pool.is_closed + + # an instance closed before it ever spoke fails as a closed pool does + unused = aai.TextToSpeech(api_key="k") + unused.close() + with pytest.raises(RuntimeError, match="closed"): + unused.synthesize("Hi.") + + # and the shared default client's pool stays open + shared = aai.TextToSpeech() + pool = shared.client.tts_http_client + shared.close() + assert not pool.is_closed + + +class _CountingServer(http.server.ThreadingHTTPServer): + """An HTTP/1.1 server answering every POST with 20 ms of audio, counting + the TCP connections it accepts.""" + + daemon_threads = True + + def __init__(self) -> None: + super().__init__(("127.0.0.1", 0), _AudioHandler) + self.connections = 0 + + def get_request(self): # type: ignore[no-untyped-def] + request = super().get_request() + self.connections += 1 + return request + + +class _AudioHandler(http.server.BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" # keeps connections open between requests + + def do_POST(self) -> None: + self.rfile.read(int(self.headers["content-length"])) + self.send_response(200) + for name, value in _OK_HEADERS.items(): + self.send_header(name, value) + self.send_header("content-length", str(len(_AUDIO))) + self.end_headers() + self.wfile.write(_AUDIO) + + def log_message(self, *args: object) -> None: + pass + + +@pytest.fixture +def counting_server() -> Iterator[_CountingServer]: + server = _CountingServer() + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + yield server + finally: + server.shutdown() + server.server_close() + thread.join(timeout=10) + + +def test_a_request_within_the_tts_keepalive_expiry_reuses_the_connection( + counting_server: _CountingServer, +): + # Given: a Text-to-Speech pool that keeps idle connections 1 s + host, port = counting_server.server_address[:2] + settings = aai.Settings( + api_key="test", tts_base_url=f"http://{host}:{port}", tts_keepalive_expiry=1.0 + ) + client = aai.Client(settings=settings) + tts = aai.TextToSpeech(client=client) + + try: + # When: two requests follow each other + tts.synthesize("One.") + tts.synthesize("Two.") + + # Then: the second reuses the first one's connection + assert counting_server.connections == 1 + + # When: the next request comes after the connection's idle time + time.sleep(1.5) + with tts.stream("Three.") as speech: + assert speech.read() == _AUDIO + + # Then: it opens a new connection + assert counting_server.connections == 2 + finally: + client.tts_http_client.close() diff --git a/tests/unit/test_tts_async.py b/tests/unit/test_tts_async.py index b8c3cda..a15962c 100644 --- a/tests/unit/test_tts_async.py +++ b/tests/unit/test_tts_async.py @@ -304,3 +304,55 @@ async def test_a_shared_client_is_left_open_on_aclose(httpx_mock: HTTPXMock): async def test_wrong_config_type_is_a_type_error(): with pytest.raises(TypeError): aai.AsyncTextToSpeech(config=aai.DictationConfig()) # type: ignore[arg-type] + + +async def test_the_tts_pool_keeps_idle_connections_120_s(): + client = aai.AsyncClient(settings=aai.Settings(api_key="test")) + pool = client.tts_http_client + + # A pool of its own, speaking HTTP/1.1, holding idle connections 120 s + assert pool is not client.http_client + assert pool is client.tts_http_client + assert pool._transport._pool._keepalive_expiry == 120.0 # type: ignore[attr-defined] + assert pool._transport._pool._http2 is False # type: ignore[attr-defined] + # while the pool of the other APIs keeps httpx's default + assert client.http_client._transport._pool._keepalive_expiry == 5.0 # type: ignore[attr-defined] + await client.aclose() + + +async def test_requests_go_through_the_tts_pool(httpx_mock: HTTPXMock): + _mock_ok(httpx_mock) + _mock_ok(httpx_mock) + client = aai.AsyncClient(api_key="test") + assert client._tts_http_client is None # built on first use only + # The pool of the other APIs is not needed + await client.http_client.aclose() + + async with aai.AsyncTextToSpeech(client=client) as tts: + await tts.synthesize("Hi.") + async with tts.stream("Hi.") as speech: + await speech.aread() + + assert len(httpx_mock.get_requests()) == 2 + assert client.last_response is not None + assert str(client.last_response.request.url) == SPEECH_URL + await client.aclose() + + +async def test_aclose_closes_both_pools(httpx_mock: HTTPXMock): + _mock_ok(httpx_mock) + tts = aai.AsyncTextToSpeech(api_key="k") + await tts.synthesize("Hi.") + pools = (tts.client.http_client, tts.client.tts_http_client) + + await tts.aclose() + + assert all(pool.is_closed for pool in pools) + + +async def test_a_client_closed_before_it_spoke_fails_as_a_closed_pool_does(): + tts = aai.AsyncTextToSpeech(api_key="k") + await tts.aclose() + + with pytest.raises(RuntimeError, match="closed"): + await tts.synthesize("Hi.") From 1c9b29b2cda7b5dc1a19f0e8a92f3b3a09a56c3d Mon Sep 17 00:00:00 2001 From: Aleksandar Mitov <140423361+aleks-mitov@users.noreply.github.com> Date: Sun, 25 Oct 2026 00:57:32 +0000 Subject: [PATCH 3/7] fix(tts): build the TTS pool with the instance on one shared TLS context Review of #284. Building the Text-to-Speech pool on first use moved httpx's TLS context build (it loads certifi's CA bundle, about 15 ms with OpenSSL 3.0) into the first request of every TextToSpeech / AsyncTextToSpeech, on top of the context the other APIs' pool still builds at construction. - TextToSpeech and AsyncTextToSpeech build their client's TTS pool when they are constructed. A Client or AsyncClient that never speaks still builds none. - Every TTS pool in the process verifies with one context, httpx's own default context built once (_shared_tts_ssl_context), so a new pool loads no certificates. The other APIs' pool is unchanged. - A forked child drops the TTS pools it inherited (os.register_at_fork), so it never shares the parent's idle connections, now kept 120 s. - Client.close() closes both pools; TextToSpeech.close() uses it. - Docs: the client never reuses a connection idle longer than tts_keepalive_expiry (there is no idle reaper); keepalive_expiry does not apply to TTS requests (README, MIGRATION, CLAUDE.md). - Tests: a server that closes idle connections before the expiry (sync and async), the reuse and expiry tests split so neither depends on timing, a fork test, the shared context, construction-time build and closed clients. The local server moves to tests/unit/tts_http_server.py. Fixes the mypy errors CI's mypy job reports on the changed test files. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01KcyY6MAEUVjqBt5pMpetvD --- CLAUDE.md | 5 +- MIGRATION.md | 2 + README.md | 4 +- assemblyai/async_client.py | 35 +++- assemblyai/client.py | 123 +++++++++++++- assemblyai/tts/v1/async_client.py | 3 + assemblyai/tts/v1/client.py | 17 +- assemblyai/types.py | 4 +- tests/unit/test_streaming_tls.py | 6 +- tests/unit/test_tts.py | 261 +++++++++++++++++++++--------- tests/unit/test_tts_async.py | 63 ++++++++ tests/unit/tts_http_server.py | 85 ++++++++++ 12 files changed, 504 insertions(+), 104 deletions(-) create mode 100644 tests/unit/tts_http_server.py diff --git a/CLAUDE.md b/CLAUDE.md index 946d4cc..2e787d9 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -52,7 +52,7 @@ 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). Requests use the client's own TTS pool (`client.tts_http_client`), which keeps idle connections `aai.settings.tts_keepalive_expiry` seconds (120), so consecutive sentences reuse one connection +- `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). Requests use the client's own TTS pool (`client.tts_http_client`, built when the `TextToSpeech` is), which keeps idle connections `aai.settings.tts_keepalive_expiry` seconds (120; `keepalive_expiry` does not apply), so consecutive sentences reuse one connection. Every TTS pool shares one process-wide TLS context. A passed `client=` stays the caller's: `client.close()` closes both pools - `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.TTSSession` — Text-to-Speech WebSocket session: one connection, text in as it is written, audio out. `with` connects and waits for `Begin` (`session.begin`); `speak(text)` yields one request's audio; `generate(text)` / `flush()` -> flush_id / `audio(flush_id)` for streamed-in text and pipelining (`audio()` with no id follows the request being written, from another thread); `cancel()` for barge-in (ends the iteration at once); `keep_alive()`; `on(TTSEvents.X, handler(event))` for `Begin`, `Audio` (switches audio from pull to push), `FlushDone`, `WordBoundaries`, `Cancelled`, `Termination`, `Error`. Exit sends `Terminate` and sets `session.termination` (`None` after a failure). Refusals and failures raise `TTSError` (`status_code` None, `request_id` = session id). Raw PCM/G.711 only; 2,000 chars per generate unless `Begin` reports another `limits.max_generate_text_length`; ~100 chars/s sustained. `token=` for temporary tokens - `aai.AsyncTTSSession` — Asyncio counterpart: `async with`, coroutine sends, `async for chunk in session.speak(text)` @@ -325,7 +325,8 @@ it ends," raise `keepalive_expiry` (e.g. to 120, the sync audio cap) so one `war covers the whole clip; otherwise call `warm()` shortly before `transcribe()` (it is idempotent, so re-warming to refresh is fine). It only helps the *first* request: sync calls share the client's connection pool, so later `transcribe()` calls already reuse the -open connection. Note `keepalive_expiry` applies to the whole client (sync and async). +open connection. Note `keepalive_expiry` applies to the whole client (sync and async) except +Text-to-Speech requests, which use a pool of their own and `tts_keepalive_expiry` (120s). ## Streaming (real-time) diff --git a/MIGRATION.md b/MIGRATION.md index d493cfe..ea5997c 100644 --- a/MIGRATION.md +++ b/MIGRATION.md @@ -222,6 +222,8 @@ transcriber.warm() # while the clip is still being recorded `AsyncSyncTranscriber.warm()` is the awaitable equivalent. +Text-to-Speech requests (`aai.TextToSpeech`, `aai.AsyncTextToSpeech`) use a pool of their own, kept by `settings.tts_keepalive_expiry` (120 seconds by default), and `keepalive_expiry` does not apply to them. If you lowered `keepalive_expiry` because your network drops idle connections, lower `tts_keepalive_expiry` as well. + ### Pass `pathlib.Path` directly Local files can be a `Path`, a `str` path, raw `bytes`/`bytearray`, or an open binary file — no `str()` wrapping needed: diff --git a/README.md b/README.md index ada0066..80864e7 100644 --- a/README.md +++ b/README.md @@ -894,7 +894,7 @@ except aai.DictationError as error: `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. They share a connection pool of their own that keeps an idle connection for `aai.settings.tts_keepalive_expiry` seconds (120 by default), so a request that follows the previous one within that time, such as a voice agent's next sentence, reuses its connection instead of opening a new one. +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. They share a connection pool of their own that keeps an idle connection for `aai.settings.tts_keepalive_expiry` seconds (120 by default), so a request that follows the previous one within that time, such as a voice agent's next sentence, reuses its connection instead of opening a new one. `aai.settings.keepalive_expiry` does not apply to this pool: if you lowered it because your network drops idle connections, lower `tts_keepalive_expiry` as well. A client you pass as `client=` stays yours to close: `client.close()` closes both of its pools.
Synthesize a WAV file @@ -2042,7 +2042,7 @@ aai.settings.dictation_http_timeout = 300.0 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 -aai.settings.tts_keepalive_expiry = 120.0 # seconds an idle connection stays in the TTS pool +aai.settings.tts_keepalive_expiry = 120.0 # seconds an idle connection stays in the TTS pool (keepalive_expiry does not apply to it) ``` 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`. diff --git a/assemblyai/async_client.py b/assemblyai/async_client.py index cd7c3e6..cd5b81f 100644 --- a/assemblyai/async_client.py +++ b/assemblyai/async_client.py @@ -8,9 +8,11 @@ from .client import ( _CLOSED_CLIENT_ERROR, _MISSING_API_KEY_ERROR, + _TTS_POOL_HOLDERS, _build_headers, _build_limits, _build_tts_limits, + _shared_tts_ssl_context, ) @@ -77,8 +79,8 @@ async def _store_response(response: httpx.Response) -> None: event_hooks=self._event_hooks, ) - # Built by `tts_http_client` on first use, so a client that never - # speaks pays nothing for it. + # Built when an `AsyncTextToSpeech` takes this client, or on first + # use, so a client that never speaks pays nothing for it. self._tts_http_client: Optional[httpx.AsyncClient] = None self._closed = False @@ -117,7 +119,8 @@ def http_client(self) -> httpx.AsyncClient: @property def tts_http_client(self) -> httpx.AsyncClient: """ - The HTTP client of the Text-to-Speech API, built on first use. + The HTTP client of the Text-to-Speech API, built when an + `AsyncTextToSpeech` takes this client or on first use. A connection pool of its own, whose idle connections stay `settings.tts_keepalive_expiry` seconds (120 by default); see @@ -128,20 +131,40 @@ def tts_http_client(self) -> httpx.AsyncClient: built. """ - client = self._tts_http_client + client = self._ensure_tts_http_client() if client is None: - if self._closed: - raise RuntimeError(_CLOSED_CLIENT_ERROR) + raise RuntimeError(_CLOSED_CLIENT_ERROR) + + return client + + def _ensure_tts_http_client(self) -> Optional[httpx.AsyncClient]: + """ + Returns the Text-to-Speech pool, building it if it is not built yet, + or `None` if the client was closed before the pool was ever built. + + `AsyncTextToSpeech` calls it when it is constructed, so its first + request does not wait for the build. + """ + + client = self._tts_http_client + if client is None and not self._closed: client = self._tts_http_client = httpx.AsyncClient( base_url=self._settings.tts_base_url, headers=_build_headers(self._settings), timeout=self._settings.tts_http_timeout, limits=_build_tts_limits(self._settings), event_hooks=self._event_hooks, + verify=_shared_tts_ssl_context(), ) + _TTS_POOL_HOLDERS.add(self) return client + def _forget_tts_http_client(self) -> None: + """Drops the Text-to-Speech pool unclosed; see `client._forget_tts_pools_in_child`.""" + + self._tts_http_client = None + async def aclose(self) -> None: """Closes the underlying HTTP connection pools.""" diff --git a/assemblyai/client.py b/assemblyai/client.py index 169bf1f..64c4da4 100644 --- a/assemblyai/client.py +++ b/assemblyai/client.py @@ -1,6 +1,9 @@ +import os +import ssl import sys import threading -from typing import ClassVar, Dict, Optional +import weakref +from typing import Any, ClassVar, Dict, Optional import httpx @@ -51,6 +54,72 @@ def _build_tts_limits(settings: types.Settings) -> httpx.Limits: _CLOSED_CLIENT_ERROR = "Cannot send a request, as the client has been closed." +# The context `_shared_tts_ssl_context()` returns once it has been built. +_tts_ssl_context: Optional[ssl.SSLContext] = None +_TTS_SSL_CONTEXT_LOCK = threading.Lock() + + +def _shared_tts_ssl_context() -> ssl.SSLContext: + """ + Returns the TLS context of every Text-to-Speech pool in the process, + built on the first call. Shared with `AsyncClient`. + + It is the context httpx would build for the pool by default (certifi's CA + bundle, or `SSL_CERT_FILE` / `SSL_CERT_DIR` when set), built once instead + of once per pool. A build loads and parses the CA bundle, about 15 ms on an + idle machine with OpenSSL 3.0 and several times that on a busy one, and + each `AsyncTextToSpeech` built without `client=`, like each `TextToSpeech` + built from `api_key=`, has a pool of its own. One context serves any + number of pools, threads and event loops; each connection keeps its own + TLS state. + + Only HTTP/1.1 pools may use it: httpcore sets the context's ALPN protocols + at every connect, to `http/1.1` for an HTTP/1.1 pool. + + The CA certificates and environment it reflects are those in place at the + first call; a later change takes effect in a new process. + """ + + global _tts_ssl_context + context = _tts_ssl_context + if context is None: + with _TTS_SSL_CONTEXT_LOCK: + context = _tts_ssl_context + if context is None: + context = _tts_ssl_context = httpx.create_ssl_context() + return context + + +# Every `Client` and `AsyncClient` that has built a Text-to-Speech pool, for +# `_forget_tts_pools_in_child()`. +_TTS_POOL_HOLDERS: "weakref.WeakSet[Any]" = weakref.WeakSet() + + +def _forget_tts_pools_in_child() -> None: + """ + Runs in a child process right after `fork()`, and drops every + Text-to-Speech pool the child inherited, so the child's next request + builds a pool of its own. + + An inherited pool holds the parent's idle connections, kept up to + `tts_keepalive_expiry` seconds (120 by default): a request made before + the fork, such as a warm-up in a `gunicorn --preload` app, leaves one + there. Processes sending on one connection at once garble each other's + responses. The pools are dropped rather than closed: another thread of the + parent may have held a pool's lock at the moment of the fork, and an async + pool needs its event loop. Each process closes its own socket descriptors, + so the parent's connections stay intact. + """ + + for holder in list(_TTS_POOL_HOLDERS): + holder._forget_tts_http_client() + _TTS_POOL_HOLDERS.clear() + + +if hasattr(os, "register_at_fork"): # POSIX only + os.register_at_fork(after_in_child=_forget_tts_pools_in_child) + + _MISSING_API_KEY_ERROR = ( "Please provide an API key: set the ASSEMBLYAI_API_KEY environment variable, " "set aai.settings.api_key, or pass api_key= to the transcriber or client." @@ -103,8 +172,8 @@ def _store_response(response): event_hooks=self._event_hooks, ) - # Built by `tts_http_client` on first use, so a client that never - # speaks pays nothing for it. + # Built when a `TextToSpeech` takes this client, or on first use, so + # a client that never speaks pays nothing for it. self._tts_http_client: Optional[httpx.Client] = None self._tts_http_client_closed = False self._tts_http_client_lock = threading.Lock() @@ -144,36 +213,60 @@ def http_client(self) -> httpx.Client: @property def tts_http_client(self) -> httpx.Client: """ - The HTTP client of the Text-to-Speech API, built on first use. + The HTTP client of the Text-to-Speech API, built when a + `TextToSpeech` takes this client or on first use. A connection pool of its own, whose idle connections stay `settings.tts_keepalive_expiry` seconds (120 by default): a voice agent's next sentence reuses the connection of the previous one instead of opening a new one. `http_client` and its `settings.keepalive_expiry` are unaffected. Responses update - `last_response` as those of `http_client` do. + `last_response` as those of `http_client` do. `close()` closes it + with `http_client`. Raises: RuntimeError: if the pool was closed before it was ever built. """ + client = self._ensure_tts_http_client() + if client is None: + raise RuntimeError(_CLOSED_CLIENT_ERROR) + + return client + + def _ensure_tts_http_client(self) -> Optional[httpx.Client]: + """ + Returns the Text-to-Speech pool, building it if it is not built yet, + or `None` if the pool was closed before it was ever built. + + `TextToSpeech` calls it when it is constructed, so its first request + does not wait for the build. + """ + client = self._tts_http_client if client is None: with self._tts_http_client_lock: client = self._tts_http_client - if client is None: - if self._tts_http_client_closed: - raise RuntimeError(_CLOSED_CLIENT_ERROR) + if client is None and not self._tts_http_client_closed: client = self._tts_http_client = httpx.Client( base_url=self._settings.tts_base_url, headers=_build_headers(self._settings), timeout=self._settings.tts_http_timeout, limits=_build_tts_limits(self._settings), event_hooks=self._event_hooks, + verify=_shared_tts_ssl_context(), ) + _TTS_POOL_HOLDERS.add(self) return client + def _forget_tts_http_client(self) -> None: + """Drops the Text-to-Speech pool unclosed; see `_forget_tts_pools_in_child`.""" + + self._tts_http_client = None + # Another thread of the parent may have held the lock at the fork. + self._tts_http_client_lock = threading.Lock() + def _close_tts_http_client(self) -> None: """Closes the Text-to-Speech pool, and keeps it from being built later.""" @@ -183,6 +276,20 @@ def _close_tts_http_client(self) -> None: if client is not None: client.close() + def close(self) -> None: + """ + Closes both HTTP connection pools, `http_client` and `tts_http_client`. + + A request after it raises httpx's closed-client `RuntimeError`. Leave + the default client (`Client.get_default()`) open: every instance built + without `client=` or `api_key=` shares it. + """ + + try: + self._http_client.close() + finally: + self._close_tts_http_client() + @classmethod def get_default(cls, api_key_required: bool = True): """ diff --git a/assemblyai/tts/v1/async_client.py b/assemblyai/tts/v1/async_client.py index 4d5f303..c3c7801 100644 --- a/assemblyai/tts/v1/async_client.py +++ b/assemblyai/tts/v1/async_client.py @@ -65,6 +65,9 @@ def __init__( 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() + # Build the client's TTS pool now, so the first request does not wait + # for it. A closed client builds none; its first request raises. + self._client._ensure_tts_http_client() @property def config(self) -> TTSConfig: diff --git a/assemblyai/tts/v1/client.py b/assemblyai/tts/v1/client.py index b848ed0..9503e27 100644 --- a/assemblyai/tts/v1/client.py +++ b/assemblyai/tts/v1/client.py @@ -55,7 +55,9 @@ def __init__( Creates a `TextToSpeech`. Args: - client: The HTTP client to use. Defaults to the shared default client. + client: The HTTP client to use. Defaults to the shared default + client. A client you pass stays yours to close: `client.close()` + closes both of its pools. 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 @@ -73,6 +75,9 @@ def __init__( self._owns_client = api_key is not None self._client = _client._resolve_client(client, api_key) self.config = config or TTSConfig() + # Build the client's TTS pool now, so the first request does not wait + # for it. A closed client builds none; its first request raises. + self._client._ensure_tts_http_client() @property def config(self) -> TTSConfig: @@ -158,13 +163,13 @@ def stream( ) def close(self) -> None: - """Closes the HTTP pools this instance owns. A shared `client` is left open.""" + """ + Closes the HTTP pools this instance owns. A `client` you passed is + left open: `client.close()` closes both of its pools. + """ if self._owns_client: - try: - self._client.http_client.close() - finally: - self._client._close_tts_http_client() + self._client.close() def __enter__(self) -> "TextToSpeech": return self diff --git a/assemblyai/types.py b/assemblyai/types.py index acd7859..d2b992e 100644 --- a/assemblyai/types.py +++ b/assemblyai/types.py @@ -132,7 +132,7 @@ class Settings(BaseSettings): "The default HTTP timeout for general requests" keepalive_expiry: Optional[float] = None - "Seconds an idle connection is kept in the HTTP pool before being closed. None uses httpx's default (5s). Raise it (e.g. to 120) so a connection opened by `SyncTranscriber.warm()` stays reusable across an in-progress recording." + "Seconds an idle connection is kept in the HTTP pool before being closed. None uses httpx's default (5s). Raise it (e.g. to 120) so a connection opened by `SyncTranscriber.warm()` stays reusable across an in-progress recording. Text-to-Speech requests use `tts_keepalive_expiry` instead." base_url: str = "https://api.assemblyai.com" "The base URL for the AssemblyAI API" @@ -172,7 +172,7 @@ class Settings(BaseSettings): "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." tts_keepalive_expiry: float = Field(default=120.0, ge=0.0) - "Seconds an idle connection stays in the Text-to-Speech HTTP pool (`TextToSpeech`/`AsyncTextToSpeech`), a pool of its own, so `keepalive_expiry` and the other APIs are unaffected. A voice agent's turns are often more than httpx's default 5 seconds apart; a request within this time of the previous one reuses its connection and skips the TCP and TLS handshakes (a few milliseconds in the API's region, two network round trips from farther away). Well under the hour the API keeps an idle connection open, so the client closes an idle connection before the API does." + "Seconds an idle connection stays in the Text-to-Speech HTTP pool (`TextToSpeech`/`AsyncTextToSpeech`). It is a pool of its own, which `keepalive_expiry` does not apply to: if you lowered `keepalive_expiry` because your network drops idle connections, lower this as well. A voice agent's turns are often more than httpx's default 5 seconds apart; a request within this time of the previous one reuses its connection and skips the TCP and TLS handshakes (a few milliseconds in the API's region, at least two network round trips from farther away). The client never reuses a connection idle longer than this, well under the API load balancer's one-hour idle 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_streaming_tls.py b/tests/unit/test_streaming_tls.py index 48f956c..8f20293 100644 --- a/tests/unit/test_streaming_tls.py +++ b/tests/unit/test_streaming_tls.py @@ -42,9 +42,9 @@ @pytest.fixture(scope="module") def tls_cert(tmp_path_factory: pytest.TempPathFactory) -> Tuple[str, str]: """A throwaway self-signed certificate and key for ``127.0.0.1``.""" - openssl = shutil.which("openssl") - if openssl is None: - pytest.skip("the openssl CLI is needed to generate a test certificate") + openssl = shutil.which("openssl") or pytest.skip( + "the openssl CLI is needed to generate a test certificate" + ) directory = tmp_path_factory.mktemp("tls") cert = str(directory / "cert.pem") key = str(directory / "key.pem") diff --git a/tests/unit/test_tts.py b/tests/unit/test_tts.py index 424b113..2f0e1d1 100644 --- a/tests/unit/test_tts.py +++ b/tests/unit/test_tts.py @@ -4,8 +4,9 @@ reads audio back; the response headers carry the metadata. """ -import http.server import json +import os +import ssl import threading import time import warnings @@ -16,7 +17,9 @@ from pytest_httpx import HTTPXMock import assemblyai as aai +from assemblyai import client as client_module from assemblyai.tts.v1 import api +from tests.unit.tts_http_server import serving aai.settings.api_key = "test" @@ -419,7 +422,7 @@ def test_error_envelope_becomes_a_tts_error(httpx_mock: HTTPXMock): aai.TextToSpeech().synthesize("Hi.", aai.TTSConfig(sample_rate=12345)) # THEN every field of the envelope is on the error - err = exc_info.value + err: aai.TTSError = exc_info.value assert isinstance(err, aai.AssemblyAIError) assert str(err) == _ERROR_BODY["message"] assert err.status_code == 422 @@ -673,6 +676,12 @@ def _keepalive_expiry(http: httpx.Client) -> float: return http._transport._pool._keepalive_expiry # type: ignore[attr-defined] +def _ssl_context(http: httpx.Client) -> ssl.SSLContext: + """The TLS context `http`'s connection pool connects with (httpcore's own field).""" + + return http._transport._pool._ssl_context # type: ignore[attr-defined] + + def test_the_tts_pool_keeps_idle_connections_120_s_by_default(): assert aai.Settings().tts_keepalive_expiry == 120.0 @@ -696,15 +705,65 @@ def test_the_tts_keepalive_expiry_is_a_setting(): aai.Settings(tts_keepalive_expiry=-1.0) -def test_the_tts_pool_is_built_on_first_use_only(httpx_mock: HTTPXMock): - _mock_ok(httpx_mock) +def test_the_tts_pool_is_built_with_the_text_to_speech_not_with_the_client(): client = aai.Client(api_key="test") - tts = aai.TextToSpeech(client=client) assert client._tts_http_client is None # a client that never speaks - tts.synthesize("Hi.") + aai.TextToSpeech(client=client) + # Built before the first request, which then does not wait for it assert client._tts_http_client is not None + # and an instance on the shared default client builds its pool as well + assert aai.TextToSpeech().client._tts_http_client is not None + + +def test_every_tts_pool_shares_one_verifying_tls_context(): + first = aai.TextToSpeech(api_key="k") + second = aai.TextToSpeech(api_key="k") + + try: + context = _ssl_context(first.client.tts_http_client) + # One context for every TTS pool, so a new pool loads no certificates + assert context is _ssl_context(second.client.tts_http_client) + assert context is client_module._shared_tts_ssl_context() + # httpx's own verifying context + assert context.verify_mode == ssl.CERT_REQUIRED + assert context.check_hostname is True + # The pool of the other APIs keeps a context of its own, as before + assert _ssl_context(first.client.http_client) is not context + finally: + first.close() + second.close() + + +def test_concurrent_first_calls_build_one_tls_context(monkeypatch: pytest.MonkeyPatch): + monkeypatch.setattr(client_module, "_tts_ssl_context", None) + built: List[ssl.SSLContext] = [] + real_create = httpx.create_ssl_context + + def slow_create() -> ssl.SSLContext: + time.sleep(0.05) # widens the window two builders would race in + built.append(real_create()) + return built[-1] + + monkeypatch.setattr(httpx, "create_ssl_context", slow_create) + callers = 8 + start = threading.Barrier(callers) + contexts: List[ssl.SSLContext] = [] + + def take() -> None: + start.wait() + contexts.append(client_module._shared_tts_ssl_context()) + + threads = [threading.Thread(target=take) for _ in range(callers)] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=10) + + assert len(built) == 1 + assert len(contexts) == callers + assert all(context is built[0] for context in contexts) def test_requests_go_through_the_tts_pool(httpx_mock: HTTPXMock): @@ -751,12 +810,9 @@ def test_closing_an_owned_client_closes_its_tts_pool(httpx_mock: HTTPXMock): tts.synthesize("Hi.") pool = tts.client.tts_http_client assert pool.is_closed - - # an instance closed before it ever spoke fails as a closed pool does - unused = aai.TextToSpeech(api_key="k") - unused.close() + assert tts.client.http_client.is_closed with pytest.raises(RuntimeError, match="closed"): - unused.synthesize("Hi.") + tts.synthesize("Hi.") # and the shared default client's pool stays open shared = aai.TextToSpeech() @@ -765,76 +821,131 @@ def test_closing_an_owned_client_closes_its_tts_pool(httpx_mock: HTTPXMock): assert not pool.is_closed -class _CountingServer(http.server.ThreadingHTTPServer): - """An HTTP/1.1 server answering every POST with 20 ms of audio, counting - the TCP connections it accepts.""" - - daemon_threads = True - - def __init__(self) -> None: - super().__init__(("127.0.0.1", 0), _AudioHandler) - self.connections = 0 - - def get_request(self): # type: ignore[no-untyped-def] - request = super().get_request() - self.connections += 1 - return request - - -class _AudioHandler(http.server.BaseHTTPRequestHandler): - protocol_version = "HTTP/1.1" # keeps connections open between requests - - def do_POST(self) -> None: - self.rfile.read(int(self.headers["content-length"])) - self.send_response(200) - for name, value in _OK_HEADERS.items(): - self.send_header(name, value) - self.send_header("content-length", str(len(_AUDIO))) - self.end_headers() - self.wfile.write(_AUDIO) +def test_client_close_closes_both_pools(): + client = aai.Client(api_key="test") + tts = aai.TextToSpeech(client=client) + pools = (client.http_client, client.tts_http_client) - def log_message(self, *args: object) -> None: - pass + # A passed client stays the caller's to close + tts.close() + assert not any(pool.is_closed for pool in pools) + client.close() + assert all(pool.is_closed for pool in pools) -@pytest.fixture -def counting_server() -> Iterator[_CountingServer]: - server = _CountingServer() - thread = threading.Thread(target=server.serve_forever, daemon=True) - thread.start() - try: - yield server - finally: - server.shutdown() - server.server_close() - thread.join(timeout=10) +def test_a_client_closed_before_it_spoke_fails_as_a_closed_pool_does(): + client = aai.Client(api_key="test") + client.close() -def test_a_request_within_the_tts_keepalive_expiry_reuses_the_connection( - counting_server: _CountingServer, -): - # Given: a Text-to-Speech pool that keeps idle connections 1 s - host, port = counting_server.server_address[:2] - settings = aai.Settings( - api_key="test", tts_base_url=f"http://{host}:{port}", tts_keepalive_expiry=1.0 - ) - client = aai.Client(settings=settings) + # Construction builds no pool on a closed client and does not raise tts = aai.TextToSpeech(client=client) + assert client._tts_http_client is None - try: - # When: two requests follow each other - tts.synthesize("One.") - tts.synthesize("Two.") - - # Then: the second reuses the first one's connection - assert counting_server.connections == 1 + with pytest.raises(RuntimeError, match="closed"): + tts.synthesize("Hi.") - # When: the next request comes after the connection's idle time - time.sleep(1.5) - with tts.stream("Three.") as speech: - assert speech.read() == _AUDIO - # Then: it opens a new connection - assert counting_server.connections == 2 - finally: - client.tts_http_client.close() +def test_a_request_within_the_tts_keepalive_expiry_reuses_the_connection(): + # Given: a Text-to-Speech pool with the default 120 s idle time + with serving(_AUDIO, _OK_HEADERS) as server: + client = aai.Client( + settings=aai.Settings(api_key="test", tts_base_url=server.url) + ) + tts = aai.TextToSpeech(client=client) + try: + # When: three requests follow each other + tts.synthesize("One.") + tts.synthesize("Two.") + with tts.stream("Three.") as speech: + assert speech.read() == _AUDIO + + # Then: they all use the first one's connection + assert server.connections == 1 + finally: + client.close() + + +def test_a_request_after_the_tts_keepalive_expiry_opens_a_new_connection(): + # Given: a Text-to-Speech pool that keeps idle connections 0.2 s + with serving(_AUDIO, _OK_HEADERS) as server: + settings = aai.Settings( + api_key="test", tts_base_url=server.url, tts_keepalive_expiry=0.2 + ) + client = aai.Client(settings=settings) + tts = aai.TextToSpeech(client=client) + try: + tts.synthesize("One.") + + # When: the next request comes after the connection's idle time + time.sleep(0.5) + with tts.stream("Two.") as speech: + assert speech.read() == _AUDIO + + # Then: it opens a new connection + assert server.connections == 2 + finally: + client.close() + + +def test_a_connection_the_server_closed_while_idle_is_not_reused(): + # Given: a server (or load balancer) that closes each connection 0.2 s + # after its response, well before the pool's 120 s idle time runs out + with serving(_AUDIO, _OK_HEADERS, close_after=0.2) as server: + settings = aai.Settings( + api_key="test", tts_base_url=server.url, tts_keepalive_expiry=120.0 + ) + client = aai.Client(settings=settings) + tts = aai.TextToSpeech(client=client) + try: + tts.synthesize("One.") + + # When: the next request comes after the server closed it + time.sleep(0.5) + response = tts.synthesize("Two.") + + # Then: it succeeds on a new connection, with no + # RemoteProtocolError from sending on the closed one + assert response.audio == _AUDIO + assert server.connections == 2 + finally: + client.close() + + +@pytest.mark.skipif(not hasattr(os, "fork"), reason="needs os.fork") +def test_a_forked_child_does_not_share_the_parents_tts_connections(): + with serving(_AUDIO, _OK_HEADERS) as server: + settings = aai.Settings(api_key="test", tts_base_url=server.url) + client = aai.Client(settings=settings) + tts = aai.TextToSpeech(client=client) + try: + # Given: a parent whose pool holds an idle connection + tts.synthesize("Warm-up.") + parent_pool = client.tts_http_client + assert server.connections == 1 + + # When: a forked child makes a request + with warnings.catch_warnings(): + # Python 3.12 warns about forking a process that has threads. + warnings.simplefilter("ignore", DeprecationWarning) + pid = os.fork() + if pid == 0: + code = 1 + try: + dropped = client._tts_http_client is None + response = tts.synthesize("Child.") + fresh = client.tts_http_client is not parent_pool + code = 0 if dropped and fresh and response.audio else 2 + finally: + os._exit(code) + _, status = os.waitpid(pid, 0) + + # Then: the child used a pool and a connection of its own + assert os.WIFEXITED(status) and os.WEXITSTATUS(status) == 0 + assert server.connections == 2 + # and the parent's connection is intact and still reused + assert client.tts_http_client is parent_pool + tts.synthesize("Parent.") + assert server.connections == 2 + finally: + client.close() diff --git a/tests/unit/test_tts_async.py b/tests/unit/test_tts_async.py index a15962c..d7430f9 100644 --- a/tests/unit/test_tts_async.py +++ b/tests/unit/test_tts_async.py @@ -5,6 +5,7 @@ plus pool ownership. """ +import asyncio import json from typing import AsyncIterator @@ -13,7 +14,9 @@ from pytest_httpx import HTTPXMock import assemblyai as aai +from assemblyai import client as client_module from assemblyai.tts.v1 import async_api +from tests.unit.tts_http_server import serving pytestmark = pytest.mark.asyncio @@ -356,3 +359,63 @@ async def test_a_client_closed_before_it_spoke_fails_as_a_closed_pool_does(): with pytest.raises(RuntimeError, match="closed"): await tts.synthesize("Hi.") + + +async def test_the_tts_pool_is_built_with_the_text_to_speech_not_with_the_client(): + client = aai.AsyncClient(api_key="test") + assert client._tts_http_client is None # a client that never speaks + + aai.AsyncTextToSpeech(client=client) + + # Built before the first request, which then does not wait for it, with + # the TLS context every TTS pool of the process shares + assert client._tts_http_client is not None + pool = client._tts_http_client._transport._pool # type: ignore[attr-defined] + assert pool._ssl_context is client_module._shared_tts_ssl_context() + await client.aclose() + + +async def test_a_closed_client_builds_no_pool_and_its_first_request_raises(): + client = aai.AsyncClient(api_key="test") + await client.aclose() + + tts = aai.AsyncTextToSpeech(client=client) + assert client._tts_http_client is None + + with pytest.raises(RuntimeError, match="closed"): + await tts.synthesize("Hi.") + + +async def test_a_connection_the_server_closed_while_idle_is_not_reused(): + # Given: a server (or load balancer) that closes each connection 0.2 s + # after its response, well before the pool's 120 s idle time runs out + with serving(_AUDIO, _OK_HEADERS, close_after=0.2) as server: + settings = aai.Settings( + api_key="test", tts_base_url=server.url, tts_keepalive_expiry=120.0 + ) + async with aai.AsyncClient(settings=settings) as client: + tts = aai.AsyncTextToSpeech(client=client) + await tts.synthesize("One.") + + # When: the next request comes after the server closed it + await asyncio.sleep(0.5) + response = await tts.synthesize("Two.") + + # Then: it succeeds on a new connection, with no + # RemoteProtocolError from sending on the closed one + assert response.audio == _AUDIO + assert server.connections == 2 + + +async def test_a_forked_child_drops_the_inherited_tts_pool(): + client = aai.AsyncClient(api_key="test") + parent_pool = client.tts_http_client + + # What runs in a child right after os.fork() + client_module._forget_tts_pools_in_child() + + # The child builds a pool of its own on its next request + assert client._tts_http_client is None + assert client.tts_http_client is not parent_pool + await parent_pool.aclose() + await client.aclose() diff --git a/tests/unit/tts_http_server.py b/tests/unit/tts_http_server.py new file mode 100644 index 0000000..e79772e --- /dev/null +++ b/tests/unit/tts_http_server.py @@ -0,0 +1,85 @@ +"""A local HTTP/1.1 Text-to-Speech host for the connection tests of +`test_tts.py` and `test_tts_async.py`: it answers every POST with audio and +counts the TCP connections it accepts, so a test can tell a reused +connection from a new one.""" + +import contextlib +import http.server +import threading +import time +from typing import Dict, Iterator, Optional, cast + + +class CountingTTSServer(http.server.ThreadingHTTPServer): + """ + Answers every POST with `audio` and `headers`, and keeps the connection + open for the next request (HTTP/1.1 keep-alive). + + With `close_after`, the server closes each connection that many seconds + after its response, as a load balancer or NAT closes an idle connection + while the client still holds it. + """ + + daemon_threads = True + + def __init__( + self, + audio: bytes, + headers: Dict[str, str], + close_after: Optional[float] = None, + ) -> None: + super().__init__(("127.0.0.1", 0), _AudioHandler) + self.audio = audio + self.response_headers = headers + self.close_after = close_after + self.connections = 0 + + @property + def url(self) -> str: + return f"http://127.0.0.1:{self.server_port}" + + def get_request(self): # type: ignore[no-untyped-def] + request = super().get_request() + self.connections += 1 + return request + + +class _AudioHandler(http.server.BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" # keeps connections open between requests + + def do_POST(self) -> None: + server = cast(CountingTTSServer, self.server) + self.rfile.read(int(self.headers["content-length"])) + self.send_response(200) + for name, value in server.response_headers.items(): + self.send_header(name, value) + self.send_header("content-length", str(len(server.audio))) + self.end_headers() + self.wfile.write(server.audio) + if server.close_after is not None: + # The response carries no `Connection: close`, so the client keeps + # the connection in its pool; the server then closes it. + time.sleep(server.close_after) + self.close_connection = True + + def log_message(self, *args: object) -> None: + pass + + +@contextlib.contextmanager +def serving( + audio: bytes, + headers: Dict[str, str], + close_after: Optional[float] = None, +) -> Iterator[CountingTTSServer]: + """Runs a `CountingTTSServer` on a background thread for the block.""" + + server = CountingTTSServer(audio, headers, close_after) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + yield server + finally: + server.shutdown() + server.server_close() + thread.join(timeout=10) From 431b7412ddc197848f3b2bb50d61795f130187db Mon Sep 17 00:00:00 2001 From: Aleksandar Mitov <140423361+aleks-mitov@users.noreply.github.com> Date: Mon, 26 Oct 2026 04:31:37 +0000 Subject: [PATCH 4/7] fix(tts): require httpx 0.22 so a forked child drops TTS pools without closing them httpx 0.19 to 0.21 close a client in Client.__del__, which takes the pool's lock, so the fork handler's dropped pool could deadlock the child when it was collected. Keeping the pools referenced only moves that to the child's interpreter exit (reproduced on 0.19, 0.20 and 0.21.3) and keeps the parent's sockets open in every child. httpx 0.22, the lowest version the test matrix already runs, has no __del__, so dropping is safe there. Tests: a dropped sync pool that has sent a request is collected without close(), a dropped async pool without the unclosed-client warning, and setup.py's floor stays at 0.22 or later. The first two fail on httpx 0.19, 0.20 and 0.21.3. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01KcyY6MAEUVjqBt5pMpetvD --- assemblyai/client.py | 7 +++++ setup.py | 2 +- tests/unit/test_tts.py | 53 ++++++++++++++++++++++++++++++++++++ tests/unit/test_tts_async.py | 26 ++++++++++++++++++ tox.ini | 2 +- 5 files changed, 88 insertions(+), 2 deletions(-) diff --git a/assemblyai/client.py b/assemblyai/client.py index 64c4da4..d671841 100644 --- a/assemblyai/client.py +++ b/assemblyai/client.py @@ -109,6 +109,13 @@ def _forget_tts_pools_in_child() -> None: parent may have held a pool's lock at the moment of the fork, and an async pool needs its event loop. Each process closes its own socket descriptors, so the parent's connections stay intact. + + Dropping relies on httpx not closing a client when it is garbage + collected, which holds from httpx 0.22, the floor in `setup.py`. httpx + 0.19 to 0.21 close it in `Client.__del__`, which takes the pool's lock. + Keeping the dropped pools referenced would not help there: the child's + interpreter would close them at exit, and its copies of the parent's + sockets would stay open until then. """ for holder in list(_TTS_POOL_HOLDERS): diff --git a/setup.py b/setup.py index bbde86d..980965a 100644 --- a/setup.py +++ b/setup.py @@ -22,7 +22,7 @@ def get_version() -> str: package_data={"assemblyai": ["py.typed"]}, zip_safe=False, install_requires=[ - "httpx>=0.19.0", + "httpx>=0.22.0", "pydantic>=2.0; python_version>='3.14'", "pydantic>=1.10.17; python_version<'3.14'", "pydantic-settings>=2.0; python_version>='3.14'", diff --git a/tests/unit/test_tts.py b/tests/unit/test_tts.py index 2f0e1d1..fb1b611 100644 --- a/tests/unit/test_tts.py +++ b/tests/unit/test_tts.py @@ -4,12 +4,16 @@ reads audio back; the response headers carry the metadata. """ +import gc import json import os +import re import ssl import threading import time import warnings +import weakref +from pathlib import Path from typing import Iterator, List import httpx @@ -949,3 +953,52 @@ def test_a_forked_child_does_not_share_the_parents_tts_connections(): assert server.connections == 2 finally: client.close() + + +def test_a_forked_child_drops_the_inherited_tts_pool_without_closing_it( + monkeypatch: pytest.MonkeyPatch, +): + # Closing a pool takes its lock, which another thread of the parent may + # have held at the fork, so a child that closes it can wait forever. + closed: List[str] = [] + close = httpx.Client.close + + def recording_close(self: httpx.Client) -> None: + closed.append(repr(self)) + close(self) + + with serving(_AUDIO, _OK_HEADERS) as server: + settings = aai.Settings(api_key="test", tts_base_url=server.url) + client = aai.Client(settings=settings) + tts = aai.TextToSpeech(client=client) + try: + # Given: a pool that has sent a request and holds an idle + # connection, as a warm-up before the fork leaves it + tts.synthesize("Warm-up.") + inherited = weakref.ref(client.tts_http_client) + monkeypatch.setattr(httpx.Client, "close", recording_close) + + # When: the fork handler runs in the child, and the pool it + # dropped is collected + with warnings.catch_warnings(): + # The collected pool's sockets warn that nothing closed them. + warnings.simplefilter("ignore", ResourceWarning) + client_module._forget_tts_pools_in_child() + gc.collect() + + # Then: nothing closed the pool, and it is gone + assert closed == [] + assert inherited() is None + finally: + monkeypatch.undo() + client.close() + + +def test_the_httpx_floor_excludes_releases_that_close_a_collected_client(): + # httpx 0.19 to 0.21 close a client in `Client.__del__`, so on them the + # pool the fork handler drops would take its lock when it is collected. + setup_py = (Path(__file__).parents[2] / "setup.py").read_text() + floor = re.search(r'"httpx>=([0-9.]+)"', setup_py) + + assert floor is not None + assert tuple(int(part) for part in floor.group(1).split(".")) >= (0, 22) diff --git a/tests/unit/test_tts_async.py b/tests/unit/test_tts_async.py index d7430f9..3f30921 100644 --- a/tests/unit/test_tts_async.py +++ b/tests/unit/test_tts_async.py @@ -6,7 +6,10 @@ """ import asyncio +import gc import json +import warnings +import weakref from typing import AsyncIterator import httpx @@ -419,3 +422,26 @@ async def test_a_forked_child_drops_the_inherited_tts_pool(): assert client.tts_http_client is not parent_pool await parent_pool.aclose() await client.aclose() + + +async def test_a_forked_child_drops_the_inherited_tts_pool_without_a_warning(): + # Given: an open pool, as one that has sent a request before the fork is + client = aai.AsyncClient(api_key="test") + pool = client.tts_http_client + await pool.__aenter__() + inherited = weakref.ref(pool) + del pool + gc.collect() # leaves only the pool to collect below + + # When: the fork handler runs in the child, and the pool it dropped is + # collected + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + warnings.simplefilter("ignore", ResourceWarning) + client_module._forget_tts_pools_in_child() + gc.collect() + + # Then: the pool is gone, with no warning about an unclosed client + assert inherited() is None + assert [str(warning.message) for warning in caught] == [] + await client.aclose() diff --git a/tox.ini b/tox.ini index 7a7ab9b..1928a55 100644 --- a/tox.ini +++ b/tox.ini @@ -1,7 +1,7 @@ # Per-axis matrix, not a cartesian product: `latest` envs resolve the newest # deps via setup.py install_requires; legacy floors are pinned on one Python # (3.11) so each dependency's lower bound is still exercised without a 480-env -# blow-up. Floors track setup.py (pydantic>=1.10.17, websockets>=11, httpx>=0.19). +# blow-up. Floors track setup.py (pydantic>=1.10.17, websockets>=11, httpx>=0.22). [tox] envlist = py{39,310,311,312}-latest From 8d071ca9c94b611aa1ab963b52b8ea4f306fcf16 Mon Sep 17 00:00:00 2001 From: Aleksandar Mitov <140423361+aleks-mitov@users.noreply.github.com> Date: Mon, 26 Oct 2026 04:31:53 +0000 Subject: [PATCH 5/7] docs(tts): say what tts_keepalive_expiry controls, without naming server infrastructure Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01KcyY6MAEUVjqBt5pMpetvD --- assemblyai/types.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/assemblyai/types.py b/assemblyai/types.py index d2b992e..ef08673 100644 --- a/assemblyai/types.py +++ b/assemblyai/types.py @@ -172,7 +172,7 @@ class Settings(BaseSettings): "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." tts_keepalive_expiry: float = Field(default=120.0, ge=0.0) - "Seconds an idle connection stays in the Text-to-Speech HTTP pool (`TextToSpeech`/`AsyncTextToSpeech`). It is a pool of its own, which `keepalive_expiry` does not apply to: if you lowered `keepalive_expiry` because your network drops idle connections, lower this as well. A voice agent's turns are often more than httpx's default 5 seconds apart; a request within this time of the previous one reuses its connection and skips the TCP and TLS handshakes (a few milliseconds in the API's region, at least two network round trips from farther away). The client never reuses a connection idle longer than this, well under the API load balancer's one-hour idle timeout." + "Seconds an idle connection stays in the Text-to-Speech HTTP pool (`TextToSpeech`/`AsyncTextToSpeech`). It is a pool of its own, which `keepalive_expiry` does not apply to: if you lowered `keepalive_expiry` because your network drops idle connections, lower this as well. A voice agent's turns are often more than httpx's default 5 seconds apart; a request within this time of the previous one reuses its connection and skips the TCP and TLS handshakes (a few milliseconds in the API's region, at least two network round trips from farther away). The client never reuses a connection that has been idle longer than this: lower it if a proxy or NAT on your network closes idle connections sooner." 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)" From 66434532a79e568741c79501900ca841a3990e3b Mon Sep 17 00:00:00 2001 From: Aleksandar Mitov <140423361+aleks-mitov@users.noreply.github.com> Date: Mon, 26 Oct 2026 04:32:20 +0000 Subject: [PATCH 6/7] docs(tts): move the TTS pool note out of the 1.0 migration guide 1.8 users upgrading to 1.9 do not read "Migrating to 1.0". The note goes to the 1.9.0 release notes instead (drafted in the PR description), with the other changes a 1.9 upgrade brings. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01KcyY6MAEUVjqBt5pMpetvD --- MIGRATION.md | 2 -- 1 file changed, 2 deletions(-) diff --git a/MIGRATION.md b/MIGRATION.md index ea5997c..d493cfe 100644 --- a/MIGRATION.md +++ b/MIGRATION.md @@ -222,8 +222,6 @@ transcriber.warm() # while the clip is still being recorded `AsyncSyncTranscriber.warm()` is the awaitable equivalent. -Text-to-Speech requests (`aai.TextToSpeech`, `aai.AsyncTextToSpeech`) use a pool of their own, kept by `settings.tts_keepalive_expiry` (120 seconds by default), and `keepalive_expiry` does not apply to them. If you lowered `keepalive_expiry` because your network drops idle connections, lower `tts_keepalive_expiry` as well. - ### Pass `pathlib.Path` directly Local files can be a `Path`, a `str` path, raw `bytes`/`bytearray`, or an open binary file — no `str()` wrapping needed: From 0803e4a842585e05c5e92f72c0f0e3d07f2e7a01 Mon Sep 17 00:00:00 2001 From: Aleksandar Mitov <140423361+aleks-mitov@users.noreply.github.com> Date: Mon, 26 Oct 2026 04:34:22 +0000 Subject: [PATCH 7/7] fix(tts): keep a user-set shorter keepalive_expiry for TTS connections tts_keepalive_expiry is now None by default. Unset, the TTS pool keeps idle connections 120 s, or keepalive_expiry when that is set shorter, so someone who lowered keepalive_expiry because their network drops idle connections gets the same on TTS. A set tts_keepalive_expiry still wins. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01KcyY6MAEUVjqBt5pMpetvD --- CLAUDE.md | 5 ++-- README.md | 4 +-- assemblyai/async_client.py | 4 +-- assemblyai/client.py | 54 +++++++++++++++++++++++------------- assemblyai/tts/v1/client.py | 6 ++-- assemblyai/types.py | 6 ++-- tests/unit/test_tts.py | 33 ++++++++++++++++++++-- tests/unit/test_tts_async.py | 11 ++++++++ 8 files changed, 90 insertions(+), 33 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 2e787d9..ce03621 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -52,7 +52,7 @@ 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). Requests use the client's own TTS pool (`client.tts_http_client`, built when the `TextToSpeech` is), which keeps idle connections `aai.settings.tts_keepalive_expiry` seconds (120; `keepalive_expiry` does not apply), so consecutive sentences reuse one connection. Every TTS pool shares one process-wide TLS context. A passed `client=` stays the caller's: `client.close()` closes both pools +- `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). Requests use the client's own TTS pool (`client.tts_http_client`, built when the `TextToSpeech` is), which keeps idle connections 120 s by default (`aai.settings.tts_keepalive_expiry`; unset, a shorter `keepalive_expiry` applies), so consecutive sentences reuse one connection. Every TTS pool shares one process-wide TLS context. A passed `client=` stays the caller's: `client.close()` closes both pools - `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.TTSSession` — Text-to-Speech WebSocket session: one connection, text in as it is written, audio out. `with` connects and waits for `Begin` (`session.begin`); `speak(text)` yields one request's audio; `generate(text)` / `flush()` -> flush_id / `audio(flush_id)` for streamed-in text and pipelining (`audio()` with no id follows the request being written, from another thread); `cancel()` for barge-in (ends the iteration at once); `keep_alive()`; `on(TTSEvents.X, handler(event))` for `Begin`, `Audio` (switches audio from pull to push), `FlushDone`, `WordBoundaries`, `Cancelled`, `Termination`, `Error`. Exit sends `Terminate` and sets `session.termination` (`None` after a failure). Refusals and failures raise `TTSError` (`status_code` None, `request_id` = session id). Raw PCM/G.711 only; 2,000 chars per generate unless `Begin` reports another `limits.max_generate_text_length`; ~100 chars/s sustained. `token=` for temporary tokens - `aai.AsyncTTSSession` — Asyncio counterpart: `async with`, coroutine sends, `async for chunk in session.speak(text)` @@ -326,7 +326,8 @@ covers the whole clip; otherwise call `warm()` shortly before `transcribe()` (it idempotent, so re-warming to refresh is fine). It only helps the *first* request: sync calls share the client's connection pool, so later `transcribe()` calls already reuse the open connection. Note `keepalive_expiry` applies to the whole client (sync and async) except -Text-to-Speech requests, which use a pool of their own and `tts_keepalive_expiry` (120s). +Text-to-Speech requests, whose own pool keeps idle connections 120s, or `keepalive_expiry` when +that is shorter (`tts_keepalive_expiry` overrides both). ## Streaming (real-time) diff --git a/README.md b/README.md index 80864e7..525f5d6 100644 --- a/README.md +++ b/README.md @@ -894,7 +894,7 @@ except aai.DictationError as error: `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. They share a connection pool of their own that keeps an idle connection for `aai.settings.tts_keepalive_expiry` seconds (120 by default), so a request that follows the previous one within that time, such as a voice agent's next sentence, reuses its connection instead of opening a new one. `aai.settings.keepalive_expiry` does not apply to this pool: if you lowered it because your network drops idle connections, lower `tts_keepalive_expiry` as well. A client you pass as `client=` stays yours to close: `client.close()` closes both of its pools. +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. They share a connection pool of their own that keeps an idle connection for 120 seconds by default, so a request that follows the previous one within that time, such as a voice agent's next sentence, reuses its connection instead of opening a new one. If you set `aai.settings.keepalive_expiry` shorter than that, this pool uses it too; `aai.settings.tts_keepalive_expiry` sets this pool's idle time on its own. A client you pass as `client=` stays yours to close: `client.close()` closes both of its pools.
Synthesize a WAV file @@ -2042,7 +2042,7 @@ aai.settings.dictation_http_timeout = 300.0 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 -aai.settings.tts_keepalive_expiry = 120.0 # seconds an idle connection stays in the TTS pool (keepalive_expiry does not apply to it) +aai.settings.tts_keepalive_expiry = 120.0 # seconds an idle connection stays in the TTS pool (unset: 120, or keepalive_expiry when shorter) ``` 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`. diff --git a/assemblyai/async_client.py b/assemblyai/async_client.py index cd5b81f..321b6a2 100644 --- a/assemblyai/async_client.py +++ b/assemblyai/async_client.py @@ -122,8 +122,8 @@ def tts_http_client(self) -> httpx.AsyncClient: The HTTP client of the Text-to-Speech API, built when an `AsyncTextToSpeech` takes this client or on first use. - A connection pool of its own, whose idle connections stay - `settings.tts_keepalive_expiry` seconds (120 by default); see + A connection pool of its own, whose idle connections stay 120 + seconds by default (see `settings.tts_keepalive_expiry`); see `Client.tts_http_client`. `aclose()` closes it with `http_client`. Raises: diff --git a/assemblyai/client.py b/assemblyai/client.py index d671841..7a1c862 100644 --- a/assemblyai/client.py +++ b/assemblyai/client.py @@ -39,14 +39,34 @@ def _build_limits(settings: types.Settings) -> httpx.Limits: ) +# Seconds the Text-to-Speech pool keeps an idle connection when neither +# `tts_keepalive_expiry` nor `keepalive_expiry` is set. +_DEFAULT_TTS_KEEPALIVE_EXPIRY = 120.0 + + +def _tts_keepalive_expiry(settings: types.Settings) -> float: + """ + Returns the seconds the Text-to-Speech pool keeps an idle connection: + `settings.tts_keepalive_expiry` when set, else 120, or + `settings.keepalive_expiry` when that is set lower (someone who lowered + it because their network drops idle connections gets the same here). + """ + + if settings.tts_keepalive_expiry is not None: + return settings.tts_keepalive_expiry + if settings.keepalive_expiry is not None: + return min(settings.keepalive_expiry, _DEFAULT_TTS_KEEPALIVE_EXPIRY) + return _DEFAULT_TTS_KEEPALIVE_EXPIRY + + def _build_tts_limits(settings: types.Settings) -> httpx.Limits: """ Builds the limits of the Text-to-Speech pool: no cap on connections, as from `_build_limits`, and idle connections kept - `settings.tts_keepalive_expiry` seconds. Shared with `AsyncClient`. + `_tts_keepalive_expiry(settings)` seconds. Shared with `AsyncClient`. """ - return httpx.Limits(keepalive_expiry=settings.tts_keepalive_expiry) + return httpx.Limits(keepalive_expiry=_tts_keepalive_expiry(settings)) # What httpx raises from a request on a closed client, raised as well when a @@ -101,21 +121,18 @@ def _forget_tts_pools_in_child() -> None: Text-to-Speech pool the child inherited, so the child's next request builds a pool of its own. - An inherited pool holds the parent's idle connections, kept up to - `tts_keepalive_expiry` seconds (120 by default): a request made before - the fork, such as a warm-up in a `gunicorn --preload` app, leaves one - there. Processes sending on one connection at once garble each other's - responses. The pools are dropped rather than closed: another thread of the - parent may have held a pool's lock at the moment of the fork, and an async - pool needs its event loop. Each process closes its own socket descriptors, - so the parent's connections stay intact. + An inherited pool holds the parent's idle connections, kept up to 120 + seconds by default: a request made before the fork, such as a warm-up in + a `gunicorn --preload` app, leaves one there. Processes sending on one + connection at once garble each other's responses. The pools are dropped + rather than closed: another thread of the parent may have held a pool's + lock at the moment of the fork, and an async pool needs its event loop. + Each process closes its own socket descriptors, so the parent's + connections stay intact. Dropping relies on httpx not closing a client when it is garbage collected, which holds from httpx 0.22, the floor in `setup.py`. httpx 0.19 to 0.21 close it in `Client.__del__`, which takes the pool's lock. - Keeping the dropped pools referenced would not help there: the child's - interpreter would close them at exit, and its copies of the parent's - sockets would stay open until then. """ for holder in list(_TTS_POOL_HOLDERS): @@ -223,13 +240,12 @@ def tts_http_client(self) -> httpx.Client: The HTTP client of the Text-to-Speech API, built when a `TextToSpeech` takes this client or on first use. - A connection pool of its own, whose idle connections stay - `settings.tts_keepalive_expiry` seconds (120 by default): a voice + A connection pool of its own, whose idle connections stay 120 + seconds by default (see `settings.tts_keepalive_expiry`): a voice agent's next sentence reuses the connection of the previous one - instead of opening a new one. `http_client` and its - `settings.keepalive_expiry` are unaffected. Responses update - `last_response` as those of `http_client` do. `close()` closes it - with `http_client`. + instead of opening a new one. `http_client` is unaffected. + Responses update `last_response` as those of `http_client` do. + `close()` closes it with `http_client`. Raises: RuntimeError: if the pool was closed before it was ever built. diff --git a/assemblyai/tts/v1/client.py b/assemblyai/tts/v1/client.py index 9503e27..7659e09 100644 --- a/assemblyai/tts/v1/client.py +++ b/assemblyai/tts/v1/client.py @@ -22,9 +22,9 @@ class TextToSpeech: 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. They share the client's Text-to-Speech pool (`Client.tts_http_client`), - which keeps an idle connection `settings.tts_keepalive_expiry` seconds - (120 by default), so a request within that time of the previous one - reuses its connection. + which keeps an idle connection 120 seconds by default + (`settings.tts_keepalive_expiry`), so a request within that time of the + previous one reuses its connection. Example: ```python diff --git a/assemblyai/types.py b/assemblyai/types.py index ef08673..cb5ba27 100644 --- a/assemblyai/types.py +++ b/assemblyai/types.py @@ -132,7 +132,7 @@ class Settings(BaseSettings): "The default HTTP timeout for general requests" keepalive_expiry: Optional[float] = None - "Seconds an idle connection is kept in the HTTP pool before being closed. None uses httpx's default (5s). Raise it (e.g. to 120) so a connection opened by `SyncTranscriber.warm()` stays reusable across an in-progress recording. Text-to-Speech requests use `tts_keepalive_expiry` instead." + "Seconds an idle connection is kept in the HTTP pool before being closed. None uses httpx's default (5s). Raise it (e.g. to 120) so a connection opened by `SyncTranscriber.warm()` stays reusable across an in-progress recording. Text-to-Speech requests keep idle connections 120 seconds, or this long when it is shorter; `tts_keepalive_expiry` overrides both." base_url: str = "https://api.assemblyai.com" "The base URL for the AssemblyAI API" @@ -171,8 +171,8 @@ class Settings(BaseSettings): 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." - tts_keepalive_expiry: float = Field(default=120.0, ge=0.0) - "Seconds an idle connection stays in the Text-to-Speech HTTP pool (`TextToSpeech`/`AsyncTextToSpeech`). It is a pool of its own, which `keepalive_expiry` does not apply to: if you lowered `keepalive_expiry` because your network drops idle connections, lower this as well. A voice agent's turns are often more than httpx's default 5 seconds apart; a request within this time of the previous one reuses its connection and skips the TCP and TLS handshakes (a few milliseconds in the API's region, at least two network round trips from farther away). The client never reuses a connection that has been idle longer than this: lower it if a proxy or NAT on your network closes idle connections sooner." + tts_keepalive_expiry: Optional[float] = Field(default=None, ge=0.0) + "Seconds an idle connection stays in the Text-to-Speech HTTP pool (`TextToSpeech`/`AsyncTextToSpeech`), a pool of its own. None (the default) keeps it 120 seconds, or `keepalive_expiry` when you set that shorter. A voice agent's turns are often more than httpx's default 5 seconds apart; a request within this time of the previous one reuses its connection and skips the TCP and TLS handshakes (a few milliseconds in the API's region, at least two network round trips from farther away). The client never reuses a connection that has been idle longer than this: lower it if a proxy or NAT on your network closes idle connections sooner." 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 index fb1b611..20f0b28 100644 --- a/tests/unit/test_tts.py +++ b/tests/unit/test_tts.py @@ -14,7 +14,7 @@ import warnings import weakref from pathlib import Path -from typing import Iterator, List +from typing import Iterator, List, Optional import httpx import pytest @@ -687,7 +687,7 @@ def _ssl_context(http: httpx.Client) -> ssl.SSLContext: def test_the_tts_pool_keeps_idle_connections_120_s_by_default(): - assert aai.Settings().tts_keepalive_expiry == 120.0 + assert aai.Settings().tts_keepalive_expiry is None client = aai.Client(settings=aai.Settings(api_key="test")) pool = client.tts_http_client @@ -709,6 +709,35 @@ def test_the_tts_keepalive_expiry_is_a_setting(): aai.Settings(tts_keepalive_expiry=-1.0) +@pytest.mark.parametrize( + "keepalive_expiry, tts_keepalive_expiry, expected", + [ + # Given a keepalive_expiry lowered because the network drops idle + # connections, the TTS pool keeps them no longer either + (2.0, None, 2.0), + # Given one raised past 120 s, the TTS pool keeps its own 120 s + (300.0, None, 120.0), + # Given tts_keepalive_expiry as well, it wins either way + (2.0, 30.0, 30.0), + (300.0, 30.0, 30.0), + ], +) +def test_the_tts_pool_follows_a_shorter_keepalive_expiry( + keepalive_expiry: float, tts_keepalive_expiry: Optional[float], expected: float +): + settings = aai.Settings( + api_key="test", + keepalive_expiry=keepalive_expiry, + tts_keepalive_expiry=tts_keepalive_expiry, + ) + client = aai.Client(settings=settings) + + assert _keepalive_expiry(client.tts_http_client) == expected + # and the pool of the other APIs keeps keepalive_expiry + assert _keepalive_expiry(client.http_client) == keepalive_expiry + client.close() + + def test_the_tts_pool_is_built_with_the_text_to_speech_not_with_the_client(): client = aai.Client(api_key="test") assert client._tts_http_client is None # a client that never speaks diff --git a/tests/unit/test_tts_async.py b/tests/unit/test_tts_async.py index 3f30921..944e066 100644 --- a/tests/unit/test_tts_async.py +++ b/tests/unit/test_tts_async.py @@ -326,6 +326,17 @@ async def test_the_tts_pool_keeps_idle_connections_120_s(): await client.aclose() +async def test_the_tts_pool_follows_a_shorter_keepalive_expiry(): + # Given a keepalive_expiry lowered because the network drops idle + # connections + settings = aai.Settings(api_key="test", keepalive_expiry=2.0) + client = aai.AsyncClient(settings=settings) + + # The TTS pool keeps them no longer either + assert client.tts_http_client._transport._pool._keepalive_expiry == 2.0 # type: ignore[attr-defined] + await client.aclose() + + async def test_requests_go_through_the_tts_pool(httpx_mock: HTTPXMock): _mock_ok(httpx_mock) _mock_ok(httpx_mock)