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

# Observability with Skill Call Hooks

> Observe and instrument every skill call without modifying the skills themselves

Skill call hooks are decorators that fire around every skill invocation an agent makes. In the SDK a skill is a `Tool`, so the decorators are named `@on_tool_before`, `@on_tool_after`, and `@on_tool_error`. They give you a single place to plug in logging, metrics, alerting, payload redaction, custom guardrails, and per-skill observability without touching the skills themselves. Hooks don't depend on the agent loop: they run at the SDK level, so the same hook fires whether the agent is built on Agno, OpenAI Agents SDK, LangChain, or AWS Strands, and whether the skill is a catalog skill, a custom `@register_tool` function, or a skill served over MCP.

There are three decorators:

* `@on_tool_before` runs before each skill invocation.
* `@on_tool_after` runs after a successful invocation, with the result.
* `@on_tool_error` runs when a skill invocation raises.

## Prerequisites

* **Complete the [Quickstart](/developers/quickstart)** so the CLI, SDK, and `xpander login` are already set up.
* **An agent with at least one skill attached.** Catalog skills selected in Xpander Chat, `@register_tool` functions, or skills served over MCP all work.
* **Python 3.12+** for the local handler.

## 1. Log every skill call

The smallest useful hook is a logger that records each skill the agent reaches for. Drop the three decorators in a module that's imported from your handler and they auto-register at import time:

```python hooks.py highlight={5,12,20} theme={"dark"}
from typing import Any, Dict, Optional
from loguru import logger
from xpander_sdk import on_tool_before, on_tool_after, on_tool_error, Tool

@on_tool_before
def log_invocation(tool: Tool, payload: Any,
                   payload_extension: Optional[Dict[str, Any]] = None,
                   tool_call_id: Optional[str] = None,
                   agent_version: Optional[str] = None):
    logger.info(f"-> {tool.name} called with payload {payload}")

@on_tool_after
def log_success(tool: Tool, payload: Any,
                payload_extension: Optional[Dict[str, Any]] = None,
                tool_call_id: Optional[str] = None,
                agent_version: Optional[str] = None,
                result: Any = None):
    logger.info(f"<- {tool.name} returned {type(result).__name__}")

@on_tool_error
def log_failure(tool: Tool, payload: Any,
                payload_extension: Optional[Dict[str, Any]] = None,
                tool_call_id: Optional[str] = None,
                agent_version: Optional[str] = None,
                error: Optional[Exception] = None):
    logger.error(f"x {tool.name} failed: {error}")
```

What this means in practice:

1. **`@on_tool_before`** runs immediately before the skill body executes. The `Tool` object exposes `tool.name`, `tool.id`, `tool.is_local` (true for `@register_tool` functions, false for catalog skills), and `tool.description`.
2. **`@on_tool_after`** only runs on success and adds a `result` parameter carrying whatever the skill returned. For catalog skills, that's the raw response body. For local skills, it's whatever your function returned.
3. **`@on_tool_error`** runs in place of the after-hook when the skill raises. The `error` parameter is the original exception. The agent loop still sees the failure; the hook is for your side effects (logs, alerts, traces).
4. **`tool_call_id`** is unique per invocation. Use it as the correlation key to pair before-hooks with their matching after-hooks or error-hooks.
5. **Both sync and async hooks work.** The SDK detects coroutine functions and awaits them automatically, so you can `await` an HTTP client or a DB write inside an async hook without extra wiring.

Three traits of every hook signature, regardless of which decorator you use:

| Parameter | Type | Notes |
| - | - | - |
| `tool` | `Tool` | The skill being invoked. Read `tool.name`, `tool.id`, `tool.is_local`, `tool.description`. Always set. |
| `payload` | `Any` | The arguments the LLM produced for the call. For catalog skills, typically `{"body_params": {...}, "path_params": {...}, "query_params": {...}}`. For local skills, whatever your function expects. Always set. |
| `payload_extension` | `Optional[Dict]` | The deep-merged extension you passed via `tool_call_payload_extension` on the task or `payload_extension=` on `agent.ainvoke_tool`. `None` if you didn't set one. |
| `tool_call_id` | `Optional[str]` | Stable identifier for one invocation. Pair before/after hooks with this key. |
| `agent_version` | `Optional[str]` | The deployed version of the agent that issued the call. Useful for filtering metrics by rollout. |
| `result` (after only) | `Any` | The value the skill returned on success. |
| `error` (error only) | `Optional[Exception]` | The exception raised by the skill body. |

