Decorators - xpander.ai

Decorators are how your Python module becomes a deployed agent’s run. The runtime dispatches incoming task events to your @on_task handler over SSE, calls @on_boot once at startup, runs @on_tool_* hooks around every skill invocation, and routes auth events through @on_auth_event. In the SDK a skill is a Tool, which is where the hook names come from.

Decorator	Purpose
`@on_task`	Handle incoming tasks. The primary entry point for any SDK-based agent.
`@on_boot`	Run once at startup, before the SSE listener subscribes.
`@on_shutdown`	Run once during graceful shutdown.
`@on_auth_event`	Receive MCP/OAuth authentication events as they happen.
`@on_tool_before`	Run before any skill invocation.
`@on_tool_after`	Run after a successful skill invocation.
`@on_tool_error`	Run when a skill invocation raises.

For the skill-registration decorator, see @register_tool: it lives next to the ToolsRepository API, since it constructs Tool objects rather than wiring runtime hooks.

Minimal handler

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

backend = Backend()

@on_boot
async def boot():
    print("Worker starting…")

@on_shutdown
async def shutdown():
    print("Worker stopping…")

@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

Run this module with python worker.py, and the process subscribes to the runtime’s SSE stream, picks up tasks dispatched to its agent, runs them, and saves results. Tasks are auto-marked Completed (or Error on raise) and metrics are reported automatically when task.tokens is set.

Required environment

@on_task validates required env vars eagerly at import. Missing any of these raises ModuleException:

Variable	Required for
`XPANDER_API_KEY`	Authentication.
`XPANDER_ORGANIZATION_ID`	Routing.
`XPANDER_AGENT_ID`	Registers the process as this agent’s run.

For local development without an org-managed agent, set these yourself: get XPANDER_API_KEY from Settings > API keys in Xpander Chat, and pick your own XPANDER_ORGANIZATION_ID and XPANDER_AGENT_ID. See Environment variables for where each value comes from.

@on_task

@on_task registers a function as the agent’s task executor. The decorated function runs every time the runtime dispatches a task to this process: over SSE in production, and via the embedded HTTP server (POST /invoke on port 59321) for local invocations and Cloud Run-style integrations. The runtime auto-detects whether your function is a regular handler (returns Task) or a streaming handler (async generator that yields TaskUpdateEvent).

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

Decorator forms

@on_task
def fn(task: Task) -> Task: ...

@on_task(configuration=config)
def fn(task: Task) -> Task: ...

@on_task(test_task=local_test_task)
async def fn(task: Task) -> Task: ...

Parameter	Type	Default	Description
`configuration`	`Configuration`	`None`	SDK config for the underlying `Events` module. Falls back to env vars.
`test_task`	`LocalTaskTest`	`None`	A simulated task to run locally. The runtime invokes the handler once with this task and exits, instead of subscribing to the SSE stream.

Required signature

The handler must accept exactly one parameter named task. Either positional or keyword form works.

@on_task
def handler(task: Task) -> Task:        # ✓
    ...

@on_task
async def handler(task: Task) -> Task:  # ✓
    ...

@on_task
async def handler(thing: Task):         # ✗ raises TypeError: must be named `task`
    ...

Handler types

The decorator detects which kind you wrote by inspecting whether your function is an async generator.

Regular handler

Sync or async, returns the (possibly mutated) Task.

@on_task
async def handler(task: Task) -> Task:
    task.result = "done"
    return task

The runtime persists the returned task automatically. Don’t call task.asave() yourself unless you need to checkpoint mid-handler.

Streaming handler

Async generator that yields TaskUpdateEvents. Used for token-by-token streaming responses. A streaming handler emits two kinds of events. Each token (or text chunk) from the LLM is yielded as a TaskUpdateEventType.Chunk event, which xpander forwards to subscribers in real time. After the loop ends, the handler yields one final TaskUpdateEventType.TaskFinished event carrying the completed Task object. xpander uses that final event to mark the task complete and persist its result.

from datetime import datetime, timezone
from xpander_sdk import on_task, TaskUpdateEvent, TaskUpdateEventType

