Quick Start (WebSockets)

Best for backend-side apps. This guide covers the WebSocket-based integration, which is recommended for server-side applications. If you are building a client-side (browser or mobile) app, see the WebRTC Quick Start Guide instead.

Palabra API client

Consider using the Palabra API Python client for WS-based integrations.

How it works

With the WebSocket transport, everything happens over a single connection:

  1. You connect to the WebSocket endpoint with your API Key — a streaming session is created for you automatically.
  2. You send a set_task message with your translation settings.
  3. You stream your audio to Palabra as base64-encoded chunks.
  4. Palabra transcribes, translates, and synthesizes your speech in real time, then streams the translated audio and text back through the same connection.

Step 1. Get an API Key

Create an API Key on the Palabra API Keys page. See Authentication for details.

Step 2. Connect to the WebSocket API

Connect to the endpoint below, passing your API Key as the token query parameter (or in the Authorization header). The server validates the key and creates a streaming session for the lifetime of the connection automatically.

wss://streaming.palabra.ai/streaming-api/{random_hash}/v1/speech-to-speech/stream?token=<API_KEY>

Replace the {random_hash} path segment with any random URL-safe string generated by your client for the connection — it is used to distribute connections across translation servers.

import uuid
import websockets

async def connect_websocket(api_key: str):
    ws_url = (
        "wss://streaming.palabra.ai/streaming-api/"
        f"{uuid.uuid4().hex}/v1/speech-to-speech/stream"
    )
    websocket = await websockets.connect(
        f"{ws_url}?token={api_key}", ping_interval=10, ping_timeout=30
    )
    print("🔌 Connected to WebSocket")
    return websocket

Alternative: connect with a session token

If you prefer to manage streaming sessions yourself, create a session via POST /session-storage/session. The response contains the ws_url and the publisher token: connect to ws_url, passing the publisher token as the token query parameter.

import websockets

async def connect_websocket_with_session(ws_url: str, publisher_token: str):
    full_url = f"{ws_url}?token={publisher_token}"
    websocket = await websockets.connect(full_url, ping_interval=10, ping_timeout=30)
    print("🔌 Connected to WebSocket")
    return websocket

Step 3. Configure the Translation

Send a set_task message with your translation settings. The format, sample_rate, and channels you declare here must match the audio you will send in Step 4.

See the Translation management API for the full command reference and the Translation settings breakdown for every available option.

import json

async def configure_translation(websocket, source_lang: str, target_langs: list):
    settings = {
        "message_type": "set_task",
        "data": {
            "input_stream": {
                "content_type": "audio",
                "source": {
                    "type": "ws",
                    "format": "pcm_s16le",
                    "sample_rate": 24000,
                    "channels": 1
                }
            },
            "output_stream": {
                "content_type": "audio",
                "target": {
                    "type": "ws",
                    "format": "pcm_s16le",
                    "sample_rate": 24000,
                    "channels": 1
                }
            },
            "pipeline": {
                "preprocessing": {},
                "transcription": {
                    "source_language": source_lang
                },
                "translations": [
                    {
                        "target_language": lang,
                        "speech_generation": {}
                    } for lang in target_langs
                ]
            }
        }
    }

    await websocket.send(json.dumps(settings))
    print(f"⚙️ Translation configured: {source_lang} → {target_langs}")

Give the server a moment (2–3 seconds) to apply the settings before you start streaming audio.

Step 4. Send Audio Data

Capture audio from your source (for example, a microphone) and send it as base64-encoded chunks in input_audio_data messages. The recommended chunk length is 320 ms.

Pace your audio. Send chunks at the real-time rate (one 320 ms chunk every 320 ms). Sending audio faster than real time breaks transcription quality.
import asyncio
import base64
import json
import queue
import threading
import time
import numpy as np
import sounddevice as sd

async def stream_microphone(websocket):
    sample_rate = 24000          # must match `input_stream` in set_task
    chunk_duration = 0.32        # 320 ms chunks recommended
    chunk_samples = int(sample_rate * chunk_duration)

    audio_queue = queue.Queue(maxsize=100)
    stop_event = threading.Event()

    def input_callback(indata, frames, time_info, status):
        try:
            audio_queue.put_nowait(np.frombuffer(indata, dtype=np.int16).copy())
        except queue.Full:
            pass

    def recording_thread():
        with sd.RawInputStream(
                samplerate=sample_rate,
                channels=1,
                dtype='int16',
                callback=input_callback,
                blocksize=int(sample_rate * 0.02)  # 20 ms callback
        ):
            print("🎤 Microphone started")
            while not stop_event.is_set():
                time.sleep(0.01)

    threading.Thread(target=recording_thread, daemon=True).start()

    buffer = np.array([], dtype=np.int16)
    while True:
        try:
            audio_data = audio_queue.get(timeout=0.1)
            buffer = np.concatenate([buffer, audio_data])

            while len(buffer) >= chunk_samples:
                chunk = buffer[:chunk_samples]
                buffer = buffer[chunk_samples:]

                message = {
                    "message_type": "input_audio_data",
                    "data": {
                        "data": base64.b64encode(chunk.tobytes()).decode("utf-8")
                    }
                }
                await websocket.send(json.dumps(message))

                # Important: pace audio to the real-time rate
                await asyncio.sleep(chunk_duration)

        except queue.Empty:
            await asyncio.sleep(0.001)

