> ## 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. # Decorators > Wire your code into the agent runtime (task handlers, lifecycle, and tool hooks). Decorators are how your Python module becomes a deployed agent worker. The platform dispatches incoming task events to your `@on_task` handler over SSE, calls `@on_boot` once at startup, runs `@on_tool_*` hooks around every tool invocation, and routes auth events through `@on_auth_event`. | Decorator | Purpose | | ----------------------------------------------------------------------------------------------------------- | ----------------------------------------------------------------------- | | [`@on_task`](/developers/sdk-reference/decorators#@on_task) | Handle incoming tasks. The primary entry point for any SDK-based agent. | | [`@on_boot`](/developers/sdk-reference/decorators#@on_boot-/-@on_shutdown) | Run once at startup, before the SSE listener subscribes. | | [`@on_shutdown`](/developers/sdk-reference/decorators#@on_boot-/-@on_shutdown) | Run once during graceful shutdown. | | [`@on_auth_event`](/developers/sdk-reference/decorators#@on_auth_event) | Receive MCP/OAuth authentication events as they happen. | | [`@on_tool_before`](/developers/sdk-reference/decorators#@on_tool_before-/-@on_tool_after-/-@on_tool_error) | Run before any tool invocation. | | [`@on_tool_after`](/developers/sdk-reference/decorators#@on_tool_before-/-@on_tool_after-/-@on_tool_error) | Run after a successful tool invocation. | | [`@on_tool_error`](/developers/sdk-reference/decorators#@on_tool_before-/-@on_tool_after-/-@on_tool_error) | Run when a tool invocation raises. | For the tool-registration decorator, see [`@register_tool`](/developers/sdk-reference/tools#register_tool-classmethod): it lives next to the tools API, since it constructs `Tool` objects rather than wiring runtime hooks. ## Minimal worker ```python theme={"dark"} 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: `python worker.py`: and the worker subscribes to the platform'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` | Worker registration. | For local development without an org-managed agent, set these to your dev values. ## @on\_task `@on_task` registers a function as the agent's task executor. The decorated function runs every time the platform dispatches a task to this worker: 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`). ```python theme={"dark"} from xpander_sdk import Backend, on_task from xpander_sdk.modules.tasks.sub_modules.task import Task from agno.agent import Agent as AgnoAgent backend = Backend() @on_task async def handler(task: Task) -> Task: args = await backend.aget_args( agent_id=task.agent_id, agent_version=task.agent_version, task=task, ) agno_agent = AgnoAgent(**args) result = await agno_agent.arun(input=task.to_message()) task.result = result.content return task ``` ### Decorator forms ```python theme={"dark"} @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. ```python theme={"dark"} @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`. ```python theme={"dark"} @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 `TaskUpdateEvent`s. 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 the platform forwards to subscribers in real time. After the loop ends, the handler yields one final `TaskUpdateEventType.TaskFinished` event carrying the completed `Task` object. The platform uses that final event to mark the task complete and persist its result. ```python theme={"dark"} 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`: 1. **Validates the signature** (must accept `task`). 2. **Detects handler type** (regular vs streaming). 3. **Starts an embedded HTTP server** on port `59321` for `POST /invoke` (always, in a daemon thread). Override the port with `XPANDER_STREAMING_PORT`. 4. **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 the platform'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](/developers/sdk-reference/tasks#state) 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 the platform. #### With a `LocalTaskTest` ```python theme={"dark"} 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 a worker, creates the test task, dispatches it to the handler, and exits after completion (printing the final result). Useful for local iteration without invoking from the dashboard. #### From the CLI The decorator also responds to `--invoke` / `--prompt` arguments on `sys.argv`, so you can run any `@on_task`-decorated module with: ```bash theme={"dark"} 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 (or container instances) 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 the platform: useful for warming caches, opening database pools, validating environment, or pre-fetching data. `@on_shutdown` registers a function to run during graceful shutdown. ```python theme={"dark"} 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 ```python theme={"dark"} @on_boot def fn(): ... @on_boot(configuration=config) async def fn(): ... ``` ```python theme={"dark"} @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 ```text theme={"dark"} 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 worker 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 ```python theme={"dark"} 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 ```python theme={"dark"} 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 worker takes any tasks: useful for fail-fast environment validation. #### Multiple handlers ```python theme={"dark"} @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 ```python theme={"dark"} @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 tool invocations: useful for logging, validation, caching, alerting, or analytics. The hooks run during every `Tool.ainvoke` (which means around every framework-driven tool call too). ```python theme={"dark"} 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 tool runs. | | `@on_tool_after` | `tool`, `payload`, `payload_extension`, `tool_call_id`, `agent_version`, `result` | Fires after a successful invocation. `result` is the tool'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 tool 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 tool-call. | | `agent_version` | `str \| None` | Agent version pinned for the call. | | `result` *(after only)* | `Any` | The tool's response. | | `error` *(error only)* | `Exception` | The raised exception. | ### Decorator forms ```python theme={"dark"} @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 tool call ```python theme={"dark"} 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: ```python theme={"dark"} @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 tool, validate at the framework layer or in `@register_tool` function bodies. #### Cache successful results ```python theme={"dark"} 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 ```python theme={"dark"} @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 tool invocation**, regardless of whether the tool 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 an MCP server 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. ```python theme={"dark"} 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. ```python theme={"dark"} @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 ```python theme={"dark"} @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 ```python theme={"dark"} @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. ```python theme={"dark"} @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: ```python theme={"dark"} 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.