@on_task
async def handler(task: Task):
    async for chunk in stream_from_llm(task.to_message()):
        yield TaskUpdateEvent(
            type=TaskUpdateEventType.Chunk,
            task_id=task.id,
            organization_id=task.organization_id,
            time=datetime.now(timezone.utc),
            data=chunk,
        )

    yield TaskUpdateEvent(
        type=TaskUpdateEventType.TaskFinished,
        task_id=task.id,
        organization_id=task.organization_id,
        time=datetime.now(timezone.utc),
        data=task,  # final Task object
    )

Streaming handlers are exposed via POST /invoke only: the SSE listener wraps them in an adapter that consumes the generator and tracks the final Task from the TaskFinished event.

What the runtime does

When you decorate a function with @on_task:

Validates the signature (must accept task).
Detects handler type (regular vs streaming).
Starts an embedded HTTP server on port 59321 for POST /invoke (always, in a daemon thread). Override the port with XPANDER_STREAMING_PORT.
Registers an SSE listener with the Events module. The listener:
- Reads XPANDER_API_KEY, XPANDER_ORGANIZATION_ID, XPANDER_AGENT_ID from env (raises ModuleException if any are missing).
- Subscribes to xpander’s task-dispatch stream.
- Acquires a semaphore (max_sync_workers=6 by default) before dispatching.
- Sets task status to Executing before calling your handler.
- Calls your handler.
- Persists the returned task; if your handler raised, marks the task Error with the exception string as the result.
- Reports metrics if task.tokens is set.
- Re-runs the handler with a continuation prompt if deep planning is enabled (multi-step task plans tracked in task.deep_planning) and items remain incomplete, up to MAX_PLAN_RETRIES = 5. Pre-retry the runtime triggers session compaction.

Test mode

Pass test_task (or use the CLI override) to invoke the handler once locally instead of subscribing to xpander.

With a `LocalTaskTest`

from xpander_sdk import on_task
from xpander_sdk.modules.tasks.models.task import LocalTaskTest, AgentExecutionInput
from xpander_sdk.models.shared import OutputFormat

local_test_task = LocalTaskTest(
    input=AgentExecutionInput(text="What can you do?"),
    output_format=OutputFormat.Json,
    output_schema={"capabilities": "list of capabilities"},
)

@on_task(test_task=local_test_task)
async def handler(task: Task) -> Task:
    task.result = {"capabilities": ["Search", "Summarize"]}
    return task

The runtime registers the process as the agent’s run, creates the test task, dispatches it to the handler, and exits after completion (printing the final result). Useful for local iteration without invoking from Xpander Chat.

From the CLI

The decorator also responds to --invoke / --prompt arguments on sys.argv, so you can run any @on_task-decorated module with:

python worker.py --invoke --prompt "Try this prompt" --output_format json

Supported flags:

Flag	Purpose
`--invoke`	Switches the runtime into single-task test mode.
`--prompt`	Required when `--invoke` is set. Becomes `task.input.text`.
`--output_format`	One of `json`, `markdown`, `text`.
`--output_schema`	JSON-encoded schema string.

Runtime caveats

One handler per process. The decorator subscribes to the SSE stream on import. Decorating multiple functions in the same module isn’t useful: the most recently registered handler wins. Use multiple Python processes if you need to host different agents.
HTTP server starts immediately. The POST /invoke endpoint is up before the SSE listener finishes connecting. If port 59321 is in use the runtime logs a warning and continues. Override the port with the XPANDER_STREAMING_PORT env var.
Synchronous handlers are dispatched on a thread pool. A thread pool of max_sync_workers=6 handles sync handlers concurrently. Async handlers run on the event loop.

Errors raised by the decorator

TypeError: handler doesn’t accept task, or returns/yields the wrong type.
ModuleException: required env vars are missing when the SSE listener starts.

@on_boot / @on_shutdown

@on_boot registers a function to run before the SSE listener subscribes to xpander: useful for warming caches, opening database pools, validating environment, or pre-fetching data. @on_shutdown registers a function to run during graceful shutdown.

from xpander_sdk import on_boot, on_shutdown

