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. See the guides below for compatibility details and historical test results. 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

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":
                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)

Avartha's helper sends the gateway-compatible envelope and defaults to pcm_24000. Use AsyncElevenLabs and async for for async TTS; text accepts either a regular iterable or an async iterable. Run the TTS example with --async to use that client.

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 remain inherited methods but are not supported by managed inference.

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, the official SDK sends null, meaning no context. Current preview accepts this default; existing audio-send code works unchanged, with no empty-string workaround needed.

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.preview.avartha.ai/inference/serverless/openai/v1
ElevenLabs, AsyncElevenLabs	Dialect 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. 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 PortAudio and the audio extra:

python -m pip install 'avartha-python-sdk[audio]'

For development, clone the repository and run:

git clone https://github.com/avartha/avartha-python-sdk.git
cd avartha-python-sdk
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.