Backend - xpander.ai

The Backend module resolves an agent stored on xpander into arguments for your code. It does three things:

Resolves runtime arguments for your agent loop: aget_args(agent_id=..., task=task) returns kwargs you can spread directly into the Agno agent constructor (or another supported loop’s). In the SDK a skill is a Tool; the resolved tools kwarg carries them.
Invokes agents directly: ainvoke_agent(agent_id=..., prompt=...) is shorthand for “load the agent, create a task.” Useful when you don’t need control over the agent loop.
Reports externally executed runs: areport_external_task(...) lets you log a run that happened outside the runtime (e.g. in a queue consumer or another process) so it shows up in the same task history and metrics as runtime-managed runs.

from xpander_sdk import Backend

backend = Backend()

Constructor

Backend(configuration: Optional[Configuration] = None)

Parameter	Type	Default	Description
`configuration`	`Configuration`	`None`	SDK configuration. Falls back to env vars.

When called without arguments, the module reads XPANDER_API_KEY, XPANDER_ORGANIZATION_ID, and XPANDER_BASE_URL from the environment.

Methods

Method	What it does
`ainvoke_agent` / `invoke_agent`	Load an agent and create a task in one call. Returns a `Task`.
`aget_args` / `get_args`	Resolve agent loop kwargs (model, system prompt, skills, memory, …) from a stored agent configuration.
`areport_external_task` / `report_external_task`	Record an external execution (run outside xpander) so it appears in task history.

Typical pattern

from xpander_sdk import Backend, on_task
from xpander_sdk.modules.tasks.sub_modules.task import Task
from agno.agent import Agent as AgnoAgent

backend = Backend()

@on_task
async def handler(task: Task) -> Task:
    args = await backend.aget_args(
        agent_id=task.agent_id,
        agent_version=task.agent_version,
        task=task,
    )
    agno_agent = AgnoAgent(**args)
    result = await agno_agent.arun(input=task.to_message())
    task.result = result.content
    return task

@on_task listens for incoming task events from the runtime; the handler resolves the agent loop args for the current task, builds an Agno agent, runs it, and stores the result on the task. The runtime records status, metrics, and skill calls automatically.

Environment variable shortcut

Set XPANDER_AGENT_ID and you can omit agent_id from every Backend call:

export XPANDER_AGENT_ID="agent-123"

backend = Backend()
args = await backend.aget_args()                     # uses XPANDER_AGENT_ID
task = await backend.ainvoke_agent(prompt="Hello")   # same

Explicit arguments always override the env var.

get_args

Backend.aget_args returns a dictionary of keyword arguments ready for the agent loop of a stored agent. Spread the result directly into an Agno (or other supported loop’s) constructor: the dict already contains the model, system prompt, skills, memory settings, knowledge-base retrievers, and guardrails configured in Xpander Chat.

from xpander_sdk import Backend, Task, on_task
from agno.agent import Agent as AgnoAgent

@on_task
async def handler(task: Task) -> Task:
    backend = Backend(configuration=task.configuration)
    args = await backend.aget_args(task=task)

    agno_agent = AgnoAgent(**args)
    result = await agno_agent.arun(input=task.to_message())

    task.result = result.content
    return task

This is the primary integration point for code-based agents. Call it once per task, inside @on_task, passing the task so the resolved kwargs carry that run’s input, files, and output schema alongside the latest stored configuration.

Parameters

Parameter	Type	Required	Default	Description
`agent_id`	`str`	No¹	`XPANDER_AGENT_ID` env var	Agent ID. Required if `agent` is not provided.
`agent`	`Agent`	No¹	`None`	A pre-loaded `Agent` instance. Takes precedence over `agent_id`.
`agent_version`	`int`	No	`None`	Specific version to load. Latest if omitted.
`task`	`Task`	Yes²	`None`	The current task. Supplies the run’s input, files, and output schema on top of the stored configuration.
`override`	`dict[str, Any]`	No	`None`	Final overrides merged into the resolved kwargs.
`tools`	`list[Callable]`	No	`None`	Extra Python callables added to the agent’s skill list.
`is_async`	`bool`	No	`True`	(async only) Whether you’ll run the agent loop in an async context. Affects how local skills are wrapped.
`auth_events_callback`	`Callable`	No	`None`	Per-call callback for MCP/OAuth authentication events. See Authentication events.

