Python SDK

The Spatius Python SDK creates server-side avatar sessions over WebSocket. It handles session-token creation, WebSocket authentication, audio upload, motion data payload callbacks, and optional egress to LiveKit or Agora.

Repository

The Python SDK is available on GitHub: spatius-ai/spatius-sdk-python and PyPI.

pip install spatius

Install the optional Ogg Opus encoder support when you want the SDK to encode raw PCM before sending:

pip install "spatius[opus]"

The optional encoder uses opuslib, which requires a working libopus runtime on the host system.

Requirements

api_key
app_id
avatar_id
expire_at
audio bytes in a supported input format

The SDK defaults to the us-west region. See Regions for the current region list.

Region configuration

Passing a region is enough for normal production use:

session = new_avatar_session(
    api_key="your-api-key",
    app_id="your-app-id",
    avatar_id="your-avatar-id",
    region="us-west",
    expire_at=expire_at,
)

Quick Start

import asyncio
from datetime import datetime, timedelta, timezone

from spatius import new_avatar_session


async def main():
    session = new_avatar_session(
        api_key="your-api-key",
        app_id="your-app-id",
        avatar_id="your-avatar-id",
        expire_at=datetime.now(timezone.utc) + timedelta(minutes=5),
        transport_frames=lambda payload, last: print(
            f"Received payload: {len(payload)} bytes, last={last}"
        ),
        on_error=lambda err: print(f"Session error: {err}"),
        on_close=lambda: print("Session closed"),
    )

    await session.init()
    connection_id = await session.start()
    print(f"Connected: {connection_id}")

    audio_data = b"..."  # mono PCM s16le audio bytes
    request_id = await session.send_audio(audio_data, end=True)
    print(f"Sent audio request: {request_id}")

    await asyncio.sleep(10)
    await session.close()


if __name__ == "__main__":
    asyncio.run(main())

Session Configuration

Use new_avatar_session() to configure and create a session:

from datetime import datetime, timedelta, timezone

from spatius import AudioFormat, new_avatar_session


session = new_avatar_session(
    avatar_id="avatar-123",
    api_key="your-api-key",
    app_id="your-app-id",
    use_query_auth=False,
    expire_at=datetime.now(timezone.utc) + timedelta(minutes=5),
    region="us-west",
    sample_rate=16000,
    audio_format=AudioFormat.PCM_S16LE,
    transport_frames=on_frame_received,
    on_error=on_error,
    on_close=on_close,
)

use_query_auth=False sends WebSocket credentials in headers. Set use_query_auth=True for web-style auth that sends appId and sessionKey in the WebSocket query string.

Session Lifecycle

# 1. Initialize and request a session token
await session.init()

# 2. Start the WebSocket connection
connection_id = await session.start()

# 3. Send audio data
request_id = await session.send_audio(audio_bytes, end=True)

# 4. Receive motion data payloads through transport_frames

# 5. Close the session
await session.close()

For audio source and timing guidance, see Audio.

Audio Format

The SDK supports two session-level input formats:

AudioFormat.PCM_S16LE - mono 16-bit PCM bytes
AudioFormat.OGG_OPUS - one continuous Ogg Opus stream per request ID

PCM Input

Sample rate: one of 8000, 16000, 22050, 24000, 32000, 44100, 48000
Channels: 1 (mono)
Bit depth: 16-bit
Format: raw PCM bytes

from spatius import AudioFormat, new_avatar_session


session = new_avatar_session(
    ...,
    sample_rate=16000,
    audio_format=AudioFormat.PCM_S16LE,
)

with open("audio.pcm", "rb") as f:
    audio_data = f.read()

await session.send_audio(audio_data, end=True)

Ogg Opus Input

Sample rate: one of 8000, 12000, 16000, 24000, 48000
Channels: 1 (mono)
Format: Ogg Opus pages/chunks
Request contract: each request ID must carry one continuous Ogg Opus stream across one or more send_audio() calls, and the final chunk must use end=True

from spatius import AudioFormat, new_avatar_session


session = new_avatar_session(
    ...,
    sample_rate=24000,
    bitrate=32000,
    audio_format=AudioFormat.OGG_OPUS,
)

with open("audio.ogg", "rb") as f:
    while chunk := f.read(4096):
        await session.send_audio(chunk, end=False)

await session.send_audio(b"", end=True)

Built-In PCM to Ogg Opus Encoder

If you want the session to negotiate AudioFormat.OGG_OPUS but still provide raw PCM bytes to send_audio(), enable the optional internal encoder.

from spatius import AudioFormat, OggOpusEncoderConfig, new_avatar_session


encoded_outputs = []

session = new_avatar_session(
    ...,
    sample_rate=24000,
    bitrate=32000,
    audio_format=AudioFormat.OGG_OPUS,
    ogg_opus_encoder=OggOpusEncoderConfig(frame_duration_ms=20),
    on_encoded_audio=lambda req_id, payload: encoded_outputs.append((req_id, payload)),
)

with open("audio_24000.pcm", "rb") as f:
    pcm_audio = f.read()

await session.send_audio(pcm_audio, end=True)

Notes:

The internal encoder is optional; if you do not install spatius[opus], keep using PCM or provide pre-encoded Ogg Opus bytes yourself.
on_encoded_audio fires when internal encoding completes for a request and receives (req_id, encoded_audio_bytes).
If audio_format=AudioFormat.OGG_OPUS and ogg_opus_encoder is unset, send_audio() forwards your pre-encoded Ogg Opus bytes unchanged.

LiveKit Egress Mode

When configured with livekit_egress, audio and motion data are streamed to a LiveKit room instead of being returned through the WebSocket connection. For end-to-end integration details, see the LiveKit Agent server side guide.

from datetime import datetime, timedelta, timezone

from spatius import LiveKitEgressConfig, new_avatar_session


session = new_avatar_session(
    avatar_id="avatar-123",
    api_key="your-api-key",
    app_id="your-app-id",
    region="us-west",
    expire_at=datetime.now(timezone.utc) + timedelta(minutes=5),
    livekit_egress=LiveKitEgressConfig(
        url="wss://livekit.example.com",
        api_token="livekit-token",
        room_name="my-room",
        publisher_id="avatar-publisher",
    ),
)

api_key and api_secret remain supported for backward compatibility, but they are deprecated. Prefer api_token for new integrations. When LiveKit egress is enabled:

the server streams output to the specified LiveKit room
the transport_frames callback is not invoked
audio and motion data are published under the configured publisher ID

Agora Egress Mode

When configured with agora_egress, audio and motion data are streamed to an Agora channel instead of being returned through the WebSocket connection.

from datetime import datetime, timedelta, timezone

from spatius import AgoraEgressConfig, new_avatar_session


session = new_avatar_session(
    avatar_id="avatar-123",
    api_key="your-api-key",
    app_id="your-app-id",
    region="us-west",
    expire_at=datetime.now(timezone.utc) + timedelta(minutes=5),
    agora_egress=AgoraEgressConfig(
        channel_name="my-channel",
        token="agora-token",
        uid=0,
        publisher_id="avatar-publisher",
    ),
)

Interrupt

The interrupt() method sends an interrupt signal to stop current audio processing. This is only available when using egress mode.

request_id = await session.send_audio(audio_data, end=True)

# Later, interrupt the most recent in-flight request.
interrupted_id = await session.interrupt()
print(f"Interrupted request: {interrupted_id}")

The interrupt uses the most recent request ID, even after end=True has been sent.

Callbacks

`transport_frames`

Receives motion data payloads from the server:

def on_frame_received(payload: bytes, is_last: bool):
    print(f"Received payload: {len(payload)} bytes")
    if is_last:
        print("This is the last payload")

`on_error`

Handles structured SDK errors from the session:

from spatius import AvatarSDKError


def on_error(error: Exception):
    print(f"Session error: {error}")

    if isinstance(error, AvatarSDKError):
        print("  code:", error.code.value)
        print("  phase:", error.phase)
        print("  http_status:", error.http_status)
        print("  server_code:", error.server_code)
        print("  server_detail:", error.server_detail)

The SDK reports structured AvatarSDKError instances for token creation failures, WebSocket upgrade rejections, handshake failures, runtime ServerError messages, and unexpected connection drops.

`on_close`

Called when the session closes:

def on_close():
    print("Session has been closed")

Error Handling

Use SessionTokenError for token creation failures and AvatarSDKError for all other structured SDK errors:

from spatius import AvatarSDKError, SessionTokenError


try:
    await session.init()
    await session.start()
except SessionTokenError as error:
    print("token failed", error.code.value, error.server_detail)
except AvatarSDKError as error:
    print("sdk error", error.code.value, error.phase, error.server_detail)

AvatarSDKError and SessionTokenError expose these fields:

code - stable SDK error code
message - human-readable message
phase - failure phase such as session_token, websocket_connect, websocket_handshake, websocket_runtime, or websocket_send
http_status - HTTP status for token or WebSocket upgrade failures
server_code - server-provided error code, including runtime protobuf ServerError.code
server_title / server_detail - parsed server error details when available
connection_id / req_id - server correlation identifiers when available
raw_body - raw HTTP rejection body for token or WebSocket upgrade failures
close_code / close_reason - WebSocket close details for unexpected disconnects

Common `AvatarSDKErrorCode` Values

sessionTokenExpired - session token expired or unauthorized
sessionTokenInvalid - invalid or empty session token
appIDUnrecognized - App ID is not recognized by the server
appIDMismatch - session token belongs to a different app
avatarNotFound - avatar does not exist
billingRequired - session denied by billing checks
creditsExhausted - runtime or connect-time credits exhausted
sessionDurationExceeded - billing-enforced session timeout reached
unsupportedSampleRate - handshake rejected unsupported audio sample rate
invalidEgressConfig - egress config is invalid
egressUnavailable - egress service is unavailable or not configured
idleTimeout - server closed the session after input inactivity
upstreamError - internal upstream service failed
protocolError - invalid protobuf or unexpected message sequence
connectionFailed - transport-level connection failure
connectionClosed - unexpected WebSocket close
serverError - server-side failure that did not match a more specific mapping
invalidRequest - other client-side request validation errors
unknown - fallback when the SDK cannot classify the failure

