breeze-blue 0.5.1

Python SDK for the Breeze Blue Developer API.

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="bluebell-v1",
    ) 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)

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.

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="bluebell-v1",
)

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

If a realtime 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.

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.