Breeze Blue Python SDK

Python SDK for the Breeze Blue Developer API. Covers text-to-speech, voice management, voice preview generation, history audio, models, and account usage.

Install

uv add breeze-blue

or:

pip install breeze-blue

API Key

Create an API key in the Breeze Blue Developer Console, then export it:

export BREEZE_API_KEY=brz_...

The SDK sends the key with the xi-api-key header.

Quickstart

from breeze_blue import BreezeBlue, save

client = BreezeBlue()

audio = client.text_to_speech.convert(
    voice_id="voc_...",
    text="Hello from Breeze Blue.",
    output_format="mp3",
)

save(audio, "hello.mp3")
print(audio.content_type)
print(audio.history_item_id)

BreezeBlue() reads BREEZE_API_KEY by default and sends requests to https://api.breeze.blue. To point at another environment, pass base_url=... or set BREEZE_BASE_URL.

client = BreezeBlue(
    api_key="brz_...",
    base_url="https://api.breeze.blue",
    timeout=120.0,
)

All resource methods accept timeout=<seconds> to override the client default for a single call.

Text to Speech

audio = client.text_to_speech.convert(
    voice_id="voc_...",
    text="Render this line.",
)

streamed_audio = client.text_to_speech.stream(
    voice_id="voc_...",
    text="Stream this line.",
)

enhanced = client.text_to_speech.enhance(
    instruction="Calm, warm, bedtime narration.",
    language_code="en",
)

Use async text-to-speech for long text, reference-heavy voices, or batch production where the caller should not hold an HTTP connection open:

job = client.text_to_speech.create_job(
    voice_id="voc_...",
    text="Render this longer script.",
    output_format="mp3",
)

status = client.generation_jobs.get(job["generation_job_id"])
if status["status"] == "ready":
    audio = client.generation_jobs.download_audio(job["generation_job_id"])
    audio.save("async.mp3")

If the job is still active, download_audio(...) raises GenerationNotReadyError; read exc.retry_after before retrying.

Use realtime text-to-speech when one WebSocket connection should handle multiple conversation turns. Realtime audio is fixed to raw pcm_s16le, 24000 Hz, mono, 16-bit frames. Start consuming before appending text: audio can arrive after flush() and may continue after end_turn(), so keep consuming until turn.done:

import asyncio

from breeze_blue import BreezeBlue, stream

client = BreezeBlue()


async def consume_audio(connection) -> tuple[bytes, dict]:
    audio = bytearray()
    audio_format = None
    async for event in connection:
        if event["type"] == "session.ready":
            audio_format = event["audio_format"]
        elif event["type"] == "audio":
            audio.extend(event["audio"])
        elif event["type"] == "error":
            raise RuntimeError(event.get("message"))
        elif event["type"] == "turn.done":
            if audio_format is None:
                raise RuntimeError("Missing realtime audio format")
            return bytes(audio), audio_format
    raise RuntimeError("Realtime session closed before turn.done")


async def speak() -> None:
    async with client.text_to_speech.connect_realtime(
        voice_id="voc_...",
        model_id="breeze-tts-2",
    ) as connection:
        consumer = asyncio.create_task(consume_audio(connection))
        await connection.start_turn("turn_1")
        await connection.append_text("Hello from Breeze.")
        await connection.flush()
        await connection.end_turn()
        audio, audio_format = await consumer

    stream(audio, audio_format=audio_format)


asyncio.run(speak())

The connection carries a single ordered message stream. Iterate the connection directly (as above), or use exactly one of connection.events() (non-audio events) and connection.audio() (raw PCM chunks) — running two consumers at the same time splits the stream between them. connection.audio() raises RealtimeError when the server reports an error event or the WebSocket closes abnormally, so audio-only consumers never mistake a failure for the end of a turn.

Not every server error ends the session: validation and protocol errors only cancel the active turn. Those raise a RealtimeError with recoverable set to True — catch it and start a new turn on the same connection:

from breeze_blue import RealtimeError

try:
    async for chunk in connection.audio():
        handle(chunk)
except RealtimeError as exc:
    if not exc.recoverable:
        raise
    # Only the current turn was cancelled; the connection is still open.
    await connection.start_turn("turn_2")
    await connection.append_text("Let me try that again.")
    await connection.flush()
    await connection.end_turn()
    async for chunk in connection.audio():
        handle(chunk)

