RinggLabs Python SDK

ringglabs is the official Python SDK for RinggLabs-compatible speech-to-text services.

It supports:

• Sync and async file transcription (REST/offline)
• Sync and async real-time transcription (streaming)
• A LiveKit Agents STT plugin (optional extra, see LiveKit Agents)

Installation

pip install ringglabs

With the LiveKit Agents plugin:

pip install "ringglabs[livekit]"

Compatibility

• Python: 3.10+
• Windows: 10+
• macOS: 11+
• Ubuntu: 20.04+

Parameter Reference

Client Initialization

Client(...) and AsyncClient(...) share the same constructor parameters.

Parameter	Type	Default	Description
base_url	str | None	"prod-api.ringg.ai"	Domain with an optional route. See Overriding base_url.
api_key	str | None	None	Default API key for all requests; can be overridden per call.
timeout	TimeoutConfig | None	TimeoutConfig()	HTTP and WS timeout budgets.
default_headers	dict[str, str] | None	None	Extra headers attached to SDK requests.

Overriding base_url

base_url takes a domain, or a domain + route. A leading http://, https://, ws:// or wss:// is stripped, and the SDK always connects with ws:// for streaming and http:// for REST and health. This means https:// and wss:// inputs do not encrypt the connection.

base_url	WebSocket URL	REST URL	Health URL
not set	ws://prod-api.ringg.ai/stt/v1/stream	http://prod-api.ringg.ai/stt/v1/transcriptions	http://prod-api.ringg.ai/health
localhost:9000	ws://localhost:9000/stt/v1/stream	http://localhost:9000/stt/v1/transcriptions	http://localhost:9000/health
wss://prod-api.ringg.ai/stt/v2/stream	ws://prod-api.ringg.ai/stt/v2/stream	http://prod-api.ringg.ai/stt/v2/stream	http://prod-api.ringg.ai/health

• A route replaces the default route for both WebSocket and REST, so the server must accept each call type you use on that route.
• With no route, the default routes are used.
• Health checks always use /health on the given domain.
• Query strings and fragments are dropped, and a trailing / is removed.
• Other schemes, an empty value, a missing host, or whitespace raise ValueError.

transcribe(...) Parameters

Client.transcribe(...) and AsyncClient.transcribe(...):

Parameter	Type	Default	Description
source	str | Path | bytes | bytearray | BinaryIO	required	Input audio source.
language	str	"hi"	Language hint sent to proxy.
enable_cap_punc	bool	True	Enable capitalization and punctuation in output.
api_key	str | None	None	Per-request API key override.
filename	str | None	auto-detected, fallback "audio.wav"	Multipart filename for upload.
content_type	str	"audio/wav"	Multipart content type.

stream(...) Parameters

Client.stream(...) and AsyncClient.stream(...):

Parameter	Type	Default	Description
sample_rate	int	16000	Audio sample rate for stream start config.
encoding	str	"int16"	Audio encoding (int16, linear16, float32, int32).
language	str	"hi"	Language hint sent to proxy.
mode	str	"stream"	Streaming mode (stream or on_final).
vad_tail_sil_ms	int	200	Server VAD tail silence in ms.
vad_confidence	float	0.55	Server VAD confidence threshold.
enable_cap_punc	bool	True	Enable capitalization/punctuation processing.
accept_client_vad_events	bool	False	Enables start_speaking()/stop_speaking() signaling from client.
api_key	str | None	None	Per-stream API key override.

Stream Session Controls

Sync session (StreamSession)

Method	Purpose
send_audio(bytes)	Send raw audio bytes chunk.
send_vad_event(state)	Send explicit VAD event (user_start_speaking, user_stop_speaking).
start_speaking()	Convenience wrapper for user_start_speaking.
stop_speaking()	Convenience wrapper for user_stop_speaking.
ping()	Send ping frame/message.
end(command="end")	Request stream finalization/termination.
recv_event()	Receive one parsed server event.
events()	Iterator over incoming parsed events.
close()	Close websocket session.

Async session (AsyncStreamSession)

Async equivalents:

