Realtime transcription | Nari Labs

Send audio as it becomes available and receive partial transcripts while the speaker talks. At the end of each utterance, receive a final transcript. One connection can transcribe multiple utterances.

For a runnable example, follow the STT quickstart. STT connections share an organization concurrency limit.

Connect and configure

Connect from your backend using your API key:

wss://api.narilabs.com/v1/realtime?intent=transcription
Authorization: Bearer YOUR_NARI_API_KEY

Send your session configuration immediately after connecting:

{
  "type": "session.configure",
  "session": {
    "model": "qwen3-asr-fast"
  }
}

Wait for session.configured before sending audio. It returns the applied settings and defaults in session, with duration and idle limits in session.limits. The configuration is fixed for this connection.

Session arguments

These fields belong inside session in the initial session.configure:

Argument	Required / default	Purpose
`model`	Required	A public STT model ID
`language`	Optional; omitted or `null` means automatic recognition	Language hint such as `en`, `es`, or `ko`
`prompt`	Optional; empty by default	Names or terminology to help recognition, applied to every utterance
`turn_detection`	Optional; omitted or `null` means manual commit	Set `{"type":"server_vad"}` for automatic turn detection
`word_timestamps`	Optional; `false` by default	Set `true` to receive word-level alignment after each completed transcript

The API reference lists all accepted language codes, VAD fields, and validation limits.

Send audio

Use 16 kHz, signed 16-bit little-endian PCM, mono, without a WAV header. Encode each chunk as base64 and send it in an input_audio_buffer.append JSON message:

await socket.send(json.dumps({
    "type": "input_audio_buffer.append",
    "audio": base64.b64encode(pcm_chunk).decode("ascii"),
}))

For example, 100 ms of audio is 3,200 bytes before base64 encoding. Each chunk must contain whole two-byte samples, and the complete JSON message must fit within 128 KiB.

Send and receive in Python

This example streams audio and receives transcripts concurrently, supporting multiple utterances on one connection.

Use Python 3.11+, set NARI_API_KEY, and place the quickstart sample hello.pcm beside stt_stream.py.

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

Download stt_stream.py

"""WebSocket transcription example. Requires Python 3.11+ and websockets 14–15."""
import asyncio
import base64
import json
import os
from websockets.asyncio.client import connect
URL = "wss://api.narilabs.com/v1/realtime?intent=transcription"
class InsufficientCredits(RuntimeError):
    """Stop this session; the caller must obtain credit before reconnecting."""
async def pcm_file(path):
    with open(path, "rb") as audio:
        while chunk := audio.read(3200):  # 100 ms at 16 kHz PCM16 mono
            yield chunk
            await asyncio.sleep(len(chunk) / 32000)
async def transcribe(audio_source, *, turn_detection=None, word_timestamps=False, on_event=print):
    """Source yields PCM bytes, or None to manually commit an utterance."""
    async with connect(
        URL,
        additional_headers={"Authorization": f"Bearer {os.environ['NARI_API_KEY']}"},
    ) as socket:
        async def receive():
            event = json.loads(await socket.recv())
            if event["type"] == "error":
                on_event(event)  # Preserve the event/request IDs for support.
                if event["error"]["code"] == "INSUFFICIENT_CREDITS":
                    raise InsufficientCredits("Add credit before starting another session")
                raise RuntimeError(event["error"])
            return event
        await socket.send(json.dumps({
            "type": "session.configure",
            "session": {
                "model": "qwen3-asr-fast",
                "turn_detection": turn_detection,
                "word_timestamps": word_timestamps,
            },
        }))
        assert (await receive())["type"] == "session.configured"
        async def send_audio():
            commit_number = 0
            async for chunk in audio_source:
                if chunk is None:
                    commit_number += 1
                    event = {"type": "input_audio_buffer.commit",
                             "event_id": f"commit_{commit_number}"}
                else:
                    if not chunk or len(chunk) % 2:
                        raise ValueError("Send nonempty chunks of whole PCM16 samples")
                    event = {"type": "input_audio_buffer.append",
                             "audio": base64.b64encode(chunk).decode("ascii")}
                await socket.send(json.dumps(event))
            # Flush any remaining audio, including when using VAD.
            await socket.send(json.dumps({
                "type": "input_audio_buffer.commit", "event_id": "end_of_input",
            }))
        async def read_results():
            pending = set()
            pending_words = set()
            end_acknowledged = False
            while True:
                event = await receive()
                kind = event["type"]
                if kind == "input_audio_buffer.committed":
                    pending.add(event["item_id"])
                elif kind == "transcript.completed":
                    pending.discard(event["item_id"])
                    if word_timestamps:
                        pending_words.add(event["item_id"])
                elif kind in ("transcript.words", "transcript.words.failed"):
                    pending_words.discard(event["item_id"])
                if kind in ("input_audio_buffer.committed", "input_audio_buffer.commit_empty"):
                    if event.get("client_event_id") == "end_of_input":
                        end_acknowledged = True
                on_event(event)
                if end_acknowledged and not pending and not pending_words:
                    return
        async with asyncio.TaskGroup() as tasks:
            tasks.create_task(send_audio())
            tasks.create_task(read_results())
