dooers-agents-server

Python agents server SDK for agents client SDK.

Install

pip install dooers-agents

How It Works

The SDK has two entry points for executing handlers:

WebSocket (real-time chat)          Everything else (REST, webhooks, cron)
─────────────────────────           ────────────────────────────────────────
Client connects via WS              Your code calls dispatch()
  ↓                                   ↓
agent_server.handle()              agent_server.dispatch()
  ↓                                   ↓
Context from WS frame               Context from your parameters
  ↓                                   ↓
  └──────────── handler(incoming, send, memory, analytics, settings) ──┘
                                   ↓
                          Same handler, same API

The handler is always the same. Whether the message came from a WebSocket or a REST endpoint, the handler receives the same parameters and yields the same events.

Quick Start — WebSocket

The primary entry point. A WebSocket connection sends messages, and the handler responds in real-time.

from fastapi import FastAPI, WebSocket
from openai import AsyncOpenAI
from dooers_agents import AgentServer, AgentConfig

app = FastAPI()
openai = AsyncOpenAI()

agent_server = AgentServer(AgentConfig(
    database_type="sqlite",
    database_name="agent.db",
    assistant_name="My Assistant",
))


async def agent_handler(incoming, send, memory, analytics, settings):
    yield send.run_start()

    history = await memory.get_history(limit=20, format="openai")

    completion = await openai.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "You are a helpful assistant."},
            *history,
        ],
    )

    yield send.text(completion.choices[0].message.content)
    yield send.update_thread(title=incoming.message[:60])
    yield send.run_end()


@app.websocket("/ws")
async def ws(websocket: WebSocket):
    await websocket.accept()
    await agent_server.handle(websocket, agent_handler)

Quick Start — Dispatch

For REST endpoints, webhooks, background jobs — anything outside a WebSocket connection. You provide the context explicitly and iterate over the handler's events.

from dooers_agents import User

@app.post("/api/webhook")
async def webhook(request: Request):
    body = await request.json()

    stream = await agent_server.dispatch(
        handler=agent_handler,       # Same handler as WebSocket
        agent_id="agent-1",
        organization_id="org-1",
        workspace_id="ws-1",
        message=body["text"],
        user=User(user_id=body["user_id"], user_name="Alice"),
        thread_id=body.get("thread_id"),   # None → creates new thread
        thread_title="Webhook conversation",
    )

    # Stream events from the handler
    async for event in stream:
        if event.send_type == "text":
            print(event.data["text"])

    return {"thread_id": stream.thread_id, "is_new": stream.is_new_thread}

Handler

Handlers are async generators that receive context and yield response events.

async def agent_handler(incoming, send, memory, analytics, settings):
    ...

incoming

The incoming message and its context.

Messages can contain text, images, and documents. incoming.message is a convenience shortcut that joins all text parts into a single string. incoming.content is the full list of content parts — use it when you need to handle images or documents.

incoming.message                     # str — all text parts joined with spaces
incoming.content                     # list[ContentPart] — full content parts

# ContentPart types:
# TextPart:     { type: "text", text: str }
# ImagePart:    { type: "image", url: str, mime_type?, alt? }
# DocumentPart: { type: "document", url: str, filename: str, mime_type: str }

incoming.context.thread_id           # str
incoming.context.event_id            # str
incoming.context.organization_id     # str
incoming.context.workspace_id        # str
incoming.context.thread_title        # str | None
incoming.context.thread_created_at   # datetime | None
incoming.context.queue               # str | None — current thread queue slug
incoming.context.queues              # list[WorkspaceQueueCatalog] — handshake roster
incoming.context.members_for_queue() # members of the current queue (or pass a slug)

# User (incoming.context.user)
incoming.context.user.user_id             # str
incoming.context.user.user_name           # str | None
incoming.context.user.user_email          # str | None
incoming.context.user.system_role         # str — "admin" | "user"
incoming.context.user.organization_role   # str — "owner" | "manager" | "member"
incoming.context.user.workspace_role      # str — "manager" | "member"

send

Yield events back to the client.

# Messages
yield send.text("Hello!")
yield send.text("Hello!", author="Support Bot")   # Override assistant_name
yield send.image(url, mime_type?, alt?, author?)
yield send.document(url, filename, mime_type, author?)

# Reasoning (collapsible "thinking" block; not fed back to the LLM history)
yield send.reasoning("Considering the user's constraints…")
yield send.reasoning(text, author?)

# Tool calls
yield send.tool_call(name, args, display_name?, id?)
yield send.tool_result(name, result, args?, display_name?, id?)

# Thread metadata
yield send.update_thread(title="New title")

# Run lifecycle
yield send.run_start(agent_id?)
yield send.run_end(status?, error?)

memory

Conversation history in multiple formats.

# LLM-formatted dicts (ready to pass to your model)
messages = await memory.get_history(limit=20, format="openai")
messages = await memory.get_history(format="anthropic")
messages = await memory.get_history(format="google")
messages = await memory.get_history(format="cohere")

# Chronological order
messages = await memory.get_history(limit=50, order="asc")

# Filter by fields
messages = await memory.get_history(filters={"actor": "user"})

