Hooks

Hooks are small functions you attach to one agent that run on either side of every tool call. Use them for checks and side effects that belong to this agent only: vetting arguments before a call, logging, metrics, alerts.

• @agent.before_tool runs before the call. Returning PermissionDecision.deny_(...) stops the call.
• @agent.after_tool runs after, with the ToolResult. Use it for logging, metrics and alerts.

@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      # no objection: the permissions layer decides next

Return values:

• PermissionDecision.deny_(reason): deny. The tool doesn’t run and the model gets a denied result carrying reason.
• Anything else (None, allow_(), ask_()): no objection. The call moves on to the permissions layer, which still applies in full.

Hooks can only deny. They can’t pre-approve a call or skip the permissions check, so an ask from the policy always reaches the approval handler. Multiple @before_tool hooks run in registration order and the first deny wins.

Each hook runs under a 5-second timeout. A before_tool hook that times out or raises denies the call, with a reason naming the hook: a check that never answered hasn’t approved anything. (Before 0.17 a timeout counted as no objection and a raising hook crashed the run.) To let timed-out calls through, pass your own registry:

from loomflow import Agent
from loomflow.security import HookRegistry
 
agent = Agent(
    "...",
    model="claude-opus-4-7",
    hooks=HookRegistry(hook_timeout_s=10.0, timeout_denies=False),
)

A hook that raises still denies with timeout_denies=False.

@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. Exceptions raised inside an after_tool hook are swallowed so they can’t break the run.

Hooks must be async

Write hooks as async def. The registry awaits every hook, so a plain def hook doesn’t work: in before_tool awaiting it raises TypeError, so every call is denied, and in after_tool it runs on the event loop and blocks it. To do blocking work (file I/O, a synchronous client), hand it to a worker thread so it doesn’t stall the event loop:

import anyio
 
def append_line(line: str) -> None:
    with open("tool-calls.log", "a") as f:
        f.write(line + "\n")
 
@agent.after_tool
async def log_to_file(call, result):
    await anyio.to_thread.run_sync(append_line, f"{call.tool} ok={result.ok}")

Redaction example

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

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

redact() masks common API-key shapes (OpenAI, Anthropic, AWS access keys, GitHub PATs). Every built-in Secrets backend has it; for other patterns, run your own regexes over the text.

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.