Tazo Python SDK

Typed, synchronous and asynchronous clients for Tazo audio decisions. Supports exactly four public operations: uploaded decisions, streaming decisions, model discovery, and health. API-key management and internal APIs are outside this package.

Release status: This repository's 0.1.0 release has not been published to PyPI. tazo-sdk is the proposed PyPI name; verify ownership before publishing. Until then, install from source.

Install

Python 3.11 or newer is required. From the repository root:

python -m pip install ./packages/tazo-sdk
export TAZO_API_KEY='your-key'

After publication, use python -m pip install tazo-sdk. Create a key in the console; the SDK never manages keys.

Uploaded decision

from pathlib import Path
from tazo import Tazo

questions = {
    "action": {
        "type": "choice",
        "instructions": "Which action should be taken?",
        "criteria": {
            "get_weather": "Get weather for a city.",
            "NO_TOOL": "No tool is needed.",
        },
    }
}

with Tazo() as client:
    result = client.decisions.create(
        audio=Path("request.wav"),
        state="Choose the action matching the spoken request.",
        questions=questions,
    )
    print(result.answers["action"])
    print(result.request_id, result.idempotency_key)

model defaults to tazo-large. Choice and noul dictionaries or the optional ChoiceQuestion / NoulQuestion models are accepted. Check models.list() for model-specific capabilities. Score questions aren't supported by this SDK release.

Audio can be a path, bytes, or a binary stream. For bytes, pass filename and content_type when known. Uploads are snapshotted once, up to 16 MiB, so retries use the same bytes and MIME type. Caller-provided streams aren't closed; seekable streams are restored to their initial position. Nonseekable streams are consumed once. No prompt word limit is imposed by the SDK.

Models and health

from tazo import Tazo

with Tazo() as client:
    print(client.health.check().status)  # ok or unavailable; no auth sent
    for model in client.models.list().models:
        print(model.id, model.ready, model.capabilities.question_types)

Credentials are required only for protected calls. HTTP 503 with the documented unavailable health body returns HealthStatus(status="unavailable").

Async

import asyncio
from tazo import AsyncTazo


async def main():
    async with AsyncTazo() as client:
        print((await client.models.list()).models)


asyncio.run(main())

Await decisions.create, models.list, and health.check. Async streaming is an async context manager, not an awaitable factory.

Streaming

Both clients expose decisions.stream(state=..., questions=...). Entering the context authenticates in session.start and waits for session.ready. Inspect session.ready for limits. Send raw mono 16 kHz PCM16 little-endian bytes, at most max_audio_message_bytes per message. The wav_chunks helper strips WAV headers and validates the format; it does not resample or pace audio.

Send audio in one task while another consumes events:

import asyncio
from tazo import AsyncTazo, wav_chunks


async def stream_recording(questions):
    async with AsyncTazo() as client:
        async with client.decisions.stream(
            state="Choose an action.", questions=questions, audio_scope="turn"
        ) as session:

            async def send():
                for chunk in wav_chunks("request.wav"):
                    await asyncio.sleep(len(chunk) / 32000)  # microphone pacing
                    await session.send_audio(chunk)
                await session.commit()

            async def receive():
                async for event in session:
                    if event.type in {"decision.updated", "decision.finalized"}:
                        print(event.model_dump())
                    if event.type == "decision.finalized":
                        await session.finish()

            async with asyncio.TaskGroup() as group:
                group.create_task(send())
                group.create_task(receive())

The executable examples/stream.py also propagates producer failures immediately. For synchronous streaming, use a sender thread and iterate the session in the receiving thread. Never receive concurrently from multiple tasks or threads.

commit() sends a control message; consume decision.finalized before another turn. update(state=..., questions=...) is allowed between turns; consume session.updated before audio. finish() requests finalization and closure; continue iterating through session.closed. Context exit closes immediately. Fatal server errors raise StreamingError; recoverable errors are ErrorEvents. Future event types become UnknownEvent, preserving their fields.

Default audio_scope="turn" evaluates cumulative audio. chunk evaluates only new audio; its final result covers the last window, not the whole turn. Each evaluation counts against the account's per-model limits. Long recordings must be divided into turns within the server's advertised duration. There is no automatic reconnect, audio replay, silence detection, or microphone dependency.

Retries and errors

Each decisions.create() call generates a UUID operation key, unless you supply idempotency_key. All automatic attempts reuse the key, JSON, and audio. A new method call creates a new operation unless you explicitly reuse a retained key. Use a caller-generated key if you need recovery across process crashes or async cancellation. Accepted durable work can continue after a disconnect.

from pathlib import Path
from uuid import uuid4
from tazo import DecisionPendingError, Tazo

key = str(uuid4())  # persist alongside your original request before submission
with Tazo() as client:
    try:
        result = client.decisions.create(
            audio=Path("request.wav"),
            state="Choose an action.",
            questions=questions,
            idempotency_key=key,
        )
    except DecisionPendingError as error:
        print(error.request_id, error.idempotency_key)
        # Later: call create with this key and the exact same input and MIME type.

Default max_retries=2 means at most three attempts. Retryable service/network errors use bounded exponential backoff with jitter and respect Retry-After. Authentication, invalid input, conflicts, and oversized uploads are not retried. decision_pending may be retried with the retained key. There is no job-status endpoint. Recovery requires calling create again with the original input.

Exceptions expose status_code, code, request_id, idempotency_key, retryable, and retry_after where available. Invalid server responses raise ProtocolError. HTTP errors without JSON receive a safe generic message. The SDK does not log request content, credentials, or audio. Don't enable wire debug logging in applications that must keep these private.

Configuration and lifecycle

Defaults: base_url="https://api.tazolabs.com", connect timeout 5 seconds, read timeout 65 seconds, write timeout 30 seconds, pool timeout 5 seconds, and operation_timeout=180 seconds. Set a float or an httpx.Timeout for timeout.

The operation timeout bounds the HTTP retry budget and caps each network phase by the remaining time. Sync calls cannot forcibly interrupt a response that keeps delivering chunks or spans multiple phases; async HTTP additionally enforces a hard per-attempt deadline within the remaining budget. Upload snapshotting is before this budget. Streaming has its own event_timeout (default 65 seconds), 5-second connection timeout, and 2-second close timeout. All are seconds.

Custom base_url supports local HTTP and path-prefixed proxies; omit /v1. Use max_retries=0 with direct-inference development servers that don't persist idempotency keys. Safe replay requires a target honoring durable idempotency. WebSocket URLs are derived as ws/wss from that URL. Redirects aren't followed. Clients are reusable connection pools; close them after their streaming contexts. An injected http_client remains caller-owned. Injected clients should not have default authorization headers if health requests must remain unauthenticated.

Development and release

From the repository root, uv sync --all-packages --all-groups prepares development. CI uses an isolated environment and installs this package directly, preventing backend dependencies from masking missing SDK dependencies.

uv run --package tazo-sdk --group dev pytest packages/tazo-sdk/tests
uv run --package tazo-sdk --group dev ruff check packages/tazo-sdk
uv run --package tazo-sdk --group dev mypy --config-file packages/tazo-sdk/pyproject.toml packages/tazo-sdk/src/tazo

See RELEASING.md before publishing. API v1 and SDK versions are independent. This beta may change its Python interface before 1.0; changes are recorded in CHANGELOG.md. Response fields are additive-compatible.