• await send_audio(...)
• await send_vad_event(...)
• await start_speaking()
• await stop_speaking()
• await ping()
• await end(...)
• await recv_event()
• async for event in session.events(): ...
• await close()

Timeout Configuration

TimeoutConfig fields:

Field	Default	Description
connect	10.0	HTTP connect timeout (seconds).
read	10.0	HTTP read timeout (seconds).
write	10.0	HTTP write timeout (seconds).
pool	10.0	HTTP pool timeout (seconds).
ws_open	10.0	WebSocket open timeout (seconds).
ws_recv	30.0	WebSocket receive timeout (seconds). Use None to wait indefinitely for long idle streams.
ws_close	10.0	WebSocket close timeout (seconds).
ws_ping_interval	30.0	SDK JSON heartbeat interval. Sends {"type":"ping"} to keep proxy app-level session timeout alive during silence. Use None or 0 to disable.
ws_ping_timeout	10.0	Reserved heartbeat pong timeout budget for applications that monitor pong events.
ws_max_queue	None	Passed to websockets.connect(max_queue=...).
ws_max_size	None	Passed to websockets.connect(max_size=...).

The SDK heartbeat uses the proxy JSON ping/pong protocol, not websocket protocol-level ping frames. pong responses are delivered through the normal event receive path, so heartbeat does not consume application events.

Response Models

RestTranscriptionResult

Field	Type
status	str
transcription	str
is_final	bool
language	str
duration_seconds	float
processing_time_seconds	float
request_id	str
raw	dict

WebSocket Event Types

• ready
• transcript
• ack
• pong
• error

All events include raw with original server payload.

transcript event fields may include:

• transcription
• is_final
• language
• request_id
• segment_idx
• segments
• compute_latency_ms
• audio_duration_sec
• transcribed_audio_duration_sec
• processing_time_ms

Health Check

Sync:

from ringglabs.stt import Client

with Client(api_key="rk_live_xxx") as client:
    print(client.health())

Async:

from ringglabs.stt import AsyncClient

health = await AsyncClient(api_key="rk_live_xxx").health()
print(health)

Examples (Stream, Offline)

Helper Functions

import wave


def load_wav_mono_int16_16k(path: str) -> tuple[int, bytes]:
    with wave.open(path, "rb") as wf:
        channels = wf.getnchannels()
        sample_width = wf.getsampwidth()
        sample_rate = wf.getframerate()
        frames = wf.getnframes()
        audio = wf.readframes(frames)

    if channels != 1:
        raise ValueError("Expected mono WAV (1 channel).")
    if sample_width != 2:
        raise ValueError("Expected 16-bit PCM WAV (sample width = 2).")
    if sample_rate != 16000:
        raise ValueError("Expected 16 kHz WAV for these full-audio stream examples.")

    return sample_rate, audio


def iter_pcm_chunks(audio: bytes, sample_rate: int, chunk_ms: int = 20):
    samples_per_chunk = max(1, int(sample_rate * (chunk_ms / 1000.0)))
    bytes_per_chunk = samples_per_chunk * 2  # int16 mono
    for i in range(0, len(audio), bytes_per_chunk):
        yield audio[i : i + bytes_per_chunk]

1) Sync Stream (mode="stream")

from ringglabs.stt import Client, TimeoutError as SdkTimeoutError


def main() -> None:
    sample_rate, audio = load_wav_mono_int16_16k("sample.wav")
    transcripts: list[str] = []

    with Client(api_key="rk_live_xxx").stream(
        sample_rate=sample_rate,
        encoding="int16",
        language="en",
        mode="stream",
        enable_cap_punc=True,
        accept_client_vad_events=False,
    ) as session:
        for chunk in iter_pcm_chunks(audio, sample_rate, chunk_ms=20):
            session.send_audio(chunk)
        session.end()

        try:
            for event in session.events():
                if event.type == "transcript" and event.transcription.strip():
                    transcripts.append(event.transcription.strip())
        except SdkTimeoutError:
            pass

    print("segment transcripts:", transcripts)


if __name__ == "__main__":
    main()

2) Async Stream (mode="stream")

import asyncio
from ringglabs.stt import AsyncClient, TimeoutError as SdkTimeoutError


