gnani-vachana

Official Python client by Gnani.ai. Build multilingual voice workflows with Speech-to-Text (STT) and Text-to-Speech (TTS) across REST, SSE streaming, and real-time WebSockets.

Gnani.ai is a production-ready speech AI platform with high-accuracy STT and low-latency TTS for 10+ Indian languages, with Timbre voices across REST, SSE, and WebSocket APIs.

Installation

pip install gnani-vachana

Or with uv:

uv add gnani-vachana

Requires Python 3.9+.

Quick Start

STT REST (file-based transcription)

from gnani.stt import GnaniSTTClient

client = GnaniSTTClient(api_key="your-api-key")

result = client.transcribe("audio.wav", language_code="hi-IN")
print(result["transcript"])

STT Batch (many or long files)

from gnani.stt import GnaniSTTBatchClient

client = GnaniSTTBatchClient()  # reads GNANI_API_KEY
result = client.transcribe(["call1.wav", "call2.wav"], language_code="hi-IN")

print(result[0].text)
result.save("./outputs")  # one JSON per file

See Batch STT usage for step-by-step control, diarization, and webhooks.

Realtime Streaming (WebSocket)

import asyncio
from gnani.stt import GnaniSTTStreamClient, StreamTranscriptEvent

async def main():
    async with GnaniSTTStreamClient(api_key="your-api-key", language_code="hi-IN") as stream:
        # Send audio chunks (raw PCM, 16-bit LE, 16 kHz, mono)
        with open("audio.pcm", "rb") as f:
            while chunk := f.read(1024):
                await stream.send_audio(chunk)
                await asyncio.sleep(0.032)  # real-time pacing (32 ms per frame)

        # Iterate over events
        async for event in stream:
            if isinstance(event, StreamTranscriptEvent):
                print(event.text)

asyncio.run(main())

TTS REST (single response)

from gnani.tts import GnaniTTSClient, AudioConfig

client = GnaniTTSClient(api_key="your-api-key")

# timbre-v2.0 (default model)
audio = client.synthesize(
    "नमस्ते, आप कैसे हैं?",
    voice="Pranav",
    model="timbre-v2.0",
    audio_config=AudioConfig(sample_rate=48000, encoding="linear_pcm", container="wav"),
)

# timbre-v2.5 — language control + speed
audio = client.synthesize(
    "नमस्ते, यह एक परीक्षण है।",
    voice="Nalini",
    model="timbre-v2.5",
    language="hi-IN",
    speed=1.0,
    audio_config=AudioConfig(container="mp3", sample_rate=48000, bitrate="128k"),
)

with open("tts_output.mp3", "wb") as f:
    f.write(audio)

TTS SSE Streaming (lower latency)

from gnani.tts import GnaniTTSStreamClient, AudioConfig

client = GnaniTTSStreamClient(api_key="your-api-key")
audio = client.synthesize(
    "Hello from Gnani TTS",
    voice="Pranav",
    model="timbre-v2.0",
    audio_config=AudioConfig(sample_rate=48000, encoding="linear_pcm", container="wav"),
    output_file="tts_sse.wav",
)

TTS Realtime WebSocket (lowest latency)

import asyncio
from gnani.tts import GnaniTTSRealtimeClient, AudioConfig

async def main():
    async with GnaniTTSRealtimeClient(api_key="your-api-key") as client:
        audio = await client.synthesize_and_collect(
            "Hello from Gnani TTS",
            voice="Pranav",
            model="timbre-v2.0",
            audio_config=AudioConfig(sample_rate=48000, encoding="linear_pcm", container="wav"),
            output_file="tts_realtime.wav",
        )

asyncio.run(main())

Authentication

All APIs (STT REST, STT Realtime, TTS) require a single API key:

Obtaining Credentials

You need a Gnani API key. Gnani APIs have this.

Passing Credentials

Option 1 -- Constructor argument:

from gnani.stt import GnaniSTTClient, GnaniSTTStreamClient
from gnani.tts import GnaniTTSClient, GnaniTTSRealtimeClient, GnaniTTSStreamClient

client = GnaniSTTClient(api_key="your-api-key")
stream = GnaniSTTStreamClient(api_key="your-api-key")
tts_rest = GnaniTTSClient(api_key="your-api-key")
tts_stream = GnaniTTSStreamClient(api_key="your-api-key")
tts_realtime = GnaniTTSRealtimeClient(api_key="your-api-key")