# Raw ThreadEvent objects (for manual conversion)
events = await memory.get_history_raw(limit=50)
events = await memory.get_history_raw(limit=50, order="asc", filters={"type": "message"})

analytics

await analytics.track("event.name", data={"key": "value"})
await analytics.like("event", target_id, reason?)
await analytics.dislike("event", target_id, reason?)

settings

value = await settings.get("field_id")
all_values = await settings.get_all()
all_values = await settings.get_all(exclude=["avatar_base64"])
await settings.set("field_id", new_value)

Dispatch API

Full signature for programmatic handler execution.

from dooers_agents import User

stream = await agent_server.dispatch(
    handler=my_handler,
    agent_id="agent-1",
    organization_id="org-1",
    workspace_id="ws-1",
    message="Hello from webhook",
    user=User(                      # optional — defaults to User(user_id="")
        user_id="user-1",
        user_name="Alice",
        user_email="alice@example.com",
        system_role="user",         # "admin" | "user"
        organization_role="member", # "owner" | "manager" | "member"
        workspace_role="member",    # "manager" | "member"
    ),
    thread_id=None,                 # None → creates new thread
    thread_title="Webhook",         # title for new threads
    content=None,                   # optional list of ContentPart
)

# Available immediately (before iterating)
stream.thread_id       # str
stream.event_id        # str
stream.is_new_thread   # bool

# Iterate to run the handler
async for event in stream:
    if event.send_type == "text":
        print(event.data["text"])

# Or collect all events at once
events = await stream.collect()

Error Handling

from dooers_agents import DispatchError, HandlerError

try:
    stream = await agent_server.dispatch(...)
except DispatchError:
    # Setup failed (bad parameters, DB error)
    pass

try:
    async for event in stream:
        ...
except HandlerError as e:
    # Handler raised during execution
    # Pipeline already cleaned up (logged, run marked failed, error event created)
    print(e.original)  # The original exception

Repository

Direct database access for threads, events, runs, and settings.

repo = await agent_server.repository()

# Threads
threads = await repo.list_threads(filter={"agent_id": "w1", "organization_id": "org1"})
thread = await repo.get_thread(thread_id)
thread = await repo.create_thread(agent_id, organization_id, workspace_id, user_id, title?)
thread = await repo.update_thread(thread_id, title="New title")
await repo.remove_thread(thread_id)

# Events
events = await repo.list_events(filter={"thread_id": "t1"}, order={"direction": "asc"})
event = await repo.get_event(event_id)
event = await repo.create_event(thread_id, type="message", actor="user", content=[...])
await repo.remove_event(event_id)

# Runs
runs = await repo.list_runs(filter={"thread_id": "t1"})
run = await repo.get_run(run_id)

# Settings
values = await repo.get_settings(agent_id)
await repo.update_settings(agent_id, {"key": "value"})

Standalone Utilities

Access memory, settings, and analytics outside of handlers.

memory = await agent_server.memory(thread_id)
history = await memory.get_history(limit=20, format="openai")

settings = await agent_server.settings(agent_id)
value = await settings.get("model")

analytics = await agent_server.analytics(agent_id, thread_id="t1")
await analytics.track("custom.event")

Settings Schema

Full reference: docs/agent-settings.md (field parameters, internal vs public fields, frontend rendering, WebSocket frames).

Define configurable settings for your agent.

from dooers_agents import (
    AgentConfig,
    AgentServer,
    SettingsSchema,
    SettingsField,
    SettingsFieldGroup,
    SettingsFieldType,
    SettingsFieldVisibility,
    SettingsSelectOption,
)

schema = SettingsSchema(
    fields=[
        SettingsField(
            id="model",
            type=SettingsFieldType.SELECT,
            label="Model",
            value="gpt-4o-mini",
            options=[
                SettingsSelectOption(value="gpt-4o-mini", label="GPT-4o Mini"),
                SettingsSelectOption(value="gpt-4o", label="GPT-4o"),
            ],
        ),
        SettingsFieldGroup(
            id="advanced",
            label="Advanced Settings",
            collapsible="closed",
            visibility=SettingsFieldVisibility.USER,
            fields=[
                SettingsField(
                    id="temperature",
                    type=SettingsFieldType.NUMBER,
                    label="Temperature",
                    value=0.7,
                    min=0,
                    max=2,
                ),
            ],
        ),
        SettingsField(
            id="api_key",
            type=SettingsFieldType.PASSWORD,
            label="API Key",
            visibility=SettingsFieldVisibility.INTERNAL,  # handler-only; not sent over WebSocket
        ),
    ]
)

agent_server = AgentServer(AgentConfig(
    database_type="sqlite",
    database_name="agent.db",
    settings_schema=schema,
))

Field Types

• TEXT — Single-line text input
• NUMBER — Numeric input with optional min/max
• SELECT — Dropdown selection
• CHECKBOX — Boolean toggle
• TEXTAREA — Multi-line text input
• PASSWORD — Password input (hidden)
• EMAIL — Email input
• DATE — Date picker
• IMAGE — Display-only image (e.g., QR codes)

Field Groups

Groups organize related fields with an optional collapsible UI:

