Palabra AI Python Client

Simple Python client for Palabra AI real-time streaming APIs: speech-to-speech translation, speech-to-text, and low-latency text-to-speech.

uv add palabra-ai          # or: pip install palabra-ai

Full API documentation: docs.palabra.ai.

Palabra has three separate streaming APIs, and the client mirrors that at the top level:

Authentication, regions, errors and reconnection are shared by all three.

Authentication

Everything is driven by two values — an API Key and a region; all endpoints are derived from them automatically. Create your API Key on the Palabra API Keys page and set it via the environment or the constructor:

export PALABRA_API_KEY=...
export PALABRA_REGION=eu   # optional, defaults to "eu"

from palabra_ai import Palabra

palabra = Palabra()                            # reads the env vars
palabra = Palabra(api_key="...", region="eu")  # or explicit

The API Key authorizes the WebSocket connection directly: a streaming session is created server-side when you connect and cleaned up when the connection ends — there is nothing to manage.

Regions

Availability per region (more regions and endpoints are being added over time):

Opening a stream for a product that is not available in the configured region raises ValueError with the list of regions where it is.

Speech-to-Speech Translation API

You continuously push audio chunks; Palabra streams back transcripts, translations, and synthesized speech.

Quick start

Audio comes from your source — a microphone, a VoIP call leg, a telephony bridge — anything that hands you PCM chunks. Push them into the session and consume events:

import asyncio
import time
from palabra_ai import Palabra, Transcript, Audio

CHUNK_MS = 320  # ~320 ms of PCM (s16le, 24 kHz, mono)

async def main():
    palabra = Palabra()

    async with palabra.translation(source="en", targets=["es"]) as session:

        async def send_audio():
            next_send = time.monotonic()
            while chunk := await audio_buffer.get():  # your audio source
                await session.send_audio(chunk)
                next_send += CHUNK_MS / 1000
                await asyncio.sleep(max(0, next_send - time.monotonic()))
            await session.end(eos_timeout=4)

        sender = asyncio.create_task(send_audio())

        async for event in session:
            match event:
                case Transcript():
                    print(event)
                case Audio():
                    play(event.pcm)

        await sender

asyncio.run(main())

async with palabra.translation(...) does everything for you: connects the WebSocket (your API Key authorizes it; a streaming session is created server-side automatically), sends the translation task, waits until the pipeline actually confirms it, and cleans up on exit.

Two rules for the input stream:

1. Chunks must match the format declared in the task (default: PCM s16le, 24 kHz, mono; ~320 ms per chunk is optimal).
2. Push at real-time rate — faster/slower pacing triggers ServerWarning (AUDIO_STREAM_TOO_FAST/TOO_SLOW/STALLED) and degrades quality. If your source is a live device or call, pacing comes for free.

WebRTC (browser / client-side apps)

This client uses WebSocket transport, which is the recommended option for server-side applications. For browser and mobile apps Palabra recommends the WebRTC transport with a JavaScript client: follow the WebRTC Quick Start, or start from the official TypeScript example. WebRTC handles microphone capture, pacing, jitter, etc. in the browser natively.

Events

Iterating the session yields typed events:

Session control

await session.send_audio(chunk)                      # one raw chunk (pace it yourself)

await session.speak("Hola!", "es")                   # speak text into the stream (note: you must have this language as one of the target languages)
await session.speak("Hi all!", "en", translate=True) # translate to all targets first

await session.flush()                                # drop the current transcription and audio (interruption)
await session.pause();  await session.resume()       # pause and resume your session (stops billing)
await session.set_task(new_task)                     # change settings on the fly
await session.end(eos_timeout=4)                     # graceful finish: waits for the tail, emits StreamEnd

session.speak(text, lang) (the tts_task command) speaks through the translation pipeline and is unrelated to the standalone Realtime TTS API.

Settings

Common options are keyword arguments of translation(...); anything beyond that — build the task dict yourself:

from palabra_ai import build_task, Palabra

# common options inline
session = Palabra().translation(
    source="auto",
    targets=["es", "fr"],
    translate_partials=True,
    silence_threshold=0.8
)

# or full control, including per-target overrides and any server option
task = build_task(
    "en",
    {"es": {"speech_generation": {"voice_id": "default_high"}}, "fr": {}},
    input_sample_rate=48000,
)
task["pipeline"]["transcription"]["silence_threshold"] = 0.75

async with Palabra().translation(task=task) as session:
    ...