if __name__ == "__main__":
    try:
        asyncio.run(transcribe(pcm_file("hello.pcm")))
    except* InsufficientCredits:
        print("Transcription stopped: add credit before reconnecting. Keep received final transcripts.")

python stt_stream.py

For live audio, replace pcm_file(...) with an async iterator yielding PCM bytes. In this example, yielding None commits an utterance manually. Use on_event to handle incoming events.

See Transcripts and turn detection for callback and VAD examples, or Errors and troubleshooting for error handling.

Read the results

Partial transcripts may change as more audio arrives, including words returned earlier. After an utterance ends, the server returns a completed transcript that replaces all partial text for that item_id.

A partial result:

{
  "type": "transcript.partial",
  "event_id": "event_1",
  "item_id": "utterance_1",
  "transcript": "Hello, welcome",
  "revision": 1
}

The completed result for the same item_id:

{
  "type": "transcript.completed",
  "event_id": "event_3",
  "item_id": "utterance_1",
  "transcript": "Hello, welcome to Nari Labs.",
  "language": "en",
  "commit_reason": "manual",
  "usage": {
    "input_audio_seconds": 2.48
  }
}

Field	Use
`item_id`	Associate results with an utterance
`transcript`	Replace the current partial text or store the completed text
`language`	Recognized language code, or `null` if undetermined
`commit_reason`	What ended the utterance: `manual`, `vad`, or `max_duration`
`usage.input_audio_seconds`	Input audio duration assigned to this utterance, not billable duration; usage rules explain the difference

To receive word timing, configure "word_timestamps": true. The server first sends the authoritative transcript.completed, then sends one transcript.words event with the same item_id:

{
  "type": "transcript.words",
  "event_id": "event_4",
  "item_id": "utterance_1",
  "words": [
    {"word": "Hello", "start": 0.0, "end": 0.32},
    {"word": "welcome", "start": 0.35, "end": 0.78}
  ]
}

Alignment is asynchronous. Keep the completed transcript even if the terminal event is transcript.words.failed, and correlate timestamp events by item_id rather than assuming events for multiple utterances finish in commit order.

Transcripts and turn detection explains how to update text, finalize utterances, and reuse the connection. For timeouts and reconnecting, use WebSocket failures.

Session limits

These limits apply to all connections. The session.configured event returns the applied duration and idle limits in session.limits.

Limit	Default	Behavior
Utterance duration	36 seconds of input audio	Finalizes the utterance with `commit_reason: "max_duration"`; the connection stays open
Idle timeout	60 seconds without nonempty audio	Closes the connection; waiting for a final result is exempt
WebSocket message size	128 KiB per complete JSON message	Rejects oversized audio append events; use smaller chunks

See automatic utterance finalization and WebSocket errors for details.