Estuary Python SDK

Python SDK for the Estuary real-time AI conversation platform. Supports text chat, streaming voice (WebSocket and LiveKit WebRTC), vision, and memory.

Installation

pip install estuary-sdk

Install with optional extras:

pip install estuary-sdk[audio]    # Microphone recording + speaker playback
pip install estuary-sdk[livekit]  # LiveKit WebRTC voice
pip install estuary-sdk[all]      # Everything

Or with PDM:

pdm install              # Core only
pdm install -G audio     # + audio
pdm install -G livekit   # + LiveKit
pdm install -G all       # Everything

Requires Python 3.11+.

Getting Your Credentials

To use the SDK you need an API key and a character ID from the Estuary Dashboard:

1. Sign up or log in at app.estuary-ai.com
2. Create a character — go to Characters and click Create Character. Configure your character's name, personality, and voice, then save.
3. Copy the character ID — on the character's page, copy the UUID shown under the character name (or from the URL).
4. Generate an API key — go to Settings → API Keys and click Create Key. Copy the key (it starts with est_).

Use these values for api_key and character_id in the examples below.

Quick Start

import asyncio
from estuary_sdk import EstuaryClient, EstuaryConfig, BotResponse

async def main():
    config = EstuaryConfig(
        server_url="https://api.estuary-ai.com",
        api_key="est_...",
        character_id="your-character-uuid",
        player_id="player-1",
    )

    async with EstuaryClient(config) as client:
        client.on("bot_response", lambda r: print(r.text, end="" if not r.is_final else "\n"))

        await client.connect()
        client.send_text("Hello!")

        await asyncio.sleep(5)  # Wait for response

asyncio.run(main())

Voice

Continuous Mode

Audio streams continuously. The server uses VAD (voice activity detection) to detect turn boundaries.

from estuary_sdk import VoiceMode

await client.start_voice(VoiceMode.CONTINUOUS)

# Send raw PCM16 audio (16-bit signed, mono, 16kHz)
await client.send_audio(pcm_bytes)

await client.stop_voice()

Push-to-Talk

You control when audio is captured and when the turn ends.

await client.start_voice(VoiceMode.PUSH_TO_TALK)

await client.start_recording()
await client.send_audio(pcm_bytes)
await client.stop_recording()  # Triggers end-of-turn

await client.stop_voice()

With Microphone (requires audio extra)

from estuary_sdk.audio import AudioRecorder, AudioPlayer

recorder = AudioRecorder(on_audio=client.send_audio)
player = AudioPlayer()

client.on("bot_voice", player.enqueue)

await client.start_voice()
await recorder.start()

VoiceSession

VoiceSession is a high-level wrapper that owns the AudioPlayer, AudioRecorder, and event handlers as an async context manager:

from estuary_sdk import VoiceSession, VoiceMode

async with EstuaryClient(config) as client:
    await client.connect()

    async with VoiceSession(client, mode=VoiceMode.CONTINUOUS) as session:
        # Microphone and speaker are wired up, voice is active.
        await asyncio.Event().wait()

For push-to-talk, control recording within the session:

async with VoiceSession(client, mode=VoiceMode.PUSH_TO_TALK) as session:
    await session.start_recording()   # User presses button
    await asyncio.sleep(3)
    await session.stop_recording()    # User releases button

Requires the audio extra (pip install estuary-sdk[audio]).

Utilities

client.toggle_mute()                        # Toggle microphone mute
print(client.is_muted)                      # Check mute state
print(client.is_voice_active)               # Check if voice is running
client.interrupt()                          # Interrupt current bot response
client.notify_audio_playback_complete()     # Tell server playback finished

# Script a specific line (skips LLM, goes straight to TTS)
client.say_line("Welcome back, friend.")
client.say_line("Thinking...", text_only=True)  # No TTS audio, just chat history

Vision

Send a camera image for multimodal VLM processing:

client.send_camera_image(image_base64, "image/jpeg")

# Respond to server-initiated capture requests
client.on("camera_capture_request", lambda req: client.send_camera_image(b64, "image/jpeg", request_id=req.request_id))

Streaming Responses

Bot responses stream token-by-token. Use is_final to detect the complete message.

