1.5　应用 Agent SDK 的 Hooks 拦截工具调用并规范化数据

这一条考什么

考你会不会用 Claude Agent SDK 的 Hooks 在两个关键位置插入确定性逻辑：工具调用发出前（PreToolUse，用于拦截违规操作并重定向到替代流程）和工具结果交给模型前（PostToolUse，用于规范化来自不同 MCP 工具的异构数据）。核心判断题是：当业务规则要求必然遵守时，选 Hook 而不是提示词。

核心概念

Hook 是什么

Hook 是 Agent SDK 在 Agent 循环的固定节点上调用的回调函数，由你的代码执行，而不是由模型决定是否执行。常见事件有 PreToolUse（工具执行前）、PostToolUse（工具执行后）、UserPromptSubmit、Stop 等。注册时用匹配器（matcher）按工具名过滤，例如只匹配 mcp__billing__process_refund。MCP 工具在 SDK 中的名字形如 mcp__<服务器名>__<工具名>。

from claude_agent_sdk import ClaudeAgentOptions, HookMatcher

options = ClaudeAgentOptions(
    hooks={
        "PreToolUse": [HookMatcher(matcher="mcp__billing__process_refund", hooks=[guard_refund])],
        "PostToolUse": [HookMatcher(matcher="mcp__crm__.*", hooks=[normalize_result])],
    }
)

回调签名是 async def hook(input_data, tool_use_id, context) -> dict。input_data 中包含 tool_name、tool_input，PostToolUse 时还包含 tool_response。

PreToolUse：拦截出站调用、强制合规

典型场景："超过 500 美元的退款必须人工审批"。PreToolUse Hook 在 process_refund 真正执行前检查 amount，超过阈值就返回拒绝决定，并通过拒绝原因告诉模型该走哪条替代流程：

async def guard_refund(input_data, tool_use_id, context):
    amount = float(input_data["tool_input"].get("amount", 0))
    if amount > 500:
        return {
            "hookSpecificOutput": {
                "hookEventName": "PreToolUse",
                "permissionDecision": "deny",
                "permissionDecisionReason": (
                    f"退款金额 {amount} 超过 500 美元授权上限。"
                    "请调用 escalate_to_human，并附上客户 ID、订单号、金额与原因。"
                ),
            }
        }
    return {}  # 放行

模型收到拒绝原因后，会按提示改为调用升级工具。Hook 还可以直接在代码里创建审批工单，实现真正的"重定向"。

PostToolUse：在模型处理前规范化数据

多个 MCP 服务器往往各有各的格式：CRM 返回 Unix 时间戳 1718000000，订单系统返回 ISO 8601 2024-06-10T08:13:20Z，物流系统用数字状态码 3 表示"已签收"。如果原样交给模型，它要在每一步自己换算，既浪费 token 又容易算错、比较出错。

PostToolUse Hook 在结果进入模型上下文之前把它们统一：时间统一成 ISO 8601（含时区），状态码映射成可读枚举（delivered），金额统一单位。实现方式上，可以通过 Hook 输出中的附加上下文字段把规范化后的结果提供给模型；对于 MCP 工具，SDK/Claude Code 也支持用 Hook 输出替换工具返回内容（具体字段名以所用 SDK 版本文档为准）。如果工具是你自己写的，在工具内部输出前规范化同样可行，但当数据来自多个第三方 MCP 服务器时，集中在一个 PostToolUse Hook 中处理最省事、最一致。

确定性保证 vs. 概率性遵守

• 提示词（含强调语气、few-shot）：模型"通常"遵守，适合风格、偏好、软性指导。
• Hook：每一次匹配的调用都会经过你的代码，结果确定，适合合规、资金、安全、数据格式等不允许出错的规则。

判断口诀："必须""绝不""合规""审计"→ Hook；"尽量""倾向""风格"→ 提示。

正确做法与反模式

• 正确：用 PreToolUse Hook 检查金额并拒绝超限退款，在拒绝原因中引导到人工升级。反模式：在系统提示里写"超过 500 美元绝不能退款"并寄希望于模型遵守。
• 正确：用 PostToolUse Hook 把 Unix 时间戳、ISO 8601、数字状态码统一成一种格式再交给模型。反模式：在系统提示里教模型"遇到数字时间戳请自行换算"。
• 正确：Hook 用 matcher 精准匹配目标工具，其余调用直接放行。反模式：在每个 Hook 里写一大堆与该工具无关的判断，造成误拦截。
• 正确：拒绝时给出可执行的替代路径。反模式：只返回"拒绝"，让模型反复重试同一调用。
• 正确：合规规则由 Hook 兜底，提示词说明规则以减少被拦截次数。反模式：认为两者二选一。

