Python WebSocket SDK

koldan-websocket-sdk is the official asyncio-based Python SDK for Koldan real-time speech recognition. It provides typed events, handles the complete WebSocket lifecycle, discovers streaming-capable models, validates audio frames, and lets applications stream audio without implementing the wire protocol.

The package supports Python 3.10 and newer.

Install

Download and extract the SDK archive:

Download koldan-websocket-sdk-python-9.1.3.zip

Install the prebuilt wheel from the extracted archive:

py -m venv .venv
.\.venv\Scripts\Activate.ps1
py -m pip install .\koldan-websocket-sdk-python-9.1.3\packages\koldan_websocket_sdk-9.1.3-py3-none-any.whl

python3 -m venv .venv
source .venv/bin/activate
python -m pip install ./koldan-websocket-sdk-python-9.1.3/packages/koldan_websocket_sdk-9.1.3-py3-none-any.whl

When the package is available from your configured Python package index, the equivalent command is:

python -m pip install koldan-websocket-sdk

The SDK has one required runtime dependency, websockets. Model discovery, WAV parsing, and audio framing use the Python standard library.

Configure

Keep credentials in environment variables:

$env:KOLDAN_BASE_URL = "https://koldan.dixilang.com"
$env:KOLDAN_API_KEY = "YOUR_API_KEY"
$env:KOLDAN_MODEL = "general"
$env:KOLDAN_WAV_FILE = "C:\path\to\speech-16kHz.wav"

export KOLDAN_BASE_URL="https://koldan.dixilang.com"
export KOLDAN_API_KEY="YOUR_API_KEY"
export KOLDAN_MODEL="general"
export KOLDAN_WAV_FILE="/path/to/speech-16kHz.wav"

The API key needs the speech:sessions:write scope. Model discovery also needs speech:model-aliases:read.

Stream a WAV File

transcribe() is the simplest API. It sends audio and receives recognition events concurrently, sends session.end when the audio source is exhausted, and closes the connection after completion.

import asyncio
import os

from koldan_websocket import (
    ErrorMessage,
    FinalResult,
    KoldanStreamingClient,
    PartialResult,
    SessionCompleted,
    SessionInfo,
    StreamingSessionOptions,
    wav_file_frames,
)


async def main() -> None:
    options = StreamingSessionOptions(
        model=os.environ.get("KOLDAN_MODEL", "general"),
        language="he-IL",
    )

    async with KoldanStreamingClient(
        base_url=os.environ["KOLDAN_BASE_URL"],
        api_key=os.environ["KOLDAN_API_KEY"],
    ) as client:
        async for event in client.transcribe(
            wav_file_frames(os.environ["KOLDAN_WAV_FILE"]),
            options,
            frame_interval_seconds=0.1,
        ):
            if isinstance(event, SessionInfo):
                print(f"Session: {event.session_id}")
            elif isinstance(event, PartialResult):
                print(f"\rPartial: {event.transcript}", end="", flush=True)
            elif isinstance(event, FinalResult):
                print(f"\nFinal[{event.segment_index}]: {event.transcript}")
            elif isinstance(event, ErrorMessage):
                label = "Warning" if event.is_warning else "Error"
                print(f"\n{label} {event.code}: {event.message}")
            elif isinstance(event, SessionCompleted):
                print(
                    f"\nCompleted: {event.total_segments} segments, "
                    f"{event.total_words} words"
                )


asyncio.run(main())

frame_interval_seconds=0.1 matches the helper's default 100 ms frames and paces file playback in real time. A live microphone source is already paced by the capture device and should omit this argument.

Select a Channel From a Multichannel WAV

wav_file_frames() accepts uncompressed 16-bit PCM WAV files and returns mono PCM frames from one zero-based channel:

# Stream only the first channel from a stereo recording.
audio = wav_file_frames("stereo-16kHz.wav", channel=0)

The helper validates a 16 kHz sample rate by default. It does not silently resample or convert compressed audio.

Stream a Microphone

Install the optional capture dependency:

python -m pip install "koldan-websocket-sdk[microphone]"

The downloaded SDK archive includes a complete examples/microphone.py program. Run it after setting the environment variables:

python examples/microphone.py

It captures 16 kHz, 16-bit, mono frames from the default input device and streams them until you press Enter. The SDK accepts any synchronous or asynchronous source of bytes, so you can integrate another audio framework without changing the recognition code.

Discover a Streaming Model

models = await client.list_streaming_models()
for model in models:
    print(model.id, model.display_name, model.capabilities.languages)

This calls GET /api/v1/speech-services/models with the configured credential and returns only aliases with supportsStreaming: true.

Configure All Streaming Features

from koldan_websocket import DiarizationOptions, StreamingSessionOptions

options = StreamingSessionOptions(
    model="general",
    language="he-IL",
    enable_endpoint_detection=True,
    record_audio=True,
    inverse_text_normalization=True,
    diarization=DiarizationOptions(
        enabled=True,
        max_speakers=-1,
    ),
    metadata={
        "externalId": "case-4821",
        "department": "radiology",
    },
)

max_speakers may be omitted, set to -1 for automatic speaker-count detection, or set to a positive integer.

Events

All events are immutable dataclasses with snake-case Python attributes.

Each WordTiming contains word, start_seconds, end_seconds, confidence, and optional speaker_tag.

RECORDING_UNAVAILABLE is a non-fatal warning and the session continues without recording. Other ErrorMessage events end the stream. An error that rejects startup raises KoldanServerError; its .error property contains the typed ErrorMessage.

Lower-Level Session Control

Use start_session() when your application manages producer and consumer tasks itself:

import asyncio

session = await client.start_session(options)


async def send_audio() -> None:
    for frame in wav_file_frames("speech-16kHz.wav"):
        await session.send_audio(frame)
        await asyncio.sleep(0.1)
    await session.finish()


sender = asyncio.create_task(send_audio())
try:
    async for event in session:
        print(event)
finally:
    await session.close()
    await sender

Only one coroutine may call receive() or iterate over a session. Sending and receiving from separate coroutines is supported and recommended.

Audio Requirements

At 16 kHz, 16-bit mono, 100 ms is 3,200 bytes. Use pcm_file_frames() for an existing raw PCM file or pcm_frame_size() to calculate a frame size.

Authentication, TLS, and Timeouts

Pass either api_key or bearer_token. If both are present, the API key takes precedence. Authentication is sent inside session.start; credentials are never placed in the URL or WebSocket upgrade headers.

TLS certificate and hostname verification are enabled by default. For a private certificate authority, load it into an ssl.SSLContext and pass that context as ssl_context. Do not disable certificate verification in production.

The client also exposes session_start_timeout, connect_timeout, close_timeout, http_timeout, ping_interval, and ping_timeout for deployment-specific tuning.

Exceptions

All SDK errors inherit from KoldanError.

Related Documentation

• Streaming WebSocket API
• Streaming Fundamentals
• Streaming Sessions
• Transcription Models