> ## Documentation Index
> Fetch the complete documentation index at: https://docs.spatius.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# Python SDK

> Python Server SDK reference for Backend Mode sessions, audio input, motion data callbacks, egress modes, and errors.

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](https://github.com/spatius-ai/spatius-sdk-python) and [PyPI](https://pypi.org/project/spatius/).

```bash theme={null}
pip install spatius
```

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

```bash theme={null}
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 `auto`: bootstrap selects a region and falls back to `us-west` when automatic selection is unavailable. See [Regions](/api-reference/regions).

## Telemetry

The SDK exports OpenTelemetry metrics and traces by default; it does not export logs. Telemetry resources include the App ID and resolved region. Disable export before creating a session if your application does not want SDK telemetry:

```python theme={null}
from spatius import configure_telemetry

configure_telemetry("")
```

See the [SDK repository](https://github.com/spatius-ai/spatius-sdk-python#telemetry) for endpoint configuration and shutdown behavior.

## Region configuration

Passing a region is enough for normal production use:

```python theme={null}
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

```python theme={null}
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:

```python theme={null}
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

```python theme={null}
# 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](/concepts/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

```python theme={null}
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`

```python theme={null}
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.

```python theme={null}
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

<Note>
  LiveKit and Agora egress are low-level Server SDK capabilities. They are not additional integration paths; LiveKit Agents and Agora Convo AI remain the supported packaged integrations.
</Note>

When configured with `livekit_egress`, audio and motion data are streamed to a LiveKit room instead of being returned through the WebSocket connection.

```python theme={null}
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.

```python theme={null}
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.

```python theme={null}
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:

```python theme={null}
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:

```python theme={null}
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:

```python theme={null}
def on_close():
    print("Session has been closed")
```

## Error Handling

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

```python theme={null}
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 recovery guidance, see [Server Error Handling](/resources/server-error).

## 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 `auto`; see [Regions](/api-reference/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` lets Agora assign one
* `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](https://github.com/spatius-ai/spatius-sdk-python/releases)