For server-side error details, see Server Error Codes.

SDK Reference

This page includes usage guidance. The SDK source also keeps developer-facing docstrings on public classes and functions so generated API reference pages can stay close to the code.

`new_avatar_session(...)`

Creates an AvatarSession from typed keyword parameters. It is the primary public factory.

`AvatarSession`

Main class for managing avatar sessions.

Methods

async init() - initialize the session and obtain a token
async start() -> str - start the WebSocket connection and return a connection_id
async send_audio(audio: bytes, end: bool = False) -> str - send audio data and return a request ID
async interrupt() -> str - interrupt current audio processing in egress mode
async close() - close the session and clean up resources
config -> SessionConfig - current session configuration

`SessionConfig`

Configuration dataclass for avatar sessions.

Fields

avatar_id: str - avatar identifier
api_key: str - API key for authentication
app_id: str - application identifier
use_query_auth: bool - send WebSocket auth in query params instead of headers
expire_at: datetime - session expiration time
sample_rate: int - audio sample rate, default 16000
bitrate: int - audio bitrate, default 0
audio_format: AudioFormat - negotiated input format
ogg_opus_encoder: Optional[OggOpusEncoderConfig] - optional PCM to Ogg Opus encoder config
on_encoded_audio: Optional[Callable[[str, bytes], None]] - callback for encoded Ogg output
transport_frames: Callable[[bytes, bool], None] - frame callback
on_error: Callable[[Exception], None] - error callback
on_close: Callable[[], None] - close callback
region: str - Spatius region. Defaults to us-west; see Regions.
console_endpoint_url: str - optional explicit Console API URL
ingress_endpoint_url: str - optional explicit ingress WebSocket URL
livekit_egress: Optional[LiveKitEgressConfig] - LiveKit egress configuration
agora_egress: Optional[AgoraEgressConfig] - Agora egress configuration

`LiveKitEgressConfig`

Configuration for streaming to a LiveKit room.

Fields

url: str - LiveKit server URL, for example wss://livekit.example.com
api_key: str - deprecated LiveKit API key
api_secret: str - deprecated LiveKit API secret
api_token: str - preferred pre-generated LiveKit access token
room_name: str - room name to join
publisher_id: str - publisher identity in the room
extra_attributes: dict[str, str] - extra participant attributes
idle_timeout: int - idle timeout in seconds, 0 uses server defaults

`AgoraEgressConfig`

Configuration for streaming to an Agora channel.

Fields

channel_name: str - Agora channel name
token: str - Agora token
uid: int - publisher UID, 0 for auto-assign
publisher_id: str - publisher identity/name

Utility Functions

generate_log_id() -> str - generate a unique log ID in the format YYYYMMDDHHMMSS_<nanoid>

Exceptions

AvatarSDKError - structured SDK error with stable code and context fields
SessionTokenError - subclass of AvatarSDKError raised when session token creation fails

Changelog

spatius-sdk-python releases

Reference Overview

Client SDKs

Server SDKs

Session Token API

Regions & Endpoints

Error Codes

Repository

Requirements

Region configuration

Quick Start

Session Configuration

Session Lifecycle

Audio Format

PCM Input

Ogg Opus Input

Built-In PCM to Ogg Opus Encoder

LiveKit Egress Mode

Agora Egress Mode

Interrupt

Callbacks

`transport_frames`

`on_error`

`on_close`

Error Handling

Common `AvatarSDKErrorCode` Values

SDK Reference

`new_avatar_session(...)`

`AvatarSession`

Methods

`SessionConfig`

Fields

`LiveKitEgressConfig`

Fields

`AgoraEgressConfig`

Fields

Utility Functions

Exceptions

Changelog

​Repository

​Requirements

​Region configuration

​Quick Start

​Session Configuration

​Session Lifecycle

​Audio Format

​PCM Input

​Ogg Opus Input

​Built-In PCM to Ogg Opus Encoder

​LiveKit Egress Mode

​Agora Egress Mode

​Interrupt

​Callbacks

​transport_frames

​on_error

​on_close

​Error Handling

​Common AvatarSDKErrorCode Values

​SDK Reference

​new_avatar_session(...)

​AvatarSession

​Methods

​SessionConfig

​Fields

​LiveKitEgressConfig

​Fields

​AgoraEgressConfig

​Fields

​Utility Functions

​Exceptions

​Changelog

Repository

Requirements

Region configuration

Quick Start

Session Configuration

Session Lifecycle

Audio Format

PCM Input

Ogg Opus Input

Built-In PCM to Ogg Opus Encoder

LiveKit Egress Mode

Agora Egress Mode

Interrupt

Callbacks

`transport_frames`

`on_error`

`on_close`

Error Handling

Common `AvatarSDKErrorCode` Values

SDK Reference

`new_avatar_session(...)`

`AvatarSession`

Methods

`SessionConfig`

Fields

`LiveKitEgressConfig`

Fields

`AgoraEgressConfig`

Fields

Utility Functions

Exceptions

Changelog