The client does not validate settings — invalid options are rejected by the server (TaskError is raised on async with, with the server's reason). The full list of options, their constraints, and tuning advice live in the docs: see Recommended Settings.

Offline files (utility)

For testing and batch jobs there are file helpers — but keep in mind this is a real-time service: the input is paced to real time, so translating a file takes roughly as long as the audio itself. For UX experiments and pipeline debugging it's convenient; for bulk offline processing it's the wrong tool.

palabra.translate_file(
    "speech_en.wav",
    source="en",
    targets="es",
    output="speech_es.wav",
    on_transcript=print
)
# mp3/ogg/resampling need: uv add "palabra-ai[audio]"

Related helpers: session.send_file(path), session.send_pcm(pcm) (chunking + real-time pacing built in), load_pcm / read_wav / write_wav.

Speech-to-Text API

Standalone transcription API (with optional translation). You push raw audio frames in and receive incremental transcriptions back; set translate_languages to also get a translation of each finalized segment.

All settings are sent as query parameters, audio goes out as raw binary frames.

Quick start

import asyncio
from palabra_ai import Palabra, SttTranscript

async def main():
    palabra = Palabra()

    async with palabra.stt(language="en") as stt:

        async def feed():
            # any audio source: PCM s16le, 16 kHz, mono, ~320 ms per chunk, real-time paced
            while chunk := await audio_buffer.get():
                await stt.send_audio(chunk)

        feeder = asyncio.create_task(feed())

        async for event in stt:
            if isinstance(event, SttTranscript):
                print(event) # "~ [en] partial" / "[en] final"

        await feeder

asyncio.run(main())

Iteration ends when the server closes the connection (or raises SessionError if the receive loop crashed).

Events

Iterating the session yields SttTranscript events (anything unrecognized comes through as Raw):

With the filler filter enabled the segment tail may be rewritten midsegment, so render text whole on each message rather than appending delta.

Settings

All settings are keyword arguments of stt(...) and become URL query parameters:

async with Palabra().stt(
    language="en",                       # source; defaults to auto-detect
    format="pcm_s16le",                  # see the audio-formats table in the docs
    sample_rate=16000,                   # required for raw PCM other than 16 kHz pcm_s16le
    translate_languages=["es", "de"],    # also emit translated_transcription per target
    enable_filler_filter=True,           # server default: True for every language but ja
) as stt:
    ...

Realtime TTS API

Standalone synthesis, no translation pipeline. Designed for incremental text (LLM token streams): send pieces as they come, audio chunks come back with minimal latency.

Two methods. send_text() -- incremental streaming, e.g. straight from an LLM token stream; mark the end of each sentence with eos=True and consume TtsChunk events as they arrive:

async with palabra.tts(language="en", voice_id="default_low") as tts:
    await tts.send_text("The sun was setting over the mountains,")
    await tts.send_text(" casting long golden shadows.", eos=True)

    async for chunk in tts: # TtsChunk: audio, generation_id, last_chunk, audio_len
        play(chunk.audio)
        if chunk.last_chunk:
            break

    await tts.cancel() # stop current synthesis, session stays open

Each send_text() message is limited to 256 characters (the server limit); longer text raises ValueError -- splitting is up to you.

synthesize() -- one sentence in, audio bytes out:

async with palabra.tts(language="en", voice_id="default_low") as tts:
    pcm = await tts.synthesize("Curious minds think alike.")   # bytes (pcm s16le by default)

Options & limits

All palabra.tts(...) options (languages, voices, speed, output formats, sample rates), rate limits, and constraints are described in the Realtime TTS API docs. Per-message voice overrides can be passed as keyword arguments of send_text()/synthesize().

TTS is currently available in the eu and us regions.

Common reference

Errors

Shared by all APIs:

• AuthError — missing/invalid API Key.
• SessionError — WebSocket connection problems (including a crashed receive loop — the original exception is attached as __cause__).
• NotReadyError — the pipeline didn't confirm set_task in time (translation only).
• TaskError — the server rejected set_task (raised immediately on async with, with the server's code/desc), or raised by session.raise_on_error(event) for server error messages; by default in-stream errors are delivered as ServerError events so a long-running stream survives recoverable errors.
• ValueError — the requested product is not available in the configured region, or an unknown region was set.

Reconnection

There is no automatic WebSocket reconnect, by design: a session is tied to one connection and to server-side pipeline state, so a transparent resume would silently lose the audio in flight and the transcription context. When the connection drops, iteration simply ends (or SessionError is raised if the receive loop crashed).

If your application needs resilience, build the retry loop on top — you control what state to restore:

while True:
    try:
        async with palabra.translation(source="en", targets=["es"]) as session:
            ...  # feed audio, consume events
        break  # finished normally
    except (SessionError, NotReadyError):
        await asyncio.sleep(1)  # reconnect with your own backoff policy

Examples

Development

uv sync --dev   # editable install + pytest/ruff
make check      # ruff check + tests + format check

Migrating from 1.x

Migrating from 0.x