> ## Documentation Index
> Fetch the complete documentation index at: https://docs.xpander.ai/llms.txt
> Use this file to discover all available pages before exploring further.
# Agents
> List, load, and interact with agents stored in the xpander.ai platform.
The `Agents` module is your gateway to agents stored on the platform. Use it to discover what agents exist, load a specific one, and from there create tasks, invoke tools, attach knowledge bases, and inspect sessions.
```python theme={"dark"}
from xpander_sdk import Agents
agents = Agents()
all_agents = await agents.alist()
agent = await agents.aget("agent-123")
task = await agent.acreate_task(prompt="Hello!")
```
## Constructor
```python theme={"dark"}
Agents(configuration: Optional[Configuration] = None)
```
| Parameter | Type | Default | Description |
| --------------- | --------------- | ------- | ------------------------------------------ |
| `configuration` | `Configuration` | `None` | SDK configuration. Falls back to env vars. |
## Module methods
| Method | Returns | What it does |
| --------------------------------------------------------- | ---------------------- | ------------------------------------------------ |
| [`alist` / `list`](/developers/sdk-reference/agents#list) | `list[AgentsListItem]` | Summary list of every agent your org can access. |
| [`aget` / `get`](/developers/sdk-reference/agents#get) | `Agent` | Load one agent fully (config + graph + tools). |
## `Agent` instance methods
Once you've loaded an `Agent`, use its instance methods for the actual work:
| Method | What it does |
| -------------------------------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------- |
| [`acreate_task` / `create_task`](/developers/sdk-reference/agents#create_task) | Create a new task and run it. |
| [`ainvoke_tool` / `invoke_tool`](/developers/sdk-reference/agents#invoke_tool) | Invoke a single tool bound to this agent. |
| [`aget_knowledge_bases` / `get_knowledge_bases`](/developers/sdk-reference/agents#knowledge-bases) | Load all knowledge bases attached to the agent. |
| [`attach_knowledge_base`](/developers/sdk-reference/agents#attach_knowledge_base) | Link a knowledge base to the agent (in-memory only). |
| [`knowledge_bases_retriever`](/developers/sdk-reference/agents#knowledge_bases_retriever) | Returns a `search(query, num_documents)` callable for use as a framework retriever. |
| [`aget_db` / `get_db`](/developers/sdk-reference/agents#sessions) | Returns the Agno PG client backing the agent's session storage. |
| [`aget_user_sessions` / `get_user_sessions`](/developers/sdk-reference/agents#sessions) | All sessions for a given end-user. |
| [`aget_session` / `get_session`](/developers/sdk-reference/agents#sessions) | Load a single session. |
| [`adelete_session` / `delete_session`](/developers/sdk-reference/agents#sessions) | Delete a session. |
| [`aget_streaming_spec` / `get_streaming_spec`](/developers/sdk-reference/agents#streaming-spec) | Get the deployed agent's streaming URL + auth key. |
| [`aget_connection_string` / `get_connection_string`](/developers/sdk-reference/agents#connection-string) | Get the agent's DB connection details. |
See the [`Agent` class reference](/developers/sdk-reference/agents#agent-class) for attribute documentation.
## Quick patterns
### Find and load
```python theme={"dark"}
all_agents = await agents.alist()
for item in all_agents:
print(item.id, item.name, item.status)
agent = await all_agents[0].aload() # AgentsListItem.aload returns the full Agent
# equivalent to: await agents.aget(all_agents[0].id)
```
### Pin to a specific version
```python theme={"dark"}
agent = await agents.aget("agent-123", version=4)
```
Versioning is opt-in. Without `version`, you get the latest.
### Default agent via env var
```bash theme={"dark"}
export XPANDER_AGENT_ID="agent-123"
```
```python theme={"dark"}
agent = await agents.aget() # uses XPANDER_AGENT_ID
```
This works because `Agents.aget()` falls back to `Configuration.agent_id` and then to `XPANDER_AGENT_ID`.
## list
`Agents.alist` returns a list of summary objects (`AgentsListItem`) for every agent visible to the configured organization. Each item carries enough metadata for display (id, name, icon, status, instructions, access scope) and a `.aload()` shortcut to fetch the full `Agent`.
```python theme={"dark"}
from xpander_sdk import Agents
agents = Agents()
items = await agents.alist()
for item in items:
print(f"{item.id}\t{item.status.value}\t{item.name}")
```
#### Parameters
None.
#### Returns `list[AgentsListItem]`
`AgentsListItem` is a lightweight summary: not the full `Agent`. Notable fields:
| Field | Type | Description |
| ----------------- | ------------------- | -------------------------------------------------------- |
| `id` | `str` | Agent ID. |
| `name` | `str` | Display name. |
| `icon` | `str` | Emoji or icon identifier. |
| `instructions` | `AgentInstructions` | `role`, `goal`, `general` text. |
| `status` | `AgentStatus` | `DRAFT`, `ACTIVE`, or `INACTIVE`. |
| `organization_id` | `str` | Owning organization. |
| `created_at` | `datetime` | Creation timestamp. |
| `description` | `str \| None` | Agent description. |
| `access_scope` | `AgentAccessScope` | `Personal` or `Organizational`. |
| `created_by` | `str \| None` | Creator user id. |
| `type` | `AgentType \| None` | `Manager`, `Regular`, `A2A`, `Curl`, or `Orchestration`. |
To load the full agent (graph, tools, model config, …) call `.aload()` on the item, or pass `item.id` to `agents.aget()`.
### Examples
#### Filter active agents
```python theme={"dark"}
items = await agents.alist()
active = [i for i in items if i.status.value == "ACTIVE"]
```
#### Load all in parallel
```python theme={"dark"}
import asyncio
items = await agents.alist()
full = await asyncio.gather(*(item.aload() for item in items))
```
For a large org this can be heavy: prefer loading only the agents you'll actually use.
#### Sync version
```python theme={"dark"}
items = agents.list()
```
Same return type; blocks until the list is fetched.
### Errors
`alist` raises [`ModuleException`](/developers/sdk-reference/overview#error-handling) on failure:
| Status | Cause |
| ------ | -------------------------------- |
| 401 | Missing or invalid `api_key`. |
| 403 | Wrong `organization_id`. |
| 500 | Server error or network failure. |
## get
`Agents.aget` loads one agent's complete configuration: instructions, model settings, the execution graph, all attached tools, knowledge-base links, and framework settings. Use this whenever you need to actually invoke an agent or inspect its config in code.
```python theme={"dark"}
from xpander_sdk import Agents
agents = Agents()
agent = await agents.aget("agent-123")
print(agent.name, agent.model_provider, agent.model_name)
```
#### Parameters
| Parameter | Type | Required | Default | Description |
| ---------- | ----- | -------- | --------------------------------------------- | ------------------------------------ |
| `agent_id` | `str` | No¹ | `Configuration.agent_id` → `XPANDER_AGENT_ID` | Agent to load. |
| `version` | `int` | No | `None` | Specific version. Latest if omitted. |
¹ The method falls back to `Configuration.agent_id` and then `XPANDER_AGENT_ID`. If none of those are set, you'll hit a 404 from the cloud.
#### Returns `Agent`
A full `Agent` instance. See [`Agent` class reference](/developers/sdk-reference/agents#agent-class) for attributes (instructions, framework, graph, deployment\_type, tools, knowledge\_bases, agno\_settings, etc.).
### Examples
#### Latest version
```python theme={"dark"}
agent = await agents.aget("agent-123")
```
#### Pinned version
```python theme={"dark"}
agent = await agents.aget("agent-123", version=4)
print(agent.version) # 4
```
The version corresponds to deployment snapshots in the Workbench. Pin a version in production to avoid surprises when someone edits the agent.
#### Use a default agent ID
```bash theme={"dark"}
export XPANDER_AGENT_ID="agent-123"
```
```python theme={"dark"}
agent = await agents.aget() # uses XPANDER_AGENT_ID
```
This is the pattern used inside `@on_task` handlers: you don't usually pass an agent ID explicitly because the runtime injects it.
#### Sync version
```python theme={"dark"}
agent = agents.get("agent-123")
```
Same parameters; blocks until the agent is loaded.
### What gets loaded
When you call `aget`, the SDK:
1. Fetches the agent record (`GET /agents/:id`).
2. Builds the `AgentGraph` from the graph items returned in the response.
3. Initializes a `ToolsRepository` populated with the agent's tools.
4. If any local tools (Python `@register_tool` functions) need syncing to the platform's graph, kicks off a background `sync_local_tools` task: non-blocking.
You can read `agent.graph` to inspect the tool/connector wiring, `agent.tools.list` for all tools, `agent.knowledge_bases` for KB links, and `agent.mcp_servers` for the configured MCP servers.
### Errors
`aget` raises [`ModuleException`](/developers/sdk-reference/overview#error-handling) on failure:
| Status | Cause |
| ------ | -------------------------------------------------- |
| 404 | Agent not found, or wrong organization. |
| 403 | Agent exists but isn't accessible to your account. |
| 500 | Server error or network failure. |
## create\_task
`Agent.acreate_task` creates a new task on the platform and returns a `Task` object you can save, stream, or stop. This is the primary way to invoke an agent from code.
```python theme={"dark"}
from xpander_sdk import Agents
agent = await Agents().aget("agent-123")
task = await agent.acreate_task(prompt="Summarize this document.")
print(task.id, task.status.value)
```
The newly-created task is queued by the platform; the agent worker (either a managed cloud worker or your local `@on_task` handler) picks it up and executes it.
#### Parameters
| Parameter | Type | Required | Default | Description |
| ------------------------------ | ------------------------ | -------- | --------- | ----------------------------------------------------------------------- |
| `prompt` | `str` | No | `""` | Natural-language input. |
| `existing_task_id` | `str` | No | `None` | Continue an existing task. Reuses memory thread. |
| `file_urls` | `list[str]` | No | `[]` | URLs of files to attach (PDFs, images, CSVs, …). |
| `user_details` | `User` | No | `None` | End-user context (id, email, timezone) for memory and connector auth. |
| `agent_version` | `str` | No | `None` | Pin to a specific agent version. |
| `tool_call_payload_extension` | `dict` | No | `None` | Extra fields merged into every tool-call payload. |
| `source` | `str` | No | `"sdk"` | Origin tag (`"webhook"`, `"slack"`, …). |
| `worker_id` | `str` | No | `None` | Pin execution to a specific worker. |
| `run_locally` | `bool` | No | `False` | Mark the task as locally executed (skips cloud dispatch). |
| `output_format` | `OutputFormat` | No | `None` | `Text`, `Markdown`, `Json`, or `Voice`. |
| `output_schema` | `dict` | No | `None` | JSON schema (paired with `OutputFormat.Json`). |
| `events_streaming` | `bool` | No | `False` | Enables SSE streaming via `task.aevents()`. |
| `additional_context` | `str` | No | `None` | Extra context appended to the system prompt. |
| `instructions_override` | `str` | No | `None` | Extra instructions appended for this run only. |
| `test_run_node_id` | `str` | No | `None` | (Internal) Workflow node to execute in test mode. |
| `user_oidc_token` | `str` | No | `None` | OIDC token for connector pre-auth. |
| `expected_output` | `str` | No | `None` | Description of expected output shape. |
| `mcp_servers` | `list[MCPServerDetails]` | No | `[]` | Per-task MCP servers (in addition to the agent's configured servers). |
| `triggering_agent_id` | `str` | No | `None` | ID of the agent that triggered this task (for sub-agent tracking). |
| `title` | `str` | No | `None` | Display title in the dashboard. |
| `think_mode` | `ThinkMode` | No | `Default` | `Default` or `Harder`. Toggles extended reasoning. |
| `disable_attachment_injection` | `bool` | No | `False` | Skip auto-injecting human-readable file content into the prompt. |
| `return_metrics` | `bool` | No | `False` | Return metrics in the task. Only valid for Workflow → Agent invocation. |
| `user_tokens` | `dict` | No | `None` | Pre-computed user tokens injected for MCP auth. |
#### Returns `Task`
The created `Task`. See [`Task` class reference](/developers/sdk-reference/tasks#task-class).
The task starts in status `Pending`. It moves through `Executing` → `Completed` (or `Error` / `Failed` / `Stopped`) as the worker runs it.
### Examples
#### With files
```python theme={"dark"}
task = await agent.acreate_task(
prompt="Find action items in this meeting.",
file_urls=[
"https://example.com/recording.mp3",
"https://example.com/transcript.pdf",
],
)
```
#### Structured output
```python theme={"dark"}
from xpander_sdk import OutputFormat
task = await agent.acreate_task(
prompt="Extract company name and revenue from the press release.",
file_urls=["https://example.com/press-release.pdf"],
output_format=OutputFormat.Json,
output_schema={
"type": "object",
"properties": {
"company": {"type": "string"},
"revenue_usd": {"type": "number"},
},
"required": ["company"],
},
)
```
#### Streaming
```python theme={"dark"}
task = await agent.acreate_task(
prompt="Long-running analysis…",
events_streaming=True,
)
async for event in task.aevents():
print(event.type, event.data)
```
`events_streaming=True` is required to use `task.aevents()`. See [`task.events`](/developers/sdk-reference/tasks#events).
#### Continue a thread
```python theme={"dark"}
task = await agent.acreate_task(
existing_task_id="task_abc123",
prompt="What was the conclusion?",
)
```
The new task reuses the prior task's memory thread, so the agent has full context.
#### Per-end-user
```python theme={"dark"}
from xpander_sdk import User
task = await agent.acreate_task(
prompt="Summarize my unread email.",
user_details=User(
id="user_42",
email="alex@example.com",
first_name="Alex",
timezone="America/New_York",
),
)
```
`user_details.id` scopes user-memory and connector auth (e.g. each end user's Gmail OAuth token).
#### Add an MCP server just for this task
```python theme={"dark"}
from xpander_sdk import MCPServerDetails, MCPServerType, MCPServerTransport
task = await agent.acreate_task(
prompt="Query our internal docs.",
mcp_servers=[
MCPServerDetails(
type=MCPServerType.Remote,
name="internal-docs",
url="https://mcp.internal.acme.com",
transport=MCPServerTransport.HTTP_Transport,
),
],
)
```
Per-task MCP servers stack on top of the agent's persistent MCP configuration.
#### Pin a version
```python theme={"dark"}
task = await agent.acreate_task(
prompt="Run with the locked v3 prompt",
agent_version="3",
)
```
### Sync version
```python theme={"dark"}
task = agent.create_task(prompt="Hello!")
```
Same parameters; blocks until the task is created. Don't call from inside an event loop.
### Errors
Raises [`ModuleException`](/developers/sdk-reference/overview#error-handling) on failure:
| Status | Cause |
| --------- | ----------------------------------------------------------------- |
| 400 | Invalid input (bad output schema, malformed `mcp_servers`, etc.). |
| 401 / 403 | Auth failure. |
| 404 | Agent doesn't exist. |
| 500 | Server error. |
## invoke\_tool
`Agent.ainvoke_tool` runs a single tool from the agent's `ToolsRepository` and returns a `ToolInvocationResult`. Use it when you want to test a tool, or when you're building an agent loop yourself and the framework's tool dispatcher isn't a good fit.
```python theme={"dark"}
from xpander_sdk import Agents
agent = await Agents().aget("agent-123")
weather_tool = agent.tools.get_tool_by_id("get_weather")
result = await agent.ainvoke_tool(
tool=weather_tool,
payload={"city": "New York"},
)
print(result.is_success, result.result)
```
#### Parameters
| Parameter | Type | Required | Default | Description |
| ------------------- | ------ | -------- | ------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `tool` | `Tool` | Yes | – | A `Tool` instance, typically from `agent.tools.get_tool_by_id(...)` or `agent.tools.list`. |
| `payload` | `Any` | Yes | – | Tool input. Validated against `tool.schema` before execution. Local tools registered with `@register_tool` take their arguments flat, as in the example above. Platform connector tools take the grouped form instead — `{"body_params": {...}, "query_params": {...}, "path_params": {...}, "headers": {...}}` — matching [Invoke Connector Operation](/api-reference/v1/tools/invoke-connector-operation). Check `tool.schema` when unsure. |
| `payload_extension` | `dict` | No | `{}` | Extra fields deep-merged into the payload (e.g. headers, auth params). |
| `task_id` | `str` | No | `None` | Associate the invocation with a specific task (for activity logging). |
| `tool_call_id` | `str` | No | `None` | Correlation ID for this call (matches LLM tool-call IDs). |
#### Returns `ToolInvocationResult`
| Field | Type | Description |
| -------------- | ------------- | ------------------------------------------------------------------ |
| `tool_id` | `str` | The tool's id. |
| `tool_call_id` | `str \| None` | The correlation ID, if passed. |
| `task_id` | `str \| None` | The task ID, if passed. |
| `payload` | `Any` | The payload that was sent. |
| `result` | `Any` | The tool's response. Shape depends on the tool. |
| `is_success` | `bool` | `True` when the call returned 2xx. |
| `is_error` | `bool` | `True` when the call raised or returned an error status. |
| `status_code` | `int` | HTTP status from the remote tool, or `500` for client-side errors. |
| `is_local` | `bool` | `True` for tools registered via `@register_tool`. |
`is_success` and `is_error` are mutually exclusive in normal operation; check `is_error` first.
### Examples
#### With `payload_extension`
```python theme={"dark"}
result = await agent.ainvoke_tool(
tool=weather_tool,
payload={"city": "Tokyo"},
payload_extension={"headers": {"X-Trace-ID": "abc-123"}},
)
```
`payload_extension` is deep-merged with `payload`, so nested objects compose naturally.
#### Local Python tools
Tools registered with `@register_tool` are invoked locally (the `fn` runs in-process) and pass through the same `ToolInvocationResult` interface:
```python theme={"dark"}
from xpander_sdk import register_tool
@register_tool
def add(a: int, b: int) -> int:
"""Add two numbers."""
return a + b
# 'add' is now in the global tool registry
agent = await Agents().aget("agent-123")
add_tool = agent.tools.get_tool_by_id("add")
result = await agent.ainvoke_tool(tool=add_tool, payload={"a": 2, "b": 3})
result.result # 5
result.is_local # True
```
See [`@register_tool`](/developers/sdk-reference/tools#register_tool-classmethod) for details.
#### Inside an `@on_task` handler
```python theme={"dark"}
from xpander_sdk import on_task
from xpander_sdk.modules.tasks.sub_modules.task import Task
@on_task
async def handler(task: Task) -> Task:
agent = await Agents().aget(task.agent_id)
tool = agent.tools.get_tool_by_id("my_tool")
result = await agent.ainvoke_tool(
tool=tool,
payload={"x": 1},
task_id=task.id, # links the invocation to this task in the dashboard
)
task.result = str(result.result)
return task
```
Passing `task_id` ensures the invocation appears in `task.aget_activity_log()`.
#### Sync version
```python theme={"dark"}
result = agent.invoke_tool(tool=weather_tool, payload={"city": "Berlin"})
```
### Notes
* Payload validation: if `payload` doesn't match `tool.schema`, the SDK raises `ValueError` (not `ModuleException`). Catch it separately if you're invoking with untrusted input.
* Lifecycle hooks ([`@on_tool_before` / `@on_tool_after` / `@on_tool_error`](/developers/sdk-reference/decorators#@on_tool_before-/-@on_tool_after-/-@on_tool_error)) fire around the invocation.
* For invocations outside an agent context, use `tool.get_invocation_function()` instead: it requires a `connector_id` and `operation_id` resolved via `tools.aload_tool_by_id(...)`.
## Knowledge bases
Agents can have knowledge bases linked to them in the Workbench. From code, you can attach more, load the `KnowledgeBase` objects, or build a retriever callable for use as a framework retriever.
```python theme={"dark"}
agent = await agents.aget("agent-123")
# Inspect linked KBs
for kb_link in agent.knowledge_bases:
print(kb_link.id)
# Load full KnowledgeBase objects
kbs = await agent.aget_knowledge_bases()
for kb in kbs:
print(kb.name, kb.total_documents)
```
### `aget_knowledge_bases`
Load every `KnowledgeBase` linked to the agent, in parallel.
```python theme={"dark"}
kbs = await agent.aget_knowledge_bases()
```
#### Returns `list[KnowledgeBase]`
A list of full `KnowledgeBase` objects. See [`KnowledgeBase`](/developers/sdk-reference/knowledge-bases#knowledgebase-class) for the methods you can call on each.
#### Sync version
```python theme={"dark"}
kbs = agent.get_knowledge_bases()
```
### `attach_knowledge_base`
Link a knowledge base to the agent on the in-memory instance. Pass either a `KnowledgeBase` object or a knowledge-base ID.
```python theme={"dark"}
from xpander_sdk import KnowledgeBases
kb = await KnowledgeBases().aget("kb-456")
agent.attach_knowledge_base(knowledge_base=kb)
# or by id only
agent.attach_knowledge_base(knowledge_base_id="kb-456")
```
#### Parameters
| Parameter | Type | Required | Description |
| ------------------- | --------------- | -------- | --------------------------- |
| `knowledge_base` | `KnowledgeBase` | No | A `KnowledgeBase` instance. |
| `knowledge_base_id` | `str` | No | Knowledge-base ID. |
You must pass at least one of the two. If both are passed, `knowledge_base.id` is used.
`attach_knowledge_base` only updates the agent in memory. To persist the link, save the agent through the platform (e.g. via the API or the Workbench UI). The runtime instance will use the link for the current session, but it won't survive a re-load.
### `knowledge_bases_retriever`
Returns a `search(query, agent=None, num_documents=5)` callable that searches every linked KB and returns top-N results sorted by score. This is the form Agno expects as a custom retriever.
```python theme={"dark"}
retriever = agent.knowledge_bases_retriever()
results = retriever(query="quarterly revenue", num_documents=10)
for r in results:
print(r["score"], r["content"][:100])
```
#### Returned callable signature
```python theme={"dark"}
def search(
query: str,
agent: Optional[Any] = None, # ignored; for compat with framework retrievers
num_documents: int = 5,
**kwargs,
) -> list[dict] | None
```
| Parameter | Type | Default | Description |
| --------------- | ----- | ------- | ----------------------------------------------------------------------- |
| `query` | `str` | – | Search query. |
| `agent` | `Any` | `None` | Ignored. Present for compatibility with framework retriever signatures. |
| `num_documents` | `int` | `5` | Top-K to return. If `0` is passed, defaults to `10`. |
Each result is a dict from `KnowledgeBaseSearchResult.model_dump()`:
```python theme={"dark"}
{"content": "...", "score": 0.86}
```
The retriever swallows errors and returns `[]` if the search fails: useful for embedding into framework pipelines that shouldn't crash on retrieval errors.
### Patterns
#### Wire as the Agno knowledge
```python theme={"dark"}
from xpander_sdk import Backend
from agno.agent import Agent as AgnoAgent
agent = await agents.aget("agent-123")
backend = Backend()
args = await backend.aget_args(agent=agent)
# args["knowledge"] is already populated when the agent has KBs linked.
agno_agent = AgnoAgent(**args)
```
`Backend.aget_args` already wires the retriever for you: you don't usually need to call `knowledge_bases_retriever()` directly. Use it only when bypassing the Backend dispatcher.
#### Add a KB on the fly
```python theme={"dark"}
from xpander_sdk import KnowledgeBases
kbs = KnowledgeBases()
new_kb = await kbs.acreate(name="Q4 reports", description="Quarterly earnings")
await new_kb.aadd_documents([
"https://example.com/Q4-2024.pdf",
"https://example.com/Q4-2025.pdf",
])
agent.attach_knowledge_base(knowledge_base=new_kb)
```
After `attach_knowledge_base`, `agent.aget_knowledge_bases()` includes the new KB in the list.
## Sessions
Agents using the Agno framework with `agno_settings.session_storage = True` persist their session history to a Postgres database managed by the platform. The SDK exposes that database via `Agent.aget_db()` plus a small set of helpers for session CRUD.
Sessions are only available for Agno agents with session storage enabled. Calling these methods on other agents raises `NotImplementedError` (wrong framework) or `LookupError` (storage disabled). The Agno extras must be installed: `pip install xpander-sdk[agno]`.
### `aget_db`
Returns the Agno Postgres client backing the agent's session storage.
```python theme={"dark"}
db = await agent.aget_db() # async client (AsyncPostgresDb)
db_sync = await agent.aget_db(async_db=False) # sync client (PostgresDb)
```
#### Parameters
| Parameter | Type | Default | Description |
| ---------- | ------ | ------- | ---------------------------------------------------------- |
| `async_db` | `bool` | `True` | When `False`, returns the synchronous `PostgresDb` client. |
#### Returns
`agno.db.postgres.AsyncPostgresDb` (or `PostgresDb` if `async_db=False`). The client is namespaced to the agent's schema, so different agents don't share a session table.
The connection URI is fetched (and cached) via `agent.aget_connection_string()` on the first call.
#### Sync version
```python theme={"dark"}
db = agent.get_db() # equivalent to aget_db(async_db=False)
```
### `aget_user_sessions`
Load all sessions for a given end-user.
```python theme={"dark"}
sessions = await agent.aget_user_sessions(user_id="user_42")
for s in sessions:
print(s.session_id, s.created_at)
```
#### Parameters
| Parameter | Type | Required | Description |
| --------- | ----- | -------- | -------------------------------------------------------------------- |
| `user_id` | `str` | Yes | End-user identifier (matches `user_details.id` from `acreate_task`). |
#### Returns
A list of session records. The exact type comes from Agno's `db.get_sessions(...)`: fields include `session_id`, `user_id`, `agent_id` (or `team_id` for Team mode), and timestamps. The query caps results at 50 sessions per call.
For Team agents (`agent.is_a_team == True`), this returns `SessionType.TEAM` records; otherwise `SessionType.AGENT`.
#### Sync version
```python theme={"dark"}
sessions = agent.get_user_sessions(user_id="user_42")
```
### `aget_session`
Load a single session by ID.
```python theme={"dark"}
session = await agent.aget_session(session_id="sess_456")
```
#### Parameters
| Parameter | Type | Required | Description |
| ------------ | ----- | -------- | ------------------- |
| `session_id` | `str` | Yes | Session identifier. |
#### Returns
The session record, or `None` if it doesn't exist.
#### Sync version
```python theme={"dark"}
session = agent.get_session(session_id="sess_456")
```
### `adelete_session`
Delete a session.
```python theme={"dark"}
await agent.adelete_session(session_id="sess_456")
```
#### Parameters
| Parameter | Type | Required | Description |
| ------------ | ----- | -------- | ------------------ |
| `session_id` | `str` | Yes | Session to delete. |
#### Sync version
```python theme={"dark"}
agent.delete_session(session_id="sess_456")
```
### Errors
All session methods raise:
* `NotImplementedError`: agent isn't using Agno.
* `LookupError`: Agno is configured but `agno_settings.session_storage` is `False`.
* `ImportError`: Agno extras not installed (run `pip install xpander-sdk[agno]`).
* `ValueError`: connection URI couldn't be resolved.
### Patterns
#### Inspect recent sessions for a user
```python theme={"dark"}
sessions = await agent.aget_user_sessions(user_id="user_42")
for s in sorted(sessions, key=lambda x: x.created_at, reverse=True)[:5]:
print(f"{s.session_id}: {s.created_at}")
```
#### Wipe all sessions for a user
```python theme={"dark"}
sessions = await agent.aget_user_sessions(user_id="user_42")
for s in sessions:
await agent.adelete_session(session_id=s.session_id)
```
`aget_user_sessions` returns at most 50 records per call, so loop until empty for users with many sessions.
#### Drop into the Agno DB directly
```python theme={"dark"}
db = await agent.aget_db()
# Now use any AsyncPostgresDb method: read messages, custom queries, etc.
session = await db.get_session(session_id="sess_456", session_type=...)
```
Use this when the four convenience methods don't cover what you need; the Agno DB client has a richer API.
## Agent class
`Agent` is the rich object returned by `Agents.aget()` and `AgentsListItem.aload()`. It carries the agent's full stored configuration, plus methods for creating tasks, invoking tools, and managing sessions.
```python theme={"dark"}
from xpander_sdk import Agents
agent = await Agents().aget("agent-123")
print(agent.name)
print(agent.model_provider, agent.model_name)
print(agent.deployment_type.value) # "serverless" or "container"
print(len(agent.tools.list))
```
### Class methods
#### `Agent.aload(agent_id, configuration=None, version=None) -> Agent`
Load an agent by ID. This is what `Agents.aget()` calls internally. Use the module-level `agents.aget(...)` in normal code; `Agent.aload(...)` is for places where you have a `Configuration` but no `Agents` instance.
```python theme={"dark"}
from xpander_sdk import Agent, Configuration
config = Configuration(api_key="...", organization_id="...")
agent = await Agent.aload(agent_id="agent-123", configuration=config)
```
| Parameter | Type | Required | Default | Description |
| --------------- | --------------- | -------- | ------- | ------------------------------------- |
| `agent_id` | `str` | Yes | – | Agent to load. |
| `configuration` | `Configuration` | No | `None` | SDK config. Uses defaults if omitted. |
| `version` | `int` | No | `None` | Specific version. Latest if omitted. |
The sync sibling is `Agent.load(...)`.
### Attributes
| Attribute | Type | Description |
| ----------------------- | ------------------------------------------------------------------------- | -------------------------------------------------------------------------------------- |
| `id` | `str` | Agent identifier. |
| `organization_id` | `str` | Owning organization. |
| `name` | `str` | Display name. |
| `description` | `str \| None` | Long description. |
| `unique_name` | `str` | URL-safe slug. |
| `icon` | `str` | Emoji or icon. Defaults to `"🚀"`. |
| `status` | `AgentStatus` | `DRAFT`, `ACTIVE`, or `INACTIVE`. |
| `version` | `int` | Loaded version. |
| `created_by` | `str \| None` | Creator user id. |
| `created_at` | `datetime \| None` | Creation timestamp. |
| `access_scope` | `AgentAccessScope` | `Personal` or `Organizational`. |
| `type` | `AgentType \| None` | `Manager`, `Regular`, `A2A`, `Curl`, `Orchestration`. |
| `framework` | `Framework` | Always `Agno` today. Other values reserved. |
| `deployment_type` | `AgentDeploymentType` | `Serverless` or `Container`. |
| `instructions` | `AgentInstructions` | `role: list[str]`, `goal: list[str]`, `general: str`. |
| `graph` | `AgentGraph` | Execution graph (source nodes, tools, sub-agents, MCP). |
| `tools` | `ToolsRepository` | Repository of attached tools. See [Tools Repository](/developers/sdk-reference/tools). |
| `knowledge_bases` | `list[AgentKnowledgeBase]` | Linked knowledge bases (id pointers). |
| `model_provider` | `str` | e.g. `"openai"`, `"anthropic"`, `"bedrock"`. |
| `model_name` | `str` | e.g. `"gpt-4o"`, `"claude-sonnet-4"`. |
| `llm_reasoning_effort` | `LLMReasoningEffort` | `Low`, `Medium`, or `High`. |
| `llm_api_base` | `str \| None` | Override for the LLM endpoint (Bedrock / Azure / Ollama). |
| `llm_extra_headers` | `dict[str, str]` | Extra headers for LLM calls. |
| `output_format` | `OutputFormat` | Default output format. |
| `output_schema` | `dict \| None` | Default JSON schema (used when `output_format == Json`). |
| `expected_output` | `str` | Description of the expected output. |
| `agno_settings` | `AgnoSettings` | Memory, tool-call limits, guardrails, reasoning settings. |
| `webhook_url` | `str \| None` | Configured webhook for completed tasks. |
| `connectivity_details` | `AIAgentConnectivityDetailsA2A \| AIAgentConnectivityDetailsCurl \| dict` | Wire details for `A2A` and `Curl` agent types. |
| `voice_id` | `str \| None` | Voice ID for `OutputFormat.Voice`. |
| `task_level_strategies` | `TaskLevelStrategies \| None` | Retry / iterative / stop strategies and per-day caps. |
| `orchestration_nodes` | `list[OrchestrationNode]` | Sub-nodes for orchestration agents. |
| `notification_settings` | `NotificationSettings` | Alerting config. |
| `use_oidc_pre_auth` | `bool` | Whether the agent uses OIDC pre-auth tokens. |
| `pre_auth_audiences` | `list[str]` | OIDC audiences for pre-auth. |
| `using_nemo` | `bool` | Whether NeMo is in use. |
| `deep_planning` | `bool` | Whether deep planning is enabled. |
| `enforce_deep_planning` | `bool` | Strict plan enforcement. |
| `llm_credentials` | `LLMCredentials \| None` | Per-agent LLM credentials, if overridden. |
### Computed properties
| Property | Type | Description |
| ------------------------- | ------------------------ | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `mcp_servers` | `list[MCPServerDetails]` | MCP servers configured in the graph. |
| `output` | `AgentOutput` | Resolved output configuration (Pydantic schema, markdown flag, JSON-mode flag). |
| `search_knowledge` | `bool` | `True` when the agent has at least one linked knowledge base. |
| `is_a_team` | `bool` | Truthy when the agent has sub-agents (Manager / Team mode). Returns an empty list rather than `False` when there are none, so test it for truthiness rather than comparing to `False`. |
| `is_active` | `bool` | `agent.status == AgentStatus.ACTIVE`. |
| `sanitized_name` | `str` | `name` munged to a valid Python identifier. |
| `strands_tools` | `list` | Strands-formatted tool wrappers. |
| `openai_agents_sdk_tools` | `list` | OpenAI Agents SDK-formatted tool wrappers. |
### Instance methods
| Method | Reference |
| ---------------------------------------------- | ----------------------------------------------------------------------------- |
| `acreate_task` / `create_task` | [Create a task](/developers/sdk-reference/agents#create_task) |
| `ainvoke_tool` / `invoke_tool` | [Invoke a tool](/developers/sdk-reference/agents#invoke_tool) |
| `aget_knowledge_bases` / `get_knowledge_bases` | [Knowledge bases](/developers/sdk-reference/agents#knowledge-bases) |
| `attach_knowledge_base` | [Knowledge bases](/developers/sdk-reference/agents#attach_knowledge_base) |
| `knowledge_bases_retriever` | [Knowledge bases](/developers/sdk-reference/agents#knowledge_bases_retriever) |
| `aget_db` / `get_db` | [Sessions](/developers/sdk-reference/agents#sessions) |
| `aget_user_sessions` / `get_user_sessions` | [Sessions](/developers/sdk-reference/agents#sessions) |
| `aget_session` / `get_session` | [Sessions](/developers/sdk-reference/agents#sessions) |
| `adelete_session` / `delete_session` | [Sessions](/developers/sdk-reference/agents#sessions) |
#### Streaming spec
`agent.aget_streaming_spec() -> StreamingSpecResponse`: returns the deployed agent's streaming URL and an API key for the container's `/invoke` endpoint. Useful when you want to bypass the SSE channel and POST directly to a deployed worker.
```python theme={"dark"}
spec = await agent.aget_streaming_spec()
print(spec.url) # https://.containers.xpander.ai/invoke
print(spec.api_key) # auth token for the /invoke endpoint
```
`StreamingSpecResponse` has two fields: `url` (str | None) and `api_key` (str | None). Both can be `None` if the agent isn't deployed yet.
#### Connection string
`agent.aget_connection_string() -> DatabaseConnectionString`: returns DB connection details for agents using session storage. The result is cached on the instance after the first call.
```python theme={"dark"}
conn = await agent.aget_connection_string()
print(conn.connection_uri.uri)
```
`DatabaseConnectionString` has `id`, `name`, `organization_id`, and `connection_uri.uri` (a Postgres-compatible URI).
#### sync\_local\_tools
`await agent.sync_local_tools(tools=[...])`: pushes locally-decorated `@register_tool(add_to_graph=True)` tools to the platform's graph. Called automatically when `Agent.aload` detects unsynced tools, so you rarely need to invoke it directly.