Agno

Agno is the framework xpander.ai has the deepest integration with. In this guide, we’ll build a production-ready agent with credentials, instructions, tools, knowledge-base access, session DB, memory, guardrails, and a context-optimization pipeline.

Prerequisites

Complete the Quickstart so the CLI, SDK, and xpander login are already set up.
Python 3.12+ for the local handler.
An LLM provider key in your shell like OPENAI_API_KEY, ANTHROPIC_API_KEY. (These keys will only be used locally.)

1. Install

# Quote the package name so zsh doesn't expand the brackets.
pip install "xpander-sdk[agno]"

2. Set up scaffolding

# Create an agent named "my-first-agent"
xpander agent new \
    --name "my-first-agent" \
    --framework "agno" \
    --folder "."

These files get created:

./
│
├── xpander_handler.py        # Your @on_task entry point. The file you'll edit most.
├── xpander_config.json       # Agent ID, organization ID, API key, framework selection.
├── agent_instructions.json   # role / goal / general (the agent's system prompt).
├── requirements.txt          # Python dependencies (xpander-sdk[agno] is pinned here).
└── .env                      # XPANDER_API_KEY, XPANDER_ORGANIZATION_ID, XPANDER_AGENT_ID.

`xpander_config.json` reference

xpander_config.json

{
  "agent_id": "agt_01H...",
  "organization_id": "org_01H...",
  "api_key": "xpd_...",
  "framework": "agno"
}

3. Create task handler

The full pattern, wrapped in @on_task so the platform routes tasks to it:

xpander_handler.py

from dotenv import load_dotenv
load_dotenv()  # loads XPANDER_API_KEY and friends before any sdk import

from xpander_sdk import on_task, Task, Backend, Tokens
from agno.agent import Agent

@on_task
async def handler(task: Task) -> Task:
    # 1. Fetch the agent's full configuration from xpander's control plane.
    backend = Backend(configuration=task.configuration)
    agno_args = await backend.aget_args(task=task)

    # 2. Build the framework's own Agent with that config.
    agno_agent = Agent(**agno_args, debug_mode=True)

    # 3. Run the LLM loop with the task's input, files, and images.
    result = await agno_agent.arun(
        input=task.to_message(),
        files=task.get_files(),
        images=task.get_images(),
    )

    # 4. Write the result back so the platform can store and display it.
    task.result = result.content
    task.tokens = Tokens(
        prompt_tokens=result.metrics.input_tokens,
        completion_tokens=result.metrics.output_tokens,
    )
    task.used_tools = [t.tool_name for t in (result.tools or [])]
    return task

Here’s what’s happening:

Backend(configuration=task.configuration) picks up the API key, organization ID, and base URL from the active task. No need to read .env directly.
await backend.aget_args(task=task) calls the control plane and returns a dict with the full agent configuration (instructions, tools, model, knowledge bases, session storage, memory, guardrails). Always pass task=task inside an @on_task handler so task-level overrides (instructions_override, expected_output, output_schema) are merged in.
Agent(**agno_args, debug_mode=True) splats that dict into Agno’s own Agent class. debug_mode prints tool calls and token usage; remove it for production.
task.to_message() flattens the prompt text, file URLs, and any inline-readable file content into a single string ready for Agno. task.get_files() and task.get_images() return Agno-typed agno.media.File and agno.media.Image objects.
Reporting task.tokens and task.used_tools is optional. Skipping them just means the metrics view in Agent Studio shows “no usage data” for that run.

`backend.aget_args` reference

Input parameters

backend.aget_args accepts these arguments:

Key	Type	Notes
`task`	`Task`	The active task inside an `@on_task` handler. Pass it so task-level overrides (`instructions_override`, `expected_output`, `output_schema`) are merged into the resolved args. Required inside `@on_task`; optional in scripts and notebooks where you can pass `agent_id=` instead.
`override`	`dict`	`dict.update()`-merged onto the resolved args after everything else is built. Accepts any key Agno’s `Agent.__init__` takes (see below). Optional. Defaults to `None`.
`tools`	`list[Callable]`	Appended to the resolved tools list (connectors plus custom plus MCP). Use it to inject ephemeral tools without touching the agent’s graph. Optional. Defaults to `[]`.

What can override change? override accepts any key Agno’s Agent.__init__ takes. Two ways to use it:

Replace any value the SDK already resolves: any key from the output-params table below (model, instructions, tools, db, knowledge_retriever, pre_hooks, output_schema, and so on).
Add Agno-native kwargs the SDK doesn’t set itself. Common ones:

Key	What it controls
`temperature`	Sampling temperature for the model.
`max_tokens`	Cap on tokens generated per response.
`show_tool_calls`	Print tool calls during the run.
`debug_mode`	Print full LLM payloads and timings.
`use_json_mode`	Force JSON-mode responses.
`markdown`	Render responses as Markdown.

See the Agno Agent reference for the full surface. Setting override["model"] skips the SDK’s own model resolution entirely, so use it whenever you want a different model client without re-implementing credential handling. Example using override to A/B-test two models against the same agent definition:

