数据截至 (上游 commit a4eaba4a56f9)
扩展点 — 进程内工具、权限回调、钩子
本章讲 SDK 给你的三个"插手"口子。它们都建立在第 02 章那条控制协议上:CLI 发反向请求,SDK 调你的 Python 代码,把结果送回去。
1. 三个口子先分清
| 口子 | 你提供什么 | 什么时候触发 | 目的 |
|---|---|---|---|
@tool + create_sdk_mcp_server | 一个 async 函数 | 模型决定调这个工具时 | 给模型加能力 |
can_use_tool | 一个权限回调 | CLI 权限规则判成"要问"时 | 决定工具准不准用 |
hooks | 一批钩子函数 | 生命周期节点(工具前/后、提交、停止…) | 观察/干预流程 |
关键区别(权限 vs 钩子),来自 types.py:2110-2139 的 docstring:
can_use_tool只在 CLI 权限规则评估为"ask"时触发。已被allowed_tools、permission_mode、settings 里allow规则放行的调用,压根不会问它。- 想观察/拦截每一个工具调用(不管权限规则),用
PreToolUse钩子。
2. 进程内工具:@tool
2.1 它要解决的小问题
"我想让模型能调用我 Python 里的一个函数(比如查我自己的数据库),而不用另起一个 MCP 服务器进程。"
2.2 直觉:进程内 vs 外部 MCP
传统 MCP 服务器是独立进程,靠 IPC 通信。SDK MCP 服务器跑在你的 Python 进程里(__init__.py:313 docstring),好处:没 IPC 开销、单进程好部署、能直接访问你应用的状态。
2.3 原理演示
# 示意,非源码:定义一个进程内工具服务器
from claude_agent_sdk import tool, create_sdk_mcp_server, ClaudeAgentOptions
@tool("add", "Add two numbers", {"a": float, "b": float})
async def add(args): # 函数必须是 async
return {"content": [{"type": "text", "text": f"{args['a'] + args['b']}"}]}
calc = create_sdk_mcp_server(name="calc", tools=[add])
options = ClaudeAgentOptions(mcp_servers={"calc": calc}, allowed_tools=["add"])
真实可运行版见 examples/mcp_calculator.py。
2.4 真实实现:两步
第一步,tool 装饰器(__init__.py:169)只是把元数据打包成一个 SdkMcpTool dataclass(__init__.py:158):name、description、input_schema、handler。不做别的。
第二步,create_sdk_mcp_server(__init__.py:310)才是重头:
- 建一个真正的
mcp.server.Server实例。 - 把每个工具的输入 schema 预计算成 JSON Schema(
_build_schema,__init__.py:402),创建时算一次、缓存。 - 注册
list_tools和call_tool两个 handler;call_tool里把你返回的{"content":[...]}翻译成 MCP 的CallToolResult(__init__.py:454-520)。 - 返回一个
McpSdkServerConfig(type="sdk",带instance)。
2.5 精妙处:Python 类型 → JSON Schema
_python_type_to_json_schema(__init__.py:238)让你能用 {"a": float} 甚至 TypedDict 当 schema,而不必手写 JSON Schema。它递归处理 Annotated(取描述)、Optional/Union(单个非 None 就解包)、list/dict、嵌套 TypedDict(_typeddict_to_json_schema,__init__.py:292)。这是纯粹的"开发体验"打磨。
2.6 实例怎么留在本进程
回顾第 02 章 2.3:传给 CLI 的 --mcp-config 里,SDK 服务器被剥掉 instance 字段(subprocess_cli.py:662-667),只告诉 CLI"有这么个 sdk 服务器"。真正的 Server 实例留在 Query.sdk_mcp_servers,当 CLI 发 mcp_message 反向请求时,由 _handle_sdk_mcp_request 转给包着它的桥接会话执行(query.py:645-674)。这就是"进程内"的实现真相。
3. 权限回调:can_use_tool
3.1 权限模式先看
permission_mode(types.py:1991)是粗粒度总开关:
| 模式 | 行为 |
|---|---|
default | 危险操作要问 |
acceptEdits | 自动接受文件编辑 |
bypassPermissions | 全部放行(慎用) |
plan | 只规划、不执行工具 |
dontAsk | 不问;没预批准的就拒 |
auto | 模型分类器逐个批/拒(PermissionMode 字面量见 types.py:25-27) |
3.2 求值顺序
工具调用来了
├─ 在 allowed_tools 里? ──── 是 ─▶ 直接放行
├─ permission_mode 定了? ── acceptEdits/bypass ─▶ 放行
├─ settings 的 allow 规则命中? ─▶ 放行
└─ 都没 → 规则评估为 "ask" ─▶ 调 can_use_tool 回调
依据:README 的 permissions 说明 + types.py:2110-2126 的 can_use_tool docstring。
3.3 回调签名与返回
CanUseTool 类型(types.py:257):(tool_name, input, context) -> PermissionResult。返回二选一:
PermissionResultAllow(types.py:238):可选updated_input(改写入参)、updated_permissions(顺带更新权限规则)。PermissionResultDeny(types.py:247):带message,可选interrupt=True(顺便打断)。
_handle_control_request 的 can_use_tool 分支(query.py:478-530)负责把这两种结果翻译成控制协议的 dict。ToolPermissionContext(types.py:202)给回调带了很多上下文:tool_use_id、blocked_path、title(现成的权限提示句)、suggestions(CLI 建议的权限更新)等。
3.4 一个硬约束
can_use_tool 和 permission_prompt_tool_name 互斥,同时设置直接抛 ValueError;只设回调时,SDK 自动把 permission_prompt_tool_name 设成 "stdio" 走控制协议。这套校验由 query() 与 ClaudeSDKClient.connect() 共用的 _configure_can_use_tool(types.py:1896-1921)完成,两个入口分别在哪调用见 _internal/client.py:81 与 client.py:142。(旧版还要求 prompt 必须是 AsyncIterable、传 str 就抛错;新版已取消这条 streaming 限制。)
4. 钩子:hooks
4.1 能挂哪些事件
HookEvent(types.py:263)列了十种:PreToolUse、PostToolUse、PostToolUseFailure、UserPromptSubmit、Stop、SubagentStop、PreCompact、Notification、SubagentStart、PermissionRequest。
4.2 怎么配
用 HookMatcher(types.py:589):matcher 是工具名匹配模式(如 "Bash",或 None 匹配全部),hooks 是回调列表。真实例子(examples/hooks.py:162):
# 示意,非源码:拦截含 foo.sh 的 bash 命令
options = ClaudeAgentOptions(
allowed_tools=["Bash"],
hooks={"PreToolUse": [HookMatcher(matcher="Bash", hooks=[check_bash_command])]},
)
钩子函数返回一个 HookJSONOutput,里面 hookSpecificOutput 可带 permissionDecision: "deny" 来拦(examples/hooks.py:62-68)。
4.3 两个坑
- 并发派发:同一事件注册的多个 matcher 是 CLI 并行触发的,不保证顺序(
types.py:2134-2139)。每个钩子要设计成独立的。 - 字段名转换:Python 侧用
async_/continue_(避开关键字),写回 CLI 前_convert_hook_output_for_cli(query.py:83)把它们转回async/continue。
4.4 initialize 时的注册
钩子不走命令行,而是在 initialize 里为每个回调分配 hook_{id}、建立 id→函数表(query.py:245-259)。之后 CLI 触发时发 hook_callback 控制请求带 callback_id,SDK 靠这张表找到函数(query.py:532-546)。
5. 代码地图
| 主题 | 文件路径 | 符号名 |
|---|---|---|
| 工具装饰器 | src/claude_agent_sdk/__init__.py | tool SdkMcpTool |
| 建服务器 | 同上 | create_sdk_mcp_server _build_schema |
| 类型转 schema | 同上 | _python_type_to_json_schema _typeddict_to_json_schema |
| 权限类型 | src/claude_agent_sdk/types.py | CanUseTool PermissionResultAllow PermissionResultDeny ToolPermissionContext |
| 权限模式 | 同上 | ClaudeAgentOptions.permission_mode |
| 钩子类型 | 同上 | HookEvent HookMatcher |
| 反向请求处理 | src/claude_agent_sdk/_internal/query.py | _handle_control_request _handle_sdk_mcp_request |
| 钩子字段转换 | 同上 | _convert_hook_output_for_cli |