Ensoul Python SDK

Official Python SDK for the Ensoul API. Build AI personas and NPCs with memory and personality that evolve through real conversation. Scale to thousands of personas, and run simulations where they grow and change over time.

Installation

Requires Python 3.11 or newer. On Python 3.9 or 3.10, pip install ensoul fails with no compatible version. Check your version with python --version and upgrade if needed.

pip install ensoul

Quick Start

A persona needs no world. This one stands alone, and talks to one of your own end users:

from ensoul import Ensoul

client = Ensoul(api_key="YOUR_API_KEY")

# A persona needs no world. This one stands alone.
persona = client.personas.create(name="Ada")

# user_id is your own id for the person talking.
# Each end user gets their own conversations with the persona.
reply = client.chat.send(
    persona.id,
    "Hi! Who are you?",
    user_id="YOUR_END_USER_ID",
)
print(reply.response)

# End the conversation when the person is done.
client.chat.end_conversation(
    persona.id,
    reply.conversation_id,
    user_id="YOUR_END_USER_ID",
)

Pass world="my_world" to create to place a persona in a world instead.

The API key can also be set via the ENSOUL_API_KEY environment variable, in which case no api_key argument is needed.

The client supports context managers for automatic cleanup:

with Ensoul(api_key="your-api-key") as client:
    persona = client.personas.get("persona-id")

Async Usage

import asyncio
from ensoul import AsyncEnsoul

async def main():
    async with AsyncEnsoul(api_key="your-api-key") as client:
        persona = await client.personas.create(name="Jordan", world="my_domain")
        response = await client.chat.send(persona.id, "Tell me about yourself.")
        print(response.response)

asyncio.run(main())

Streaming

Chat supports server-sent events (SSE) for streaming responses. Each event's data is a JSON object. The delta text lives in the chunk field:

Use parse_chat_event to turn each raw SSEEvent into a typed ChatStreamEvent, then print chunk as it arrives:

from ensoul import Ensoul
from ensoul.streaming import parse_chat_event

client = Ensoul(api_key="your-api-key")

stream = client.chat.stream("persona-id", "What do you think about music?")
try:
    for event in stream.events():
        parsed = parse_chat_event(event)
        print(parsed.chunk, end="", flush=True)  # print text as it streams
        if parsed.is_final:
            print()  # newline after the stream completes
            if parsed.token_usage:
                print(f"tokens: {parsed.token_usage}")
finally:
    stream.close()

The async client returns an AsyncSSEStream that works the same way with async for:

import asyncio
from ensoul import AsyncEnsoul
from ensoul.streaming import parse_chat_event

async def main() -> None:
    client = AsyncEnsoul(api_key="your-api-key")
    stream = await client.chat.stream("persona-id", "What do you think about music?")
    async for event in stream:
        parsed = parse_chat_event(event)
        print(parsed.chunk, end="", flush=True)
        if parsed.is_final:
            print()

asyncio.run(main())

Pagination

List endpoints return a SyncPage (or AsyncPage) object. Use .auto_paging_iter() to iterate through all pages automatically:

# Manual page access
page = client.personas.list(per_page=50)
print(page.items)    # current page items
print(page.total)    # total count across all pages

# Automatic pagination: fetches subsequent pages as needed
for persona in client.personas.list().auto_paging_iter():
    print(persona.name)

Async pagination works the same way with async for.

Error Handling

All SDK errors inherit from EnsoulError. HTTP errors are subclasses of APIError:

from ensoul.errors import (
    EnsoulError,
    APIError,
    AuthenticationError,
    AuthorizationError,
    NotFoundError,
    RateLimitError,
    ValidationError,
    EndUserRequiredError,
    PaymentRequiredError,
    PersonaCallsLimitError,
    ConflictError,
    TurnInFlightError,
    ConversationEndedError,
    EndUserForgottenError,
    ServerError,
)

try:
    persona = client.personas.get("nonexistent-id")
except NotFoundError:
    print("Persona not found")
except AuthenticationError:
    print("Invalid or missing API key")
except RateLimitError:
    print("Rate limit exceeded, back off and retry")
except EndUserRequiredError:
    print("A persona-bound key must name user_id")
except PersonaCallsLimitError as e:
    print(f"Persona {e.persona_id} is past its monthly call limit")
except ConversationEndedError:
    print("That conversation has ended; start a new one")
except APIError as e:
    print(f"API error {e.status_code}: {e.message}")

EndUserRequiredError, PersonaCallsLimitError, ConversationEndedError, and the other new 0.3.0 error types each subclass an existing class above (ValidationError, PaymentRequiredError, ConflictError), so a broader except written for 0.2.6 keeps catching them. See MIGRATION.md.

Configuration

The client reads two environment variables as defaults:

Demo API: the current hosted demo is available at:

export ENSOUL_BASE_URL="https://api.demo.ensoul-ai.com"
export ENSOUL_API_KEY="your-api-key"

With these set, Ensoul() connects to the demo with no constructor arguments.

You can also pass the base URL explicitly:

client = Ensoul(api_key="ens_...", base_url="https://api.demo.ensoul-ai.com")

Authentication

API key (recommended):

client = Ensoul(api_key="ens_...")
# or set ENSOUL_API_KEY in the environment
client = Ensoul()

Bearer token:

client = Ensoul(bearer_token="eyJ...")

OAuth2 token exchange (password flow, form-encoded):

token_response = client.auth.token(
    username="you@example.com",
    password="your-password",
)
authed_client = Ensoul(bearer_token=token_response.access_token)

Resources

Requirements

• Python 3.11+
• httpx: HTTP transport
• pydantic 2.x: request and response models