# Run the same agent against a different model for one task.
from agno.models.anthropic import Claude

args = await backend.aget_args(
    task=task,
    override={"model": Claude(id="claude-sonnet-4-5", api_key="...")},
)

Example using override to tune Agno-native sampling parameters:

# Lower temperature and bound max tokens for a deterministic run.
args = await backend.aget_args(
    task=task,
    override={"temperature": 0.2, "max_tokens": 800},
)

Example using tools to inject an ephemeral test tool:

def _local_clock() -> str:
    """Return the developer's local clock for debugging."""
    from datetime import datetime
    return datetime.now().isoformat()

args = await backend.aget_args(task=task, tools=[_local_clock])

For anything more invasive (a new pre-hook, a different DB), grab the args dict and mutate it directly before splatting into Agent(...). The dict is yours; the SDK won’t reach back in.

Output parameters

Calling backend.aget_args returns these fields:

Key	Type	Notes
`id`	`str`	Agent ID from `xpander_config.json`. Always set.
`name`	`str`	Agent name from the control plane. Always set.
`description`	`str`	Sourced from `agent_instructions.json` `general`. Always set.
`model`	`agno.models.base.Model`	Instantiated with credentials resolved. The agent’s custom LLM key wins on cloud, your shell env var wins locally. Always set.
`instructions`	`str`	The Agent Studio instructions (or `task.instructions_override`), with context-optimization, workspace-output, and compact-tool guidance appended. Always set.
`tools`	`list[Callable]`	Connectors, custom `@register_tool` functions, MCP servers (with OAuth handled), `xpcompact_context`, plus Agno’s `think` / `analyze` when `reasoning_tools_enabled=True`. Always set.
`tool_hooks`	`list[Callable]`	One internal hook for retries on transient errors, stuck-loop detection, activity reporting, and Layer 1 microcompaction. Append to it, don’t replace. Always set.
`compression_manager`	`XPanderContextOptimizer`	Layered context-optimization pipeline (microcompaction, auto-compaction, manual via `xpcompact_context`, emergency, pre-retry). Always set.
`add_datetime_to_context`	`bool`	Inject the current datetime into context on every run. Always set. Defaults to `True`.
`store_events`	`bool`	Persist Agno run events. Always set. Defaults to `True`.
`db`	`AsyncPostgresDb` or `PostgresDb`	Scoped to a per-agent schema. Set when `session_storage`, `user_memories`, or `agent_memories` is on.
`add_history_to_context`, `session_id`, `user_id`, `num_history_runs`, `max_tool_calls_from_history`, `enable_session_summaries`	various	Driven by `agno_settings.session_storage` (see settings table below). Set when `session_storage=True`.
`enable_user_memories`, `memory_manager`, `enable_agentic_memory`	various	Per-user facts that persist across sessions. Set when `user_memories=True`.
`add_culture_to_context`, `update_cultural_knowledge`, `enable_agentic_culture`	`bool`	Org-wide facts. Single-agent only (skipped on Teams). Set when `agent_memories=True`.
`learning`	`bool`	Set when `agno_settings.learning=True`.
`tool_call_limit`	`int`	Cap on tool calls per run. Set when configured on the agent.
`knowledge_retriever`, `search_knowledge`	callable, `bool`	Agno calls the retriever automatically during the loop. Set when KBs are attached.
`pre_hooks`	`list`	`PIIDetectionGuardrail`, `PromptInjectionGuardrail`, `OpenAIModerationGuardrail` per `agno_settings`. Set when guardrails are on.
`output_schema`, `use_json_mode`, `markdown`	various	Structured output and Markdown formatting. Task-level output format and schema overrides applied. Set per output settings.
`expected_output`, `additional_context`	`str`	Forwarded from the agent definition and task. Set when present.
`members`, `add_member_tools_to_context`, `share_member_interactions`, `show_members_responses`	various	The SDK recursively builds each sub-agent as `AgnoAgent` or `AgnoTeam`. Set when `agent.is_a_team`.

4. Edit the agent’s system prompt

agent_instructions.json contains the agent’s system prompt and has exactly three fields:

agent_instructions.json

{
  "role": [
    "You are a customer support assistant for Acme.",
    "Always confirm the customer's account ID before taking any action."
  ],
  "goal": [
    "Resolve the customer's issue in as few turns as possible.",
    "Escalate to a human if the request involves a refund over $500."
  ],
  "general": "Be concise, professional, and friendly. Never invent policy details; if you don't know something, say so and offer to escalate."
}

Save the file and the next xpander agent dev syncs it to the control plane.

5. Set up streaming (optional)

For token-by-token output, decorate an async def that yields TaskUpdateEvent objects instead of returning a Task. The decorator detects the difference automatically.

streaming_handler.py

from datetime import datetime, timezone
from xpander_sdk import on_task, Task, Backend, TaskUpdateEvent, TaskUpdateEventType
from agno.agent import Agent
from agno.run.agent import RunEvent, RunOutput