def on_bot_response(response: BotResponse):
    if response.is_final:
        print(f"[{response.message_id}] {response.text}")
    else:
        print(response.text, end="", flush=True)

client.on("bot_response", on_bot_response)

Send and Wait

For simple request-response flows, send_text_and_wait() sends a message and returns the final BotResponse directly — no manual event wiring needed:

response = await client.send_text_and_wait("What is the capital of France?")
print(response.text)  # "The capital of France is Paris."

# With options
response = await client.send_text_and_wait(
    "Summarize our conversation.",
    text_only=True,   # Suppress voice response
    timeout=30.0,     # Max seconds to wait (default: 20)
)

Raises asyncio.TimeoutError if no final response arrives within the timeout. The listener is cleaned up automatically whether the call succeeds or times out.

Memory

The REST memory API is available after connect() via client.memory.

# Semantic search (limit is 1 to 50, server default 10)
found = await client.memory.search("favorite color", limit=5)
for hit in found.results:
    print(hit.memory_id, f"{hit.similarity:.2f}", hit.content)

# List with filters
memories = await client.memory.get_memories(memory_type="preference", limit=20)

# Other endpoints
stats = await client.memory.get_stats()
facts = await client.memory.get_core_facts()
timeline = await client.memory.get_timeline(group_by="week")
graph = await client.memory.get_graph()

# Delete all memories
result = await client.memory.delete_all(confirm=True)
print(result.deleted_count, result.message)

Memory, player, character generation and 3D model calls all use the canonical /api/v1/characters/... routes. search() sends POST with a JSON body and each hit carries memory_id, similarity and content; earlier releases sent GET ?q=, which the gateway always answered with 405.

Real-time memory extraction events:

from estuary_sdk import MemoryUpdatedEvent

def on_memory(event: MemoryUpdatedEvent):
    print(f"Extracted {event.memories_extracted} memories")

client.on("memory_updated", on_memory)

Characters

Manage characters via client.characters (available immediately — no connect() required):

char = await client.characters.create("My Character", personality="Friendly helper")
chars = await client.characters.list()
char = await client.characters.get("character-uuid")
char = await client.characters.update("character-uuid", name="New Name")
await client.characters.delete("character-uuid")

New characters default to Inworld inworld-tts-2-flash. For the Inworld TTS-2 family (inworld-tts-2 and inworld-tts-2-flash), pass the optional tts_delivery_mode to create() or update():

char = await client.characters.create("Narrator", tts_delivery_mode="BALANCED")
char = await client.characters.update(char.id, tts_delivery_mode="CREATIVE")

Values are STABLE, BALANCED, or CREATIVE (case-insensitive on input). The response returns the uppercase value, or None when unset. Other TTS models ignore this setting.

Players

Query player data via client.players (available after connect()):

stats = await client.players.get_stats()
players = await client.players.list(limit=20)
player = await client.players.get("player-123")
messages = await client.players.get_messages("player-123")
deleted = await client.players.delete("player-123")
print(deleted["message"])  # "Player conversation deleted"; server answers 204

Character Generation

Generate a character from a photo via client.generate (available immediately):

with open("photo.jpg", "rb") as f:
    character = await client.generate.image_to_character(f.read())

# Start the animated humanoid pipeline (omit rigged=True for a static model)
started = await client.generate.generate_model(character.id, rigged=True)
print(started["modelStatus"])  # "posing"

# Poll 3D model progress
status = await client.generate.get_model_status(character.id)
print(status.model_status, f"{status.progress}%", status.rigged)

Use wait_for_model() to poll until the 3D model is ready (or fails), with an optional progress callback:

from estuary_sdk import ModelStatus

status = await client.generate.wait_for_model(
    character.id,
    poll_interval=2.0,   # Seconds between polls (default: 2)
    timeout=600.0,       # Rigged generation can take several minutes (default: 300)
    on_progress=lambda s: print(f"{s.model_status} {s.progress}%"),
)
print("Final:", status.model_status, status.model_url)
if status.rigged:
    print("Animation clips:", status.animations)

rigged=True works for humanoid characters and produces one skinned GLB with animation clips. ModelStatus.animations contains full glTF clip names such as preset:biped:idle. The character action name is the suffix (idle). If wait_for_model() returns rig_failed or animation_failed, check model_url: the generated static model may still be usable. The default generate_model(id) continues to request a static model.

