Observability with Tool Hooks

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.

• @on_tool_before runs before each tool invocation.
• @on_tool_after runs after a successful invocation, with the result.
• @on_tool_error runs when a tool invocation raises.

​

Prerequisites

• Complete the Quickstart so the CLI, SDK, and xpander login are already set up.
• An agent with at least one tool attached. Connectors selected in Agent Studio, @register_tool functions, or MCP-server tools all work.
• Python 3.12+ for the local handler.

​

1. Log every tool call

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}")

1. @on_tool_before runs immediately before the tool body executes. The Tool object exposes tool.name, tool.id, tool.is_local (true for @register_tool functions, false for connectors), and tool.description.
2. @on_tool_after only runs on success and adds a result parameter carrying whatever the tool returned. For connector tools, that’s the raw response body. For local tools, it’s whatever your function returned.
3. @on_tool_error runs in place of the after-hook when the tool raises. The error parameter is the original exception. The agent’s framework 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.

​

2. Time and instrument every tool call

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")

1. tool_call_id is the correlation key. Concurrent tool 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 containers.
3. Embed tool.name in the metric name. Per-tool dashboards drop out of this naming scheme without per-tool boilerplate.

​

3. Redact payloads and add custom guardrails

• 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 tool itself still runs, but the alert reaches your error-tracking system.

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})

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 tool, not just the ones you remember to instrument.

​

4. Alert on failures of business-critical tools

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",
    )

1. The name allowlist is what keeps alert volume sane. Without it, every transient connector 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

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)

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 tool 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 tool types. Connector calls, custom @register_tool calls, and MCP tools all hit this hook with the same extension.

​

6. Where hooks fit in your project

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

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 tool fires.
3. Hooks coexist with framework-level 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 tool-invocation layer, so they run alongside (not instead of) any framework callback you’ve already wired up.

​

How hooks fire

1. Schema validation runs first if the tool has a Pydantic schema.
2. All @on_tool_before hooks run, in registration order.
3. The tool body executes (the connector HTTP call, the local @register_tool function, or the MCP server call).
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 Agent Studio happens after hooks return, so your hooks see the call before the platform’s metrics view does.

• Exceptions inside a hook are caught by the SDK and logged. They don’t prevent the tool 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

​

Next steps