Overview - xpander.ai

The Python SDK is in preview. It’s for building in-product agents — agents you write in Python and embed in your own product — and custom agents built on other frameworks (Agno, LangChain, OpenAI Agents, AWS Strands). To invoke or integrate xpander agents from an application, use the REST API, which is stable and works from any language.

The xpander.ai SDK is a typed Python client for the xpander.ai platform. It exposes six core modules (Backend, Agents, Tasks, ToolsRepository, KnowledgeBases, Events), plus a set of decorators (@on_task, @on_boot, @on_shutdown, @on_auth_event, @on_tool_*, @register_tool) for wiring your code into the agent runtime. Every method comes in two flavors: an async coroutine (e.g. aget, acreate_task) and a synchronous wrapper (get, create_task). Use the async forms in production; sync wrappers are convenient for scripts and notebooks.

Install

# Requires Python 3.9+
pip install xpander-sdk

For Agno (the recommended framework), install the optional extras:

pip install xpander-sdk[agno]

Authenticate

The SDK reads credentials from environment variables by default:

export XPANDER_API_KEY="your-api-key"
export XPANDER_ORGANIZATION_ID="your-org-id"
# Optional:
export XPANDER_BASE_URL="https://inbound.xpander.ai"   # cloud default
export XPANDER_AGENT_ID="agent-123"                    # avoids passing agent_id everywhere

Or instantiate Configuration explicitly and pass it into any module:

from xpander_sdk import Configuration, Agents

config = Configuration(
    api_key="your-api-key",
    organization_id="your-org-id",
)

agents = Agents(configuration=config)

See the Configuration reference for the full attribute list and self-hosted setup.

Initialize and call

import asyncio
from xpander_sdk import Agents

async def main():
    agents = Agents()                          # picks up env vars

    agent = await agents.aget("agent-123")     # load agent config
    task = await agent.acreate_task(
        prompt="Summarize the latest sales report",
        file_urls=["https://example.com/sales.csv"],
    )
    print(task.id, task.status)

asyncio.run(main())

For framework integration (Agno, OpenAI Agents, LangChain, Google ADK, AWS Strands), use Backend.aget_args() to resolve framework-specific kwargs from a stored agent configuration:

from xpander_sdk import Backend
from agno.agent import Agent as AgnoAgent

backend = Backend()
args = await backend.aget_args(agent_id="agent-123", task=task)

agno_agent = AgnoAgent(**args)
result = await agno_agent.arun(input="Hello!")

Async vs sync

Every coroutine has a sync sibling: same parameters, blocks until complete:

Async	Sync
`agents.aget(...)`	`agents.get(...)`
`agent.acreate_task(...)`	`agent.create_task(...)`
`task.aevents()`	`task.events()`
`tools.aload_tool_by_id(...)`	`tools.load_tool_by_id(...)`

Async methods are coroutines: wrap them in async def and call with await. Sync wrappers internally call run_sync(), so don’t call them from inside an existing event loop (FastAPI, asyncio scripts): use the a* form there.

Self-hosted

Point base_url at your Agent Controller endpoint and use the API key generated during your Helm install:

config = Configuration(
    api_key="agent-controller-api-key",
    organization_id="your-org-id",
    base_url="https://agent-controller.your-company.com",
)

The SDK auto-detects when the URL points at a self-hosted controller (agent-controller in the host or port 9016) and prefixes paths with the organization ID. See the Configuration reference for details.

Where to next

Backend

Resolve framework args, invoke agents directly, report external runs.

Agents

List, load, and interact with agents.

Tasks

Create, stream, and inspect task executions.

Decorators

@on_task, @on_boot, @on_shutdown, and tool hooks.

Configuration

Configuration is the credentials object every SDK module accepts. By default each module instantiates its own Configuration from environment variables, so most code never touches it directly. Construct one explicitly when you need to override credentials or point at a self-hosted controller.

from xpander_sdk import Configuration

config = Configuration(
    api_key="your-api-key",
    organization_id="your-org-id",
)

Constructor

