Avartha Python SDK

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

Drop-in Python interfaces for OpenAI Realtime and ElevenLabs speech. Change the client import and configure Avartha credentials, URLs, and model IDs. Keep the upstream resource methods, request types, response objects, and event loops:

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

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

These classes extend the official SDKs. Compatibility depends on the server's supported endpoints. Avartha managed inference is WebSocket-only; legacy Chat Completions, Responses, and HTTP speech calls are retired. Preview testing also found ASR and TTS mismatches with the official ElevenLabs client defaults. Migration details · Protocol audit and triage · Initial live test results

Documentation

• Drop-in migration from OpenAI and ElevenLabs
• Compatibility and endpoint coverage
• Protocol audit: defaults, message sequences, errors, and triage
• Platform management
• Runnable examples
• Contributing

Installation

The repository is private and the package has not been published to PyPI. With GitHub access configured, install from source:

git clone https://github.com/avartha/avartha-python-sdk.git
python -m pip install ./avartha-python-sdk

Set your Avartha API key and platform root:

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

For preview, set AVARTHA_BASE_URL=https://platform.preview.avartha.ai. 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":
                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", []))

The preview catalog currently includes Gemma for openai_realtime, Qwen for elevenlabs_tts, and Voxtral for elevenlabs_asr. Availability is workspace- and tier-specific; discovery is authoritative for your key.

Async usage

Use AsyncOpenAI, await operations, and iterate with async for. Request parameters and event types remain the upstream SDK's own.

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":
                    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. Its method names and wire messages come from the official ElevenLabs package.

Text to speech

Discover voices for the selected TTS model using the upstream 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 official TTS helper currently fails on preview: Vajra requires auto_mode=true and rejects the helper's hardcoded try_trigger_generation and generation_config fields. Raw single- and multi-context TTS both succeeded with native pacing, so this is a client/runtime contract mismatch, not a model outage. See the diagnosis. The example above preserves the vendor interface and requires that mismatch to be resolved on the service.

convert_realtime is synchronous, matching upstream. The pinned ElevenLabs SDK has no multi-context helper; the platform's multi-stream-input WebSocket is a separate vendor protocol surface. HTTP text_to_speech.stream and .convert remain inherited methods but are not supported by managed inference. Runnable TTS example.

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

Preview workaround: pass previous_text="" in each connection.send data dictionary when you have no context. The official SDK otherwise sends previous_text: null, which preview currently rejects. Both sync-client and async-client ASR sessions passed with the empty-string workaround. This is an application option, not a change to the SDK's wire protocol.

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

Agents

The SDK retains the complete ElevenLabs conversational_ai namespace, including agent creation, tools, knowledge-base documents, and conversations. Managed Avartha agent CRUD does not currently implement the ElevenLabs endpoints. Its native control API uses a different schema. Preview's ElevenLabs-shaped conversation-list endpoint did respond successfully.

Against a service that implements ElevenLabs agent management, the original calls work unchanged:

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)

Conversation, AsyncConversation, and ClientTools are the upstream classes, also available from avartha.conversational_ai. AsyncConversation takes a sync ElevenLabs client, matching upstream. Agent session execution has local contract coverage; it was not exercised against a published preview agent.

• Agent, tool, and knowledge-base example
• Conversation with client tools

Using types

Continue importing types and exceptions from the upstream packages. The adapter returns their original objects:

from avartha import OpenAI
from openai.types import Model

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

Use openai.types and elevenlabs.types for their full type catalogs. avartha.types is not a replacement namespace. Platform management responses are JSON dictionaries and lists.

Handling errors

Keep existing vendor exception handlers. For HTTP discovery:

import openai
from avartha import OpenAI

with OpenAI() as client:
    try:
        client.models.list()
    except openai.APIConnectionError as error:
        print("Connection failed:", error)
    except openai.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 HTTP/TTS helper errors remain elevenlabs.core.api_error.ApiError; ASR also emits RealtimeEvents.ERROR. The upstream TTS helper can omit a server error's details when the gateway closes the socket.

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

Retries and timeouts

Upstream constructor options pass through:

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 pinned OpenAI client accepts 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 does not bound every WebSocket receive; the live smoke runner isolates its synchronous TTS helper to enforce a deadline. Control-plane writes are never retried automatically.

Configuration

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

Without an explicit URL, protocol clients derive their URL from AVARTHA_BASE_URL. 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. Keep custom upstream transports and type imports where needed; OpenAI 3 uses HTTPX2. Migration configuration.

Always close clients or use context managers. OpenAI retains its upstream HTTP-client ownership behavior. ElevenLabs and Control close only 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 is a shortcut to the OpenAI resource. AsyncAvartha, Control, and AsyncControl are available too. Endpoint creation, readiness waiting, scaling, routing, stopping, and deletion are described in the management guide.

Examples and development

For local microphone/speaker support, install the audio extra from the repository (python -m pip install '.[audio]') and PortAudio.

python -m venv .venv
. .venv/bin/activate
make install-dev
make check

On images without ensurepip, install uv and use make check BUILD_FLAGS=--installer=uv. CI runs on Python 3.12–3.14. Default tests use mock HTTP transports and local WebSockets without cloud inference. make build produces wheel and source archives; publication is separate.

Live testing is explicit and consumes inference credits. See preview testing for the runner and current results.