Skip to main content

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 create a session and receive a WebSocket URL and an access token.
  2. You connect to the WebSocket and 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 API Credentials​

Visit the Palabra API Keys page to obtain your Client ID and Client Secret. See Obtaining API Keys for details.

Step 2. Create a Session​

Call POST /session-storage/session with your credentials. The response contains the ws_url and the publisher token you need to connect.


import httpx

async def create_session(client_id: str, client_secret: str) -> dict:
    url = "https://api.palabra.ai/session-storage/session"
    headers = {"ClientId": client_id, "ClientSecret": client_secret}
    payload = {"data": {}}

    async with httpx.AsyncClient() as client:
        response = await client.post(url, json=payload, headers=headers)
        response.raise_for_status()
        return response.json()

Response Example​

{
  "data": {
    "publisher": "eyJhbGciOiJIU...Gxr2gjWSA4",
    "webrtc_url": "https://streaming-0.palabra.ai/livekit/",
    "ws_url": "wss://streaming-0.palabra.ai/streaming-api/v1/speech-to-speech/stream",
    "id": "7f99b553-4697...7d450728"
  }
}

You need two fields from the response:

  • ws_url — the WebSocket endpoint for streaming and control.
  • publisher — the JWT token used to authenticate the WebSocket connection.

Step 3. Connect to the WebSocket API​

Connect to ws_url, passing the publisher token as the token query parameter.

import websockets

async def connect_websocket(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 4. 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 5.

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 5. 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 6. 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 credentials
    client_id = os.getenv("PALABRA_CLIENT_ID")
    client_secret = os.getenv("PALABRA_CLIENT_SECRET")

    # Step 2: Create session
    session = await create_session(client_id, client_secret)
    ws_url = session["data"]["ws_url"]
    publisher_token = session["data"]["publisher"]

    # Step 3: Connect to WebSocket
    websocket = await connect_websocket(ws_url, publisher_token)

    # Step 4: 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 5 & 6: 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.
  • 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 [email protected].