@on_boot
async def warm_caches():
    print("Boot: warming caches…")
    await prefetch_models()

@on_shutdown
async def cleanup():
    print("Shutting down…")
    await close_connections()

Both decorators accept either sync or async functions, and you can register multiple handlers: the runtime invokes them in registration order.

Decorator forms

@on_boot
def fn(): ...

@on_boot(configuration=config)
async def fn(): ...

@on_shutdown
def fn(): ...

@on_shutdown(configuration=config)
async def fn(): ...

Parameter	Type	Default	Description
`configuration`	`Configuration`	`None`	Reserved for future use. Currently only the `Events` module reads config; boot/shutdown handlers receive no arguments.

Lifecycle ordering

Process start
  ↓
@on_boot handlers (in registration order, await one at a time)
  ↓
Events.start(): subscribes to SSE, begins task dispatch
  ↓
… task dispatch ongoing …
  ↓
SIGINT / SIGTERM received
  ↓
Events.stop(): cancels tracked tasks, closes thread pool
  ↓
@on_shutdown handlers (in registration order, errors logged but don't block)
  ↓
Process exit

If a @on_boot handler raises, the process fails to start: the exception propagates and the process exits. If a @on_shutdown handler raises, the error is logged but the runtime continues with the next shutdown handler.

Examples

Open a DB pool at boot, close on shutdown

import asyncpg
from xpander_sdk import on_boot, on_shutdown

pool = None

@on_boot
async def open_pool():
    global pool
    pool = await asyncpg.create_pool(dsn="postgresql://…")

@on_shutdown
async def close_pool():
    if pool:
        await pool.close()

Validate env at boot

import os
from xpander_sdk import on_boot

@on_boot
def check_env():
    for var in ("OPENAI_API_KEY", "REDIS_URL"):
        if not os.environ.get(var):
            raise RuntimeError(f"Missing required env var: {var}")

Raising in @on_boot aborts startup before the process takes any tasks: useful for fail-fast environment validation.

Multiple handlers

@on_boot
def first():
    print("1")

@on_boot
def second():
    print("2")

Both run, in declaration order: 1 then 2. The runtime awaits each one before moving on.

Sync and async mix

@on_boot
def synchronous():
    print("sync boot")

@on_boot
async def asynchronous():
    print("async boot")
    await asyncio.sleep(0)

Sync handlers are called directly; async ones are awaited. Either works.

Notes

Boot/shutdown handlers are class-level on Events. They register globally per process: there’s no scoping to a specific Events instance. This is fine for a one-process-one-agent model.
Shutdown handlers run after the SSE listener is stopped but before final cleanup. By then the process is no longer accepting new tasks; in-flight tasks are cancelled.

@on_tool_before / @on_tool_after / @on_tool_error

@on_tool_before, @on_tool_after, and @on_tool_error register hooks that fire around skill invocations: useful for logging, validation, caching, alerting, or analytics. The hooks run during every Tool.ainvoke (which means around every skill call the agent loop makes too).

from xpander_sdk import on_tool_before, on_tool_after, on_tool_error

@on_tool_before
async def log_invoke(tool, payload, payload_extension, tool_call_id, agent_version):
    print(f"→ {tool.name} {payload}")

@on_tool_after
async def log_success(tool, payload, payload_extension, tool_call_id, agent_version, result):
    print(f"✓ {tool.name} → {result}")

@on_tool_error
async def log_error(tool, payload, payload_extension, tool_call_id, agent_version, error):
    print(f"✗ {tool.name} failed: {error}")

You can register multiple hooks of each type: they fire in registration order. Sync and async functions are both supported.

Required signatures

Decorator	Parameters	Notes
`@on_tool_before`	`tool`, `payload`, `payload_extension`, `tool_call_id`, `agent_version`	Fires before the skill runs.
`@on_tool_after`	`tool`, `payload`, `payload_extension`, `tool_call_id`, `agent_version`, `result`	Fires after a successful invocation. `result` is the skill’s response.
`@on_tool_error`	`tool`, `payload`, `payload_extension`, `tool_call_id`, `agent_version`, `error`	Fires when the invocation raises. `error` is the exception.

Parameter	Type	Description
`tool`	`Tool`	The skill being invoked.
`payload`	`Any`	The input payload after schema validation.
`payload_extension`	`dict \| None`	Extra fields that will be deep-merged into the payload.
`tool_call_id`	`str \| None`	Correlation id matching the LLM’s `tool_call_id`.
`agent_version`	`str \| None`	Agent version pinned for the call.
`result` (after only)	`Any`	The skill’s response.
`error` (error only)	`Exception`	The raised exception.

Decorator forms

@on_tool_before
def fn(...): ...

@on_tool_before(configuration=config)
def fn(...): ...

The same form applies to @on_tool_after and @on_tool_error. The optional configuration parameter is reserved for future use; hooks don’t currently read it.

Examples

Log every skill call

import time
import contextvars

start_time = contextvars.ContextVar("start_time")

@on_tool_before
def begin(tool, payload, *_):
    start_time.set(time.perf_counter())
    print(f"[{tool.name}] start")

@on_tool_after
def done(tool, payload, *_, result=None):
    elapsed = time.perf_counter() - start_time.get()
    print(f"[{tool.name}] done in {elapsed*1000:.0f}ms")

@on_tool_error
def failed(tool, payload, *_, error=None):
    elapsed = time.perf_counter() - start_time.get()
    print(f"[{tool.name}] failed in {elapsed*1000:.0f}ms: {error}")

Block invocations matching a payload pattern

Hooks can’t cancel invocations directly (the runtime catches their exceptions and continues), but you can mutate state or short-circuit by raising a guard before the call:

@on_tool_before
def block_test_payloads(tool, payload, *_):
    if isinstance(payload, dict) and payload.get("test_mode"):
        raise RuntimeError("test_mode payloads are blocked")

Hooks that raise are caught by the runtime: the exception is logged but doesn’t propagate to the caller. To genuinely block a skill, validate inside the agent loop or in @register_tool function bodies.

Cache successful results

cache: dict = {}

@on_tool_after
def cache_result(tool, payload, _ext, _id, _ver, result):
    if hasattr(payload, "items"):
        key = (tool.id, frozenset(payload.items()))
        cache[key] = result

Alert on failures

@on_tool_error
async def alert(tool, payload, _ext, _id, _ver, error):
    await pagerduty.fire(f"tool {tool.id} failed: {error}")

Notes

Hooks fire once per skill invocation, regardless of whether the skill is local or remote.
Exceptions in hooks are caught and logged via loguru: they never crash the invocation pipeline.
Hooks are class-level on ToolHooksRegistry (process-wide singleton). There’s no scoping to a specific repository.
The auto-emitted ToolCallRequest / ToolCallResult events on task.aevents() are independent of these hooks. The hooks fire even when report_activity=False in Tool.ainvoke.

@on_auth_event

@on_auth_event registers a handler that fires whenever the agent runtime emits an authentication event: for example, when a skill served over MCP requires the end-user to log in via OAuth. Use it to surface the OAuth URL to your UI, kick off external auth flows, or log auth attempts.

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

@on_auth_event
async def handle_auth(agent: Agent, task: Task, event: TaskUpdateEvent):
    print(f"Auth required for {agent.name}")
    print(f"Task: {task.id}")
    print(f"Auth data: {event.data}")

The handler is auto-registered globally: you don’t pass it anywhere. Every Backend.aget_args(...) call routes auth events through registered handlers automatically.

Required signature

The function must accept exactly three parameters: agent, task, event. Names are flexible; arity matters. Sync or async are both accepted.

@on_auth_event
async def handler(agent, task, event):     # ✓
    ...

@on_auth_event
def handler(agent, task, event):           # ✓ (sync)
    ...

@on_auth_event
def handler(agent, task):                  # ✗ TypeError: needs 3 params
    ...

Parameter	Type	Description
`agent`	`Agent`	The agent processing the task.
`task`	`Task`	The current task.
`event`	`TaskUpdateEvent`	`event.type == "auth_event"`. `event.data` holds the auth payload.

event.type is always "auth_event" for handlers registered with this decorator.

Event payload shapes

event.data is one of the MCP OAuth response variants. Switch on event.data.type:

`type`	`event.data.data`
`MCPOAuthResponseType.LOGIN_REQUIRED` (`"login_required"`)	`MCPOAuthGetTokenLoginRequiredResponse(url, server_url, server_name)`
`MCPOAuthResponseType.TOKEN_READY` (`"token_ready"`)	`MCPOAuthGetTokenTokenReadyResponse(access_token)`
`MCPOAuthResponseType.TOKEN_ISSUE` (`"token_issue"`)	`MCPOAuthGetTokenGenericResponse(message)`
`MCPOAuthResponseType.NOT_SUPPORTED` (`"not_supported"`)	`MCPOAuthGetTokenGenericResponse(message)`

Examples

Show the OAuth URL to the user

@on_auth_event
async def show_oauth_url(agent, task, event):
    payload = event.data
    if payload.type == "login_required":
        login = payload.data
        await my_ui.show(f"Please authenticate with {login.server_name}: {login.url}")

Multiple handlers stack

@on_auth_event
async def log_auth(agent, task, event):
    audit_log.info(f"auth event for {agent.id}: {event.data}")

@on_auth_event
async def notify_user(agent, task, event):
    await pubsub.publish(f"user:{task.input.user.id}", event.data)

Both handlers fire on every auth event. Order is registration order.

Combine with a per-call callback

You can pass an additional one-off callback to Backend.aget_args(auth_events_callback=...): it runs in addition to every globally-registered handler.

@on_auth_event
async def global_handler(agent, task, event):
    audit(event)

# Per-call extra:
async def per_call(agent, task, event):
    await my_ui.show_special_state(event)

args = await backend.aget_args(
    agent_id="agent-123",
    auth_events_callback=per_call,
)

Both global_handler and per_call fire when an auth event happens during the resulting agent’s run.

Helper functions

The module also exposes two helpers for testing and management:

from xpander_sdk.modules.backend.decorators.on_auth_event import (
    get_registered_handlers,
    clear_handlers,
)

# Inspect what's registered
handlers = get_registered_handlers()

# Wipe registered handlers (useful between tests)
clear_handlers()

These aren’t re-exported from the top-level xpander_sdk package: import from the decorator module directly.

​Minimal handler

​Required environment

​@on_task

​Decorator forms

​Required signature

​Handler types

​Regular handler

​Streaming handler

​What the runtime does

​Test mode

​With a LocalTaskTest

​From the CLI

​Runtime caveats

​Errors raised by the decorator

​@on_boot / @on_shutdown

​Decorator forms

​Lifecycle ordering

​Examples

​Open a DB pool at boot, close on shutdown

​Validate env at boot

​Multiple handlers

​Sync and async mix

​Notes

​@on_tool_before / @on_tool_after / @on_tool_error

​Required signatures

​Decorator forms

​Examples

​Log every skill call

​Block invocations matching a payload pattern

​Cache successful results

​Alert on failures

​Notes

​@on_auth_event

​Required signature

​Event payload shapes

​Examples

​Show the OAuth URL to the user

​Multiple handlers stack

​Combine with a per-call callback

​Helper functions

Minimal handler

Required environment

@on_task

Decorator forms

Required signature

Handler types

Regular handler

Streaming handler

What the runtime does

Test mode

With a `LocalTaskTest`

From the CLI

Runtime caveats

Errors raised by the decorator

@on_boot / @on_shutdown

Decorator forms

Lifecycle ordering

Examples

Open a DB pool at boot, close on shutdown

Validate env at boot

Multiple handlers

Sync and async mix

Notes

@on_tool_before / @on_tool_after / @on_tool_error

Required signatures

Decorator forms

Examples

Log every skill call

Block invocations matching a payload pattern

Cache successful results

Alert on failures

Notes

@on_auth_event

Required signature

Event payload shapes

Examples

Show the OAuth URL to the user

Multiple handlers stack

Combine with a per-call callback

Helper functions