Agents - xpander.ai

The Agents module is your entry point to agents stored on the runtime. Use it to discover what agents exist, load a specific one, and from there create tasks, invoke skills, attach knowledge bases, and inspect sessions. In the SDK a skill is a Tool; agent.tools lists them.

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

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`	`list[AgentsListItem]`	Summary list of every agent your org can access.
`aget` / `get`	`Agent`	Load one agent fully (config + graph + skills).

`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`	Create a new task and run it.
`ainvoke_tool` / `invoke_tool`	Invoke a single skill bound to this agent.
`aget_knowledge_bases` / `get_knowledge_bases`	Load all knowledge bases attached to the agent.
`attach_knowledge_base`	Link a knowledge base to the agent (in-memory only).
`knowledge_bases_retriever`	Returns a `search(query, num_documents)` callable for use as a retriever in your agent loop.
`aget_db` / `get_db`	Returns the Agno PG client backing the agent’s session storage.
`aget_user_sessions` / `get_user_sessions`	All sessions for a given end-user.
`aget_session` / `get_session`	Load a single session.
`adelete_session` / `delete_session`	Delete a session.
`aget_streaming_spec` / `get_streaming_spec`	Get the deployed agent’s streaming URL + auth key.
`aget_connection_string` / `get_connection_string`	Get the agent’s DB connection details.

See the Agent class reference for attribute documentation.

Quick patterns

Find and load

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

agent = await agents.aget("agent-123", version=4)

Versioning is opt-in. Without version, you get the latest.

Default agent via env var

export XPANDER_AGENT_ID="agent-123"

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.

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, skills, model config, …) call .aload() on the item, or pass item.id to agents.aget().

Examples

Filter active agents

items = await agents.alist()
active = [i for i in items if i.status.value == "ACTIVE"]

Load all in parallel

import asyncio

items = await agents.alist()
full = await asyncio.gather(*(item.aload() for item in items))

Sync version

items = agents.list()

Same return type; blocks until the list is fetched.

Errors

alist raises ModuleException 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 skills, knowledge-base links, and agent loop settings. Use this whenever you need to actually invoke an agent or inspect its config in code.

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 for attributes (instructions, framework, graph, deployment_type, tools, knowledge_bases, agno_settings, etc.).

Examples

Latest version

agent = await agents.aget("agent-123")

Pinned version

agent = await agents.aget("agent-123", version=4)
print(agent.version)   # 4

The version corresponds to deployment snapshots in Xpander Chat. Pin a version in production to avoid surprises when someone edits the agent.

Use a default agent ID

export XPANDER_AGENT_ID="agent-123"

agent = await agents.aget()        # uses XPANDER_AGENT_ID

This is the pattern used inside @on_task handlers: you don’t pass an agent ID explicitly, because the runtime injects it.

Sync version

agent = agents.get("agent-123")

Same parameters; blocks until the agent is loaded.

What gets loaded

When you call aget, the SDK:

Fetches the agent record (GET /agents/:id).
Builds the AgentGraph from the graph items returned in the response.
Initializes a ToolsRepository populated with the agent’s skills.
If any local skills (Python @register_tool functions) need syncing to the runtime’s graph, kicks off a background sync_local_tools task: non-blocking.

You can read agent.graph to inspect the skill wiring, agent.tools.list for all skills, agent.knowledge_bases for KB links, and agent.mcp_servers for the configured MCP endpoints.

Errors

aget raises ModuleException 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 xpander and returns a Task object you can save, stream, or stop. This is the primary way to invoke an agent from code.

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 runtime; the agent’s run (either on the managed execution tier or in 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 the user’s connected accounts.
`agent_version`	`str`	No	`None`	Pin to a specific agent version.
`tool_call_payload_extension`	`dict`	No	`None`	Extra fields merged into every skill call payload.
`source`	`str`	No	`"sdk"`	Origin tag (`"webhook"`, `"slack"`, …).
`worker_id`	`str`	No	`None`	Pin execution to a specific execution tier instance.
`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 skill pre-auth.
`expected_output`	`str`	No	`None`	Description of expected output shape.
`mcp_servers`	`list[MCPServerDetails]`	No	`[]`	Per-task MCP endpoints (in addition to the agent’s configured ones).
`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. The task starts in status Pending. It moves through Executing → Completed (or Error / Failed / Stopped) as the agent’s run progresses.

Examples

With files

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

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

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.

Continue a thread

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

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 the user’s connected accounts (e.g. each end user’s Gmail OAuth token).

Add an MCP endpoint just for this task

from xpander_sdk import MCPServerDetails, MCPServerType
from xpander_sdk.modules.tools_repository.models.mcp import 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 endpoints stack on top of the agent’s persistent MCP configuration.

Pin a version

task = await agent.acreate_task(
    prompt="Run with the locked v3 prompt",
    agent_version="3",
)

Sync version

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 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 skill from the agent’s ToolsRepository and returns a ToolInvocationResult. Use it when you want to test a skill, or when you’re building an agent loop yourself and the loop’s own dispatcher isn’t a good fit.

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	-	Skill input. Validated against `tool.schema` before execution. Local skills registered with `@register_tool` take their arguments flat, as in the example above. Catalog skills take the grouped form instead, `{"body_params": {...}, "query_params": {...}, "path_params": {...}, "headers": {...}}`, matching Invoke Skill 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 the LLM’s `tool_call_id`).

Returns `ToolInvocationResult`

Field	Type	Description
`tool_id`	`str`	The skill’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 skill’s response. Shape depends on the skill.
`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 skill, or `500` for client-side errors.
`is_local`	`bool`	`True` for skills registered via `@register_tool`.