¹ Either agent_id or agent must be resolvable. If neither is passed, the method falls back to XPANDER_AGENT_ID. ² Optional in the signature, but pass it. The resolved kwargs are built from the task, so call aget_args from inside @on_task where a Task is in scope.

Returns `dict[str, Any]`

A dictionary of kwargs ready for the agent loop. The exact keys depend on the agent’s framework field (currently always Agno):

{
    "name": "Sales Assistant",
    "agent_id": "agent-123",
    "model": <agno Model instance>,
    "instructions": "...",
    "tools": [<callable>, <callable>, ...],
    "knowledge": <agno Knowledge instance | None>,
    "memory_manager": <agno MemoryManager | None>,
    "db": <agno PostgresDb | None>,
    "guardrails": [...],
    "user_id": "...",
    "session_id": "...",
    # ...framework-specific keys
}

You don’t need to read these: pass them straight to the agent loop’s constructor.

Examples

Inside `@on_task`

from xpander_sdk import Backend, on_task
from xpander_sdk.modules.tasks.sub_modules.task import Task
from agno.agent import Agent as AgnoAgent

backend = Backend()

@on_task
async def handler(task: Task) -> Task:
    args = await backend.aget_args(
        agent_id=task.agent_id,
        agent_version=task.agent_version,
        task=task,
    )
    agno_agent = AgnoAgent(**args)

    result = await agno_agent.arun(
        input=task.to_message(),
        files=task.get_files(),
        images=task.get_images(),
    )
    task.result = result.content
    return task

Passing task=task enriches the args with task-specific context (session id, user id, output schema), so memory and structured output work transparently.

Pre-loaded agent (avoid re-fetching)

from xpander_sdk import Agents, Backend

agents = Agents()
backend = Backend()

agent = await agents.aget("agent-123")  # fetched once, reused

# Many tasks for the same agent: pass `agent` instead of `agent_id`
for prompt in prompts:
    args = await backend.aget_args(agent=agent, task=task, override={"name": "batch-worker"})
    ...

agent takes precedence over agent_id. This avoids a fresh GET /agents/:id call each time you build the agent loop args.

Override resolved kwargs

args = await backend.aget_args(
    agent_id="agent-123",
    task=task,
    override={
        "show_tool_calls": True,
        "markdown": False,
    },
)

override is shallow-merged after the agent loop dispatcher builds its kwargs, so anything you set wins.

Add extra skills

def lookup_internal_id(employee_email: str) -> str:
    """Look up an employee's internal ID from email."""
    ...

args = await backend.aget_args(
    agent_id="agent-123",
    task=task,
    tools=[lookup_internal_id],
)

The callable is appended to the agent’s skill list for this run only: it isn’t persisted to the runtime. For permanent skills, register with @register_tool instead.

Authentication events

When an agent uses skills that require OAuth, whether served over MCP or from the catalog, the SDK emits auth_event updates while the user authenticates. Register a callback to handle the OAuth URL or token-ready signals:

Per-call callback

from xpander_sdk import Backend
from xpander_sdk.modules.agents.sub_modules.agent import Agent
from xpander_sdk.modules.tasks.sub_modules.task import Task, TaskUpdateEvent

async def on_auth(agent: Agent, task: Task, event: TaskUpdateEvent):
    print(f"Auth required: {event.data}")
    # Show event.data['url'] to the user, or kick off your OAuth flow

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

Globally with `@on_auth_event`

from xpander_sdk import on_auth_event

@on_auth_event
async def handle_auth(agent, task, event):
    print(f"[GLOBAL] {agent.name}: {event.data}")

# No need to pass it: the decorator auto-registers
args = await backend.aget_args(agent_id="agent-123", task=task)

You can combine both: decorated handlers are always invoked, and auth_events_callback adds a one-off handler on top. See @on_auth_event for details.

Sync version

args = backend.get_args(agent_id="agent-123", task=task)

Same parameters minus is_async (sync wrapper hard-codes it to False so local skills are wrapped synchronously). Don’t call from inside a running event loop.

invoke_agent

