Nexara Python SDK

Python SDK for the Nexara speech-to-text API: transcription, speaker diarization, speaker role tagging, structured LLM post-processing, realtime streaming over WebSocket, and account billing. Full API documentation lives at docs.nexara.ru.

Requires Python 3.10+.

pip install nexara

Quickstart

from nexara import Nexara

client = Nexara(api_key="...")  # or set NEXARA_API_KEY

text = client.transcriptions.create(file="audio.mp3").text

Pass exactly one of file= (path, bytes, or a binary file object — paths are streamed from disk, not loaded into memory) or url=.

Diarization

call = client.transcriptions.create(file="call.mp3", task="diarize")
for segment in call.segments:
    print(f"{segment.speaker}: {segment.text}")

Add meaningful speaker labels with roles — "auto" lets the model invent labels, a list restricts them, a dict adds descriptions:

call = client.transcriptions.create(
    file="call.mp3",
    task="diarize",
    roles=["client", "agent"],
)

Emotions

emotions=True attaches an emotion to each diarized segment — label (one of angry, sad, neutral, positive), confidence, and the full probs distribution when the server sends it:

call = client.transcriptions.create(
    file="call.mp3",
    task="diarize",
    model="nexara-ru",
    emotions=True,
)
for segment in call.segments:
    if segment.emotion:
        print(segment.speaker, segment.emotion.label, segment.emotion.confidence)

The scoring runs inside the ASR model, so it requires task="diarize", model="nexara-ru" and a JSON response format; anything else raises NexaraValidationError before the upload. Not every segment can be scored, so check segment.emotion rather than assuming it is there. It carries a per-second surcharge, charged only when emotion was actually returned.

Long audio: deferred jobs

create_job() submits the audio and returns immediately; the result is fetched by polling. A failed job is never billed, so resubmitting is free.

job = client.transcriptions.create_job(file="long_recording.mp3")
result = job.wait()  # polls; default timeout 1800s

# ...or pick it up later, even from another process:
job = client.transcriptions.retrieve_job(job_id)

Job results live for 12 hours from creation; up to 200 jobs may be in progress per API key. In this SDK "async" always means asyncio — the deferred mode is create_job(), not AsyncNexara.

LLM post-processing

Pass prompt= to run an LLM over the transcript, and optionally json_schema= to force structured output:

result = client.transcriptions.create(
    file="meeting.mp3",
    prompt="Summarize the key decisions",
    json_schema={"type": "object", "properties": {"decisions": {"type": "array"}}},
)
print(result.llm_output)          # dict, validated against your schema
print(result.transcription.text)  # the transcript it was derived from

Realtime streaming

Stream raw audio in, get words back as they are recognised. Every word the server sends is final — there are no interim results, so append event.text and never look back. Requires the websockets extra: pip install nexara[realtime].

import asyncio
from nexara import AsyncNexara

async def main():
    client = AsyncNexara()
    async with client.realtime.connect(sample_rate=16000, diarize=True) as session:
        async for event in session.stream(microphone()):   # any async iterator of PCM bytes
            print(f"spk{event.speaker} {event.words[0].start / 1000:6.2f}s {event.text}")
        print(session.ended.text)                          # full transcript + billing summary

asyncio.run(main())

Audio is raw PCM — int16 little-endian, 16 kHz mono by default — in binary frames of roughly 20 ms to 1 s. connect() takes the session parameters: encoding (pcm_s16le, pcm_f32le, pcm_mulaw, pcm_alaw), sample_rate (8–48 kHz, resampled server-side), channels, multichannel (transcribe stereo channels separately), diarize, delay_ms (emission delay: longer is more accurate, shorter is faster; default 480), turn_detection (see below) and client_id. Language is auto-detected.

stream() pumps an iterator in and yields RealtimeTranscript events out. When sending from your own task, use send_audio() and then finish(), which sends end-of-audio and returns the session.ended summary once the server has flushed the model's delay line:

async with client.realtime.connect() as session:
    async def send():
        async for chunk in microphone():
            await session.send_audio(chunk)
        await session.finish()

    asyncio.create_task(send())
    async for event in session:
        print(event.text, end="", flush=True)

Word timestamps are integer milliseconds on the audio clock (since the first sample you sent) at 80 ms resolution. With diarize=True, speaker is 0–3 by order of first appearance, or None when the diarizer could not tell. A session with no audio for 30 s is closed unless you call keepalive().

Turn detection

For voice agents, turn_detection="smart" tells you when the user starts speaking, stops, and has finished their turn. The session then yields three more event types alongside the transcripts (the iterator is typed as the RealtimeEvent union, so narrow on event.type; without turn_detection it stays RealtimeTranscript), and every transcript carries its turn number:

async with client.realtime.connect(turn_detection="smart") as session:
    async for event in session.stream(microphone()):
        match event.type:
            case "speech.started":   # ~100 ms after the user starts talking
                agent.interrupt()
            case "speech.ended":     # silence now — not necessarily the end of the turn
                pass
            case "turn.end":         # the turn's full text, ready for the LLM
                reply = await llm(event.text)
            case "transcript":
                print(event.turn, event.text)

turn.end arrives after every transcript of its turn and before any of the next, so event.text is complete — roughly 720 ms after the user stops at the default delay. event.confidence is the end-of-turn probability: ≥ 0.5 means the model decided the speaker was done, < 0.5 that the 3 s silence limit closed the turn, None that finish() did. A speech.ended followed by speech.started with the same turn was a pause inside the turn. Works with any delay_ms, with diarize and with multichannel (turns are counted per channel, and every event carries channel).

A session that the server ends raises from the iterator: RealtimeError with .code and .reason (session_full, idle_timeout, audio_backlog, service_restart, …), or InsufficientBalanceError if the wallet ran out mid-session. Everything the server had finalised is delivered before the error, so what you received is what you were billed for. A rejected handshake (bad key, no funds, too many sessions) raises the same status-mapped exceptions as the REST API.

Balance and usage

client.billing reports what is on the account and what it has been spent on. Both endpoints cover the whole account, not just the key you authenticate with:

balance = client.billing.balance()
print(balance.balance, balance.currency, balance.rate_per_min)

# One page of billed calls, newest first.
page = client.billing.usage(limit=20)
for item in page.items:
    print(item.timestamp, item.task, item.cost, item.api_key.name)

# ...or let the SDK walk the pages. History is unbounded — bound it.
for item in client.billing.iter_usage(max_items=200):
    print(item.request_id, item.seconds, item.cost)

Paging is keyset-based, not offset-based: pass a page's next_cursor as cursor= to get the next (older) page, so calls arriving mid-walk cannot shift rows across a page boundary. item.cost is None — not 0 — for rows written before per-request costs were recorded, and rate_per_min covers plain transcription only (profanity_filter, roles, emotions and prompt are surcharges on top of it).

asyncio

AsyncNexara is the same interface under await:

from nexara import AsyncNexara

async with AsyncNexara() as client:
    result = await client.transcriptions.create(file="audio.mp3")
    print(result.text)

Errors and validation

Requests that the server would reject — or, worse, accept, charge for, and silently do something else with — fail client-side with NexaraValidationError before any network call. Server errors map to typed exceptions by status code:

from nexara import NexaraValidationError, InsufficientBalanceError, RateLimitError

try:
    result = client.transcriptions.create(file="audio.mp3")
except InsufficientBalanceError as e:  # 402
    print(e.detail)

429 and connection/timeout failures are retried with exponential backoff (honoring Retry-After). 500 is deliberately not retried: on the synchronous path the request may already have been billed, so a blind retry could pay twice. Deferred jobs bill only on success, which makes create_job() the safe path for retry-heavy workloads.

Not yet available

• Webhooks — job results are fetched by polling.

Development

The package is fully typed (py.typed, mypy strict). Offline tests (pytest, no network needed) and runnable examples live in the repository; NEXARA_USE_MOCK=1 runs everything against an in-memory mock transport.

License

MIT