async def main() -> None:
    sample_rate, audio = load_wav_mono_int16_16k("sample.wav")
    transcripts: list[str] = []

    async with AsyncClient(api_key="rk_live_xxx").stream(
        sample_rate=sample_rate,
        encoding="int16",
        language="en",
        mode="stream",
        enable_cap_punc=True,
        accept_client_vad_events=False,
    ) as session:
        for chunk in iter_pcm_chunks(audio, sample_rate, chunk_ms=20):
            await session.send_audio(chunk)
        await session.end()

        try:
            async for event in session.events():
                if event.type == "transcript" and event.transcription.strip():
                    transcripts.append(event.transcription.strip())
        except SdkTimeoutError:
            pass

    print("segment transcripts:", transcripts)


if __name__ == "__main__":
    asyncio.run(main())

3) Sync Stream (mode="on_final")

from ringglabs.stt import Client, TimeoutError as SdkTimeoutError


def main() -> None:
    sample_rate, audio = load_wav_mono_int16_16k("sample.wav")
    partials: list[str] = []
    finals: list[str] = []

    with Client(api_key="rk_live_xxx").stream(
        sample_rate=sample_rate,
        encoding="int16",
        language="en",
        mode="on_final",
        enable_cap_punc=True,
        accept_client_vad_events=True,
    ) as session:
        session.start_speaking()
        for chunk in iter_pcm_chunks(audio, sample_rate, chunk_ms=20):
            session.send_audio(chunk)
        session.stop_speaking()
        session.end()

        try:
            for event in session.events():
                if event.type != "transcript":
                    continue
                text = event.transcription.strip()
                if not text:
                    continue
                if event.is_final:
                    finals.append(text)
                else:
                    partials.append(text)
        except SdkTimeoutError:
            pass

    print("partials:", partials)
    print("finals:", finals)
    if finals:
        print("final transcript:", finals[-1])


if __name__ == "__main__":
    main()

4) Async Stream (mode="on_final")

import asyncio
from ringglabs.stt import AsyncClient, TimeoutError as SdkTimeoutError


async def main() -> None:
    sample_rate, audio = load_wav_mono_int16_16k("sample.wav")
    partials: list[str] = []
    finals: list[str] = []

    async with AsyncClient(api_key="rk_live_xxx").stream(
        sample_rate=sample_rate,
        encoding="int16",
        language="en",
        mode="on_final",
        enable_cap_punc=True,
        accept_client_vad_events=True,
    ) as session:
        await session.start_speaking()
        for chunk in iter_pcm_chunks(audio, sample_rate, chunk_ms=20):
            await session.send_audio(chunk)
        await session.stop_speaking()
        await session.end()

        try:
            async for event in session.events():
                if event.type != "transcript":
                    continue
                text = event.transcription.strip()
                if not text:
                    continue
                if event.is_final:
                    finals.append(text)
                else:
                    partials.append(text)
        except SdkTimeoutError:
            pass

    print("partials:", partials)
    print("finals:", finals)
    if finals:
        print("final transcript:", finals[-1])


if __name__ == "__main__":
    asyncio.run(main())

5) Sync Transcribe (Offline)

from ringglabs.stt import Client


def main() -> None:
    with Client(api_key="rk_live_xxx") as client:
        result = client.transcribe(
            "sample.wav",
            language="en",
            enable_cap_punc=True,
            content_type="audio/wav",
        )
        print("request_id:", result.request_id)
        print("transcription:", result.transcription)


if __name__ == "__main__":
    main()

6) Async Transcribe (Offline)

import asyncio
from ringglabs.stt import AsyncClient


async def main() -> None:
    async with AsyncClient(api_key="rk_live_xxx") as client:
        result = await client.transcribe(
            "sample.wav",
            language="hi",
            enable_cap_punc=True,
            content_type="audio/wav",
        )
        print("request_id:", result.request_id)
        print("transcription:", result.transcription)


if __name__ == "__main__":
    asyncio.run(main())

7) Sync Transcribe from bytes and BinaryIO

from io import BytesIO
from pathlib import Path
from ringglabs.stt import Client


