Avartha Python SDK

Synchronous and asynchronous Python clients for Avartha Realtime LLMs, streaming speech, and platform management. Requires Python 3.12+.

Familiar Python interfaces for OpenAI Realtime and ElevenLabs speech. Change client, type, and exception imports, then configure Avartha credentials, URLs, and model/voice IDs. The supported APIs use the same method names and event loops:

-from openai import OpenAI, AsyncOpenAI
+from avartha import OpenAI, AsyncOpenAI

-from elevenlabs.client import ElevenLabs, AsyncElevenLabs
+from avartha import ElevenLabs, AsyncElevenLabs

Neither vendor SDK needs to be installed. Avartha supports Realtime text, streaming speech, HTTP discovery, and platform management. Managed inference is WebSocket-only; Chat Completions, Responses, and HTTP speech inference are unavailable. See the guides below for supported features and migration steps.

Documentation

• Migrating from OpenAI and ElevenLabs
• Compatibility and endpoint coverage
• Platform management
• Runnable examples
• Contributing

Installation

Install from PyPI:

python -m pip install avartha-python-sdk

Set your Avartha API key and platform root. Update AVARTHA_BASE_URL to match your environment:

export AVARTHA_API_KEY='avk_...'
export AVARTHA_BASE_URL='https://platform.preview.avartha.ai'

Clients default to https://platform.preview.avartha.ai and the serverless inference tier. Set AVARTHA_BASE_URL to use another platform root. Both protocol clients derive their service URLs from this root. Use model and voice IDs from the selected environment; vendor IDs are not mapped automatically.

Usage

Use the OpenAI Realtime interface for LLM inference. The client reads AVARTHA_API_KEY from the environment; api_key can also be passed explicitly.

from avartha import OpenAI

with OpenAI() as client:
    with client.realtime.connect(model="google/gemma-4-26b-a4b-it") as connection:
        connection.session.update(session={"type": "realtime", "output_modalities": ["text"]})
        connection.conversation.item.create(
            item={
                "type": "message",
                "role": "user",
                "content": [{"type": "input_text", "text": "Explain Python dictionaries."}],
            }
        )
        connection.response.create()
        for event in connection:
            if event.type == "response.output_text.delta":
                print(event.delta, end="", flush=True)
            elif event.type == "response.done":
                if event.response.status != "completed":
                    raise RuntimeError(f"Response ended with status: {event.response.status}")
                break
            elif event.type == "error":
                raise RuntimeError(event.error.message)

Keep the connection open for additional turns. Create another conversation item and response on the same connection to preserve conversation state.

To discover available models and their enabled protocols:

from avartha import OpenAI

with OpenAI() as client:
    for model in client.models.list():
        print(model.id, model.to_dict().get("protocols", []))

Model availability depends on your workspace, environment, and tier. Replace the example IDs with models returned for your key and the protocol you need: openai_realtime, elevenlabs_tts, or elevenlabs_asr.

Async usage

Use AsyncOpenAI, await operations, and iterate with async for. The sync and async clients accept the same request parameters and return the same event types.

import asyncio
from avartha import AsyncOpenAI


async def main():
    async with AsyncOpenAI() as client:
        async with client.realtime.connect(model="google/gemma-4-26b-a4b-it") as connection:
            await connection.session.update(
                session={"type": "realtime", "output_modalities": ["text"]}
            )
            await connection.conversation.item.create(
                item={
                    "type": "message",
                    "role": "user",
                    "content": [{"type": "input_text", "text": "Explain Python dictionaries."}],
                }
            )
            await connection.response.create()
            async for event in connection:
                if event.type == "response.output_text.delta":
                    print(event.delta, end="", flush=True)
                elif event.type == "response.done":
                    if event.response.status != "completed":
                        raise RuntimeError(f"Response ended with status: {event.response.status}")
                    break
                elif event.type == "error":
                    raise RuntimeError(event.error.message)


asyncio.run(main())

Complete Realtime example.

Streaming speech

ElevenLabs() defaults to the same Avartha platform root and serverless tier. It provides ElevenLabs-compatible realtime speech and discovery APIs.

Text to speech

Discover voices for the selected TTS model using request_options:

from avartha import ElevenLabs