With the low-level connect_realtime(...) API, the session ends after inactivity_timeout_seconds (30 seconds by default) without client messages. During idle gaps between turns, send await connection.ping() as a keepalive; the server replies with a pong event. connect_realtime(timeout=...) bounds the WebSocket handshake and defaults to 30 seconds.

For a long-running conversation, use the opt-in managed connection. It keeps the logical connection available across bounded physical WebSockets:

async with client.text_to_speech.connect_managed(
    voice_id="voc_...",
    model_id="breeze-tts-2",
) as connection:
    consumer = asyncio.create_task(consume_audio(connection))
    await connection.start_turn("turn_1")
    await connection.append_text("This conversation can continue across sessions.")
    await connection.flush()
    await connection.end_turn()
    audio, audio_format = await consumer

connect_managed(...) waits for session.ready, derives a safe heartbeat interval (including while an active turn is waiting for first audio), and requires an inbound server message within five seconds after each managed heartbeat. A missing acknowledgement triggers an idle reconnect; during an active turn it raises TURN_INTERRUPTED without replaying turn commands. The manager marks rotation pending at the earlier of the server's safe max_session_seconds deadline and a 600-second physical-session age threshold, or when the server sends session.expiring. It switches only after turn.done or turn.cancelled, so an active turn can keep the old epoch past that threshold; the server's 1,800-second hard lifetime still applies. If a new turn starts on the old socket while its replacement is opening, that turn keeps the old socket and rotation resumes after its terminal event. An idle reconnect uses fresh credentials and bounded equal-jitter backoff (three attempts by default); the physical close is hidden and the next visible epoch starts with another session.ready. A replacement epoch sends its first heartbeat immediately, then resumes the derived cadence after an inbound acknowledgement. If start_turn(...) arrives while that idle replacement is already in progress, it waits for the same bounded transition and sends turn.start once on the new epoch. Other turn commands are not buffered. Session setup retries transport failures and HTTP 408, 425, 429, and 5xx responses; other 4xx policy or configuration responses remain terminal. After a planned rotation, the replacement is available immediately while the old physical socket remains open for at most five seconds to forward a late usage.committed event. That event is best-effort; use history and account usage APIs as the durable source of truth.

The manager never buffers turn content or replays a command. The bounded start_turn(...) wait happens before its first WebSocket write. If a physical connection ends during an active turn, iteration raises RealtimeError with code == "TURN_INTERRUPTED"; rebuild the turn from application conversation state. Recoverable validation errors remain visible on the same socket, known transient upstream failures trigger an idle replacement without ending the logical audio iterator, and non-retryable policy or billing failures raise RealtimeError. A positive timeout is a shared setup deadline for obtaining fresh credentials, completing the WebSocket handshake, and receiving session.ready on every epoch. Set heartbeat_interval_seconds=0 to disable the automatic heartbeat. heartbeat_timeout_seconds must be positive and defaults to 5 seconds. Set max_physical_session_seconds=0 to disable the SDK age threshold (the server lifetime still applies), or pass a positive number to override the default 600 seconds. Tune rotation_margin_seconds, max_reconnect_attempts, and reconnect_base_delay_seconds when needed. Treat the manager as an active realtime call, not a presence channel. Exit the async context when the call ends, the user leaves, or the application enters a long-lived background state; otherwise its heartbeat intentionally keeps a server WebSocket slot occupied.

Every managed epoch creates a session through the SDK by default. If another service brokers short-lived credentials, pass a synchronous or async session_factory that returns a fresh client_secret or websocket_url on every call. Session options such as model_id must then be applied by that factory when it creates the session. Prefer an async factory for broker I/O. A synchronous factory runs through asyncio.to_thread; a manager setup timeout or close cannot forcibly stop a worker thread that has already started. Synchronous callbacks must therefore configure their own bounded network I/O timeouts.

To reduce time to first audio, create the session ahead of time (for example while waiting for user input) and connect with its client_secret when the first turn starts, keeping session setup off the critical path:

session = client.text_to_speech.create_realtime_session(
    "voc_...",
    model_id="breeze-tts-2",
)

async with client.text_to_speech.connect_realtime(
    voice_id="voc_...",
    client_secret=session["client_secret"],
) as connection:
    ...

