Realtime Speech-to-Text (ASR) API

wss://stream.palabra.ai/asr/v1/speech-to-text/stream?token=<API_KEY>&language=en&format=pcm_s16le&sample_rate=16000

{ "message_type": "finalize" }

{ "message_type": "transcription", "is_eos": true, "segment": { "text": "Hello world.<fin>", ... }, ... }

{ "message_type": "transcription", "is_eos": false, "segment": { "text": "<fin>", ... }, ... }

{
  "message_type": "transcription",
  "transcription_id": "a1b2c3d4",
  "language": "en",
  "is_eos": false,
  "segment": {
    "text": "Hello world how are",
    "start_time": 0.32,
    "end_time": 1.84
  },
  "delta": {
    "text": "how are",
    "start_time": 1.20,
    "end_time": 1.84
  }
}

{
  "message_type": "translated_transcription",
  "transcription_id": "a1b2c3d4",
  "language": "es",
  "is_eos": true,
  "segment": {
    "text": "Hola mundo, ¿cómo estás?",
    "start_time": 0.32,
    "end_time": 1.84
  }
}

pip install pyaudio websockets
export PALABRA_API_KEY=...        # from Step 1
export PALABRA_LANGUAGE=en        # source language

import json
import os
import asyncio
import threading
import queue

import pyaudio
import websockets

WS_URL = "wss://stream.palabra.ai/asr/v1/speech-to-text/stream"
LANGUAGE = os.environ.get("PALABRA_LANGUAGE", "en")

SAMPLE_RATE = 16000
CHANNELS = 1
CHUNK = 5120  # samples ≈ 320 ms at 16 kHz (recommended chunk size)


def mic_reader(audio_queue: queue.Queue, stop_event: threading.Event):
    pa = pyaudio.PyAudio()
    stream = pa.open(
        format=pyaudio.paInt16,
        channels=CHANNELS,
        rate=SAMPLE_RATE,
        input=True,
        frames_per_buffer=CHUNK,
    )
    print("Microphone open, speak now...")
    try:
        while not stop_event.is_set():
            audio_queue.put(stream.read(CHUNK, exception_on_overflow=False))
    finally:
        stream.stop_stream()
        stream.close()
        pa.terminate()


async def stream(key: str):
    url = (
        f"{WS_URL}?token={key}&language={LANGUAGE}"
        f"&format=pcm_s16le&sample_rate={SAMPLE_RATE}"
    )

    audio_queue: queue.Queue = queue.Queue()
    stop_event = threading.Event()
    threading.Thread(
        target=mic_reader, args=(audio_queue, stop_event), daemon=True
    ).start()

    async with websockets.connect(url) as ws:
        print("Connected")

        async def send_audio():
            loop = asyncio.get_event_loop()
            while True:
                data = await loop.run_in_executor(None, audio_queue.get)
                await ws.send(data)  # raw binary frame

        async def receive():
            async for message in ws:
                msg = json.loads(message)
                msg_type = msg.get("message_type")

                if msg_type == "transcription":
                    text = msg["segment"]["text"]
                    tid = msg.get("transcription_id", "")
                    if msg.get("is_eos"):
                        print(f"\n[EOS] {text} [{tid}]")
                    else:
                        # segment.text is the source of truth — render it whole
                        print(f"\r      {text}", end="", flush=True)

                elif msg_type == "translated_transcription":
                    lang = msg.get("language", "?")
                    tid = msg.get("transcription_id", "")
                    print(f"\n[{lang}] {msg['segment']['text']} [{tid}]")

        try:
            await asyncio.gather(send_audio(), receive())
        finally:
            stop_event.set()


if __name__ == "__main__":
    try:
        asyncio.run(stream(os.environ["PALABRA_API_KEY"]))
    except KeyboardInterrupt:
        print("\nStopped.")

pip install websockets
export PALABRA_API_KEY=...        # from Step 1
export PALABRA_LANGUAGE=en        # source language
python transcribe_file.py speech.wav

import asyncio
import json
import os
import sys
import wave

import websockets

WS_URL = "wss://stream.palabra.ai/asr/v1/speech-to-text/stream"
LANGUAGE = os.environ.get("PALABRA_LANGUAGE", "en")

CHUNK_MS = 320  # recommended chunk duration


async def transcribe_file(token: str, path: str):
    wav = wave.open(path, "rb")
    assert wav.getnchannels() == 1 and wav.getsampwidth() == 2, "need 16-bit mono WAV"
    sample_rate = wav.getframerate()
    chunk_frames = sample_rate * CHUNK_MS // 1000

    url = (
        f"{WS_URL}?token={token}&language={LANGUAGE}"
        f"&format=pcm_s16le&sample_rate={sample_rate}"
        "&finalization_mode=manual"
    )

    async with websockets.connect(url) as ws:
        print("Connected")

        async def send_audio():
            while True:
                data = wav.readframes(chunk_frames)
                if not data:
                    break
                await ws.send(data)  # raw binary frame
                await asyncio.sleep(CHUNK_MS / 1000)  # real-time pace
            # The file is over — no trailing silence will ever arrive,
            # so tell the recognizer the turn is done:
            await ws.send(json.dumps({"message_type": "finalize"}))
            print("\n(finalize sent)")

        async def receive():
            async for message in ws:
                msg = json.loads(message)
                if msg.get("message_type") != "transcription":
                    continue
                text = msg["segment"]["text"]
                if msg.get("is_eos"):
                    print(f"[final] {text}")
                else:
                    print(f"\r        {text}", end="", flush=True)
                if "<fin>" in text:
                    # the answer to our finalize — everything sent
                    # before it has now been transcribed
                    return

        await asyncio.gather(send_audio(), receive())


if __name__ == "__main__":
    asyncio.run(
        transcribe_file(os.environ["PALABRA_API_KEY"], sys.argv[1])
    )