## 2. Time and instrument every skill call

Once you have logging, the next thing most teams want is timing and counter metrics per skill. The before/after pair is the natural fit, with `tool_call_id` as the correlation key:

```python hooks.py highlight={5,7,15} theme={"dark"}
import time
from typing import Any, Dict, Optional
from xpander_sdk import on_tool_before, on_tool_after, Tool

starts: dict[str, float] = {}

@on_tool_before
def record_start(tool: Tool, payload: Any,
                 payload_extension: Optional[Dict[str, Any]] = None,
                 tool_call_id: Optional[str] = None,
                 agent_version: Optional[str] = None):
    if tool_call_id:
        starts[tool_call_id] = time.time()

@on_tool_after
def record_duration(tool: Tool, payload: Any,
                    payload_extension: Optional[Dict[str, Any]] = None,
                    tool_call_id: Optional[str] = None,
                    agent_version: Optional[str] = None,
                    result: Any = None):
    started = starts.pop(tool_call_id, None) if tool_call_id else None
    if started is not None:
        metrics_client.timing(f"tool.{tool.name}.duration_ms", (time.time() - started) * 1000)
        metrics_client.increment(f"tool.{tool.name}.calls")
```

What this means in practice:

1. **`tool_call_id` is the correlation key.** Concurrent skill calls run on the same agent, so a global timestamp would clobber. The id stays stable from the before hook to the matching after or error hook.
2. **Pop, don't peek.** Removing the entry on the after hook keeps memory bounded across long-running processes.
3. **Embed `tool.name` in the metric name.** Per-skill dashboards drop out of this naming scheme without per-skill boilerplate.

Mirror the increment in `@on_tool_error` so success and failure counters add up to the total call count.

## 3. Redact payloads and add custom guardrails

Hooks are observe-only by design. The SDK calls them, but ignores any return value, so you cannot mutate the payload or rewrite the result from a hook. What you can do is:

* **Redact at the sink.** Strip secrets from the copy of the payload you log or send to a tracing backend.
* **Detect and alert.** Match the payload against a guardrail policy and emit an alert or a metric when it trips.
* **Raise to fail loud.** A hook that raises has its exception logged by the SDK; the skill itself still runs, but the alert reaches your error-tracking system.

```python hooks.py highlight={5,12-13,15-17} theme={"dark"}
import copy
from typing import Any, Dict, Optional
from xpander_sdk import on_tool_before, Tool

SENSITIVE_KEYS = {"api_key", "password", "ssn", "credit_card"}

@on_tool_before
def redact_and_log(tool: Tool, payload: Any,
                   payload_extension: Optional[Dict[str, Any]] = None,
                   tool_call_id: Optional[str] = None,
                   agent_version: Optional[str] = None):
    # Deep copy so we never touch the live payload the tool will receive.
    safe = copy.deepcopy(payload) if isinstance(payload, dict) else payload
    if isinstance(safe, dict):
        for key in list(safe.get("body_params", {})):
            if key.lower() in SENSITIVE_KEYS:
                safe["body_params"][key] = "***"
    audit_log.write({"tool": tool.name, "tool_call_id": tool_call_id, "payload": safe})
```

What this means in practice:

1. **`copy.deepcopy(payload)`** is the safety net. Even though hook return values are ignored, mutating a shared dict in place could affect other observers reading the same object. Copy first, redact the copy.
2. **`SENSITIVE_KEYS`** is your project's policy. Extend it with whatever your security team flags.
3. **`audit_log.write(...)`** is a stand-in for whatever sink you ship to (S3, Datadog, OpenTelemetry). Hooks are the right place for this work because they fire on every skill, not just the ones you remember to instrument.

To enforce a policy that should *block* a call, do it inside the skill function itself. Hooks fire before the skill body runs, but raising from a hook only logs the exception, it doesn't cancel the invocation.

## 4. Alert on failures of business-critical skills

