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

# Backend

> The Backend module bridges your code and a stored agent configuration.

The `Backend` module is the bridge between your code and an agent stored in the xpander.ai platform. It does three things:

1. **Resolves runtime arguments** for your AI framework: `aget_args(agent_id=..., task=task)` returns kwargs you can spread directly into Agno (or another supported framework's) agent constructor.
2. **Invokes agents directly**: `ainvoke_agent(agent_id=..., prompt=...)` is shorthand for "load the agent, create a task." Useful when you don't need framework-level control.
3. **Reports externally executed runs**: `areport_external_task(...)` lets you log a run that happened outside the platform (e.g. in a queue worker or another runtime) so it shows up in the same task history and metrics as platform-managed runs.

```python theme={"dark"}
from xpander_sdk import Backend

backend = Backend()
```

## Constructor

```python theme={"dark"}
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`](/developers/sdk-reference/backend#invoke_agent)                         | Load an agent and create a task in one call. Returns a `Task`.                                                |
| [`aget_args` / `get_args`](/developers/sdk-reference/backend#get_args)                                     | Resolve framework-specific kwargs (model, system prompt, tools, memory, …) from a stored agent configuration. |
| [`areport_external_task` / `report_external_task`](/developers/sdk-reference/backend#report_external_task) | Record an external execution (run outside the platform) so it appears in task history.                        |

## Typical pattern

```python theme={"dark"}
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 platform; the handler resolves framework args for the *current* task, builds an Agno agent, runs it, and stores the result on the task. The platform records status, metrics, and tool calls automatically.

## Environment variable shortcut

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

```bash theme={"dark"}
export XPANDER_AGENT_ID="agent-123"
```

```python theme={"dark"}
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 framework-ready keyword arguments for a stored agent. Spread the result directly into an Agno (or other supported framework's) constructor: the dict already contains the model, system prompt, tools, memory settings, knowledge-base retrievers, and guardrails configured in the Workbench.

```python theme={"dark"}
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 tool list.                                                        |
| `is_async`             | `bool`           | No       | `True`                     | (async only) Whether you'll run the framework agent in an async context. Affects how local tools are wrapped. |
| `auth_events_callback` | `Callable`       | No       | `None`                     | Per-call callback for MCP/OAuth authentication events. See [Authentication events](#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 framework-ready kwargs. The exact keys depend on the agent's framework (currently always Agno):

```python theme={"dark"}
{
    "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 framework's constructor.

### Examples

#### Inside `@on_task`

```python theme={"dark"}
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)

```python theme={"dark"}
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 framework args.

#### Override resolved kwargs

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

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

#### Add extra tools

```python theme={"dark"}
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 tool list for this run only: it isn't persisted to the platform. For permanent tools, register with [`@register_tool`](/developers/sdk-reference/tools#register_tool-classmethod) instead.

### Authentication events

When an agent uses MCP servers or connectors that require OAuth, the framework emits `auth_event` updates while the user authenticates. Register a callback to handle the OAuth URL or token-ready signals:

#### Per-call callback

```python theme={"dark"}
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`

```python theme={"dark"}
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`](/developers/sdk-reference/decorators#@on_auth_event) for details.

### Sync version

```python theme={"dark"}
args = backend.get_args(agent_id="agent-123", task=task)
```

Same parameters minus `is_async` (sync wrapper hard-codes it to `False` so local tools 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.

<Note>
  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 the platform passes task continuations through chat-style integrations.
</Note>

```python theme={"dark"}
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 tool-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 worker.                                                       |
| `run_locally`                 | `bool`         | No       | `False`                    | Mark the task as locally executed (skips cloud worker dispatch).                          |
| `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 frameworks for self-verification). |

#### Returns `Task`

The created `Task` object, populated with the platform-assigned `id` and an initial `status` of `pending`. See [`Task`](/developers/sdk-reference/tasks#task-class) for the full attribute list.

### Examples

#### Attach files

```python theme={"dark"}
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

```python theme={"dark"}
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

```python theme={"dark"}
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

```python theme={"dark"}
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

```python theme={"dark"}
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 connector authentication (e.g. OAuth tokens for Gmail or Calendar) to the right end user.

### Sync version

```python theme={"dark"}
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 worker, a Lambda, or a different process) so it shows up in the same task history, metrics, and observability views as platform-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.

```python theme={"dark"}
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 tools 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 platform-side `Task` record after persistence. Re-using `id` returns the updated record; new `id`s create a fresh task.

### Examples

#### Idempotent updates

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

```python theme={"dark"}
# 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

```python theme={"dark"}
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

```python theme={"dark"}
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

```python theme={"dark"}
backend.report_external_task(
    agent_id="agent-123",
    id="ext-001",
    input="...",
    result="...",
)
```

Same signature; blocks until the report is persisted.