Skip to navigation
Get started

STT quickstart

Stream your first audio sample and receive a transcript.
View as Markdown

In this quickstart, you will stream a short PCM audio sample over WebSocket and print its final transcript.

1. Create an API key

Create a key in API Keys and copy it when it is shown. Store it in your server environment:

export NARI_API_KEY="your-api-key"

Keys are server-side credentials. The authentication guide covers key management and browser applications.

2. Download the sample audio

The example expects raw 16 kHz, mono PCM16 audio. Download this sample saying “Hello, welcome to Nari Labs.”

Download hello.pcm

3. Install the WebSocket client

Use Python 3.11+ for this example.

python -m pip install "websockets>=14,<16"

4. Create the transcription script

Save the following code as transcribe.py in the same folder as hello.pcm:

import asyncio
import base64
import json
import os
from pathlib import Path
from websockets.asyncio.client import connect
URL = "wss://api.narilabs.com/v1/realtime?intent=transcription"
class InsufficientCredits(RuntimeError):
    pass
async def receive(socket, expected=None):
    event = json.loads(await socket.recv())
    if event["type"] == "error":
        if event["error"]["code"] == "INSUFFICIENT_CREDITS":
            raise InsufficientCredits("Add credit before reconnecting")
        raise RuntimeError(event["error"]["code"] + ": " + event["error"]["message"])
    if event["type"] == "input_audio_buffer.commit_empty":
        raise RuntimeError("No pending audio to transcribe")
    if expected and event["type"] != expected:
        raise RuntimeError(f"Expected {expected}, received {event['type']}")
    return event
async def main():
    pcm = Path("hello.pcm").read_bytes()
    async with connect(
        URL,
        additional_headers={"Authorization": f"Bearer {os.environ['NARI_API_KEY']}"},
    ) as socket:
        await socket.send(json.dumps({
            "type": "session.configure",
            "session": {"model": "qwen3-asr-fast", "language": "en", "turn_detection": None},
        }))
        await receive(socket, "session.configured")
        async def send_audio():
            for offset in range(0, len(pcm), 3200):  # 100 ms
                chunk = pcm[offset:offset + 3200]
                await socket.send(json.dumps({
                    "type": "input_audio_buffer.append",
                    "audio": base64.b64encode(chunk).decode("ascii"),
                }))
                await asyncio.sleep(len(chunk) / 32000)
            await socket.send(json.dumps({"type": "input_audio_buffer.commit"}))
        async def read_results():
            while True:
                event = await receive(socket)
                if event["type"] == "transcript.partial":
                    print("Partial:", event["transcript"])
                elif event["type"] == "transcript.completed":
                    print("Final:", event["transcript"])
                    return
        async with asyncio.TaskGroup() as tasks:
            tasks.create_task(send_audio())
            tasks.create_task(read_results())
try:
    asyncio.run(main())
except* InsufficientCredits:
    print("Transcription stopped: add credit before reconnecting. Keep received final transcripts.")

5. Run the script

Run the script using the API key you exported in step 1:

python transcribe.py

The final output should resemble:

Final: Hello, welcome to Nari Labs.

Partial results may appear first; wording and punctuation can vary.

Why this model

Qwen3-ASR is best suited to streaming transcription applications where low time-to-final-segment (TTFS) matters.

Next steps