with ElevenLabs() as client:
    voices = client.voices.get_all(
        request_options={
            "additional_query_parameters": {"model_id": "qwen/qwen3-tts-12hz-1.7b-base"}
        }
    )
    for voice in voices.voices:
        print(voice.voice_id)

Stream text over a WebSocket with convert_realtime. Avartha returns PCM; select a sample rate supported by your model and write a WAV header for playback. This example uses Qwen's 24 kHz output:

import wave
from avartha import ElevenLabs

with ElevenLabs() as client:
    audio = client.text_to_speech.convert_realtime(
        voice_id="carol",
        model_id="qwen/qwen3-tts-12hz-1.7b-base",
        output_format="pcm_24000",
        text=iter(["Welcome. ", "How can I help?"]),
    )
    with wave.open("welcome.wav", "wb") as output:
        output.setnchannels(1)
        output.setsampwidth(2)
        output.setframerate(24000)
        for chunk in audio:
            output.writeframes(chunk)

The default output format is pcm_24000. Use AsyncElevenLabs and async for for async TTS; the async client's text accepts either a regular iterable or an async iterable. Run the TTS example with --async to use that client.

Text fragments are forwarded as supplied, without waiting for word or sentence boundaries or adding spaces. Include any intended whitespace in your text. The Avartha Platform handles buffering and segmentation. Available audio continues to arrive while the text producer is paused; when synthesis starts depends on the model.

The SDK's single-context and multi-context TTS helpers always use auto_mode=true. auto_mode=False and default_mode are not supported.

For multiple utterances on one connection, use connect_multi_context on either client. A background consumer can iterate over the session before contexts are created and between turns; messages are keyed by message.context_id. For a finite batch, create and flush contexts, then iterate over session.drain(). Session iteration waits until the connection closes via close_socket() or when its with or async with block exits. See the multi-context example and TTS lifecycle details.

HTTP text_to_speech.stream and .convert are not provided by this SDK; managed inference uses the realtime methods above.

Speech to text

Realtime ASR uses await client.speech_to_text.realtime.connect(...) on both ElevenLabs client variants. Send mono PCM16 audio and await a committed transcript. The complete example reads a WAV file, registers transcript/error callbacks, streams chunks, commits, and closes the connection:

python examples/realtime_asr.py mistralai/voxtral-mini-4b-realtime-2602 speech-16k.wav

previous_text is optional text context for transcription. When omitted from connection.send data, Avartha sends null, meaning no context. You can omit this field when streaming audio without prior text.

File-based HTTP speech_to_text.convert is retired on managed inference.

Agents

Agent management, tools, knowledge bases, and conversation sessions/history are currently unavailable. Accessing client.conversational_ai raises NotImplementedError.

ElevenLabs-compatible conversational_ai support is planned for an upcoming release, including agent creation, tools, knowledge-base documents, and conversations.

The following example previews the planned API. It requires that future SDK release and a service implementing the corresponding agent endpoints:

from avartha import ElevenLabs

with ElevenLabs(base_url="https://your-compatible-agent-service") as client:
    agent = client.conversational_ai.agents.create(
        name="Support",
        conversation_config={
            "agent": {
                "first_message": "What can I help you with?",
                "prompt": {"prompt": "Help customers find concise, accurate answers."},
            },
            "tts": {"voice_id": "your-voice"},
        },
    )
    print(agent.agent_id)

The planned avartha.conversational_ai namespace will also include Conversation, AsyncConversation, and ClientTools.

Using types

Import types and exceptions from Avartha. The SDK provides its own classes for the supported APIs, preserving upstream fields and serialization methods. Neither openai nor elevenlabs needs to be installed:

from avartha import OpenAI
from avartha.types import Model

with OpenAI() as client:
    models: list[Model] = client.models.list().data
    print([model.to_dict() for model in models])

Use avartha.types.realtime for OpenAI-compatible Realtime events and avartha.types.speech for ElevenLabs-compatible speech types. These namespaces cover the APIs supported by Avartha. Platform management responses are JSON dictionaries and lists.

ElevenLabs-compatible transcription options are also available from the root:

from avartha import AudioFormat, CommitStrategy, RealtimeAudioOptions