is_success and is_error are mutually exclusive in normal operation; check is_error first.

Examples

With `payload_extension`

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 skills

Skills registered with @register_tool are invoked locally (the fn runs in-process) and pass through the same ToolInvocationResult interface:

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 for details.

Inside an `@on_task` handler

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

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) 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 Xpander Chat. From code, you can attach more, load the KnowledgeBase objects, or build a retriever callable for use in your agent loop.

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.

kbs = await agent.aget_knowledge_bases()

Returns `list[KnowledgeBase]`

A list of full KnowledgeBase objects. See KnowledgeBase for the methods you can call on each.

Sync version

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.

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 xpander (e.g. via the API or Xpander Chat 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.

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

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 agent loop 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():

{"content": "...", "score": 0.86}

The retriever swallows errors and returns [] if the search fails: useful for embedding into agent loop pipelines that shouldn’t crash on retrieval errors.

Patterns

Wire as the Agno knowledge

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 need to call knowledge_bases_retriever() directly. Use it only when bypassing the Backend dispatcher.

Add a KB on the fly

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 running on the Agno loop with agno_settings.session_storage = True persist their session history to a Postgres database managed by the runtime. 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.

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

db = agent.get_db()                # equivalent to aget_db(async_db=False)

`aget_user_sessions`

Load all sessions for a given end-user.

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

sessions = agent.get_user_sessions(user_id="user_42")

`aget_session`

Load a single session by ID.

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

session = agent.get_session(session_id="sess_456")

`adelete_session`

Delete a session.

await agent.adelete_session(session_id="sess_456")

Parameters

Parameter	Type	Required	Description
`session_id`	`str`	Yes	Session to delete.

Sync version

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

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

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

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 skills, and managing sessions.

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"
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.

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`.
`instructions`	`AgentInstructions`	`role: list[str]`, `goal: list[str]`, `general: str`.
`graph`	`AgentGraph`	Execution graph (source nodes, skills, sub-agents, MCP).
`tools`	`ToolsRepository`	Repository of attached skills. See `ToolsRepository`.
`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, skill 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 endpoints 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 skill wrappers.
`openai_agents_sdk_tools`	`list`	OpenAI Agents SDK-formatted skill wrappers.

Instance methods

Method	Reference
`acreate_task` / `create_task`	Create a task
`ainvoke_tool` / `invoke_tool`	Invoke a skill
`aget_knowledge_bases` / `get_knowledge_bases`	Knowledge bases
`attach_knowledge_base`	Knowledge bases
`knowledge_bases_retriever`	Knowledge bases
`aget_db` / `get_db`	Sessions
`aget_user_sessions` / `get_user_sessions`	Sessions
`aget_session` / `get_session`	Sessions
`adelete_session` / `delete_session`	Sessions

Streaming spec

agent.aget_streaming_spec() -> StreamingSpecResponse: returns the deployed agent’s streaming URL and an API key for its /invoke endpoint. Useful when you want to bypass the SSE channel and POST directly to the deployed agent’s run.

spec = await agent.aget_streaming_spec()
print(spec.url)        # https://<agent>.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.

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) skills to the runtime’s graph. Called automatically when Agent.aload detects unsynced skills, so you rarely need to invoke it directly.