Configuration(
    api_key: Optional[str] = None,           # XPANDER_API_KEY
    base_url: Optional[str] = None,          # XPANDER_BASE_URL
    organization_id: Optional[str] = None,   # XPANDER_ORGANIZATION_ID
    agent_id: Optional[str] = None,          # XPANDER_AGENT_ID
)

Attributes

Attribute	Type	Default	Description
`api_key`	`str \| None`	`XPANDER_API_KEY` env var	API key for authentication. Required for any cloud call.
`organization_id`	`str \| None`	`XPANDER_ORGANIZATION_ID` env var	Organization identifier. Required for self-hosted controllers; optional but recommended for cloud.
`base_url`	`str \| None`	Auto-detected from env	API endpoint. Defaults to `https://inbound.xpander.ai` for cloud. Set to your Agent Controller URL for self-hosted.
`agent_id`	`str \| None`	`None`	Optional default agent ID. When set, methods that take `agent_id` can be called without arguments. Falls back to `XPANDER_AGENT_ID`.

Environment variables

The SDK reads these on import. Setting them once via .env or your shell removes the need to construct Configuration manually.

Variable	Purpose
`XPANDER_API_KEY`	API key. Required.
`XPANDER_ORGANIZATION_ID`	Organization ID. Required for self-hosted; recommended for cloud.
`XPANDER_BASE_URL`	Override the API base URL.
`XPANDER_AGENT_ID`	Default agent ID. Lets `Backend()`, `Events()`, and `Agents().aget()` work without an explicit ID.

from dotenv import load_dotenv
from xpander_sdk import Configuration

load_dotenv()
config = Configuration()   # picks up XPANDER_* vars

Methods

`get_full_url() -> str`

Returns the API URL, with the organization ID appended when the controller requires it.

config = Configuration(
    base_url="https://agent-controller.acme.com",
    organization_id="org_123",
)
config.get_full_url()
# 'https://agent-controller.acme.com/org_123'

config = Configuration(
    base_url="https://inbound.xpander.ai",
    organization_id="org_123",
)
config.get_full_url()
# 'https://inbound.xpander.ai'   (cloud: no prefix)

The org ID is appended automatically when base_url contains agent-controller or port 9016. You don’t need to do this yourself.

Self-hosted

Use the Agent Controller API key generated during your Helm install (not your cloud API key) and point base_url at the controller:

config = Configuration(
    api_key="agent-controller-api-key",
    organization_id="your-org-id",
    base_url="https://agent-controller.your-company.com",
)

Or via environment:

export XPANDER_API_KEY="agent-controller-api-key"
export XPANDER_ORGANIZATION_ID="your-org-id"
export XPANDER_BASE_URL="https://agent-controller.your-company.com"

