Core Concepts

Understand the architecture, event system, and design patterns of the Estuary Python SDK.

Architecture

The EstuaryClient is the main entry point. It composes several internal modules:

• SocketManager handles the WebSocket connection, authentication, reconnection logic, and event routing.
• VoiceManager streams audio to the server. Two implementations exist: WebSocketVoiceManager and LiveKitVoiceManager.
• REST Clients provide access to HTTP APIs: MemoryClient, PlayersClient, CharactersClient, and GenerateClient. Every call they make goes to a versioned /api/v1/ route.

Each REST request also carries an X-Estuary-Client header identifying the SDK and its version, in the form estuary-python-sdk/<version>, where the version is the value of estuary_sdk.__version__. It is sent on REST requests only, never on the Socket.IO connection. See Client Identification for what the header is used for.

Async-Only Design

The SDK is built entirely on asyncio. All I/O operations are async:

import asyncio
from estuary_sdk import EstuaryClient, EstuaryConfig

async def main():
    async with EstuaryClient(EstuaryConfig(...)) as client:
        session = await client.connect()
        await client.start_voice()
        memories = await client.memory.get_memories()

asyncio.run(main())

The EstuaryClient supports the async with context manager, which automatically calls disconnect() on exit. This ensures connections and resources are cleaned up even if an exception occurs.

Event-Driven Design

The SDK uses an AsyncEventEmitter pattern. Register async callback functions for events:

async def on_response(response):
    # response is a BotResponse dataclass
    print(response.text)

async def on_connected(session):
    # session is a SessionInfo dataclass
    print(session.session_id)

client.on("bot_response", on_response)
client.on("connected", on_connected)

You can subscribe with on(), unsubscribe with off(), and use once() for one-time listeners:

# Listen until removed
client.on("bot_response", handler)
client.off("bot_response", handler)

# Listen once
async def on_first_connect(session):
    print("First connection:", session.session_id)

client.once("connected", on_first_connect)

Connection Lifecycle

The client tracks its connection through five states:

from estuary_sdk import ConnectionState

async def on_state_changed(state):
    if state == ConnectionState.CONNECTING:
        print("Connecting...")
    elif state == ConnectionState.CONNECTED:
        print("Connected!")
    elif state == ConnectionState.RECONNECTING:
        print("Reconnecting...")
    elif state == ConnectionState.ERROR:
        print("Connection error")

client.on("connection_state_changed", on_state_changed)

When auto_reconnect is enabled (the default), the client automatically attempts to reconnect on disconnect, with increasing delay between attempts up to max_reconnect_attempts (default: 5).

Error Handling

All SDK errors are instances of EstuaryError, which extends Exception with a typed code property:

from estuary_sdk import EstuaryError, ErrorCode

async def on_error(err):
    if isinstance(err, EstuaryError):
        if err.code == ErrorCode.AUTH_FAILED:
            print("Invalid API key")
        elif err.code == ErrorCode.QUOTA_EXCEEDED:
            print("Usage limit reached")
        elif err.code == ErrorCode.VOICE_NOT_SUPPORTED:
            print("Voice not available")
        else:
            print("Error:", err)

client.on("error", on_error)

Methods like connect(), start_voice(), and send_text() raise EstuaryError on failure. Always wrap connect() in a try/except:

try:
    await client.connect()
except EstuaryError as err:
    if err.code == ErrorCode.AUTH_FAILED:
        print("Invalid credentials")

REST Errors

Calls on the REST sub-clients (client.memory, client.players, client.characters, client.generate, client.shares) raise EstuaryError with ErrorCode.REST_ERROR when the server rejects the request. For an HTTP error response, err.details is a dict holding the status code and the raw response body text:

from estuary_sdk import EstuaryError, ErrorCode

try:
    await client.memory.delete_all(confirm=True)
except EstuaryError as err:
    if err.code == ErrorCode.REST_ERROR and isinstance(err.details, dict):
        print(err.details["status"])  # e.g. 403
        print(err.details["body"])    # response body, as text

If the request never reached the server (a network failure or a timeout), err.details is the underlying exception instead, so check the type before subscripting.

Insufficient scope

An API key carries scopes that decide which routes it can reach. A valid key that lacks the scope a route needs gets HTTP 403 with a body that names the missing scope:

{
  "detail": {
    "error": "insufficient_scope",
    "message": "This API key lacks the 'memories:write' scope",
    "requiredScope": "memories:write"
  }
}

Handle it like any other REST failure, and read requiredScope to see what the key is missing:

import json

try:
    await client.memory.delete_all(confirm=True)
except EstuaryError as err:
    if isinstance(err.details, dict) and err.details["status"] == 403:
        detail = json.loads(err.details["body"]).get("detail", {})
        if detail.get("error") == "insufficient_scope":
            print("Key needs the", detail["requiredScope"], "scope")

Keys created before scopes existed hold every scope, so older integrations are unaffected. See API Key Scopes for the full scope list and which routes need which.

Configuration

The EstuaryConfig dataclass controls all client behavior. Only server_url, api_key, character_id, and player_id are required:

from estuary_sdk import EstuaryConfig

config = EstuaryConfig(
    # Required
    server_url="https://api.estuary-ai.com",
    api_key="est_your_api_key",
    character_id="your-character-uuid",
    player_id="user-123",

    # Optional
    voice_transport="websocket",     # "websocket" | "livekit" | "auto"
    audio_sample_rate=24000,          # Hz
    auto_reconnect=True,
    max_reconnect_attempts=5,
    reconnect_delay=1.0,              # seconds
    debug=False,
    realtime_memory=False,            # Enable real-time memory extraction events
)

See the Configuration Reference for details on every option.

Next Steps

• Text Chat -- Send messages and handle streaming responses
• Voice (WebSocket) -- Add voice input and output
• API Reference -- Full reference for all classes, events, and types