Tasks - xpander.ai

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.

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

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`	`list[TasksListItem]`	Tasks for a given agent, optionally filtered.
`alist_user_tasks` / `list_user_tasks`	`list[TasksListItem]`	Tasks for a given end-user across all agents.
`aget` / `get`	`Task`	Load a full task by id.
`acreate` / `create`	`Task`	Create a new task. (Same as `Agent.acreate_task`, but accepts `agent_id` directly.)
`aupdate` / `update`	`Task`	Patch task fields (status, result, payload extension).
`astop` / `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`	Persist local mutations back to the platform.
`aset_status` / `set_status`	Set status (and optionally result) and save in one call.
`astop` / `stop`	Terminate this task.
`areload` / `reload`	Re-fetch the latest task state from the platform.
`aevents` / `events`	Stream `TaskUpdateEvent`s via SSE.
`aget_activity_log` / `get_activity_log`	Full message / tool-call / reasoning thread.
`areport_metrics` / `report_metrics`	Push token counts and other metrics back to the platform.
`get_files` / `get_images` / `get_human_readable_files` / `to_message`	Convenience helpers for Agno integration.

For the full attribute list, see the Task class reference.

Lifecycle

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.

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.

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, 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 for parameter details and examples.

Returns `Task`

The created task. Starts in Pending. See Task.

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

task = tasks.create(agent_id="agent-123", prompt="Hello!")

Errors

Raises ModuleException. 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.

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

Examples

Wait for completion

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.

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:

await task.areload()

Sync version

task = tasks.get(task_id="task_xyz")

Errors

Raises ModuleException 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.

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

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

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:

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

items = tasks.list(agent_id="agent-123")
items = tasks.list_user_tasks(user_id="user_42")

Errors

Raises ModuleException 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.

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.

Examples

Mark complete

await tasks.aupdate(
    task_id="task_xyz",
    status=AgentExecutionStatus.Completed,
    result="Done",
)

Mark errored

await tasks.aupdate(
    task_id="task_xyz",
    status=AgentExecutionStatus.Error,
    result="Connector failed: 503 Service Unavailable",
)

Update only payload extension

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

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

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:

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.

# 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

tasks.stop(task_id="task_xyz")

Errors

Raises ModuleException. Common statuses:

Status	Cause
404	Task doesn’t exist.
500	Server error.

events

Task.aevents() is an async generator over TaskUpdateEvents 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.

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

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

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

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

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

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:

The platform closes the stream (typically after TaskFinished).
The connection fails permanently (network, auth).
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.

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

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

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

reasoning = [
    m for m in log.messages
    if isinstance(m, AgentActivityThreadReasoning)
]
for r in reasoning:
    print(f"[{r.type}] {r.thought}")

Sync version

log = task.get_activity_log()

Errors

Raises ModuleException 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.

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:

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

task.report_metrics()

Errors

Raises ModuleException 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. 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.

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.

# 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 for typical usage patterns (idempotent updates, failure reporting).

Sync version

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.

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.

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

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.

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.

`get_activity_log`

task.aget_activity_log() / task.get_activity_log(): full message / tool-call / reasoning thread for the task. See activity log.

`report_metrics`

task.areport_metrics() / task.report_metrics(): pushes token counts and tool usage to the platform. See 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`

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`

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`

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`

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.

@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

​Constructor

​Module methods

​Task instance methods

​Lifecycle

​Class-level method

​create

​Parameters

​Returns Task

​When to use this vs. Agent.acreate_task

​Sync version

​Errors

​get

​Parameters

​Returns Task

​Examples

​Wait for completion

​Reload an existing instance

​Sync version

​Errors

​list

​Parameters

​Returns list[TasksListItem]

​Examples

​Filter by status

​Filter by end-user

​Tasks for a user

​Sync versions

​Errors

​update

​Parameters

​Returns Task

​Examples

​Mark complete

​Mark errored

​Update only payload extension

​Sync version

​When to use this vs. task.asave()

​Errors

​stop

​Parameters

​Returns Task

​Examples

​Stop from a running Task instance

​Idempotent stop

​Sync version

​Errors

​events

​Parameters

​Yields TaskUpdateEvent

Event types

​Examples

​Filter on event type

​Render tool calls

​Stop on first failure

​Track auth flows in-band

​Sync version

​Lifecycle

​Errors

​When not to stream

​get_activity_log

​Parameters

​Returns AgentActivityThread

​Examples

​Just the user/assistant transcript

​Audit tool usage

​Reasoning trace

​Sync version

​Errors

​report_metrics

​Parameters

​Returns

​What gets reported

​Examples

​Inside a custom event loop

​Sync version

​Errors

​report_external_task

​Parameters

​Returns Task

​When to use this vs. Backend.areport_external_task

Constructor

Module methods

`Task` instance methods

Lifecycle

Class-level method

create

Parameters

Returns `Task`

When to use this vs. `Agent.acreate_task`

Sync version

Errors

get

Parameters

Returns `Task`

Examples

Wait for completion

Reload an existing instance

Sync version

Errors

list

Parameters

Returns `list[TasksListItem]`

Examples

Filter by status

Filter by end-user

Tasks for a user

Sync versions

Errors

update

Parameters

Returns `Task`

Examples

Mark complete

Mark errored

Update only payload extension

Sync version

When to use this vs. `task.asave()`

Errors

stop

Parameters

Returns `Task`

Examples

Stop from a running `Task` instance

Idempotent stop

Sync version

Errors

events

Parameters

Yields `TaskUpdateEvent`

Examples

Filter on event type

Render tool calls

Stop on first failure

Track auth flows in-band

Sync version

Lifecycle

Errors

When not to stream

get_activity_log

Parameters

Returns `AgentActivityThread`

Examples

Just the user/assistant transcript

Audit tool usage

Reasoning trace

Sync version

Errors

report_metrics

Parameters

Returns

What gets reported

Examples

Inside a custom event loop

Sync version

Errors

report_external_task

Parameters

Returns `Task`

When to use this vs. `Backend.areport_external_task`

Sync version