> ## 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.
# Tasks
> Create, list, update, stop, and stream task executions.
The `Tasks` module manages the full lifecycle of task executions: the unit of work an agent does in response to a prompt. List existing tasks, load a specific one, create new ones, patch their state, or terminate them.
```python theme={"dark"}
from xpander_sdk import Tasks, AgentExecutionStatus
tasks = Tasks()
# List
items = await tasks.alist(agent_id="agent-123")
# Load
task = await tasks.aget(task_id="task_xyz")
# Create
task = await tasks.acreate(agent_id="agent-123", prompt="Hello!")
# Update
await tasks.aupdate(task_id=task.id, status=AgentExecutionStatus.Completed, result="done")
# Stop
await tasks.astop(task_id=task.id)
```
## Constructor
```python theme={"dark"}
Tasks(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/tasks#list) | `list[TasksListItem]` | Tasks for a given agent, optionally filtered. |
| [`alist_user_tasks` / `list_user_tasks`](/developers/sdk-reference/tasks#tasks-for-a-user) | `list[TasksListItem]` | Tasks for a given end-user across all agents. |
| [`aget` / `get`](/developers/sdk-reference/tasks#get) | `Task` | Load a full task by id. |
| [`acreate` / `create`](/developers/sdk-reference/tasks#create) | `Task` | Create a new task. (Same as `Agent.acreate_task`, but accepts `agent_id` directly.) |
| [`aupdate` / `update`](/developers/sdk-reference/tasks#update) | `Task` | Patch task fields (status, result, payload extension). |
| [`astop` / `stop`](/developers/sdk-reference/tasks#stop) | `Task` | Terminate a running task. |
## `Task` instance methods
The `Task` object returned by these calls has its own set of methods for runtime control:
| Method | What it does |
| ---------------------------------------------------------------------------------------------------------------------- | --------------------------------------------------------- |
| [`asave` / `save`](/developers/sdk-reference/tasks#save) | Persist local mutations back to the platform. |
| [`aset_status` / `set_status`](/developers/sdk-reference/tasks#set_status) | Set status (and optionally result) and save in one call. |
| [`astop` / `stop`](/developers/sdk-reference/tasks#stop) | Terminate this task. |
| [`areload` / `reload`](/developers/sdk-reference/tasks#reload) | Re-fetch the latest task state from the platform. |
| [`aevents` / `events`](/developers/sdk-reference/tasks#events) | Stream `TaskUpdateEvent`s via SSE. |
| [`aget_activity_log` / `get_activity_log`](/developers/sdk-reference/tasks#get_activity_log) | Full message / tool-call / reasoning thread. |
| [`areport_metrics` / `report_metrics`](/developers/sdk-reference/tasks#report_metrics) | Push token counts and other metrics back to the platform. |
| [`get_files` / `get_images` / `get_human_readable_files` / `to_message`](/developers/sdk-reference/tasks#agno-helpers) | Convenience helpers for Agno integration. |
For the full attribute list, see the [`Task` class reference](/developers/sdk-reference/tasks#task-class).
## Lifecycle
```text theme={"dark"}
Pending → Executing → Completed
↓ ↓
Paused Failed
↓ ↓
Error Stopped
```
| Status | Meaning |
| ----------- | ----------------------------------------------------------- |
| `Pending` | Created but not yet picked up by a worker. |
| `Executing` | Worker is running the task. |
| `Paused` | Worker paused (HITL approval, deep-planning question). |
| `Completed` | Finished successfully. |
| `Error` | Worker raised an exception. |
| `Failed` | Task didn't complete successfully (lower-level than Error). |
| `Stopped` | Manually terminated. |
The `AgentExecutionStatus` enum (used everywhere status appears) has these values: `Pending`, `Executing`, `Paused`, `Error`, `Failed`, `Completed`, `Stopped`. String values are lowercase.
## Class-level method
`Task.areport_external_task(...)` is a `@classmethod` that records an externally-executed run without going through `Backend`. It mirrors `Backend.areport_external_task` and exists for cases where you don't have a `Backend` instance handy. See [`report_external_task`](/developers/sdk-reference/tasks#report_external_task).
## create
`Tasks.acreate` creates a new task. It's equivalent to `Agent.acreate_task` but accepts `agent_id` directly, so you don't need to load the `Agent` first. Use this when you only have an agent ID and want to skip the extra round-trip.
```python theme={"dark"}
from xpander_sdk import Tasks
tasks = Tasks()
task = await tasks.acreate(
agent_id="agent-123",
prompt="Summarize the latest sales report",
file_urls=["https://example.com/sales.csv"],
)
print(task.id, task.status.value)
```
#### Parameters
Identical to [`Agent.acreate_task`](/developers/sdk-reference/agents#create_task), with `agent_id` required up front:
| Parameter | Type | Required | Default | Description |
| ------------------------------ | ------------------------ | -------- | --------- | --------------------------------------------------- |
| `agent_id` | `str` | Yes | – | Agent to invoke. |
| `existing_task_id` | `str` | No | `None` | Continue an existing task. |
| `prompt` | `str` | No | `""` | Natural-language input. |
| `file_urls` | `list[str]` | No | `[]` | URLs of files to attach. |
| `user_details` | `User` | No | `None` | End-user context. |
| `agent_version` | `str` | No | `None` | Pin a specific agent version. |
| `tool_call_payload_extension` | `dict` | No | `None` | Extra fields merged into every tool-call payload. |
| `source` | `str` | No | `None` | Origin tag. |
| `worker_id` | `str` | No | `None` | Pin to a specific worker. |
| `run_locally` | `bool` | No | `False` | Mark as locally executed. |
| `output_format` | `OutputFormat` | No | `None` | Output format. |
| `output_schema` | `dict` | No | `None` | JSON schema. |
| `events_streaming` | `bool` | No | `False` | Enables `task.aevents()`. |
| `additional_context` | `str` | No | `None` | Extra context appended to system prompt. |
| `instructions_override` | `str` | No | `None` | Extra instructions for this run. |
| `test_run_node_id` | `str` | No | `None` | (Internal) Test workflow node. |
| `user_oidc_token` | `str` | No | `None` | OIDC token for connector pre-auth. |
| `expected_output` | `str` | No | `None` | Expected output description. |
| `mcp_servers` | `list[MCPServerDetails]` | No | `[]` | Per-task MCP servers. |
| `triggering_agent_id` | `str` | No | `None` | Triggering agent ID. |
| `title` | `str` | No | `None` | Display title. |
| `think_mode` | `ThinkMode` | No | `Default` | `Default` or `Harder`. |
| `disable_attachment_injection` | `bool` | No | `False` | Skip auto-injection of human-readable file content. |
| `return_metrics` | `bool` | No | `False` | Return metrics (Workflow → Agent only). |
| `user_tokens` | `dict` | No | `None` | Pre-computed user tokens for MCP auth. |
See [`Agent.acreate_task`](/developers/sdk-reference/agents#create_task) for parameter details and examples.
#### Returns `Task`
The created task. Starts in `Pending`. See [`Task`](/developers/sdk-reference/tasks#task-class).
### When to use this vs. `Agent.acreate_task`
| Use | When |
| ------------------------------------ | ---------------------------------------------------------------------------------------------------------------- |
| `Tasks().acreate(agent_id=..., ...)` | You only have an `agent_id` (e.g. from a webhook or env var). One fewer round-trip than loading the agent first. |
| `Agent.acreate_task(...)` | You already have the loaded `Agent` (e.g. inspecting its tools or graph first). |
Both produce identical results; pick the one that matches what you have on hand.
### Sync version
```python theme={"dark"}
task = tasks.create(agent_id="agent-123", prompt="Hello!")
```
### Errors
Raises [`ModuleException`](/developers/sdk-reference/overview#error-handling). Common statuses:
| Status | Cause |
| --------- | ----------------------------------------------------------------- |
| 400 | Invalid input (bad output schema, malformed `mcp_servers`, etc.). |
| 401 / 403 | Auth failure. |
| 404 | Agent doesn't exist. |
| 500 | Server error. |
## get
`Tasks.aget` loads a full `Task` object: input, status, result, token usage, deep-planning state, attached files, and everything else stored against the task record.
```python theme={"dark"}
from xpander_sdk import Tasks
tasks = Tasks()
task = await tasks.aget(task_id="task_xyz")
print(task.status.value, task.result)
```
#### Parameters
| Parameter | Type | Required | Description |
| --------- | ----- | -------- | ----------- |
| `task_id` | `str` | Yes | Task ID. |
#### Returns `Task`
A full `Task`. See the [`Task` class reference](/developers/sdk-reference/tasks#task-class) for attributes.
### Examples
#### Wait for completion
```python theme={"dark"}
import asyncio
from xpander_sdk import AgentExecutionStatus
terminal = {
AgentExecutionStatus.Completed,
AgentExecutionStatus.Failed,
AgentExecutionStatus.Error,
AgentExecutionStatus.Stopped,
}
while True:
task = await tasks.aget(task_id=task.id)
if task.status in terminal:
break
await asyncio.sleep(2)
print(task.result)
```
For finer-grained progress, use `task.aevents()` instead: see [streaming events](/developers/sdk-reference/tasks#events).
#### Reload an existing instance
If you already have a `Task` instance and want fresh data, prefer `task.areload()` over a fresh `aget`: it preserves any in-flight `deep_planning` state that the API might briefly return as empty:
```python theme={"dark"}
await task.areload()
```
#### Sync version
```python theme={"dark"}
task = tasks.get(task_id="task_xyz")
```
### Errors
Raises [`ModuleException`](/developers/sdk-reference/overview#error-handling) on failure. Common statuses:
| Status | Cause |
| ------ | ------------------------------------------- |
| 404 | Task doesn't exist (or wrong organization). |
| 403 | Task exists but isn't accessible. |
| 500 | Server error. |
## list
`Tasks.alist` returns task summaries for a given agent. `Tasks.alist_user_tasks` does the same scoped to a single end-user across all agents.
```python theme={"dark"}
from xpander_sdk import Tasks
tasks = Tasks()
items = await tasks.alist(agent_id="agent-123")
for item in items:
print(item.id, item.status.value, item.title or "(no title)")
```
#### Parameters
| Parameter | Type | Required | Default | Description |
| ---------- | ------ | -------- | ------- | --------------------------------------------------------------------------------------------------------------- |
| `agent_id` | `str` | Yes | – | Agent whose tasks to list. |
| `filters` | `dict` | No | `None` | Query filters. Supported keys: `user_id`, `parent_task_id`, `triggering_agent_id`, `status`, `internal_status`. |
#### Returns `list[TasksListItem]`
| Field | Type | Description |
| --------------------- | ---------------------- | ---------------------------------------------- |
| `id` | `str` | Task ID. |
| `agent_id` | `str` | Owning agent. |
| `user_id` | `str \| None` | End-user the task was created for. |
| `parent_task_id` | `str \| None` | If this is a sub-task. |
| `triggering_agent_id` | `str \| None` | If this task was kicked off by another agent. |
| `organization_id` | `str` | Owning org. |
| `status` | `AgentExecutionStatus` | Current status. |
| `created_at` | `datetime \| None` | Creation timestamp. |
| `updated_at` | `datetime \| None` | Last-update timestamp. |
| `source_node_type` | `str \| None` | Trigger origin (`webhook`, `slack`, `sdk`, …). |
| `result` | `str \| None` | Final result if completed. |
| `title` | `str \| None` | Task title. |
To load the full `Task` (including input, files, deep-planning state, tokens, activity log access), call `.aload()` on the item or pass `item.id` to `tasks.aget()`.
### Examples
#### Filter by status
```python theme={"dark"}
from xpander_sdk import AgentExecutionStatus
failed = await tasks.alist(
agent_id="agent-123",
filters={"status": AgentExecutionStatus.Failed.value},
)
```
The `status` filter accepts the lowercase string value: `"pending"`, `"executing"`, `"completed"`, `"failed"`, etc.
#### Filter by end-user
```python theme={"dark"}
mine = await tasks.alist(
agent_id="agent-123",
filters={"user_id": "user_42"},
)
```
#### Tasks for a user
`alist_user_tasks` searches across every agent the user has interacted with:
```python theme={"dark"}
items = await tasks.alist_user_tasks(user_id="user_42")
```
| Parameter | Type | Required | Description |
| --------- | ------ | -------- | ------------------------------------------- |
| `user_id` | `str` | Yes | End-user identifier. |
| `filters` | `dict` | No | Same filter keys as above, minus `user_id`. |
#### Sync versions
```python theme={"dark"}
items = tasks.list(agent_id="agent-123")
items = tasks.list_user_tasks(user_id="user_42")
```
### Errors
Raises [`ModuleException`](/developers/sdk-reference/overview#error-handling) on failure. Common statuses:
| Status | Cause |
| --------- | --------------------------------------------- |
| 401 / 403 | Auth failure or wrong organization. |
| 404 | Agent doesn't exist (or you can't access it). |
| 500 | Server error. |
## update
`Tasks.aupdate` patches selected fields on a task. Pass only the fields you want to change: anything left as `None` is ignored.
For finer control over save semantics (especially when mutating `Task` attributes locally first), use `task.asave()` on the instance.
```python theme={"dark"}
from xpander_sdk import Tasks, AgentExecutionStatus
tasks = Tasks()
updated = await tasks.aupdate(
task_id="task_xyz",
status=AgentExecutionStatus.Completed,
result="Final summary: …",
)
print(updated.status.value)
```
#### Parameters
| Parameter | Type | Required | Description |
| ----------------------------- | ---------------------- | -------- | --------------------------------------------------------------- |
| `task_id` | `str` | Yes | Task to update. |
| `tool_call_payload_extension` | `dict` | No | Extra fields merged into every tool-call payload going forward. |
| `source` | `str` | No | Update the origin tag. |
| `status` | `AgentExecutionStatus` | No | New status. |
| `last_executed_node_id` | `str` | No | Move the task's "current node" cursor (workflow agents). |
| `result` | `str` | No | Final result string. |
Fields with value `None` are stripped from the request: only what you pass is sent.
#### Returns `Task`
The freshly-loaded task with updates applied. See [`Task`](/developers/sdk-reference/tasks#task-class).
### Examples
#### Mark complete
```python theme={"dark"}
await tasks.aupdate(
task_id="task_xyz",
status=AgentExecutionStatus.Completed,
result="Done",
)
```
#### Mark errored
```python theme={"dark"}
await tasks.aupdate(
task_id="task_xyz",
status=AgentExecutionStatus.Error,
result="Connector failed: 503 Service Unavailable",
)
```
#### Update only payload extension
```python theme={"dark"}
await tasks.aupdate(
task_id="task_xyz",
tool_call_payload_extension={"headers": {"X-Trace-ID": "abc-123"}},
)
```
This is rarely needed: payload extensions are usually set at task creation. Use this when you need to inject context after the task starts.
### Sync version
```python theme={"dark"}
tasks.update(task_id="task_xyz", status=AgentExecutionStatus.Completed)
```
### When to use this vs. `task.asave()`
| Use | When |
| ----------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `Tasks().aupdate(task_id=..., field=...)` | You only need to set a couple of fields and don't have a `Task` instance loaded. |
| `task.asave()` | You've mutated the `Task` instance directly (e.g. set `task.result = "..."`). `asave` PATCHes everything except `configuration` and (by default) `deep_planning`. |
The `@on_task` runtime auto-saves the task at the end of the handler: you don't usually call `aupdate` or `asave` yourself inside a handler.
### Errors
Raises [`ModuleException`](/developers/sdk-reference/overview#error-handling). Common statuses:
| Status | Cause |
| ------ | --------------------------------------------- |
| 404 | Task doesn't exist. |
| 409 | Concurrent update conflict. Reload and retry. |
| 500 | Server error. |
## stop
`Tasks.astop` cancels a running task. The platform marks it `Stopped`, signals the worker to abandon its current step, and returns the updated record.
```python theme={"dark"}
from xpander_sdk import Tasks
tasks = Tasks()
stopped = await tasks.astop(task_id="task_xyz")
print(stopped.status.value, stopped.is_manually_stopped) # "stopped" True
```
#### Parameters
| Parameter | Type | Required | Description |
| --------- | ----- | -------- | ------------- |
| `task_id` | `str` | Yes | Task to stop. |
#### Returns `Task`
The task record after the stop request was processed. `status` is `Stopped`, `is_manually_stopped` is `True`, and `finished_at` is set.
### Examples
#### Stop from a running `Task` instance
If you already have the `Task` object, prefer the instance method:
```python theme={"dark"}
await task.astop()
print(task.status.value) # "stopped"
```
`Task.astop()` and `Tasks().astop(task_id=task.id)` are equivalent: the instance method also updates the in-memory object.
#### Idempotent stop
Calling `astop` on an already-terminal task is safe: the platform returns the existing record without changing state.
```python theme={"dark"}
# First call: stops it
await tasks.astop(task_id="task_xyz")
# Second call: no-op, returns the same record
await tasks.astop(task_id="task_xyz")
```
### Sync version
```python theme={"dark"}
tasks.stop(task_id="task_xyz")
```
### Errors
Raises [`ModuleException`](/developers/sdk-reference/overview#error-handling). Common statuses:
| Status | Cause |
| ------ | ------------------- |
| 404 | Task doesn't exist. |
| 500 | Server error. |
## events
`Task.aevents()` is an async generator over `TaskUpdateEvent`s emitted by the platform as the task runs: message chunks, tool-call requests/results, reasoning steps, sub-agent triggers, plan updates, and auth events. Use it to build live UIs or to react to specific event types in real time.
```python theme={"dark"}
from xpander_sdk import TaskUpdateEventType
# Task must be created with events_streaming=True
task = await agent.acreate_task(
prompt="Long-running research task",
events_streaming=True,
)
async for event in task.aevents():
print(event.type, event.task_id)
if event.type == TaskUpdateEventType.TaskFinished:
break
```
The task **must** be created with `events_streaming=True`. Calling `aevents()` on a non-streaming task raises `ValueError` immediately.
#### Parameters
None.
#### Yields `TaskUpdateEvent`
Each event has:
| Field | Type | Description |
| ----------------- | --------------------- | ------------------------------ |
| `type` | `TaskUpdateEventType` | What kind of event. See below. |
| `task_id` | `str` | Task this event belongs to. |
| `organization_id` | `str` | Owning organization. |
| `time` | `datetime` | Event timestamp. |
| `data` | `Any` | Event-specific payload. |
##### Event types
| `TaskUpdateEventType` | When | `data` payload |
| --------------------- | ------------------------------------------------------------ | ----------------------------------------------------------------------- |
| `TaskCreated` | Task is created. | A `Task` object. |
| `TaskUpdated` | Any non-final task mutation (status change, result patched). | A `Task` object. |
| `TaskFinished` | Task reaches a terminal state. | A `Task` object. |
| `Chunk` | Streaming text from the LLM. | `str` (text chunk). |
| `AuthEvent` | Auth required (MCP OAuth, ECA). | Auth-specific dict. Also dispatched to `@on_auth_event` handlers. |
| `ToolCallRequest` | Agent invokes a tool. | A `ToolCallRequest` (operation\_id, tool\_name, payload, reasoning, …). |
| `ToolCallResult` | Tool finishes. | A `ToolCallResult` (operation\_id, result, is\_error, …). |
| `SubAgentTrigger` | A sub-agent task is created. | Triggering details. |
| `Think` | Reasoning step (think tool). | Reasoning data. |
| `Analyze` | Reasoning step (analyze tool). | Reasoning data. |
| `PlanUpdated` | Deep-planning state changes. | A `DeepPlanning` object. |
| `TaskCompactization` | Auto-compaction event (Layer 2). | A `TaskCompactizationEvent`. |
If a typed payload fails to parse (e.g. due to schema drift), the SDK still yields the event with `data` set to the raw dict: the envelope (`type`, `task_id`, `time`) always propagates.
### Examples
#### Filter on event type
```python theme={"dark"}
async for event in task.aevents():
if event.type == TaskUpdateEventType.Chunk:
print(event.data, end="", flush=True)
elif event.type == TaskUpdateEventType.TaskFinished:
print()
break
```
`Chunk` events have a string `data` payload (the streaming text), making it easy to render token-by-token output.
#### Render tool calls
```python theme={"dark"}
async for event in task.aevents():
if event.type == TaskUpdateEventType.ToolCallRequest:
req = event.data
print(f"→ {req.tool_name}({req.payload})")
elif event.type == TaskUpdateEventType.ToolCallResult:
res = event.data
status = "✗" if res.is_error else "✓"
print(f"{status} {res.tool_name}: {res.result}")
```
#### Stop on first failure
```python theme={"dark"}
async for event in task.aevents():
if event.type == TaskUpdateEventType.ToolCallResult and event.data.is_error:
print("Tool failed, stopping.")
await task.astop()
break
```
#### Track auth flows in-band
```python theme={"dark"}
async for event in task.aevents():
if event.type == TaskUpdateEventType.AuthEvent:
print(f"Auth required: {event.data}")
# Show URL to user, etc.
```
If you've registered an `@on_auth_event` handler, it fires automatically: the in-band event lets you also handle it inline.
### Sync version
```python theme={"dark"}
for event in task.events():
print(event.type, event.data)
```
`task.events()` consumes the async generator on a synchronous executor and yields events in order. Don't call from inside a running event loop.
### Lifecycle
The generator reconnects automatically if the SSE stream drops mid-task. It exits when:
1. The platform closes the stream (typically after `TaskFinished`).
2. The connection fails permanently (network, auth).
3. You `break` out of the loop.
### Errors
* `ValueError`: task wasn't created with `events_streaming=True`.
* Network failures propagate after the SSE retry budget is exhausted.
### When *not* to stream
For batch jobs that don't care about progress, skip `events_streaming` and poll `task.areload()` (or use webhooks). Streaming holds an HTTP connection open per task: fine for one or two, expensive for hundreds.
## get\_activity\_log
`Task.aget_activity_log` returns the full thread of activity for a task: user messages, assistant messages, tool calls, tool results, reasoning steps, sub-agent triggers, and auth events. Unlike `task.aevents()` (live SSE for in-flight tasks), the activity log is a historical record you can fetch at any time, including for completed tasks.
```python theme={"dark"}
from xpander_sdk import Tasks
from xpander_sdk.models.activity import (
AgentActivityThreadMessage,
AgentActivityThreadToolCall,
AgentActivityThreadReasoning,
)
task = await Tasks().aget(task_id="task_xyz")
log = await task.aget_activity_log()
for msg in log.messages:
if isinstance(msg, AgentActivityThreadMessage):
print(f"[{msg.role}] {msg.content.text}")
elif isinstance(msg, AgentActivityThreadToolCall):
print(f"[tool] {msg.tool_name}({msg.payload}) → {msg.result}")
elif isinstance(msg, AgentActivityThreadReasoning):
print(f"[reasoning:{msg.type}] {msg.thought}")
```
#### Parameters
None.
#### Returns `AgentActivityThread`
| Field | Type | Description |
| -------------- | ---------------------------- | ----------------------------------------------------- |
| `messages` | `list[AgentActivityThread*]` | Ordered union of message types (see below). |
| (other fields) | – | Thread metadata (id, agent\_id, organization\_id, …). |
The messages list is a tagged union. Common variants:
| Type | Fields |
| ------------------------------ | ---------------------------------------------------------- |
| `AgentActivityThreadMessage` | `role`, `content.text` (string), `content` (rich content). |
| `AgentActivityThreadToolCall` | `tool_name`, `payload`, `result`. |
| `AgentActivityThreadReasoning` | `type` (`"think"` / `"analyze"`), `thought`. |
(There are additional sub-types for sub-agent triggers and auth events: branch on `isinstance` or `type(msg).__name__` to handle them.)
### Examples
#### Just the user/assistant transcript
```python theme={"dark"}
log = await task.aget_activity_log()
for msg in log.messages:
if isinstance(msg, AgentActivityThreadMessage):
print(f"{msg.role.upper()}: {msg.content.text}")
```
#### Audit tool usage
```python theme={"dark"}
log = await task.aget_activity_log()
tool_calls = [m for m in log.messages if isinstance(m, AgentActivityThreadToolCall)]
print(f"Task {task.id} called {len(tool_calls)} tools:")
for tc in tool_calls:
print(f" - {tc.tool_name}")
```
#### Reasoning trace
```python theme={"dark"}
reasoning = [
m for m in log.messages
if isinstance(m, AgentActivityThreadReasoning)
]
for r in reasoning:
print(f"[{r.type}] {r.thought}")
```
### Sync version
```python theme={"dark"}
log = task.get_activity_log()
```
### Errors
Raises [`ModuleException`](/developers/sdk-reference/overview#error-handling) if the activity log can't be retrieved:
| Status | Cause |
| ------ | ------------------------------------------------- |
| 404 | Task has no activity log (e.g. it never started). |
| 500 | Server error or network failure. |
## report\_metrics
`Task.areport_metrics` records the task's token usage, tool calls, and execution outcome to the platform's metrics store. The `@on_task` runtime calls this automatically at the end of every handler if `task.tokens` is set, so you only need to invoke it directly when you're driving a task outside the runtime.
```python theme={"dark"}
from xpander_sdk import Tokens
task.tokens = Tokens(prompt_tokens=2_140, completion_tokens=380)
task.used_tools = ["web_search", "fetch_pdf"]
task.duration = 4.7
await task.areport_metrics()
```
`task.tokens` must be set before calling `areport_metrics`. The method raises `ValueError` if `tokens` is missing.
#### Parameters
| Parameter | Type | Required | Description |
| --------------- | --------------- | -------- | ----------------------------------------------------------------------------------- |
| `configuration` | `Configuration` | No | Override SDK config. Falls back to `task.configuration`, then env-derived defaults. |
#### Returns
`None`.
### What gets reported
The metrics report includes:
* `execution_id`: task id.
* `source`: task source.
* `memory_thread_id`: same as task id (memory and execution share an id).
* `task`: input text (`task.input.text`).
* `status`: current `task.status`.
* `internal_status`: current `task.internal_status`.
* `ai_model`: fixed to `"xpander"`.
* `api_calls_made`: tool names from `task.used_tools` (or `[]` for orchestration tasks with `return_metrics=True`).
* `result`: final result string.
* `llm_tokens`: token counts wrapped in `ExecutionTokens(worker=task.tokens)`.
### Examples
#### Inside a custom event loop
If you're driving an agent without `@on_task`, report metrics manually before returning to the caller:
```python theme={"dark"}
from xpander_sdk import Backend, Tokens
from xpander_sdk.modules.tasks.sub_modules.task import Task
backend = Backend()
task = await backend.ainvoke_agent(agent_id="agent-123", prompt="...")
# ... run the agent yourself, track tokens ...
task.tokens = Tokens(prompt_tokens=1_500, completion_tokens=320)
task.used_tools = ["search", "summarize"]
task.duration = 3.1
await task.areport_metrics()
```
This makes the run show up in the dashboard with proper token accounting.
#### Sync version
```python theme={"dark"}
task.report_metrics()
```
### Errors
Raises [`ModuleException`](/developers/sdk-reference/overview#error-handling) on persistence failures, or `ValueError` if `task.tokens` is `None`.
## report\_external\_task
`Task.areport_external_task` is the `@classmethod` version of [`Backend.areport_external_task`](/developers/sdk-reference/backend#report_external_task). It logs a task that was executed outside the xpander.ai runtime so it appears in task history and metrics.
Use this form when you don't have a `Backend` instance in scope but do have a `Configuration`.
```python theme={"dark"}
from xpander_sdk import Task, Tokens, Configuration
config = Configuration() # picks up env vars
task = await Task.areport_external_task(
configuration=config,
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,
)
```
#### Parameters
| Parameter | Type | Required | Default | Description |
| --------------- | --------------- | -------- | --------------------- | ------------------------------------------------------------------- |
| `configuration` | `Configuration` | No | New `Configuration()` | SDK config (api\_key + organization\_id). |
| `agent_id` | `str` | Yes | – | Agent the run belongs to. |
| `id` | `str` | No | `None` | External task id. Re-using the same id updates the existing record. |
| `input` | `str` | No | `None` | Run input. |
| `llm_response` | `Any` | No | `None` | Raw provider response. Stored verbatim. |
| `tokens` | `Tokens` | No | `None` | Token usage. |
| `is_success` | `bool` | No | `True` | Pass/fail. |
| `result` | `str` | No | `None` | Final result. |
| `duration` | `float` | No | `0` | Wall-clock seconds. |
| `used_tools` | `list[str]` | No | `[]` | Tool names. |
#### Returns `Task`
The platform-side `Task` after persistence. Carries the `id`, `status`, and timestamps assigned by the cloud.
### When to use this vs. `Backend.areport_external_task`
Functionally identical. `Backend.areport_external_task` is more discoverable (it sits next to the other Backend methods you'd be using); this classmethod is convenient when you don't want to instantiate `Backend`.
```python theme={"dark"}
# Equivalent ways to record an external run:
# 1. Backend
backend = Backend(configuration=config)
await backend.areport_external_task(agent_id="agent-123", id="ext-001", ...)
# 2. Task classmethod
await Task.areport_external_task(configuration=config, agent_id="agent-123", id="ext-001", ...)
```
See the [`Backend` page](/developers/sdk-reference/backend#report_external_task) for typical usage patterns (idempotent updates, failure reporting).
### Sync version
```python theme={"dark"}
Task.report_external_task(configuration=config, agent_id="agent-123", id="ext-001", ...)
```
## Task class
`Task` represents a single agent execution. You get one back from `Agent.acreate_task`, `Tasks.aget`, `Tasks.acreate`, and the `@on_task` handler. It carries the full task state (input, status, result, deep-planning, tokens, …) and the methods needed to drive it forward.
```python theme={"dark"}
from xpander_sdk import Tasks
task = await Tasks().aget(task_id="task_xyz")
print(task.id, task.status.value, task.result)
```
### Class methods
#### `Task.aload(task_id, configuration=None) -> Task`
Same as `Tasks.aget` but available without a `Tasks` instance. Useful when you have a `Configuration` but don't want to instantiate the module class.
```python theme={"dark"}
from xpander_sdk import Task, Configuration
config = Configuration(api_key="...", organization_id="...")
task = await Task.aload(task_id="task_xyz", configuration=config)
```
The sync sibling is `Task.load(...)`.
#### `Task.areport_external_task(...)` (classmethod)
See the dedicated [`report_external_task`](/developers/sdk-reference/tasks#report_external_task) page.
### Attributes
#### Identity
| Attribute | Type | Description |
| --------------------- | ------------- | ------------------------------------------------ |
| `id` | `str` | Task ID. |
| `agent_id` | `str` | Owning agent. |
| `organization_id` | `str` | Owning organization. |
| `agent_version` | `str \| None` | Agent version pinned for this run. |
| `triggering_agent_id` | `str \| None` | Agent that triggered this (sub-agent flows). |
| `parent_execution` | `str \| None` | Parent task ID for sub-tasks. |
| `sub_executions` | `list[str]` | Child task IDs. |
| `source` | `str \| None` | Origin tag (`"sdk"`, `"webhook"`, `"slack"`, …). |
| `title` | `str \| None` | Display title. |
#### Input
| Attribute | Type | Description |
| ------------------------------ | ------------------------ | ----------------------------------------------------------- |
| `input` | `AgentExecutionInput` | Holds `text`, `files: list[str]`, and `user: User \| None`. |
| `payload_extension` | `dict \| None` | Extra fields merged into every tool-call payload. |
| `additional_context` | `str \| None` | Extra context appended to the system prompt. |
| `instructions_override` | `str \| None` | Extra instructions for this run. |
| `expected_output` | `str \| None` | Expected output description. |
| `mcp_servers` | `list[MCPServerDetails]` | Per-task MCP servers. |
| `user_oidc_token` | `str \| None` | OIDC token for connector pre-auth. |
| `user_tokens` | `dict \| None` | Pre-computed user tokens for MCP auth. |
| `disable_attachment_injection` | `bool` | Skip auto-inlining of human-readable file content. |
| `output_format` | `OutputFormat \| None` | Output format. |
| `output_schema` | `dict \| None` | JSON schema for structured output. |
| `voice_id` | `str \| None` | Voice ID for `OutputFormat.Voice`. |
| `think_mode` | `ThinkMode` | `Default` or `Harder`. |
#### State
| Attribute | Type | Description |
| ----------------------- | ------------------------------- | ----------------------------------------------------------------------------------- |
| `status` | `AgentExecutionStatus` | One of `Pending`, `Executing`, `Paused`, `Error`, `Failed`, `Completed`, `Stopped`. |
| `internal_status` | `str \| None` | Finer-grained status used by the platform internally. |
| `last_executed_node_id` | `str \| None` | Cursor in the execution graph (workflow agents). |
| `is_manually_stopped` | `bool` | `True` after `astop`. |
| `is_orchestration` | `bool` | `True` for orchestration tasks. |
| `events_streaming` | `bool` | Whether SSE streaming is enabled. |
| `hitl_request` | `HumanInTheLoopRequest \| None` | Set when the task is paused awaiting human approval. |
| `pending_eca_request` | `PendingECARequest \| None` | Set when an ECA (External Credential Authorization) is pending. |
| `deep_planning` | `DeepPlanning` | Deep-planning state (plan items, completion flags). |
| `execution_attempts` | `int` | Number of plan-following attempts (1 for first execution). |
| `return_metrics` | `bool` | Whether the task returns metrics. |
#### Output
| Attribute | Type | Description |
| ------------ | ---------------- | ---------------------------------- |
| `result` | `str \| None` | Final result. |
| `tokens` | `Tokens \| None` | Token usage (prompt + completion). |
| `used_tools` | `list[str]` | Names of tools the task invoked. |
| `duration` | `float` | Wall-clock duration in seconds. |
#### Timestamps
| Attribute | Type | Description |
| ------------- | ------------------ | --------------------------------- |
| `created_at` | `datetime` | Creation timestamp. |
| `started_at` | `datetime \| None` | When the worker picked it up. |
| `paused_at` | `datetime \| None` | When it last paused. |
| `finished_at` | `datetime \| None` | When it reached a terminal state. |
#### Internal
| Attribute | Type | Description |
| ------------------ | ----------------------- | ----------------------------------------------- |
| `configuration` | `Configuration \| None` | SDK configuration. Excluded from serialization. |
| `test_run_node_id` | `str \| None` | (Internal) Test workflow node id. |
### Instance methods
#### `save`
`task.asave(with_deep_plan_update: bool = False)`: PATCHes the task on the platform with the current in-memory state. By default `deep_planning` is excluded (the platform's plan state is authoritative); pass `with_deep_plan_update=True` to push your plan updates back.
```python theme={"dark"}
task.result = "Done"
task.status = AgentExecutionStatus.Completed
await task.asave()
```
After saving, the response repopulates the local instance, but `deep_planning` and `additional_context` are restored from local memory if the API returns empty values (defensive against stale API state during retries). Sync: `task.save(...)`.
#### `set_status`
`task.aset_status(status, result=None)`: convenience wrapper that sets `status` (and optionally `result`) and calls `asave()` in one step.
```python theme={"dark"}
await task.aset_status(AgentExecutionStatus.Executing)
```
Sync: `task.set_status(...)`.
#### `stop`
`task.astop()`: terminates the task and updates the local instance with the platform's response. Sync: `task.stop()`.
#### `reload`
`task.areload()`: re-fetches the task from the platform and updates the in-memory instance. `deep_planning` and `additional_context` are restored from local memory if the API returns empty values. Sync: `task.reload()`.
#### `events`
`task.aevents()` / `task.events()`: SSE stream of `TaskUpdateEvent`. Requires `events_streaming=True` at task creation. See [streaming events](/developers/sdk-reference/tasks#events).
#### `get_activity_log`
`task.aget_activity_log()` / `task.get_activity_log()`: full message / tool-call / reasoning thread for the task. See [activity log](/developers/sdk-reference/tasks#get_activity_log).
#### `report_metrics`
`task.areport_metrics()` / `task.report_metrics()`: pushes token counts and tool usage to the platform. See [report metrics](/developers/sdk-reference/tasks#report_metrics).
#### `get_plan_following_status`
`task.aget_plan_following_status() -> PlanFollowingStatus`: checks whether deep-planning tasks are complete. Returns `PlanFollowingStatus(can_finish: bool, uncompleted_tasks: list[DeepPlanningItem])`. Used by the runtime to retry the agent until the plan is satisfied. Sync: `task.get_plan_following_status()`.
#### `acompact_session_for_retry`
`task.acompact_session_for_retry() -> CompactRetryResult`: forces L2 (auto) compaction on the agent's session before a plan-following retry. Used internally by the runtime; rarely needed from user code.
#### Agno helpers
These methods help shape the task input for Agno agents.
##### `get_files`
```python theme={"dark"}
task.get_files() -> list[Any]
```
Returns PDF files from `task.input.files` as Agno `File` objects (when Agno is installed). Returns the URL strings as a fallback. Empty list if the task has no PDFs.
##### `get_images`
```python theme={"dark"}
task.get_images() -> list[Any]
```
Returns image files as Agno `Image` objects (or URL strings as fallback). Empty list if no images.
##### `get_human_readable_files`
```python theme={"dark"}
task.get_human_readable_files() -> list[dict[str, str]]
```
Fetches and parses text-based files (CSV, JSON, TXT, .py, …) and returns `[{"url": ..., "content": ...}]`. Used by `to_message()` to inline file contents into the prompt. Honors `task.disable_attachment_injection`.
##### `to_message`
```python theme={"dark"}
task.to_message(retry_count: int = 0) -> str
```
Builds the message string to pass to the framework agent. Format:
```
{input.text}
Files: url1, url2, ...
Files contents:
{json of each readable file}
```
On retries (`retry_count > 0`) and when deep planning is active, it generates a continuation prompt that lists completed/uncompleted tasks and pinpoints the next one to work on. This is the prompt format the runtime feeds to Agno when retrying after an incomplete plan.
```python theme={"dark"}
@on_task
async def handler(task: Task) -> Task:
args = await backend.aget_args(agent_id=task.agent_id, task=task)
agno_agent = AgnoAgent(**args)
result = await agno_agent.arun(
input=task.to_message(), # text + files + readable contents
files=task.get_files(), # PDFs as Agno File
images=task.get_images(), # images as Agno Image
)
task.result = result.content
return task
```