3.1 核心概念:subagent/hooks/权限/session 搭研究任务
第 2 章我们用 PydanticAI 把问答路径重写了一遍,收益是类型安全与依赖注入。但 PydanticAI 仍然是「库」——它帮你组织一次模型调用,至于子任务怎么拆、危险工具怎么拦、会话怎么续,全得自己写。这一节换一个方向:看一个把整套 Agent 运行时都内置好的 SDK。Claude Agent SDK 把 Claude Code 这个成熟产品的能力(子 Agent、钩子、权限、会话)直接暴露成 Python 接口,我们要弄清它到底给了什么。
本节把「框架」这一层的边界推进到运行时级:核对
claude-agent-sdk 0.2.165的真实 API 表面,讲清 subagent、hooks、权限、session 四块能力,并搭出一个「多 subagent + 一个 hook」的研究任务骨架。所有query()调用都需要活的 Claude 会话,本节只核对 API、不执行。
3.1.1 它不是模型 SDK,而是运行时 SDK
先纠正一个常见误解。Anthropic 官方有 anthropic 包,那是模型 SDK——你给它消息,它返回补全,一次调用一次响应。claude-agent-sdk 是另一层东西:它是Agent 运行时 SDK,封装的是本机已经装好的 claude 命令行工具。
这一点从实现上看得最清楚。SDK 启动时不是去连 api.anthropic.com,而是 spawn 一个 claude 子进程,通过 stdin/stdout 的控制协议跟它通信:
$ which claude
/opt/homebrew/bin/claude
$ ls -l /opt/homebrew/bin/claude
lrwxr-xr-x claude -> /opt/homebrew/Caskroom/claude-code/2.1.285/claude
也就是说,query() 背后真正干活的是本机那个 Claude Code 进程。这带来两个直接后果,贯穿本章:
- 好处:subagent 调度、工具权限、会话持久化、上下文压缩这些「重活」你一行都不用写,Claude Code 已经做了几年。
- 代价:你必须装
claudeCLI、必须登录、必须有额度。SDK 只是遥控器,电视还是那台 CLI。
本节所有代码都遵循这条红线:query() / ClaudeSDKClient 会 spawn 真实的 claude CLI,即真实 API 调用、真实消耗额度。本节一律不执行,只核对 API 表面。
3.1.2 API 表面现场核对
照本卷「禁止凭记忆写 API」的规矩,先看本机实测的导出符号:
import claude_agent_sdk as s
print([n for n in dir(s) if n in
("ClaudeAgentOptions", "AgentDefinition", "HookMatcher",
"query", "ClaudeSDKClient")])
['AgentDefinition', 'ClaudeAgentOptions', 'ClaudeSDKClient', 'HookMatcher', 'query']
五个核心符号都在。再看两个数据类的字段(dataclasses.fields 实测):
import dataclasses
from claude_agent_sdk import AgentDefinition, HookMatcher
for f in dataclasses.fields(AgentDefinition):
print(f.name)
description
prompt
tools
disallowedTools
model
skills
memory
mcpServers
initialPrompt
maxTurns
background
effort
permissionMode
for f in dataclasses.fields(HookMatcher):
print(f.name)
matcher
hooks
timeout
最后看两个入口的签名(inspect.signature 实测):
import inspect
from claude_agent_sdk import query, ClaudeSDKClient
print(inspect.signature(query))
print(inspect.signature(ClaudeSDKClient.__init__))
(*, prompt: str | collections.abc.AsyncIterable[dict[str, Any]], options: ClaudeAgentOptions | None = None, transport: Transport | None = None) -> collections.abc.AsyncIterator[UserMessage | AssistantMessage | SystemMessage | ResultMessage | StreamEvent | RateLimitEvent | ConversationResetMessage]
(self, options: ClaudeAgentOptions | None = None, transport: Transport | None = None)
三个事实从签名里直接读出来:query 是关键字参数(* 之后全是 kw-only)、返回的是异步迭代器(不是一次性返回值)、消息是流式多类型(AssistantMessage、ResultMessage 等)。这三点决定了 SDK 的编程模型:async for 消费、逐条处理事件。第 5.3 节会读它的调度源码,本节先把「怎么用」讲清。
3.1.3 subagent:把子任务交给独立 Agent
subagent 是 Claude Agent SDK 最省事的能力。在 minagent 里,要做一个「研究者 + 摘要者」的协作,得自己写 supervisor/worker 调度、消息总线、拓扑(卷一第 13 章整章都在讲这个)。在 SDK 里,你只要在 ClaudeAgentOptions.agents 里声明几个 AgentDefinition:
from claude_agent_sdk import AgentDefinition
agents = {
"researcher": AgentDefinition(
description="负责检索资料并返回要点,适合需要外部信息的问题",
prompt="你是资料检索员。用 WebSearch/WebFetch 找证据,返回带来源的要点。",
tools=["WebSearch", "WebFetch", "Read"],
),
"summarizer": AgentDefinition(
description="把长材料压缩成结构化摘要",
prompt="你是摘要员。把输入压成不超过 5 条的要点。",
tools=["Read"],
),
}
AgentDefinition 的必填字段只有两个:description(主 Agent 据此决定何时委派)和 prompt(子 Agent 的系统提示)。其余全部可选。注意字段是 camelCase(disallowedTools / mcpServers / maxTurns / permissionMode),这是刻意对齐 Claude Code 的 agent 配置格式,不是笔误——写 disallowed_tools 会静默无效。
调度这件事由主 Agent 自己完成:它读到各 subagent 的 description,判断某个子任务该交给谁,然后发起委派。你不需要写任何调度代码。这就是「内置省了什么」的第一个大项:subagent 的发现、匹配与调用,全部交给运行时。
3.1.4 hooks:在执行点插入你的逻辑
hooks 是 SDK 的第二个内置能力。minagent 的 on_step / on_tool_call 回调只能在事后被通知(卷一 3.3 节),而 hook 能在工具执行前拦截并改写。HookMatcher 只有三个字段:matcher(匹配哪些工具,正则字符串或 None 表示全部)、hooks(回调函数列表)、timeout(超时秒数)。
一个典型的「审计 + 拦截」hook:
from claude_agent_sdk import HookMatcher
async def audit_bash(input_data, tool_use_id, context):
command = input_data["tool_input"].get("command", "")
print(f"[audit] Bash: {command}")
if "rm -rf" in command:
return {
"hookSpecificOutput": {
"hookEventName": "PreToolUse",
"permissionDecision": "deny",
"permissionDecisionReason": "禁止执行 rm -rf",
}
}
return {}
hooks = {"PreToolUse": [HookMatcher(matcher="Bash", hooks=[audit_bash])]}
hook 回调的签名是 (input_data, tool_use_id, context) -> dict,input_data 是 PreToolUseHookInput(本机实测字段含 session_id / cwd / tool_name / tool_input / tool_use_id)。返回值用 hookSpecificOutput.permissionDecision 表达意图:allow / deny / ask / defer 四选一。这比「回调只读」强得多——hook 是可以改变执行结果的。
需要强调的是,hooks 与 agents 都不是通过命令行参数传给 claude 的。读源码可见,SDK 把它们打包进 initialize 控制请求:
# claude_agent_sdk/_internal/client.py(本机实测)
agents_dict = None
if configured_options.agents:
agents_dict = {
name: {k: v for k, v in asdict(agent_def).items() if v is not None}
for name, agent_def in configured_options.agents.items()
}
也就是说,subagent 与 hook 走的是控制协议(JSON over stdio),而 allowed_tools / permission_mode 这类则被翻译成 CLI 参数(--allowedTools / --permission-mode)。这个差别在 3.3 节讲「锁定」时会再出现一次。
3.1.5 权限与 session:两个容易忽略的内置项
权限在 3.2 节整节展开,这里先定位。ClaudeAgentOptions.permission_mode 本机实测是六档字面量:
typing.Literal['default', 'acceptEdits', 'plan', 'bypassPermissions', 'dontAsk', 'auto'] | None
default 是最保守的(每个需授权动作都问),bypassPermissions 是全放行(危险,只适合沙箱)。权限档位是运行时概念,不是模型提示——模型无法绕过它。这是 minagent 需要自己造、而 SDK 白送的东西。
session 是另一块。ClaudeAgentOptions 里与续跑相关的字段本机实测有:continue_conversation、resume、session_id、fork_session、resume_session_at。它们对应 CLI 的 --continue / --resume / --session-id / --fork-session。语义分层很清楚:
| 字段 | 含义 | 对应 CLI |
|---|---|---|
continue_conversation | 接着本目录最近一次会话继续 | --continue |
resume | 恢复指定 session id | --resume=<id> |
session_id | 显式指定本次会话 id | --session-id=<id> |
fork_session | 从现有会话分叉出新会话 | --fork-session |
读源码还发现一个安全细节:--resume 用的是 --resume=<value> 单 token 形式,而非两个 argv token,注释写明是为了防止「以短横线开头的值被解析成另一个 CLI 标志」的注入。这类细节你自研时很容易漏掉。
3.1.6 搭一个研究任务骨架
把四块拼起来,给 ResearchCopilot 的研究路径搭一个 SDK 版骨架:两个 subagent(检索 + 摘要)、一个审计 hook、只读权限档。
# research_sdk.py —— 需要活的 Claude 会话才能运行,本节未执行
import anyio
from claude_agent_sdk import (
ClaudeAgentOptions, AgentDefinition, HookMatcher, query,
AssistantMessage, ResultMessage,
)
async def audit(input_data, tool_use_id, context):
print(f"[hook] {input_data['tool_name']} 被调用")
return {}
OPTIONS = ClaudeAgentOptions(
system_prompt="你是研究助理。先检索、再摘要,最后给出带来源的结论。",
agents={
"researcher": AgentDefinition(
description="检索资料并返回带来源的要点",
prompt="用 WebSearch/WebFetch 找证据,返回要点与来源 URL。",
tools=["WebSearch", "WebFetch"],
),
"summarizer": AgentDefinition(
description="把长材料压缩成不超过 5 条要点",
prompt="只做压缩,不新增事实。",
),
},
allowed_tools=["WebSearch", "WebFetch", "Read"],
permission_mode="default",
hooks={"PreToolUse": [HookMatcher(matcher=None, hooks=[audit])]},
max_turns=12,
)
async def main():
async for msg in query(prompt="调研 2026 年 Agent 记忆系统的三种主流方案", options=OPTIONS):
if isinstance(msg, AssistantMessage):
for block in msg.content:
if getattr(block, "type", "") == "text":
print(block.text)
elif isinstance(msg, ResultMessage):
print("轮数:", msg.num_turns, "成本 USD:", msg.total_cost_usd)
anyio.run(main)
(本代码需要活的 Claude 会话才能运行,本节未执行,故不贴输出。)
这段骨架里有三处值得对照 minagent:没有循环(async for 消费,运行时的循环在 CLI 里)、没有调度器(agents 一声明,主 Agent 自己委派)、没有权限表(permission_mode 一句话)。ResultMessage 还免费带了 num_turns / total_cost_usd / usage——成本核算也是内置的,而 minagent 要自己在卷一第 15 章算。
3.1.7 这套框架替你做了什么、你又让渡了什么
| 能力 | minagent(卷一自研) | Claude Agent SDK |
|---|---|---|
| Agent 循环 | 自己写 for i in range(max_steps) | 内置在 claude CLI |
| subagent 调度 | 自己写 supervisor/worker + 消息总线 | agents= 声明即可 |
| 工具钩子 | 事后 on_tool_call 回调 | PreToolUse hook 可前置拦截 |
| 权限 | 自己写权限包装 + 审批 | permission_mode + 审批回调 |
| 会话持久化 | 自己写 checkpoint JSON | resume / continue 内置 |
| 成本核算 | 自己埋点统计 | ResultMessage.total_cost_usd 白送 |
让渡的同样清晰:你必须依赖 claude CLI(本机是 2.1.285)、模型绑定 Anthropic 家族、消息流是它定义的类型(AssistantMessage 等,不是你的 Message 契约)。换句话说,你换来了「几年打磨的运行时」,代价是把控制权交给了别人的进程。
对 ResearchCopilot 而言,取舍点在于:如果你要的正是「一个会自己查资料、自己管权限、自己记会话的 Agent」,SDK 几乎是最短路径;如果你要的是「完全掌控每一轮、把 Agent 嵌进自己的状态机」,那它反而挡路。下一节 3.2 工具与权限:allow/deny 与审批回调 会把权限这块单独拆开讲透;想回看 PydanticAI 与 minagent 的代码量对比,见 2.3 与 minagent 对照:代码量与可测性对比 。
阅读导航:上一节:2.3 与 minagent 对照:代码量与可测性对比 · 下一节:3.2 工具与权限:allow/deny 与审批回调 。
继续阅读
探索更多技术文章
浏览归档,发现更多关于系统设计、工具链和工程实践的内容。