Backend.ainvoke_agent (and its sync sibling invoke_agent) is shorthand for “load the agent and create a task.” Internally it calls Agents.aget(agent_id) followed by Agent.acreate_task(...), returning the resulting Task. Use this when you want to invoke an agent from outside the @on_task runtime: for example, from a webhook handler, a script, or another agent.

If prompt happens to be a JSON string containing xpander_task_id, the method short-circuits and returns the existing task instead of creating a new one. This is how xpander passes task continuations through chat-style integrations.

from xpander_sdk import Backend

backend = Backend()
task = await backend.ainvoke_agent(
    agent_id="agent-123",
    prompt="Summarize the latest sales report",
)
print(task.id, task.status)

Parameters

Parameter	Type	Required	Default	Description
`agent_id`	`str`	No	`XPANDER_AGENT_ID` env var	Agent to invoke. Falls back to env var if omitted.
`prompt`	`str`	No	`""`	Natural-language input for the agent.
`existing_task_id`	`str`	No	`None`	Continue an existing task instead of creating a new one.
`file_urls`	`list[str]`	No	`None`	URLs of files to attach (PDFs, images, CSVs, …).
`user_details`	`User`	No	`None`	End-user context for memory and personalization.
`agent_version`	`str`	No	`None`	Pin a specific agent version. Defaults to the latest.
`tool_call_payload_extension`	`dict`	No	`None`	Extra fields merged into every skill call payload.
`source`	`str`	No	`"sdk"`	Origin tag stored on the task (e.g. `"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 dispatch to the cloud execution tier).
`output_format`	`OutputFormat`	No	`None`	`Text`, `Markdown`, `Json`, or `Voice`.
`output_schema`	`dict`	No	`None`	JSON schema for structured output (paired with `OutputFormat.Json`).
`events_streaming`	`bool`	No	`False`	Enable SSE streaming via `task.aevents()`.
`additional_context`	`str`	No	`None`	Extra context appended to the system prompt for this run.
`expected_output`	`str`	No	`None`	Description of the expected output shape (used by some agent loops for self-verification).

Returns `Task`

The created Task object, populated with the xpander-assigned id and an initial status of pending. See Task for the full attribute list.

Examples

Attach files

task = await backend.ainvoke_agent(
    agent_id="agent-123",
    prompt="Find action items from this meeting recording.",
    file_urls=[
        "https://example.com/recordings/meeting.mp3",
        "https://example.com/transcripts/notes.pdf",
    ],
)

PDFs and images are auto-categorized by Task.get_files() / get_images() for Agno. Other file types (CSV, JSON, TXT, …) are inlined into the message via Task.to_message().

Structured output

from xpander_sdk import OutputFormat

task = await backend.ainvoke_agent(
    agent_id="agent-123",
    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"],
    },
)

Stream events

task = await backend.ainvoke_agent(
    agent_id="agent-123",
    prompt="Long-running research task",
    events_streaming=True,
)

async for event in task.aevents():
    print(event.type, event.data)

Continue an existing task

task = await backend.ainvoke_agent(
    agent_id="agent-123",
    existing_task_id="task_abc123",
    prompt="Now summarize what you found.",
)

existing_task_id reuses the same memory thread, so the agent has full context from the prior run.

Per-end-user context

from xpander_sdk import User

task = await backend.ainvoke_agent(
    agent_id="agent-123",
    prompt="What's on my calendar this afternoon?",
    user_details=User(
        id="user_42",
        email="alex@example.com",
        first_name="Alex",
        timezone="America/New_York",
    ),
)

The User object scopes user-memory storage and connected accounts (e.g. OAuth tokens for Gmail or Calendar) to the right end user.

Sync version

task = backend.invoke_agent(
    agent_id="agent-123",
    prompt="Hello!",
)

Identical signature; blocks until the task is created. Don’t call from inside a running event loop: use ainvoke_agent there.

report_external_task

Backend.areport_external_task logs a task that was executed outside the xpander.ai runtime (for example, in a queue consumer, a Lambda, or a different process) so it shows up in the same task history, metrics, and activity views as runtime-managed tasks. Use this when you’ve called an LLM yourself (outside of an @on_task handler) but still want the run to count toward agent metrics and appear in the dashboard.

from xpander_sdk import Backend, Tokens

backend = Backend()

reported = await backend.areport_external_task(
    agent_id="agent-123",
    id="ext-job-9921",
    input="Summarize Q4 earnings",
    result="Q4 revenue grew 22% YoY, driven by ...",
    tokens=Tokens(prompt_tokens=2_140, completion_tokens=380),
    duration=4.7,
    used_tools=["web_search", "fetch_pdf"],
    is_success=True,
)
print(reported.id, reported.status)

Parameters

Parameter	Type	Required	Default	Description
`agent_id`	`str`	No¹	`XPANDER_AGENT_ID` env var	Agent the run is associated with. Falls back to `agent.id` if `agent` is provided.
`agent`	`Agent`	No¹	`None`	Pre-loaded `Agent` instance. Takes precedence over `agent_id`.
`id`	`str`	No	`None`	External task identifier. Re-using the same ID updates the existing record.
`input`	`str`	No	`None`	Input prompt for the run.
`llm_response`	`Any`	No	`None`	The raw LLM response object (provider-specific). Stored verbatim for debugging.
`tokens`	`Tokens`	No	`None`	Token usage. Used by metrics.
`is_success`	`bool`	No	`True`	Whether the run completed successfully. Sets task status accordingly.
`result`	`str`	No	`None`	Final result string.
`duration`	`float`	No	`0`	Wall-clock duration in seconds.
`used_tools`	`list[str]`	No	`None`	Names of skills the run called.
`configuration`	`Configuration`	No	Module’s config	Override SDK configuration for this call.

¹ Either agent_id or agent must resolve. The method also reads XPANDER_AGENT_ID as a final fallback.

Returns `Task`

The xpander-side Task record after persistence. Re-using id returns the updated record; new ids create a fresh task.

Examples

Idempotent updates

Pass the same id to update an in-progress task as it makes progress:

# Mark started
await backend.areport_external_task(
    agent_id="agent-123",
    id="job-7842",
    input="Long-running data extraction",
    is_success=True,
    result=None,
    duration=0,
)

# ... do the work ...

# Mark completed with final tokens
await backend.areport_external_task(
    agent_id="agent-123",
    id="job-7842",
    input="Long-running data extraction",
    result="Extracted 412 rows",
    tokens=Tokens(prompt_tokens=18_400, completion_tokens=2_100),
    duration=37.6,
    is_success=True,
    used_tools=["s3_read", "snowflake_query"],
)

Using a pre-loaded agent

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

await backend.areport_external_task(
    agent=agent,
    id="ext-2025-04-25-001",
    input="Daily summary",
    result="...",
    is_success=True,
)

Reporting a failure

try:
    result = run_llm_locally(prompt)
except Exception as e:
    await backend.areport_external_task(
        agent_id="agent-123",
        id="job-failed-001",
        input=prompt,
        result=str(e),
        is_success=False,
    )
    raise

is_success=False marks the task as failed in the dashboard.

Sync version

backend.report_external_task(
    agent_id="agent-123",
    id="ext-001",
    input="...",
    result="...",
)

Same signature; blocks until the report is persisted.

​Constructor

​Methods

​Typical pattern

​Environment variable shortcut

​get_args

​Parameters

​Returns dict[str, Any]

​Examples

​Inside @on_task

​Pre-loaded agent (avoid re-fetching)

​Override resolved kwargs

​Add extra skills

​Authentication events

​Per-call callback

​Globally with @on_auth_event

​Sync version

​invoke_agent

​Parameters

​Returns Task

​Examples

​Attach files

​Structured output

​Stream events

​Continue an existing task

​Per-end-user context

​Sync version

​report_external_task

​Parameters

​Returns Task

​Examples

​Idempotent updates

​Using a pre-loaded agent

​Reporting a failure

​Sync version

Constructor

Methods

Typical pattern

Environment variable shortcut

get_args

Parameters

Returns `dict[str, Any]`

Examples

Inside `@on_task`

Pre-loaded agent (avoid re-fetching)

Override resolved kwargs

Add extra skills

Authentication events

Per-call callback

Globally with `@on_auth_event`

Sync version

invoke_agent

Parameters

Returns `Task`

Examples

Attach files

Structured output

Stream events

Continue an existing task

Per-end-user context

Sync version

report_external_task

Parameters

Returns `Task`

Examples

Idempotent updates

Using a pre-loaded agent

Reporting a failure

Sync version