Option 2 -- Environment variable:

export GNANI_API_KEY="your-api-key"

from gnani.stt import GnaniSTTClient, GnaniSTTStreamClient
from gnani.tts import GnaniTTSClient

client = GnaniSTTClient()           # picks up GNANI_API_KEY
stream = GnaniSTTStreamClient()     # picks up GNANI_API_KEY
tts = GnaniTTSClient()              # picks up GNANI_API_KEY

Supported Languages

STT Languages (Speech-to-Text)

STT uses BCP-47 locale codes (e.g. hi-IN). For the full list of supported languages, see:

• STT REST — Supported Languages
• STT Realtime — Supported Languages

TTS Voices & Models

TTS uses the Timbre engine. Two model IDs are supported:

Migration: The former model name vachana-voice-v3 has been renamed to timbre-v2.0. Update any model="vachana-voice-v3" calls to model="timbre-v2.0" (or omit model to use the default).

timbre-v2.0 voices

timbre-v2.5 voices

42 Timbre voices across Hindi, English, Tamil, Telugu, Kannada, Malayalam, Marathi, Bengali, Gujarati, Punjabi, and Hinglish. Examples: Nalini, Kaveri, Asmita, Suhana, Poorvi.

List voices programmatically:

from gnani.tts import GnaniTTSClient

print(GnaniTTSClient.supported_voices())                    # timbre-v2.0 (default)
print(GnaniTTSClient.supported_voices(model="timbre-v2.5")) # 42 voices

timbre-v2.5 optional controls

Only timbre-v2.5 accepts the language and speed parameters:

voice is required for all Timbre models unless speaker_embedding is provided.

TTS Languages (Text-to-Speech)

For the full list of supported languages, see TTS — Supported Languages.

REST Usage

Transcribe a file by path

result = client.transcribe("meeting.wav", language_code="en-IN")
print(result["transcript"])

Transcribe from a file object

with open("meeting.mp3", "rb") as f:
    result = client.transcribe(f, language_code="ta-IN")

Transcribe raw bytes

audio_bytes = download_audio_from_somewhere()
result = client.transcribe_bytes(
    audio_bytes, filename="clip.wav", language_code="kn-IN"
)

Custom request ID

result = client.transcribe(
    "call.flac", language_code="hi-IN", request_id="my-trace-123"
)

List supported languages

for code, name in GnaniSTTClient.supported_languages().items():
    print(f"{code}: {name}")

Batch STT Usage

Batch jobs run asynchronously: create, start, wait, fetch. transcribe() does all four. Limits: 100 files per job, 10 MB per file (ZIP: 50 MB). Larger files are not supported yet.

from gnani.stt import GnaniSTTBatchClient

client = GnaniSTTBatchClient()

# Files can be a path, a directory, a glob, a ZIP, bytes, a file object, or public URLs.
result = client.transcribe("./recordings/", language_code="hi-IN,en-IN", speakers=2)

for f in result:
    print(f.name, "OK" if f.ok else f"FAILED: {f.error}")
    if f.ok:
        for seg in f.segments:
            print(f"  [{seg.start:.1f}-{seg.end:.1f}] speaker {seg.speaker}: {seg.text}")

Step by step (long jobs, resuming later)

job = client.create_job("./recordings/*.wav", language_code="hi-IN")  # starts automatically
print(job.id)                      # keep this to resume
job.wait(timeout=3600, on_progress=lambda j: print(j.status, j.percent))

job = client.get_job("<job id>")   # e.g. after a restart
files = job.files()
job.save("./outputs", text_files=True)

language_code takes one code, several comma-separated (identified per speaker turn; the first is the fallback), or "auto". Other options: diarization, speakers (1-8), multi_channel, denoise, bias_words, bias_score, callback_url, start=False.

Async

AsyncGnaniSTTBatchClient has the same methods, awaited. Transcripts are downloaded concurrently, so .text and .segments are plain attributes on what transcribe() returns (after job.files() call await f.load(), or pass load=True).

from gnani.stt import AsyncGnaniSTTBatchClient

client = AsyncGnaniSTTBatchClient()
result = await client.transcribe("./recordings/", language_code="hi-IN")
print(result[0].text)
await result.save("./outputs")