​Constructor

​Module methods

​Agent instance methods

​Quick patterns

​Find and load

​Pin to a specific version

​Default agent via env var

​list

​Parameters

​Returns list[AgentsListItem]

​Examples

​Filter active agents

​Load all in parallel

​Sync version

​Errors

​get

​Parameters

​Returns Agent

​Examples

​Latest version

​Pinned version

​Use a default agent ID

​Sync version

​What gets loaded

​Errors

​create_task

​Parameters

​Returns Task

​Examples

​With files

​Structured output

​Streaming

​Continue a thread

​Per-end-user

​Add an MCP endpoint just for this task

​Pin a version

​Sync version

​Errors

​invoke_tool

​Parameters

​Returns ToolInvocationResult

​Examples

​With payload_extension

​Local Python skills

​Inside an @on_task handler

​Sync version

​Notes

​Knowledge bases

​aget_knowledge_bases

​Returns list[KnowledgeBase]

​Sync version

​attach_knowledge_base

​Parameters

​knowledge_bases_retriever

​Returned callable signature

​Patterns

​Wire as the Agno knowledge

​Add a KB on the fly

​Sessions

​aget_db

​Parameters

​Returns

​Sync version

​aget_user_sessions

​Parameters

​Returns

​Sync version

​aget_session

​Parameters

​Returns

​Sync version

​adelete_session

​Parameters

​Sync version

​Errors

​Patterns

​Inspect recent sessions for a user

​Wipe all sessions for a user

​Drop into the Agno DB directly

​Agent class

Constructor

Module methods

`Agent` instance methods

Quick patterns

Find and load

Pin to a specific version

Default agent via env var

list

Parameters

Returns `list[AgentsListItem]`

Examples

Filter active agents

Load all in parallel

Sync version

Errors

get

Parameters

Returns `Agent`

Examples

Latest version

Pinned version

Use a default agent ID

Sync version

What gets loaded

Errors

create_task

Parameters

Returns `Task`

Examples

With files

Structured output

Streaming

Continue a thread

Per-end-user

Add an MCP endpoint just for this task

Pin a version

Sync version

Errors

invoke_tool

Parameters

Returns `ToolInvocationResult`

Examples

With `payload_extension`

Local Python skills

Inside an `@on_task` handler

Sync version

Notes

Knowledge bases

`aget_knowledge_bases`

Returns `list[KnowledgeBase]`

Sync version

`attach_knowledge_base`

Parameters

`knowledge_bases_retriever`

Returned callable signature

Patterns

Wire as the Agno knowledge

Add a KB on the fly

Sessions

`aget_db`

Parameters

Returns

Sync version

`aget_user_sessions`

Parameters

Returns

Sync version

`aget_session`

Parameters

Returns

Sync version

`adelete_session`

Parameters

Sync version

Errors

Patterns

Inspect recent sessions for a user

Wipe all sessions for a user

Drop into the Agno DB directly

Agent class