Step 5. Receive and Play Translated Audio

Listen for messages on the same WebSocket. Palabra sends transcriptions, translations, and TTS audio chunks (output_audio_data).

import json
import base64
import queue
import numpy as np
import sounddevice as sd

async def receive_and_play(websocket):
    sample_rate = 24000          # must match `output_stream` in set_task
    audio_queue = queue.Queue(maxsize=100)
    buffer = np.array([], dtype=np.int16)

    def audio_callback(outdata, frames, time_info, status):
        nonlocal buffer
        while len(buffer) < frames:
            try:
                buffer = np.concatenate([buffer, audio_queue.get_nowait()])
            except queue.Empty:
                break

        if len(buffer) >= frames:
            outdata[:] = buffer[:frames].reshape(-1, 1)
            buffer = buffer[frames:]
        else:
            outdata.fill(0)

    output_stream = sd.OutputStream(
        samplerate=sample_rate,
        channels=1,
        dtype='int16',
        callback=audio_callback,
        blocksize=int(sample_rate * 0.02)
    )
    output_stream.start()
    print("🔊 Audio playback started")

    async for message in websocket:
        data = json.loads(message)

        # Some messages arrive with `data` as a JSON string — parse it
        if isinstance(data.get("data"), str):
            data["data"] = json.loads(data["data"])

        msg_type = data.get("message_type")

        if msg_type == "current_task":
            print("📝 Task confirmed")
        elif msg_type == "output_audio_data":
            audio_bytes = base64.b64decode(data["data"]["data"])
            audio_array = np.frombuffer(audio_bytes, dtype=np.int16)
            try:
                audio_queue.put_nowait(audio_array)
            except queue.Full:
                pass
        elif msg_type == "partial_transcription":
            text = data["data"]["transcription"]["text"]
            lang = data["data"]["transcription"]["language"]
            print(f"\r\033[K💬 [{lang}] {text}", end="", flush=True)
        elif msg_type == "validated_transcription":
            text = data["data"]["transcription"]["text"]
            lang = data["data"]["transcription"]["language"]
            print(f"\r\033[K✅ [{lang}] {text}")

Complete Example

A minimal runner combining all the steps.

import asyncio
import os
import signal

async def main():
    signal.signal(signal.SIGINT, lambda s, f: os._exit(0))
    print("🚀 Palabra WebSocket Client")

    # Step 1: Your API Key
    api_key = os.getenv("PALABRA_API_KEY")

    # Step 2: Connect to WebSocket (a session is created automatically)
    websocket = await connect_websocket(api_key)

    # Step 3: Configure translation and wait for it to apply
    await configure_translation(websocket, "en", ["es"])
    await asyncio.sleep(3)  # see the tip below for a deterministic check

    # Steps 4 & 5: Stream the microphone and play translated audio
    receive_task = asyncio.create_task(receive_and_play(websocket))
    stream_task = asyncio.create_task(stream_microphone(websocket))

    print("\n🎧 Listening... Press Ctrl+C to stop\n")

    try:
        await asyncio.gather(receive_task, stream_task)
    except KeyboardInterrupt:
        print("\n🛑 Shutdown complete")

if __name__ == "__main__":
    asyncio.run(main())

Summary

Once the WebSocket connection is open and your set_task settings are applied, Palabra processes your audio stream in real time: it transcribes your speech, translates it into the target languages, and streams both the text and the synthesized audio back through the same connection.

Good to know

  • The server does not acknowledge set_task directly. Instead of a fixed sleep, you can poll get_task (no more than once per 2 seconds): it returns a NOT_FOUND error until the pipeline has started, and a current_task message once your task is live.
  • Sessions created automatically on connect (API Key auth) are cleaned up for you after the connection closes. If you create sessions yourself, note that unused sessions stay active for at least 1 minute — to avoid hitting the limit on simultaneously active sessions, delete sessions you no longer need. See Session Management.
  • To pause, flush, or stop the translation, see the Translation management API.

Need help?

If you have any questions or need assistance, contact us at api.support@palabra.ai.