Skip to main content

Getting Started

Get up and running with the Estuary Python SDK in minutes. This guide walks you through installation, configuration, and sending your first message to an AI character.

For end-user deployments, use the authenticated backend token-provider flow instead of bundling a developer key. Install Python SDK 0.8.0 or later from PyPI; this flow requires the gateway's client-token endpoints.

Prerequisites​

Before you begin, ensure you have:

RequirementDetails
Python3.11 or later
Estuary AccountSign up at app.estuary-ai.com
API KeyGenerated from your Estuary dashboard (starts with est_)
Character IDThe UUID of your AI character

Installation​

Install the SDK from PyPI:

pip install estuary-sdk

Optional Extras​

The core SDK has minimal dependencies (python-socketio + aiohttp). Install optional extras for additional features:

# Audio recording and playback (sounddevice + numpy)
pip install "estuary-sdk[audio]"

# LiveKit WebRTC voice
pip install "estuary-sdk[livekit]"

# Everything
pip install "estuary-sdk[all]"

Quick Start​

Here is a minimal example that connects to a character and exchanges text messages:

import asyncio
from estuary_sdk import EstuaryClient, EstuaryConfig

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

async def main():
    async with EstuaryClient(config) as client:
        # Listen for bot responses
        async def on_response(response):
            if response.is_final:
                print("Bot:", response.text)

        client.on("bot_response", on_response)

        # Connect and send a message
        session = await client.connect()
        print("Connected! Session:", session.session_id)

        client.send_text("Hello! Tell me about yourself.")

        # Keep running to receive responses
        await asyncio.sleep(10)

asyncio.run(main())

What Happens Under the Hood​

  1. client.connect() opens a WebSocket connection to the Estuary server, authenticates with your API key, and returns a SessionInfo object.
  2. client.send_text() sends your message over the WebSocket.
  3. The server processes your message through the AI pipeline and streams the response back.
  4. The bot_response event fires for each chunk of the response. When is_final is True, the full response text is available.

Complete Example​

This example demonstrates connection lifecycle handling, streaming responses, and cleanup:

import asyncio
from estuary_sdk import EstuaryClient, EstuaryConfig, ConnectionState

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

async def main():
    async with EstuaryClient(config) as client:
        # Connection events
        async def on_connected(session):
            print("Connected! Conversation:", session.conversation_id)

        async def on_disconnected(reason):
            print("Disconnected:", reason)

        async def on_reconnecting(attempt):
            print(f"Reconnecting (attempt {attempt})...")

        # Response events
        async def on_response(response):
            if response.is_final:
                print("Bot:", response.text)
            else:
                print(response.partial, end="", flush=True)

        # Error handling
        async def on_error(err):
            print("Error:", err)

        async def on_auth_error(message):
            print("Authentication failed:", message)

        async def on_quota(data):
            print(f"Quota exceeded: {data}")

        client.on("connected", on_connected)
        client.on("disconnected", on_disconnected)
        client.on("reconnecting", on_reconnecting)
        client.on("bot_response", on_response)
        client.on("error", on_error)
        client.on("auth_error", on_auth_error)
        client.on("quota_exceeded", on_quota)

        # Connect and send
        await client.connect()
        client.send_text("Hello!")

        # Keep running
        await asyncio.sleep(30)

asyncio.run(main())

Next Steps​

Now that you have a working connection:

Last updated