Agent-to-Agent Conversations

Start a conversation between two AI characters and observe the exchange in real-time:

from estuary_sdk import AgentTurnText, AgentTurnComplete, AgentConversationComplete

client.on("agent_turn_text", lambda t: print(f"[Agent {t.agent_id}] {t.text}", end="" if not t.is_final else "\n"))
client.on("agent_turn_complete", lambda t: print(f"--- Turn {t.turn_number} complete ---"))
client.on("agent_conversation_complete", lambda c: print(f"Done after {c.total_turns} turns: {c.reason}"))

client.start_agent_conversation(
    agent_a_id="character-uuid-1",
    agent_b_id="character-uuid-2",
    conversation_context="Discuss the meaning of life.",
    max_turns=8,
    timeout_seconds=90,
)

# To stop early:
client.stop_agent_conversation()

CLI Tester

An interactive chat program is included for quick testing:

python examples/chat.py --api-key [API_KEY] --character-id [CHARACTER_ID]

Or via PDM:

pdm run chat --api-key [API_KEY] --character-id [CHARACTER_ID]

In-chat commands: /quit, /voice, /stop, /memories, /help

Configuration

EstuaryConfig fields:

Capability Declaration

Tell the server what the host device can physically do. The server hides tools that require a missing capability — e.g. request_camera_image is suppressed when camera=False, so the character never asks a headless integration to look through a camera it doesn't have.

Declare device capability, not current usage mode: a voice-capable client sets microphone=True even while it is being used for text.

from estuary_sdk import EstuaryConfig, SessionCapabilities

config = EstuaryConfig(
    server_url=..., api_key=..., character_id=..., player_id=...,
    capabilities=SessionCapabilities(
        camera=False,      # Headless service — no camera to capture from
        microphone=True,
        speaker=True,
    ),
)

All fields default to True. Omitting capabilities entirely sends {"version": "1", "camera": True, "microphone": True, "speaker": True} — identical to the server's own default for clients that declare nothing, so upgrading changes no behavior.

Events

Register listeners with client.on(), client.once(), or client.off():

client.on("bot_response", handle_response)     # Persistent listener
client.once("connected", handle_first_connect)  # One-time listener
client.off("bot_response", handle_response)     # Remove listener

Both sync and async callbacks are supported.

Error Handling

from estuary_sdk import EstuaryError, ErrorCode

try:
    await client.connect()
except EstuaryError as e:
    print(e.code)     # ErrorCode enum
    print(e.details)  # Optional extra info

Error codes:

REST failures raise EstuaryError with REST_ERROR, and e.details holds {"status", "body"}. An API key that lacks the scope a route needs (for example memories:read) gets status 403 with a body of {"detail": {"error": "insufficient_scope", "requiredScope": "..."}}. Keys created before scopes existed hold every scope.

Every REST request carries X-Estuary-Client: estuary-python-sdk/<version> (also available as estuary_sdk.__version__) so the gateway can attribute traffic by SDK version.

Optional Dependencies

Development

cd estuary-python-sdk
pdm install -G dev

pdm run test          # pytest
pdm run format        # black
pdm run lint          # flake8
pdm run sort-imports  # isort
pdm run typecheck     # mypy (strict)

Restricted end-user credentials (0.8.0)

Keep your Estuary developer key on your backend. Verify the user's account, profile and allowed characters there, then request /api/v1/client-tokens. Supply EstuaryConfig.token_provider, an async callback returning ClientCredential(token, player_id, expires_at). It uses your own backend's headers/cookies; those credentials are not sent to Estuary. Set api_key=""; the returned canonical player ID overrides the configured player.

The callback is shared by Socket.IO, REST and multipart uploads. The SDK caches credentials, renews the active socket before expiry, rejects profile changes, and releases voice after expiration/revocation or failed renewal. Changing profiles requires a new client. The backend must retain and renew the same Estuary session ID for an active call; changing operations requires reconnecting.

Restricted grants cover character/player data and permitted text/voice/vision. Administrative routes, share creation and encounters remain unavailable to these tokens. Existing trusted API-key flows are unchanged. See the authenticated backend example.

Install this flow with python -m pip install --upgrade "estuary-sdk>=0.8.0". It requires a gateway with the client-token endpoints.