def main() -> None:
    wav_bytes = Path("sample.wav").read_bytes()

    with Client(api_key="rk_live_xxx") as client:
        # bytes source
        bytes_result = client.transcribe(
            wav_bytes,
            language="en",
            enable_cap_punc=True,
            filename="sample_bytes.wav",
            content_type="audio/wav",
        )
        print("bytes transcription:", bytes_result.transcription)

        # BinaryIO source
        fileobj = BytesIO(wav_bytes)
        fileobj_result = client.transcribe(
            fileobj,
            language="en",
            enable_cap_punc=True,
            filename="sample_fileobj.wav",
            content_type="audio/wav",
        )
        print("fileobj transcription:", fileobj_result.transcription)


if __name__ == "__main__":
    main()

8) Async Transcribe from bytes and BinaryIO

import asyncio
from io import BytesIO
from pathlib import Path
from ringglabs.stt import AsyncClient


async def main() -> None:
    wav_bytes = Path("sample.wav").read_bytes()

    async with AsyncClient(api_key="rk_live_xxx") as client:
        # bytes source
        bytes_result = await client.transcribe(
            wav_bytes,
            language="hi",
            enable_cap_punc=True,
            filename="sample_bytes.wav",
            content_type="audio/wav",
        )
        print("bytes transcription:", bytes_result.transcription)

        # BinaryIO source
        fileobj = BytesIO(wav_bytes)
        fileobj_result = await client.transcribe(
            fileobj,
            language="hi",
            enable_cap_punc=True,
            filename="sample_fileobj.wav",
            content_type="audio/wav",
        )
        print("fileobj transcription:", fileobj_result.transcription)


if __name__ == "__main__":
    asyncio.run(main())

Error Handling

Common SDK exceptions:

• ApiError (includes status_code, code, payload)
• AuthenticationError
• TimeoutError
• TransportError
• ProtocolError

from ringglabs.stt import Client, ApiError, TimeoutError, TransportError


try:
    with Client(api_key="rk_live_xxx") as client:
        result = client.transcribe("sample.wav")
        print(result.transcription)
except TimeoutError:
    print("request timed out")
except TransportError:
    print("network/connection failure")
except ApiError as exc:
    print("api error:", exc.status_code, exc.code, exc.message)

Retry Wrapper Examples (retry and async_retry)

Use bounded retries for transient transport/timeouts only.

Sync bounded retry

from ringglabs.stt import Client, retry, TimeoutError, TransportError


with Client(api_key="rk_live_xxx") as client:
    def run_once():
        return client.transcribe("sample.wav", language="en", enable_cap_punc=True)

    result = retry(
        run_once,
        attempts=3,
        initial_backoff_sec=0.25,
        max_backoff_sec=1.0,
        retry_on=(TimeoutError, TransportError),
    )
    print(result.transcription)

Async bounded retry

import asyncio
from ringglabs.stt import AsyncClient, async_retry, TimeoutError, TransportError


async def main() -> None:
    async with AsyncClient(api_key="rk_live_xxx") as client:
        async def run_once():
            return await client.transcribe("sample.wav", language="en", enable_cap_punc=True)

        result = await async_retry(
            run_once,
            attempts=3,
            initial_backoff_sec=0.25,
            max_backoff_sec=1.0,
            retry_on=(TimeoutError, TransportError),
        )
        print(result.transcription)


if __name__ == "__main__":
    asyncio.run(main())

LiveKit Agents

The livekit extra adds RinggSTT, a streaming speech-to-text plugin for LiveKit Agents. It plugs into an AgentSession like any other LiveKit STT provider; the SDK handles the WebSocket, handshake and event parsing underneath.

pip install "ringglabs[livekit]"
# optional, for client-side endpointing:
pip install "livekit-agents[silero]"

Minimal Agent

import os

from livekit.agents import Agent, AgentSession, JobContext, WorkerOptions, cli
from livekit.plugins import silero

from ringglabs.integrations.livekit import RinggSTT


