estuary-sdk 0.2.5

Python SDK for the Estuary real-time AI conversation platform

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()

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)

Memory

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

# Search memories
results = await client.memory.search("favorite color")

# 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()

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)

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]

Flag	Short	Default	Description
--api-key	-k	required	API key
--character-id	-c	required	Character UUID
--server-url	-s	https://api.estuary-ai.com	Server URL
--player-id	-p	python-sdk-tester	Player ID
--debug	-d	off	Verbose logging
--text-only	-t	off	Suppress voice responses

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

Configuration

EstuaryConfig fields:

Field	Type	Default	Description
server_url	str	required	Server URL
api_key	str	required	API key
character_id	str	required	Character UUID
player_id	str	required	Player identifier
audio_sample_rate	int	16000	Audio sample rate (Hz)
auto_reconnect	bool	True	Auto-reconnect on disconnect
max_reconnect_attempts	int	5	Max reconnection attempts
reconnect_delay	float	1.0	Initial reconnect delay (seconds)
debug	bool	False	Enable debug logging
voice_transport	str	"websocket"	"websocket", "livekit", or "auto"
realtime_memory	bool	False	Enable realtime memory updates

Events

Event	Payload	Description
connected	SessionInfo	Connected and authenticated
disconnected	str	Disconnected (reason)
reconnecting	int	Reconnection attempt number
connection_state_changed	ConnectionState	State transition
bot_response	BotResponse	Streaming text response
bot_voice	BotVoice	Audio chunk (base64)
stt_response	SttResponse	Speech-to-text result
interrupt	InterruptData	Bot response interrupted
quota_exceeded	QuotaExceededData	Usage quota hit
camera_capture_request	CameraCaptureRequest	Server requests a camera image
voice_started	—	Voice session started
voice_stopped	—	Voice session stopped
memory_updated	MemoryUpdatedEvent	New memories extracted
livekit_connected	str	LiveKit room name
livekit_disconnected	—	LiveKit disconnected
error	Exception	Error occurred
auth_error	str	Authentication failed

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:

Code	Description
CONNECTION_FAILED	Could not connect to server
AUTH_FAILED	Authentication rejected
CONNECTION_TIMEOUT	Connection timed out
QUOTA_EXCEEDED	Usage quota exceeded
VOICE_NOT_SUPPORTED	Voice not supported
VOICE_ALREADY_ACTIVE	Voice session already running
VOICE_NOT_ACTIVE	No active voice session
LIVEKIT_UNAVAILABLE	LiveKit dependency not installed
NOT_CONNECTED	Not connected to server
REST_ERROR	REST API request failed
UNKNOWN	Unknown error

Optional Dependencies

Extra	Packages	Purpose
audio	sounddevice, numpy	Microphone capture + speaker playback
livekit	livekit	LiveKit WebRTC voice transport
all	All of the above	Everything
dev	pytest, black, isort, flake8, mypy	Development tools

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)