@on_task
async def handler(task: Task):
    backend = Backend(configuration=task.configuration)
    agno_agent = Agent(**(await backend.aget_args(task=task)))

    final_output = None
    # Agno emits a stream of events: chunks, tool calls, final RunOutput.
    async for event in await agno_agent.arun(
        input=task.to_message(),
        stream=True,
        stream_events=True,
        yield_run_output=True,
    ):
        if isinstance(event, RunOutput):
            final_output = event
        elif hasattr(event, "event") and event.event == RunEvent.run_content and event.content:
            # Forward each token chunk to the platform's SSE stream.
            yield TaskUpdateEvent(
                type=TaskUpdateEventType.Chunk,
                task_id=task.id,
                organization_id=task.organization_id,
                time=datetime.now(timezone.utc),
                data=event.content,
            )

    task.result = final_output.content if final_output else ""
    yield TaskUpdateEvent(
        type=TaskUpdateEventType.TaskFinished,
        task_id=task.id,
        organization_id=task.organization_id,
        time=datetime.now(timezone.utc),
        data=task,
    )

Here’s what’s happening:

stream=True, stream_events=True, yield_run_output=True tell Agno to emit events instead of buffering. The handler receives chunks, tool-call events, and a final RunOutput.
The Chunk event forwards each token to the platform’s SSE stream so clients render output as it arrives.
The TaskFinished event signals the end of the stream and carries the final task back to the platform.

A streaming handler exposes itself only through POST /invoke, returning Server-Sent Events. The platform’s SSE listener for cloud-deployed agents expects a regular handler that returns a Task. So if you need both an interactive streaming experience and platform-routed tasks, run two handlers, or have your streaming endpoint proxy through a regular handler.

6. Test local development

Run the handler with the dev server. Tasks created from any channel (REST, Slack, Agent Studio) route to your laptop:

# Starts the @on_task HTTP server and subscribes to the platform event stream.
xpander agent dev

Routing cloud traffic to a local instance is a preview feature.When a local instance is running via xpander agent dev, it takes over and all tasks route to your locally running agent instead. Only one can be active at a time.

For one-shot testing without a server:

# Calls your handler exactly once with the given prompt and exits.
python3 xpander_handler.py \
    --invoke \
    --prompt "Quick test" \
    --output_format json \
    --output_schema '{"answer":"string"}'

--output_format and --output_schema are useful for testing structured output without changing the agent’s settings in the control plane.

Troubleshooting

AttributeError: 'NoneType' object has no attribute 'instructions_override'

Backend.aget_args() reads task.instructions_override while building the args. Inside an @on_task handler, always pass the active task: await backend.aget_args(task=task). The agent_id-only form is supported outside a handler (in scripts and notebooks), but inside one the active task is the source of truth for instruction overrides.

Postgres connection errors when running the handler

Session storage is on by default, so the args dict includes a db wired to xpander’s Postgres. For cloud-hosted agents this is automatic. For self-hosted or air-gapped deployments, the database needs to be reachable from where the agent runs. Check the connection string with await agent.aget_connection_string() and confirm the host is reachable. To turn session storage off, flip agno_settings.session_storage to False in Agent Studio.

Wrong model or wrong key used at runtime

Custom LLM keys configured on the agent take precedence on cloud deployments. Locally, your shell’s OPENAI_API_KEY (or the equivalent for your provider) wins. If you want the cloud-side custom key locally too, mirror it into your .env.

pip install xpander-sdk[agno] says 'no matches found' on macOS

zsh expands the brackets. Quote the package name: pip install "xpander-sdk[agno]".

Next steps

Quickstart

The 10-minute scaffold-to-run walkthrough that produced the handler shown above.

Custom Tools

Wrap private APIs as tools with @register_tool and pass them through the args dict.

Memory & State

The deep dive on session_storage, user memories, and agent memories.

Core Concepts

The SDK class names mapped onto agents, tasks, threads, and memory.

Frameworks overview

What’s auto-wired vs. manual for Agno, OpenAI Agents SDK, LangChain, and AWS Strands.

Overview

Platform

Getting Started

Building Agents

Workflows

Agentic Automation

Observability

Deployment Channels

Self-Hosted

Security & Compliance

Building In-Product Agents

Prerequisites

1. Install

2. Set up scaffolding

`xpander_config.json` reference

3. Create task handler

`backend.aget_args` reference

Input parameters

Output parameters

4. Edit the agent’s system prompt

5. Set up streaming (optional)

6. Test local development

Troubleshooting

Next steps

Quickstart

Custom Tools

Memory & State

Core Concepts

Frameworks overview

​Prerequisites

​1. Install

​2. Set up scaffolding

​xpander_config.json reference

​3. Create task handler

​backend.aget_args reference

​Input parameters

​Output parameters

​4. Edit the agent’s system prompt

​5. Set up streaming (optional)

​6. Test local development

​Troubleshooting

​Next steps

Quickstart

Custom Tools

Memory & State

Core Concepts

Frameworks overview

Prerequisites

1. Install

2. Set up scaffolding

`xpander_config.json` reference

3. Create task handler

`backend.aget_args` reference

Input parameters

Output parameters

4. Edit the agent’s system prompt

5. Set up streaming (optional)

6. Test local development

Troubleshooting

Next steps