options = RealtimeAudioOptions(
    model_id="your-avartha-asr-model",
    audio_format=AudioFormat.PCM_16000,
    sample_rate=16000,
    commit_strategy=CommitStrategy.VAD,
)

Handling errors

Import exception classes from Avartha too; upstream exception classes will not catch Avartha's exceptions. For HTTP discovery:

from avartha import APIConnectionError, APIStatusError, OpenAI

with OpenAI() as client:
    try:
        client.models.list()
    except APIConnectionError as error:
        print("Connection failed:", error)
    except APIStatusError as error:
        print("Request failed:", error.status_code, error.request_id)

Realtime server errors are events, so handle event.type == "error" in the receive loop. Inspect response.done.response.status for completion, failure, or cancellation; a terminal event alone does not prove successful inference. Connection failures can raise WebSocket exceptions.

ElevenLabs-compatible HTTP/TTS helper errors, including speech HTTP network failures and timeouts, use avartha.ApiError. ASR also emits RealtimeEvents.ERROR.

Management failures use avartha.PlatformAPIError, retaining the HTTP status, response body, field errors, request ID, and Retry-After header. Management network failures and timeouts use avartha.ApiError.

For speech and management HTTP request failures without a server response, ApiError.status_code and ApiError.headers are None, and ApiError.body describes the failure. The original exception is available as error.__cause__.

Retries and timeouts

Set timeouts and retries for OpenAI-compatible HTTP discovery on the client:

from avartha import OpenAI

with OpenAI(timeout=30.0, max_retries=0) as client:
    print(client.models.list())

HTTP and WebSocket retry settings are separate. The OpenAI-compatible clients accept client.realtime.connect(..., max_retries=0) to disable automatic reconnection, and websocket_connection_options for transport settings. Use asyncio.timeout when an async operation needs an overall deadline. Close active sessions when cancelling. ElevenLabs timeout configures HTTP calls; it does not set a WebSocket receive deadline. Platform management writes are never retried automatically. See timeouts and cleanup.

Configuration

Client	Meaning of explicit base_url
OpenAI, AsyncOpenAI	Full inference base: https://platform.preview.avartha.ai/inference/serverless/openai/v1
ElevenLabs, AsyncElevenLabs	Speech service root before /v1: https://platform.preview.avartha.ai/inference/serverless/elevenlabs
Avartha, AsyncAvartha, Control, AsyncControl	Platform root: https://platform.preview.avartha.ai

Without an explicit URL, protocol clients derive their URL from AVARTHA_BASE_URL, defaulting to https://platform.preview.avartha.ai and the serverless tier. Set tier="dedicated" for dedicated inference; there is no automatic fallback between tiers. An explicit protocol base_url takes precedence over tier. AVARTHA_ELEVENLABS_BASE_URL overrides the derived speech root, and an explicit ElevenLabs base_url overrides that environment variable.

Keys are read at construction from AVARTHA_API_KEY or api_key. Vendor default keys and base URL variables are not substituted. Migration configuration.

Always close clients or use context managers. OpenAI manages and closes its HTTP client; custom http_client arguments are unsupported. ElevenLabs and Control close only HTTP clients they created. Close active WebSocket sessions separately.

Platform management

Use the combined Avartha client for inference plus management:

from avartha import Avartha

with Avartha() as client:
    print(client.control.catalog.models())
    print(client.control.catalog.skus(provider="modal"))
    print(client.control.organizations.list())
    print(client.control.organizations.limits("your-workspace-id"))
    print(client.control.endpoints.list(workspace_id="your-workspace-id"))

client.openai and client.elevenlabs are the protocol clients; client.realtime and client.models are shortcuts to the corresponding OpenAI resources, including model listing and retrieval. AsyncAvartha, Control, and AsyncControl are available too. Workspace administration includes organization creation, renaming, deletion, leaving, member roles, and invitation creation, revocation, and acceptance. For example, client.control.organizations.members.list("your-workspace-id") lists members. See workspace administration for sync/async usage and server permissions.

Endpoint creation, readiness waiting, scaling, routing, stopping, and deletion are described in the management guide.

Examples

Runnable examples cover Realtime text, TTS, multi-context TTS, ASR, and endpoint management. Microphone capture and speaker playback are handled by your application.

Running inference examples consumes credits. For SDK development and checks, see Contributing.