async for job in client.list_jobs(status="COMPLETED"):
    print(job.id)

wait() returns for COMPLETED and PARTIAL_FAILURE (check each BatchFile.ok) and raises BatchJobFailedError for FAILED, START_FAILED and CANCELLED, and BatchTimeoutError on timeout (the job keeps running). It polls every 10 s, the documented minimum.

Realtime Streaming Usage

Connection Flow

1. Client opens a WebSocket connection to wss://api.vachana.ai/stt/v3/stream with auth headers.
2. Server sends a connected event with the active configuration.
3. Client sends binary PCM audio frames (1024 bytes each = 512 samples at 16-bit).
4. Server detects speech via VAD and responds with processing and transcript events.
5. Either side may close the connection at any time.

PCM Specification

All audio must be sent as raw PCM binary frames. No container format (WAV, MP3, etc.) is accepted mid-stream.

• Each binary frame must be exactly 1,024 bytes.
• Frames must be sent at real-time cadence — one frame every 32 ms (16 kHz) or 64 ms (8 kHz). Do not buffer and burst; this degrades VAD accuracy.
• For 44100 and 48000 Hz sources, the server resamples internally — still send 1,024-byte frames at the appropriate cadence.

For the full WebSocket protocol reference, see STT Realtime — PCM Specification.

Using the async context manager

import asyncio
from gnani.stt import GnaniSTTStreamClient, StreamTranscriptEvent

async def main():
    async with GnaniSTTStreamClient(
        api_key="your-api-key",
        language_code="hi-IN",
        sample_rate=16000,
    ) as stream:
        print(f"Connected! Sample rate: {stream.connected_config.sample_rate}")

        with open("audio.pcm", "rb") as f:
            while chunk := f.read(1024):
                await stream.send_audio(chunk)
                await asyncio.sleep(0.032)

        async for event in stream:
            if isinstance(event, StreamTranscriptEvent):
                print(f"[{event.segment_index}] {event.text}")
                print(f"  Duration: {event.audio_duration_ms}ms, Latency: {event.latency}ms")

asyncio.run(main())

Manual connect / close

import asyncio
from gnani.stt import GnaniSTTStreamClient, StreamTranscriptEvent

async def main():
    stream = GnaniSTTStreamClient(api_key="your-api-key")
    config = await stream.connect()
    print(f"Server ready: {config.message}")

    await stream.send_audio(audio_chunk)
    transcripts = await stream.close()

    for t in transcripts:
        print(t.text)

asyncio.run(main())

High-level stream_audio helper with callbacks

import asyncio
from gnani.stt import GnaniSTTStreamClient, StreamTranscriptEvent, StreamProcessingEvent

async def main():
    async with GnaniSTTStreamClient(api_key="your-api-key") as stream:
        with open("audio.pcm", "rb") as f:
            transcripts = await stream.stream_audio(
                f,
                on_transcript=lambda t: print(f"Transcript: {t.text}"),
                on_processing=lambda p: print(f"Processing at {p.timestamp}..."),
                realtime_pace=True,
            )

    print(f"Total segments: {len(transcripts)}")

asyncio.run(main())

Using 8 kHz audio

stream = GnaniSTTStreamClient(
    api_key="your-api-key",
    language_code="en-IN",
    sample_rate=8000,
)

Event Types

All events are typed dataclasses:

Accessing the raw JSON payload

Every event includes a raw field with the full server JSON:

async for event in stream:
    print(event.raw)  # dict with the complete server response

Text-to-Speech Usage

Input text is normalized server-side; see the Text Normalization Guide for symbols, numbers, and abbreviations.

TTS Models

from gnani.tts import DEFAULT_MODEL, SUPPORTED_MODELS

print(DEFAULT_MODEL)              # timbre-v2.0
print(sorted(SUPPORTED_MODELS))   # timbre-v2.0, timbre-v2.5

TTS REST

from gnani.tts import GnaniTTSClient

client = GnaniTTSClient(api_key="your-api-key")
# Default audio_config uses sample_rate=48000
audio = client.synthesize("यह एक टेस्ट है", voice="Pranav", model="timbre-v2.0")
with open("tts_rest.wav", "wb") as f:
    f.write(audio)

TTS Streaming (SSE)

Lower latency than REST — audio is streamed via Server-Sent Events.

