Hooks

Hooks run on either side of every tool dispatch:

• @agent.before_tool. Runs before the call. Returning a PermissionDecision.deny_(...) short-circuits the dispatch.
• @agent.after_tool. Runs after, with the ToolResult. Used for logging, metrics, redaction.

@before_tool for inline review

from loomflow import Agent
from loomflow.core.types import PermissionDecision
 
agent = Agent("...", model="claude-opus-4-7", tools=[send_email])
 
@agent.before_tool
async def review(call):
    if call.tool == "send_email" and "@enemy.com" in str(call.args):
        return PermissionDecision.deny_("blocked by reviewer")
    return None      # allow — fall through to permissions

Return values:

• None. Allow; fall through to the permissions layer.
• PermissionDecision.deny_(reason). Deny and stop.
• PermissionDecision.allow_(). Allow and skip the permissions layer.
• PermissionDecision.ask_(reason). Go through the approval handler.

Multiple @before_tool hooks are called in registration order; the first non-None decision wins.

@after_tool for logging and metrics

@agent.after_tool
async def log(call, result):
    print(f"{call.tool} → ok={result.ok}, took={result.duration_ms}ms")
    if not result.ok:
        await alert(f"tool failure: {result.error}")

Hooks see the ToolResult whether the call succeeded or failed. @after_tool can’t change the result; for that, use a sandbox.

Async vs sync

Both hook types accept async def or def. Sync hooks are dispatched to a worker thread via anyio.to_thread.run_sync so they don’t block the event loop:

@agent.after_tool
def log_sync(call, result):           # sync is fine
    open("audit.log", "a").write(...)

Redaction example

Strip API keys from tool args before they hit your logger:

from loomflow.security.secrets import _apply_redaction
 
@agent.before_tool
async def redact_args(call):
    redacted = _apply_redaction(repr(call.args))
    logger.info(f"calling {call.tool}({redacted})")
    return None

_apply_redaction masks common API-key shapes (OpenAI, Anthropic, AWS, GitHub PATs). For custom redaction, pass your own regex set.

Multi-tool denial pattern

Block a list of dangerous tool names without writing a custom permission policy:

DENIED = {"format_disk", "exec_arbitrary", "shutdown"}
 
@agent.before_tool
async def block_dangerous(call):
    if call.tool in DENIED:
        return PermissionDecision.deny_(f"{call.tool} is permanently blocked")
    return None

For longer-lived deny lists, StandardPermissions(denied_tools=...) is cleaner.

Telemetry from hooks

Hooks have access to the active RunContext via get_run_context():

from loomflow import get_run_context
 
@agent.after_tool
async def emit_metric(call, result):
    ctx = get_run_context()
    metric_client.increment(
        "agent.tool.calls",
        tags={
            "tool": call.tool,
            "user_id": ctx.user_id or "anonymous",
            "ok": result.ok,
        },
    )

OTelTelemetry already emits per-tool spans; use hooks for custom metric backends or business-specific tagging.