Most skill errors are noise: an LLM produced an invalid payload, a catalog skill returned a 4xx, the agent retries. The few that should page someone (a charge that didn't go through, an auth check that broke) deserve their own hook with a name allowlist:

```python hooks.py highlight={3,8-9} theme={"dark"}
from xpander_sdk import on_tool_error, Tool

CRITICAL = {"payment_processor", "auth_service", "fraud_check"}

@on_tool_error
async def alert_on_failure(tool: Tool, payload, payload_extension=None,
                           tool_call_id=None, agent_version=None, error=None):
    if tool.name not in CRITICAL:
        return
    await alert_service.send(
        title=f"Critical tool failure: {tool.name}",
        message=f"Error: {error}\nCall ID: {tool_call_id}\nAgent: {agent_version}",
        severity="critical",
    )
```

What this means in practice:

1. **The name allowlist** is what keeps alert volume sane. Without it, every transient skill failure pages you.
2. **`agent_version`** is included in the alert so you can correlate a spike of errors with the rollout that introduced it.
3. **The hook is async**, so it can `await` an HTTP call to PagerDuty or Slack without spinning up a background thread.

## 5. Attribute cost and usage per tenant

Skill call hooks are how you build per-customer billing or per-team cost dashboards on top of agent activity. Combine [`tool_call_payload_extension`](/developers/tools/pre-built#5-pass-per-request-context-to-every-tool-call) with an `@on_tool_after` hook that reads the tenant ID off the extension and increments a counter:

```python hooks.py theme={"dark"}
from xpander_sdk import on_tool_after, Tool

@on_tool_after
async def attribute_cost(tool: Tool, payload, payload_extension=None,
                         tool_call_id=None, agent_version=None, result=None):
    tenant_id = (payload_extension or {}).get("body_params", {}).get("tenant_id")
    if tenant_id:
        await billing.increment(tenant_id, tool=tool.name, count=1)
```

What this means in practice:

1. **`payload_extension`** is the same dict you set when creating the task with `tool_call_payload_extension={"body_params": {"tenant_id": "acme-corp"}}`. Every skill call inside that task carries it through to the hook.
2. **The hook fires on every successful invocation**, so the counter reflects real usage, not LLM intentions.
3. **It works uniformly across skill kinds.** Catalog skill calls, custom `@register_tool` calls, and skills served over MCP all hit this hook with the same extension.

## 6. Where hooks fit in your project

Register hooks at module level so they're set up before any task is processed. The cleanest pattern is a `hooks.py` imported from your handler:

```python xpander_handler.py highlight={1} theme={"dark"}
import hooks  # registers logging, metrics, and audit hooks at import time

from xpander_sdk import on_task, Task, Backend
from agno.agent import Agent

@on_task
async def handler(task: Task) -> Task:
    backend = Backend(configuration=task.configuration)
    agno_agent = Agent(**(await backend.aget_args(task=task)))
    result = await agno_agent.arun(input=task.to_message())
    task.result = result.content
    return task
```

What this means in practice:

1. **The `import hooks` line is enough.** Each `@on_tool_before` / `@on_tool_after` / `@on_tool_error` decorator registers itself in a process-global registry on import. There's no `register_hooks(...)` call.
2. **Hooks compose with `@on_boot`.** Use a boot handler to construct the metrics client, alerting client, or audit-log writer that your hooks reach for, so they exist before the first skill fires.
3. **Hooks coexist with the agent loop's own callbacks.** Agno's `tool_hooks` arg, OpenAI Agents SDK's run hooks, and LangChain callbacks all keep working. xpander's hooks fire at the SDK's skill-invocation point, so they run alongside (not instead of) any loop callback you've already wired up.

## How hooks fire

The SDK runs hooks synchronously around the skill body. The order is fixed:

1. Schema validation runs first if the skill has a Pydantic schema.
2. All `@on_tool_before` hooks run, in registration order.
3. The skill body executes (the catalog skill's HTTP call, the local `@register_tool` function, or the call to the MCP endpoint).
4. **On success**, every `@on_tool_after` hook runs, in registration order, with the result.
5. **On failure**, every `@on_tool_error` hook runs, in registration order, with the exception.
6. Activity reporting to Xpander Chat happens after hooks return, so your hooks see the call before xpander's metrics view does.

A few non-obvious properties:

* **Exceptions inside a hook are caught by the SDK and logged.** They don't prevent the skill from running, don't cancel sibling hooks, and don't propagate to the agent loop. This makes hooks safe for instrumentation, but it means you can't use them to block a call.
* **Hooks observe; they don't mutate.** The SDK calls each hook and ignores its return value. Mutate the local copy you log, but don't expect hook returns to alter the live payload or rewrite the result.
* **Order matters when hooks share state.** If two `@on_tool_after` hooks both read a dict populated by a `@on_tool_before` hook, register them in the order the after-hooks need to run.

## Troubleshooting

<AccordionGroup>
  <Accordion title="My hook never fires">
    The decorator only registers the hook when the module that defines it is imported. If `hooks.py` lives next to `xpander_handler.py` but nothing ever imports it, the decorators never run. Add `import hooks` at the top of `xpander_handler.py` (or wherever your `@on_task` lives) so registration happens at boot.
  </Accordion>

  <Accordion title="My hook fires twice for one skill call">
    Hooks register globally, so importing `hooks.py` from two different modules registers each decorator twice. Pick one import site (the handler) and remove the others. Re-running `xpander agent dev` reloads the registry from a fresh process, which is the easiest way to confirm.
  </Accordion>

  <Accordion title="Async hook seems to block or never complete">
    The SDK detects coroutine functions and awaits them; sync hooks run inline. If you wrote a sync hook that calls `asyncio.run(...)` or blocks on a sync HTTP client inside an async handler, you'll stall the event loop. Either declare the hook `async def` and `await` an async client, or keep it sync and use a non-blocking client.
  </Accordion>

  <Accordion title="An exception in my hook crashed... nothing">
    Hook exceptions are caught and logged by the SDK; the skill still runs. If you need a hook failure to be loud, push the exception to your error tracker yourself (`sentry_sdk.capture_exception(e)`) inside a `try/except`. Don't rely on the exception bubbling up to the agent loop, because it won't.
  </Accordion>

  <Accordion title="`payload_extension` is `None` even though I set `tool_call_payload_extension`">
    `tool_call_payload_extension` is a per-task setting passed to `agent.acreate_task(...)`. If you're invoking a skill by hand with `agent.ainvoke_tool(...)` and didn't pass `payload_extension=...`, the hook receives `None`. Either set the extension on the task, or pass it to `ainvoke_tool` directly.
  </Accordion>

  <Accordion title="Hook return value seems to be ignored">
    It is. Hooks observe the call; they don't mutate it. The SDK ignores whatever a hook returns. To shape the payload that reaches a skill, use input schema overrides on the skill's Advanced tab in Xpander Chat. To shape the result the LLM sees, use [Output Response Filtering](/developers/tools/output-response-filtering) or filter inside your `@register_tool` function before returning.
  </Accordion>
</AccordionGroup>

## Next steps

<CardGroup cols={2}>
  <Card title="Pre-built skills" icon="plug" href="/developers/tools/pre-built">
    The other skill surface hooks observe, including `tool_call_payload_extension` for per-tenant context.
  </Card>

  <Card title="Custom skills" icon="wrench" href="/developers/tools/custom-tools">
    Wrap your own Python functions with `@register_tool`. Hooks fire for these too.
  </Card>

  <Card title="Output Response Filtering" icon="filter" href="/developers/tools/output-response-filtering">
    How large skill responses get filtered before reaching the LLM.
  </Card>

  <Card title="Lifecycle hooks" icon="play" href="/developers/sdk-reference/decorators#@on_boot-/-@on_shutdown">
    `@on_boot` and `@on_shutdown` for setting up the clients your skill call hooks reach for.
  </Card>

  <Card title="Agent loops" icon="cubes" href="/developers/frameworks">
    How skill calls flow through Agno, OpenAI Agents SDK, LangChain, and AWS Strands.
  </Card>

  <Card title="Core Concepts" icon="lightbulb" href="/developers/core-concepts">
    The SDK class names mapped onto agents, tasks, threads, and skills.
  </Card>
</CardGroup>


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.