from gnani.tts import GnaniTTSStreamClient, AudioConfig

client = GnaniTTSStreamClient(api_key="your-api-key")

# synthesize() collects all chunks and returns audio matching audio_config
audio = client.synthesize(
    "Streaming TTS response",
    voice="Pranav",
    model="timbre-v2.0",
    audio_config=AudioConfig(sample_rate=48000, encoding="linear_pcm", container="wav"),
    output_file="tts_sse.wav",
)

For chunk-by-chunk streaming (e.g. real-time playback), use synthesize_stream():

for audio_chunk in client.synthesize_stream("Hello!", voice="Pranav", model="timbre-v2.0"):
    play_audio(audio_chunk)  # format depends on audio_config (PCM, OGG, MP3, etc.)

TTS Realtime (WebSocket)

Lowest latency — audio is streamed over a persistent WebSocket connection.

import asyncio
from gnani.tts import GnaniTTSRealtimeClient, AudioConfig

async def main():
    async with GnaniTTSRealtimeClient(api_key="your-api-key") as client:
        # synthesize_and_collect() returns audio matching audio_config
        audio = await client.synthesize_and_collect(
            "Realtime TTS response",
            voice="Kaveri",
            model="timbre-v2.0",
            audio_config=AudioConfig(sample_rate=48000, encoding="linear_pcm", container="wav"),
            output_file="tts_realtime.wav",
        )

asyncio.run(main())

For chunk-by-chunk streaming (e.g. real-time playback):

async with GnaniTTSRealtimeClient(api_key="your-api-key") as client:
    async for audio_chunk in client.synthesize("Hello!", voice="Pranav", model="timbre-v2.0"):
        play_audio(audio_chunk)  # format depends on audio_config (PCM, OGG, MP3, etc.)

TTS timbre-v2.5 (expanded catalog)

from gnani.tts import GnaniTTSClient, AudioConfig

client = GnaniTTSClient(api_key="your-api-key")
audio = client.synthesize(
    "Hello, this is a Timbre test.",
    voice="Kaveri",
    model="timbre-v2.5",
    language="en-IN",
    speed=1.0,
    audio_config=AudioConfig(container="mp3", sample_rate=48000, bitrate="128k"),
)

The same model and language parameters work on GnaniTTSStreamClient (SSE) and GnaniTTSRealtimeClient (WebSocket) when using timbre-v2.5.

TTS Telephony Formats

For telephony integrations (IVR, SIP), use 8 kHz A-law or μ-law:

from gnani.tts import GnaniTTSClient, AudioConfig

client = GnaniTTSClient(api_key="your-api-key")

# A-law (G.711a) — common in European/Indian telephony
alaw_audio = client.synthesize(
    "Hello from Gnani",
    voice="Pranav",
    audio_config=AudioConfig(container="alaw", sample_rate=8000),
)

# μ-law (G.711u) — common in North American telephony
mulaw_audio = client.synthesize(
    "Hello from Gnani",
    voice="Pranav",
    audio_config=AudioConfig(container="mulaw", sample_rate=8000),
)

TTS OGG/Opus

Returns a playable OGG Opus file (OggS header). Use either form:

• container="ogg" (encoding defaults to oggopus)
• container="raw", encoding="oggopus"

Supported at all standard sample rates (8000–48000).

from gnani.tts import GnaniTTSClient, AudioConfig

client = GnaniTTSClient(api_key="your-api-key")

# Using container="ogg"
ogg_audio = client.synthesize(
    "Hello from Gnani",
    voice="Pranav",
    audio_config=AudioConfig(container="ogg", encoding="oggopus", sample_rate=48000),
)

with open("output.ogg", "wb") as f:
    f.write(ogg_audio)

SSE / WebSocket: OGG is delivered as one complete file in a single chunk (not streamable frame-by-frame). For chunk iteration, use synthesize_stream() (SSE) or async synthesize() (WebSocket) — the buffered synthesize() and synthesize_and_collect() methods correctly return the raw OGG bytes without WAV wrapping when using non-PCM encodings.

TTS Audio Format Matrix

Supported combinations of container, encoding, sample_rate, and bitrate:

Notes:

• bitrate is only applicable to container="mp3". It is ignored for all other containers.
• mulaw and alaw containers are telephony-only — they require sample_rate=8000.
• For telephony, prefer container="alaw" / container="mulaw" over encoding="pcm_alaw" / encoding="pcm_mulaw".
• container="ogg" and container="raw" with encoding="oggopus" are equivalent — both return a playable OGG Opus file.