async def entrypoint(ctx: JobContext) -> None:
    vad = silero.VAD.load()
    session = AgentSession(
        vad=vad,
        stt=RinggSTT(api_key=os.environ["RINGG_API_KEY"], language="hi", vad=vad),
        llm=...,  # any LiveKit LLM plugin
        tts=...,  # any LiveKit TTS plugin
    )
    await session.start(agent=Agent(instructions="You are a helpful assistant."), room=ctx.room)


if __name__ == "__main__":
    cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint))

RinggSTT(...) Parameters

Parameter	Type	Default	Description
api_key	str | None	None	Ringg API key. Falls back to the RINGG_API_KEY environment variable. Required unless a custom base_url is given.
base_url	str | None	None	Same semantics as the SDK client. None uses the public default. See Overriding base_url.
language	str	"hi"	Transcription language. Can be overridden per stream via stream(language=...).
vad	livekit.agents.vad.VAD | None	None	A LiveKit VAD (for example silero.VAD.load()). When given, the plugin switches to client-side endpointing. See below.
mode	str | None	None	"stream" or "on_final". Defaults to "on_final" when vad is given, otherwise "stream".
vad_tail_sil_ms	int	200	Server VAD trailing silence in milliseconds.
vad_confidence	float	0.55	Server VAD confidence threshold.
enable_cap_punc	bool	True	Server-side capitalization and punctuation.
timeout	TimeoutConfig | None	None	SDK timeout override. By default ws_open follows LiveKit's APIConnectOptions.timeout.

Audio is resampled to 16 kHz mono by LiveKit before it reaches the plugin, so room audio at 48 kHz needs no configuration.

Endpointing Modes

Server endpointing (default). Without vad, the plugin opens the stream in mode="stream" and the server decides where each turn ends using vad_tail_sil_ms and vad_confidence. Nothing else to configure.

Client endpointing. Pass the same LiveKit VAD instance you give AgentSession as vad=. The plugin then opens the stream in mode="on_final" with accept_client_vad_events=True, runs its own VAD stream on the pushed audio and sends start_speaking / stop_speaking to the server, which emits one final transcript per turn. AgentSession never forwards its VAD events to the STT stream, so this is the only way to drive client-side endpointing from LiveKit. Set mode= explicitly to override either default.

Events

Server transcripts map to LiveKit speech events:

Server event	LiveKit SpeechEventType
transcript with is_final=false	INTERIM_TRANSCRIPT
transcript with is_final=true	FINAL_TRANSCRIPT
LiveKit VAD start / end (client endpointing)	START_OF_SPEECH / END_OF_SPEECH
first transcript of a turn / its final (server endpointing)	START_OF_SPEECH / END_OF_SPEECH

Each event carries the server request_id. ack and pong events are consumed internally.

Errors and Retries

SDK errors are mapped to LiveKit APIError subclasses so the framework's retry policy (APIConnectOptions, default 3 retries) applies:

SDK error	LiveKit error
ApiError on connect, or a server error event	APIStatusError (4xx is not retried)
TimeoutError	APITimeoutError
TransportError, ProtocolError, closed socket	APIConnectionError

Audio pushed during a failed attempt is not replayed on retry, which matches other LiveKit STT plugins.

Notes

• RINGG_API_KEY is read only by this integration; the core SDK client is unchanged.
• import ringglabs and import ringglabs.stt never import LiveKit. Importing ringglabs.integrations.livekit without the extra raises an ImportError naming the extra to install.
• The WebSocket URL follows the SDK client (see Overriding base_url).
• Runnable examples live in the repository: examples/livekit_stt_smoke.py (stream a WAV file through the plugin, no LiveKit room needed) and examples/livekit_agent.py (a minimal AgentSession worker).

Production Guidance

• Reuse client instances in long-running services.
• Set explicit timeout budgets with TimeoutConfig.
• Keep long silent WebSocket streams alive with the default SDK JSON heartbeat or by calling ping() manually.
• For full-call streams that can remain idle, set ws_recv=None or a receive timeout larger than the longest expected silence.
• end() sends only the selected stop alias; the proxy handles finalization when client VAD is still speaking.
• Log result.raw and event.raw for observability.
• Use retries only for idempotent operations and transport failures.
• Keep sync and async execution models separate in production apps.