Session parameters such as model_id are fixed when the session is created; values passed to connect_realtime together with client_secret are ignored (the SDK emits a UserWarning).

When using the low-level connect_realtime(...) API, if a WebSocket is interrupted by a network change, service deployment, or upstream realtime worker restart, handle an error event with meta.reconnect == True, a session.closed event with reconnect == True, or a RealtimeError with reconnect == True by creating a new connection and starting a new turn from your own conversation state. Active turns are not resumed in place. GENERATION_CAPACITY_EXCEEDED is turn-scoped: wait for the following turn.cancelled, back off using meta.retry_after_seconds, and start a new turn on the same managed logical connection without replaying text.

The API uses the default text-to-speech model when model_id is omitted. If you need to select a model explicitly, call client.models.list() and pass one of the returned model_id values.

Audio responses are returned as AudioResponse:

audio.content          # bytes
audio.content_type     # e.g. "audio/mpeg"
audio.history_item_id  # history item id when returned by the API

audio.save("speech.mp3")

Playback helpers are available for local scripts:

from breeze_blue import play, stream

play(audio)    # ffplay, with macOS afplay fallback
stream(audio)  # mpv; accepts AudioResponse, bytes, or an iterable of byte chunks

For raw PCM audio, pass the audio_format dict returned by realtime sessions (session["audio_format"] or the session.ready event) so stream(...) plays at the right sample rate and channel count; without it, raw PCM playback falls back to 24000 Hz mono s16le.

Streaming text-to-speech defaults to pcm to reduce time to first audio. Pass output_format="wav" or output_format="mp3" when you need that wire format explicitly.

audio.stream() is a buffered helper for AudioResponse. Use the module-level stream(audio_chunks) helper when you already have chunked audio data.

Voices

Browse existing voices and inspect a single voice:

voices = client.voices.search(search="narrator")
first_voice_id = voices["voices"][0]["voice_id"]

voice = client.voices.get(first_voice_id)
settings = client.voices.get_settings(first_voice_id)

random_voice = client.voices.random()
print(random_voice["voice_id"], random_voice["name"])

Breeze voice creation is always two steps: produce a preview, let the user accept it, then save the preview as a real voice. Two ways to produce a preview:

# Option A — clone preview from an audio sample
clone_preview = client.voices.create_clone_preview(
    name="Demo voice",
    file="sample.wav",
    text="This is a short preview script.",
)
generated_voice_id = clone_preview["generated_voice_id"]

# Option B — design preview from a text description (no audio)
design = client.voices.create_design_preview(
    voice_description="Warm documentary narrator with clear articulation.",
)
generated_voice_id = design["previews"][0]["generated_voice_id"]

files is also accepted with exactly one item.

Stream the preview so the user can audition it, then save the one they pick:

audio = client.voices.stream_preview(generated_voice_id)
audio.save("preview.mp3")

saved = client.voices.save_preview(
    generated_voice_id=generated_voice_id,
    voice_name="Documentary narrator",
)

Edit, tune settings, or delete a saved voice:

client.voices.edit(saved["voice_id"], name="Renamed narrator")
client.voices.edit_settings(saved["voice_id"], guidance_scale=1.2)
client.voices.delete(saved["voice_id"])

History, Models, and Account

models = client.models.list()
balance = client.account.balance()
usage = client.account.usage(days=7)
key_usage = client.account.usage(api_key_id="key_01hprod", client_type="sdk")

history = client.history.list(page_size=10)
item = client.history.get(history["history"][0]["history_item_id"])
audio = client.history.download_audio(item["history_item_id"])

Common response shapes are exported as TypedDict types from breeze_blue.types and from the package root.

Errors

All API errors inherit from BreezeBlueError. HTTP status codes and Breeze error codes map to typed exceptions:

• BadRequestError
• AuthenticationError
• ForbiddenError
• NotFoundError
• ConflictError
• ValidationError
• RateLimitError
• BillingInsufficientCreditsError
• UpstreamError
• ServiceUnavailableError

from breeze_blue import BreezeBlue, RateLimitError

try:
    BreezeBlue().models.list()
except RateLimitError as exc:
    print(exc.retry_after)

Each ApiError exposes status_code, code, detail, meta, and retry_after.

Realtime TTS WebSocket failures raise RealtimeError (also a BreezeBlueError), which exposes code, meta, close_code, and reconnect.