File extensions: Match the output file extension to your audio_config.container — use .wav for WAV, .mp3 for MP3, .ogg for OGG/Opus, .raw for raw PCM, .alaw / .ulaw for telephony.

TTS Supported Constants

from gnani.tts import (
    SUPPORTED_ENCODINGS,     # {"linear_pcm", "oggopus", "pcm_mulaw", "pcm_alaw"}
    SUPPORTED_CONTAINERS,    # {"raw", "mp3", "wav", "ogg", "mulaw", "alaw"}
    SUPPORTED_BITRATES,      # {"32k", "64k", "96k", "128k", "192k"}
    SUPPORTED_SAMPLE_RATES,  # (8000, 16000, 22050, 24000, 44100, 48000)
    DEFAULT_SPEED,           # 1.0
    MIN_SPEED,               # 0.85
    MAX_SPEED,               # 1.15
)

TTS Voices

List voices for a specific model:

from gnani.tts import GnaniTTSClient, TIMBRE_V25_VOICES

print(GnaniTTSClient.supported_voices())                    # 4 voices (timbre-v2.0)
print(GnaniTTSClient.supported_voices(model="timbre-v2.5")) # 42 voices
print(len(TIMBRE_V25_VOICES))                               # 42

Legacy voices (Pranav, Kaveri, Shubhra, Deepak) are valid for timbre-v2.0 only.

Audio Requirements

STT REST

STT Streaming (WebSocket)

See PCM Specification for full details.

Response Format

REST

{
  "success": true,
  "request_id": "req_abc123",
  "timestamp": "20251226_143052.123",
  "transcript": "नमस्ते, आप कैसे हैं?"
}

Realtime Streaming

Connected:

{
  "type": "connected",
  "message": "STT service ready — VAD service connected",
  "timestamp": "2024-01-15T10:30:00.000Z",
  "config": { "sample_rate": 16000, "chunk_size": 512 }
}

Transcript:

{
  "type": "transcript",
  "timestamp": "2024-01-15T10:30:05.987Z",
  "text": "Hello, how are you today?",
  "audio_duration_ms": 2340,
  "segment_id": "<segment_id>",
  "segment_index": "<segment_index>",
  "latency": 320
}

Error Handling

from gnani.stt import (
    AuthenticationError,
    InvalidAudioError,
    APIError,
    StreamConnectionError,
    StreamClosedError,
    StreamError,
)

# REST errors
try:
    result = client.transcribe("audio.wav", language_code="hi-IN")
except AuthenticationError:
    print("Check your credentials")
except InvalidAudioError as e:
    print(f"Bad audio file: {e}")
except APIError as e:
    print(f"API error {e.status_code}: {e}")

# Streaming errors
try:
    async with GnaniSTTStreamClient(api_key="key") as stream:
        await stream.send_audio(chunk)
except StreamConnectionError as e:
    print(f"Connection failed: {e}")
except StreamClosedError as e:
    print(f"Stream already closed: {e}")
except StreamError as e:
    print(f"Server error: {e} (at {e.timestamp})")

TTS Errors

from gnani.tts import (
    GnaniTTSClient,
    GnaniTTSRealtimeClient,
    AuthenticationError,
    APIError,
    StreamConnectionError,
    StreamError,
)

# REST / SSE errors
try:
    audio = GnaniTTSClient(api_key="key").synthesize("Hello", voice="Pranav")
except AuthenticationError:
    print("Check your API key")
except APIError as e:
    print(f"TTS API error {e.status_code}: {e}")

# WebSocket errors
try:
    async with GnaniTTSRealtimeClient(api_key="key") as client:
        audio = await client.synthesize_and_collect("Hello", voice="Pranav")
except StreamConnectionError as e:
    print(f"WebSocket connection failed: {e}")
except StreamError as e:
    print(f"TTS stream error: {e}")

Documentation

Full API reference and guides are available at docs.gnani.ai.

• STT REST API
• STT Realtime WebSocket
• TTS REST API
• TTS Streaming (SSE)
• TTS Realtime WebSocket
• Text Normalization Guide

License

This project is licensed under the MIT License -- see the LICENSE file for details.