SettingsFieldGroup(
    id="group_id",
    label="Group Label",
    collapsible="open",    # "open" | "closed" | None (not collapsible)
    visibility=SettingsFieldVisibility.USER,
    fields=[...],
)

Field visibility (SettingsFieldVisibility)

• INTERNAL — Not sent over WebSocket; only via settings.get() / get_all() / set() in the handler.
• CREATOR — Studio / builder UI: subscribe with audience=creator (org admin/manager or agent owner). Patches require that subscription.
• USER — Default runtime config: subscribe with audience=user (default). WebSocket settings.subscribe payload supports audience and agent_owner_user_id (for creator access).

Internal Fields

Fields with visibility=INTERNAL are:

• Hidden from frontend settings snapshots
• Rejected if a WebSocket client attempts to patch them
• Suppressed from broadcast patch notifications
• Accessible normally via settings.get() and settings.get_all() in handlers

Database Configuration

Three database backends: PostgreSQL, SQLite, and Azure Cosmos DB.

PostgreSQL (default)

agent_server = AgentServer(AgentConfig(
    database_type="postgres",
    database_host="localhost",
    database_port=5432,
    database_user="postgres",
    database_name="mydb",
    database_password="secret",
    database_ssl=False,
    database_table_prefix="agent_",
    database_auto_migrate=True,
))

SQLite

agent_server = AgentServer(AgentConfig(
    database_type="sqlite",
    database_name="agent.db",
    database_table_prefix="agent_",
    database_auto_migrate=True,
))

Azure Cosmos DB

Requires the cosmos extra: pip install dooers-agents[cosmos]

agent_server = AgentServer(AgentConfig(
    database_type="cosmos",
    database_host="https://your-account.documents.azure.com:443/",
    database_name="your-database",
    database_key="your-cosmos-key",
    database_table_prefix="agent_",
    database_auto_migrate=True,
))

Environment Variables

Observability (dooers-service-otel)

Creator contract: install the extras — that is the only required step in agent code. AgentServer initializes OTEL and auto-instruments OpenAI / Anthropic / OpenAI Agents. No OpenTelemetry imports, no manual spans, no token accounting in the handler.

pip install "dooers-agents-server[observability]"
# or: uv add "dooers-agents-server[observability]"

Tracing uses Dooers platform defaults automatically (same idea as the baked analytics webhook URL). Creators do not need to set Core/otel URLs. Ensure AGENT_SEED_SECRET / core settings.seed so the runtime API key is available for export auth.

Each agent turn is exported as a trace with agent.id, thread.id, event.id, user.id, org.id, workspace.id, and agent.channel attributes. LLM calls are auto-instrumented as child spans (model + tokens) via openinference. If an instrumentor fails to load, the SDK logs a WARNING — turn requests may still appear without LLM metrics.

Each LLM span also says which kind of key paid for the call, whatever the provider (the key itself is never recorded):

Both come from the outgoing HTTP request (httpx, httpx2 or aiohttp), so OpenAI, Anthropic, Gemini (google-genai) and the Agents SDK's LiteLLM models (litellm/gemini/…, litellm/anthropic/…) are all covered. Not covered: calls with no LLM span (e.g. LiteLLM called directly, outside the Agents SDK) and clients on other transports (requests, gRPC, websockets).

The agent never talks to GCP directly. Instead, at the end of each turn it exchanges its dooers_runtime_api_key (persisted into service_secrets from the plaintext seed_secret / next_seed_secret on successful settings.seed from dooers-service-core) for a short-lived service token (POST /identity/service-token/agent, audience otel-service, scope otel:write), then POSTs that turn's spans over OTLP/HTTP to dooers-service-otel. Personal 1:1 chats omit workspaceId and mint an org-scoped write token.

Operators may override destinations with AGENT_CORE_BASE_URL / AGENT_OTEL_SERVICE_URL (e.g. staging).

User Roles and Thread Scoping

Users have three role levels that determine thread visibility:

from dooers_agents import User

user = User(
    user_id="user-1",
    user_name="Alice",
    user_email="alice@example.com",
    system_role="user",         # "admin" | "user"
    organization_role="member", # "owner" | "manager" | "member"
    workspace_role="member",    # "manager" | "member"
)

When listing threads, the SDK resolves the user's highest scope and attaches per-thread access (read / write / manage):

Elevated (non-participant) thread.subscribe requires access_reason once per UTC day (in-memory grant on the agent process). Use thread.participants.add or send.add_participant() to join and gain write.

Threads track participants automatically — each user who sends a message is added to the thread's users list. New threads also include the serving agent id.

Thread Privacy

Enable private_threads to restrict users to only their own threads:

agent_server = AgentServer(AgentConfig(
    database_type="postgres",
    database_name="mydb",
    private_threads=True,
))

When enabled:

• Thread listing filters by the connected user's user_id
• Each user only sees threads they created
• Useful for multi-tenant or personal assistant scenarios

Examples

See examples/ for complete working examples:

See Also

• WhatsApp integration — inbound/outbound, WhatsAppClient, templates, Meta 24h window
• dooers-agents-client — React SDK that connects to agents built with this server