The base_url must point at the Agent Controller endpoint (e.g. https://agent-controller.{your-domain}), not the root domain. The SDK detects this URL pattern and prefixes paths with the organization ID via get_full_url().

Pass the same Configuration to every module so they hit the same endpoint and use the same credentials:

from xpander_sdk import Configuration, Agents, Tasks, KnowledgeBases

config = Configuration(api_key="...", organization_id="...")

agents = Agents(configuration=config)
tasks = Tasks(configuration=config)
kbs = KnowledgeBases(configuration=config)

Loaded objects (Agent, Task, KnowledgeBase) carry the Configuration they were loaded with, so subsequent method calls on them reuse it automatically.

Types

A consolidated reference for every enum and Pydantic model exposed at the top level of xpander_sdk. Methods that take or return these types link here.

Enums

`OutputFormat`

from xpander_sdk import OutputFormat

Value	String	Use
`OutputFormat.Text`	`"text"`	Plain text.
`OutputFormat.Markdown`	`"markdown"`	Markdown-formatted text (default for most agents).
`OutputFormat.Json`	`"json"`	Structured JSON. Pair with `output_schema` for validation.
`OutputFormat.Voice`	`"voice"`	MP3 voice output (requires `voice_id` on the agent).

`ThinkMode`

from xpander_sdk.models.shared import ThinkMode

Value	String	Use
`ThinkMode.Default`	`"default"`	Standard reasoning.
`ThinkMode.Harder`	`"harder"`	Extended reasoning (uses more tokens).

`AgentExecutionStatus`

from xpander_sdk import AgentExecutionStatus

Value	String
`Pending`	`"pending"`
`Executing`	`"executing"`
`Paused`	`"paused"`
`Error`	`"error"`
`Failed`	`"failed"`
`Completed`	`"completed"`
`Stopped`	`"stopped"`

See the task lifecycle.

`AgentDeploymentType`

from xpander_sdk import AgentDeploymentType

Value	String
`Serverless`	`"serverless"`
`Container`	`"container"`

`AgentStatus`

from xpander_sdk.modules.agents.models.agent import AgentStatus

Value	String
`DRAFT`	`"DRAFT"`
`ACTIVE`	`"ACTIVE"`
`INACTIVE`	`"INACTIVE"`

`AgentAccessScope`

from xpander_sdk.modules.agents.models.agent import AgentAccessScope

Value	String
`Personal`	`"personal"`
`Organizational`	`"organizational"`

`AgentType`

from xpander_sdk.modules.agents.models.agent import AgentType

Value	String
`Manager`	`"manager"`
`Regular`	`"regular"`
`A2A`	`"a2a"`
`Curl`	`"curl"`
`Orchestration`	`"orchestration"`

`LLMReasoningEffort`

from xpander_sdk.modules.agents.models.agent import LLMReasoningEffort

Value	String
`Low`	`"low"`
`Medium`	`"medium"`
`High`	`"high"`

`Framework`

from xpander_sdk.models.frameworks import Framework

Value	String
`Agno`	`"agno"`
`LangChain`	`"langchain"`
`OpenAIAgents`	`"open-ai-agents"`
`GoogleADK`	`"google-adk"`
`Strands`	`"strands-agents"`
`OpenClaw`	`"open-claw"`

Today only Agno is fully wired through Backend.aget_args; other values are reserved.

`MCPServerType`, `MCPServerAuthType`, `MCPServerTransport`

See MCP types.

`TaskUpdateEventType`

See streaming events.

`KnowledgeBaseType`

from xpander_sdk.modules.knowledge_bases.models.knowledge_bases import KnowledgeBaseType

Value	String
`MANAGED`	`"managed"`
`EXTERNAL`	`"external"`

Models

`Configuration`

See Configuration.

`User`

from xpander_sdk import User

User(
    id: Optional[str] = None,
    first_name: Optional[str] = None,
    last_name: Optional[str] = None,
    email: str,                                # required
    additional_attributes: Optional[dict] = None,
    timezone: Optional[str] = None,            # IANA, e.g. "America/New_York"
)

End-user context. Pass to acreate_task(user_details=...) to scope memory and connector authentication to a specific user.

`Tokens`

from xpander_sdk import Tokens

Tokens(
    completion_tokens: int = 0,
    prompt_tokens: int = 0,
)
# tokens.total_tokens is a computed property: completion + prompt

Used in Backend.areport_external_task and Task.areport_metrics.

`ExecutionTokens`

from xpander_sdk.models.shared import ExecutionTokens

ExecutionTokens(
    inner: Tokens = Tokens(),    # inner orchestration tokens
    worker: Tokens = Tokens(),   # worker LLM tokens
)

Used internally for metrics aggregation.

`AgentInstructions`

from xpander_sdk.modules.agents.models.agent import AgentInstructions

AgentInstructions(
    role: list[str] = [],
    goal: list[str] = [],
    general: str = "",
)

Computed properties: description (= general), instructions (formatted XML block), goal_str, full.

`AgentExecutionInput`

from xpander_sdk.modules.tasks.models.task import AgentExecutionInput

AgentExecutionInput(
    text: Optional[str] = "",
    files: Optional[list[str]] = [],
    user: Optional[User] = None,
)

Stored on task.input. Constructed automatically by acreate_task.

`LocalTaskTest`

from xpander_sdk.modules.tasks.models.task import LocalTaskTest, AgentExecutionInput

LocalTaskTest(
    input: AgentExecutionInput,
    agent_version: Optional[str] = None,
    output_format: Optional[OutputFormat] = None,
    output_schema: Optional[dict] = None,
)

Pass to @on_task(test_task=...) to invoke the handler once locally instead of subscribing to the platform.

`KnowledgeBaseSearchResult`

from xpander_sdk.modules.knowledge_bases.models.knowledge_bases import KnowledgeBaseSearchResult

KnowledgeBaseSearchResult(
    content: str,
    score: float,
)

Returned by KnowledgeBase.asearch.

`KnowledgeBaseDocumentItem`

See the KnowledgeBase reference.

`AgnoSettings`

from xpander_sdk.models.frameworks import AgnoSettings

AgnoSettings(
    session_storage: bool = True,
    learning: bool = False,
    user_memories: bool = False,
    agentic_memory: bool = False,
    agent_memories: bool = False,
    agentic_culture: bool = False,
    session_summaries: bool = False,
    num_history_runs: int = 10,
    max_tool_calls_from_history: int = 0,
    tool_call_limit: Optional[int] = None,
    coordinate_mode: bool = False,
    pii_detection_enabled: bool = False,
    pii_detection_mask: bool = True,
    prompt_injection_detection_enabled: bool = False,
    openai_moderation_enabled: bool = False,
    openai_moderation_categories: Optional[list[str]] = None,
    reasoning_tools_enabled: bool = False,
)

Per-agent Agno configuration stored on agent.agno_settings. Modify in the Workbench; mutating in code only affects the in-memory instance.

`AgentsListItem` / `TasksListItem`

Summary objects returned by agents.alist() and tasks.alist(). See agents.list and tasks.list.

Exceptions

`ModuleException`

from xpander_sdk.exceptions.module_exception import ModuleException

Raised by every SDK module on API failure. Carries status_code: int and description: str. See Error Handling.

Constants

`MAX_PLAN_RETRIES`

from xpander_sdk import MAX_PLAN_RETRIES   # = 5

The maximum number of times the @on_task runtime retries a handler when deep-planning tasks remain incomplete after the first run.

Type aliases

`LLMModelT`

from xpander_sdk.models.shared import LLMModelT
# = type[ABC]

Marker type used internally for typing model classes.

Error Handling

Every SDK module raises ModuleException when an API call fails. The exception bundles the HTTP status code with a description so you can branch on the underlying cause.

from xpander_sdk import Agents
from xpander_sdk.exceptions.module_exception import ModuleException

agents = Agents()
try:
    agent = await agents.aget("non-existent-agent")
except ModuleException as e:
    print(f"[{e.status_code}] {e.description}")

`ModuleException`

from xpander_sdk.exceptions.module_exception import ModuleException

Attribute	Type	Description
`status_code`	`int`	HTTP status code from the API response, or `500` for client-side failures.
`description`	`str`	Human-readable error description (often the raw response body).

The exception’s string form is [{status_code}] {description}.

Common status codes

Status	Cause	What to do
`400`	Malformed request: invalid parameters, missing required fields, schema validation.	Inspect `description`. Fix the call site.
`401`	Missing or invalid API key.	Verify `XPANDER_API_KEY` / `Configuration.api_key`. For self-hosted, confirm you’re using the Agent Controller key, not a cloud key.
`403`	Authenticated but not authorized: wrong organization, agent owned by another user.	Verify `XPANDER_ORGANIZATION_ID`. Check the agent’s `access_scope`.
`404`	Resource doesn’t exist: bad agent/task/KB id, or the resource was deleted.	Confirm IDs. Run `agents.list()` / `tasks.list()` to discover what exists.
`409`	Conflict: duplicate creation, version mismatch on save, or worker environment collision.	Reload the resource (`task.areload()`) and retry.
`429`	Rate limit.	Back off exponentially. The SDK does not auto-retry on 429.
`500`	Server error, or any client-side exception (network, parsing, validation).	Inspect `description`. For 500 specifically from the cloud, retry once or twice with backoff.

Branching on status

try:
    task = await tasks.aget(task_id="task_xyz")
except ModuleException as e:
    if e.status_code == 404:
        print("Task not found: was it deleted?")
    elif e.status_code in (401, 403):
        print("Auth issue: check credentials")
    else:
        raise

Retry strategy

The SDK does not retry failed requests automatically (except inside Events.start(), which retries SSE connections internally). For idempotent operations, wrap calls in your own retry loop:

import asyncio
from xpander_sdk.exceptions.module_exception import ModuleException

async def with_retry(coro_fn, *, attempts: int = 3, base_delay: float = 1.0):
    for attempt in range(1, attempts + 1):
        try:
            return await coro_fn()
        except ModuleException as e:
            if e.status_code in (429, 500, 502, 503, 504) and attempt < attempts:
                await asyncio.sleep(base_delay * (2 ** (attempt - 1)))
                continue
            raise

agent = await with_retry(lambda: agents.aget("agent-123"))

Avoid retrying:

400 / 404: the request will keep failing.
acreate / acreate_task without an existing_task_id: you may create duplicate resources. Pass existing_task_id=... to make creation idempotent.

Validation errors from `Tool.ainvoke`

Tool invocations validate the payload against the tool’s auto-generated Pydantic schema before sending the request. A schema mismatch raises ValueError (not ModuleException):

try:
    result = await tool.ainvoke(agent_id="agent-123", payload={"bad": "shape"})
except ValueError as e:
    print(f"Payload didn't match schema: {e}")

Successful invocations still set is_error=True on the returned ToolInvocationResult if the remote endpoint returned an HTTP error. Check the result, not just exceptions:

result = await tool.ainvoke(agent_id="agent-123", payload={"city": "NYC"})
if result.is_error:
    print(f"Tool failed [{result.status_code}]: {result.result}")

Streaming errors

task.aevents() raises ValueError if the task wasn’t created with events_streaming=True. The underlying SSE connection logs warnings on parse failures and drops malformed events; it does not raise. Network failures propagate after the SDK’s internal retry budget is exhausted (see Events).

​Install

​Authenticate

​Initialize and call

​Async vs sync

​Self-hosted

​Where to next

Backend

Agents

Tasks

Decorators

​Configuration

​Constructor

​Attributes

​Environment variables

​Methods

​get_full_url() -> str

​Self-hosted

​Sharing config across modules

​Types

​Enums

​OutputFormat

​ThinkMode

​AgentExecutionStatus

​AgentDeploymentType

​AgentStatus

​AgentAccessScope

​AgentType

​LLMReasoningEffort

​Framework

​MCPServerType, MCPServerAuthType, MCPServerTransport

​TaskUpdateEventType

​KnowledgeBaseType

​Models

​Configuration

​User

​Tokens

​ExecutionTokens

​AgentInstructions

​AgentExecutionInput

​LocalTaskTest

​TaskUpdateEvent

​ToolInvocationResult

​MCPServerDetails

​KnowledgeBaseSearchResult

​KnowledgeBaseDocumentItem

​AgnoSettings

​AgentsListItem / TasksListItem

​Exceptions

​ModuleException

​Constants

​MAX_PLAN_RETRIES

​Type aliases

​LLMModelT

​Error Handling

​ModuleException

​Common status codes

​Branching on status

​Retry strategy

​Validation errors from Tool.ainvoke

​Streaming errors

Install

Authenticate

Initialize and call

Async vs sync

Self-hosted

Where to next

Configuration

Constructor

Attributes

Environment variables

Methods

`get_full_url() -> str`

Self-hosted

Sharing config across modules

Types

Enums

`OutputFormat`

`ThinkMode`

`AgentExecutionStatus`

`AgentDeploymentType`

`AgentStatus`

`AgentAccessScope`

`AgentType`

`LLMReasoningEffort`

`Framework`

`MCPServerType`, `MCPServerAuthType`, `MCPServerTransport`

`TaskUpdateEventType`

`KnowledgeBaseType`

Models

`Configuration`

`User`

`Tokens`

`ExecutionTokens`

`AgentInstructions`

`AgentExecutionInput`

`LocalTaskTest`

`TaskUpdateEvent`

`ToolInvocationResult`

`MCPServerDetails`

`KnowledgeBaseSearchResult`

`KnowledgeBaseDocumentItem`

`AgnoSettings`

`AgentsListItem` / `TasksListItem`

Exceptions

`ModuleException`

Constants

`MAX_PLAN_RETRIES`

Type aliases

`LLMModelT`

Error Handling

`ModuleException`

Common status codes

Branching on status

Retry strategy

Validation errors from `Tool.ainvoke`

Streaming errors