常见陷阱

• "在系统提示中加入 CRITICAL POLICY 强调措辞"——仍是概率性遵守，题目要求"保证"时一定不是最佳答案。
• "加入 400 / 500 / 600 美元的 few-shot 示例"——提升遵守率但不保证。
• "让退款工具超额时返回错误，提示请升级"——能挡住执行，但最佳答案通常是在调用执行前由 PreToolUse Hook 拦截并重定向，职责更清晰、不依赖每个后端都正确实现。
• "在系统提示里描述各种时间格式，让模型自行转换"——把确定性的数据处理交给概率性推理，易错且浪费上下文。
• "用一个计数 Hook 在达到 max_turns 的 80% 时强制升级"——用迭代次数代替业务判断，属于另一种反模式。

实操练习

目标：用 Agent SDK 搭一个进程内 MCP 工具，分别用 PreToolUse 拦截超额退款、用 PostToolUse 观察并规范化异构时间戳。

1. pip install claude-agent-sdk，保存为 hooks_lab.py。
2. 运行两次：一次请求退款 120 美元，一次请求退款 800 美元。
3. 查看终端输出中 Hook 的打印与模型的最终回复。

import asyncio, json
from datetime import datetime, timezone
from claude_agent_sdk import (
    query, ClaudeAgentOptions, HookMatcher, tool, create_sdk_mcp_server,
)

@tool("process_refund", "为订单退款。参数 order_id 与 amount（美元）。", {"order_id": str, "amount": float})
async def process_refund(args):
    return {"content": [{"type": "text", "text": json.dumps({"ok": True, **args})}]}

@tool("get_order", "查询订单，返回 created（Unix 秒）与 status_code（数字）。", {"order_id": str})
async def get_order(args):
    data = {"order_id": args["order_id"], "created": 1718000000, "status_code": 3}
    return {"content": [{"type": "text", "text": json.dumps(data)}]}

billing = create_sdk_mcp_server(name="billing", version="1.0.0", tools=[process_refund, get_order])

async def guard_refund(input_data, tool_use_id, context):
    amount = float(input_data["tool_input"].get("amount", 0))
    print(f"[PreToolUse] amount={amount}")
    if amount > 500:
        return {"hookSpecificOutput": {
            "hookEventName": "PreToolUse",
            "permissionDecision": "deny",
            "permissionDecisionReason": "超过 500 美元上限，请告知客户已转人工审批。",
        }}
    return {}

STATUS = {1: "created", 2: "shipped", 3: "delivered"}

async def normalize(input_data, tool_use_id, context):
    raw = input_data.get("tool_response")
    print("[PostToolUse] raw:", str(raw)[:200])
    try:
        text = raw[0]["text"] if isinstance(raw, list) else raw["content"][0]["text"]
        d = json.loads(text)
        d["created"] = datetime.fromtimestamp(d["created"], tz=timezone.utc).isoformat()
        d["status"] = STATUS.get(d.pop("status_code"), "unknown")
        note = "规范化后的订单数据（请以此为准）: " + json.dumps(d, ensure_ascii=False)
        return {"hookSpecificOutput": {"hookEventName": "PostToolUse", "additionalContext": note}}
    except Exception:
        return {}

async def main(prompt):
    opts = ClaudeAgentOptions(
        mcp_servers={"billing": billing},
        allowed_tools=["mcp__billing__process_refund", "mcp__billing__get_order"],
        hooks={
            "PreToolUse": [HookMatcher(matcher="mcp__billing__process_refund", hooks=[guard_refund])],
            "PostToolUse": [HookMatcher(matcher="mcp__billing__get_order", hooks=[normalize])],
        },
    )
    async for msg in query(prompt=prompt, options=opts):
        if hasattr(msg, "result"):
            print("结果:", msg.result)

asyncio.run(main("先查订单 A-1 的下单时间和状态，然后为它退款 800 美元。"))

检查结果：退款 800 时应看到 [PreToolUse] amount=800.0 且退款未执行，模型回复转人工；退款 120 时正常执行。PostToolUse 打印出原始的 Unix 时间戳，模型回复中使用的是 ISO 8601 时间与 delivered 状态。（tool_response 的具体结构随 SDK 版本可能不同，代码中做了兼容处理。）

自测清单

• PreToolUse 与 PostToolUse 分别在 Agent 循环的哪个节点触发？各适合解决什么问题？
• 如何用 PreToolUse Hook 拒绝超额退款，并把模型引导到人工升级流程？
• 为什么要在模型处理前统一时间戳和状态码，而不是在提示里教模型换算？
• 给出判断"该用 Hook 还是提示词"的标准。
• MCP 工具在 Hook matcher 中的名字长什么样？