<!-- title: Agent SDK：把Claude当库来编排 -->

## 概述

当你在命令行敲下 `claude -p` 时，调用的是一个黑盒进程；而当你 `pip install claude-agent-sdk` 后，你拿到的是**驱动 Claude Code 的整套内核**——相同的工具集、相同的 Agent 循环、相同的上下文管理——以一个 Python 库的形式嵌入你的程序。本教程聚焦 2026 年 Anthropic 官方力推的 **Claude Agent SDK**，讲解如何用 `query()`、`ClaudeSDKClient`、进程内自定义工具、应用层 Hooks 和权限回调，构建可编程、可中断、可审批的自动化 Agent 流水线。适用于已经熟悉 Claude Code CLI、希望把 Agent 能力嵌进 CI/CD、内部平台或批量任务系统的中高级开发者。

## 前置条件

- Python 3.10 及以上
- 一份有效的 Anthropic API 凭证（`ANTHROPIC_API_KEY`，或通过 Bedrock / Vertex 接入）
- 已用过 Claude Code CLI 的 `-p` 模式或交互模式，理解"工具调用"的基本概念
- 了解 Python `asyncio` / `anyio` 异步编程基础

---

## 第一步：为什么需要 Agent SDK（而非 `-p`）

很多开发者第一次接触"自动化 Claude"时，写的是这样的 shell 循环：

```bash
for repo in repos/*/; do
  claude -p "审计这个仓库的安全性" --output-format json > "report-$(basename $repo).json"
done
```

它能用，但有四道硬墙：

1. **进程边界**：每次 `claude -p` 都要重新启动 CLI、重新加载配置、重建上下文，启动开销可观。
2. **工具是封闭的**：你无法让 Claude 在同一个会话里调用你公司内部的 API（查内部 Wiki、查代码规范库、提交工单），除非把这些包装成独立 MCP 子进程。
3. **控制是黑盒的**：你无法在工具调用前插入一段确定性的 Python 逻辑（比如"凡是含 `rm -rf` 的命令一律拒绝"），也无法在工具调用后做实时审计。
4. **交互是单向的**：`-p` 是一次性问答，无法做多轮对话、并发收发、人工中途审批。

Agent SDK 把这四道墙拆了。`ClaudeSDKClient` 在你的 Python 进程里常驻，`@tool` 装饰器让一个普通 Python 函数秒变 Claude 可调用的工具（进程内 MCP 服务器，无 IPC 开销），`hooks` 让你在 Agent 循环的关键节点注入确定性逻辑，`can_use_tool` 回调让你实现真正的人工审批网关。

下面这张图说明了二者的本质差异：

```mermaid
flowchart LR
    subgraph Shell["-p 命令行模式"]
        A1[shell 循环] -->|每次启动进程| A2[CLI 进程]
        A2 -->|黑盒| A3[Claude]
    end
    subgraph SDK["Agent SDK 模式"]
        B1[Python 进程] --> B2[ClaudeSDKClient 常驻]
        B2 --> B3[bundled CLI 内核]
        B3 --> B4[Claude 模型]
        B1 -.注入.-> B5[自定义工具]
        B1 -.注入.-> B6[Hooks 拦截]
        B1 -.注入.-> B7[权限回调]
        B5 & B6 & B7 --> B2
    end
```

## 第二步：安装与首个 `query()`

安装只有一行，**CLI 内核随包捆绑，无需单独安装 Claude Code**：

```bash
pip install claude-agent-sdk
export ANTHROPIC_API_KEY=sk-ant-...
```

`query()` 是整个 SDK 的核心入口，它是一个**异步生成器**，逐条吐出会话中的消息。最简调用：

```python
import anyio
from claude_agent_sdk import query, ClaudeAgentOptions, AssistantMessage, TextBlock

async def main():
    options = ClaudeAgentOptions(
        system_prompt="你是一位严谨的代码评审员，只说结论。",
        max_turns=1,          # 限制 Agent 循环最多 1 轮，防失控
    )
    async for message in query(prompt="什么是死锁？一句话回答。", options=options):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    print(block.text)

anyio.run(main)
```

**预期输出**：一行简洁的死锁定义。

**常见问题**：
- 报 `CLINotFoundError`：旧版本需要单独装 CLI；新版已内置，若仍报错，可用 `ClaudeAgentOptions(cli_path="/path/to/claude")` 显式指定。
- 一直不返回：检查 `ANTHROPIC_API_KEY` 是否导出，或网络是否走代理。

## 第三步：解析消息流（4 类消息 × 3 类块）

`query()` 吐出的消息有四种类型，必须先理解它们才能写出可靠的解析逻辑：

| 消息类型 | 含义 | 典型内容块 |
|---------|------|-----------|
| `SystemMessage` | 系统初始化信息 | 通常忽略 |
| `AssistantMessage` | Claude 的发言 | `TextBlock`（文字）、`ToolUseBlock`（工具调用） |
| `UserMessage` | 注入回会话的内容 | `ToolResultBlock`（工具执行结果） |
| `ResultMessage` | 一轮会话结束 | 含 token 用量、耗时、成本 |

一个健壮的消息处理器长这样：

```python
from claude_agent_sdk import (
    AssistantMessage, UserMessage, ResultMessage, SystemMessage,
    TextBlock, ToolUseBlock, ToolResultBlock,
)

async def drain(async_iter, on_tool=None, on_cost=None):
    """统一消费消息流，分离文字、工具调用、工具结果与成本。"""
    async for msg in async_iter:
        if isinstance(msg, AssistantMessage):
            for block in msg.content:
                if isinstance(block, TextBlock):
                    print(f"[Claude] {block.text}")
                elif isinstance(block, ToolUseBlock):
                    if on_tool: on_tool(block.name, block.input)
        elif isinstance(msg, UserMessage):
            for block in msg.content:
                if isinstance(block, ToolResultBlock):
                    print(f"[Tool 返回] {block.content}")
        elif isinstance(msg, ResultMessage):
            if on_cost: on_cost(msg)
```

**预期结果**：你能清晰看到 Claude 何时在说话、何时在调工具、工具返回了什么、整轮花了多少 token——这是后续做"审计日志""成本上限熔断"的基础。

**常见问题**：忘记处理 `ToolResultBlock` 会导致你以为 Claude"卡住了"，其实它正在等工具结果回灌。

## 第四步：`ClaudeAgentOptions` 全配置

这是 SDK 里**唯一**的配置对象，几乎所有控制都通过它完成：

```python
options = ClaudeAgentOptions(
    system_prompt="你是只读审计员，绝不修改任何文件。",
    cwd="/home/gprshome/audit-target",     # 工作目录，等价于 --add-dir
    max_turns=8,                            # Agent 循环上限，防失控
    allowed_tools=["Read", "Grep", "Glob"], # 权限允许清单（自动批准）
    disallowed_tools=["Write", "Edit"],     # 显式禁用工具（真正从工具集移除）
    permission_mode="default",              # default / acceptEdits / plan / bypassPermissions
)
```

**关键易错点**：`allowed_tools` 是**权限允许清单**，列入其中的工具会被自动批准、不再弹权限请求；它**不会**把未列入的工具从 Claude 的工具集里删除。未列入的工具会落到 `permission_mode` 和 `can_use_tool` 来决定放不放行。要**彻底禁用**某个工具，必须用 `disallowed_tools`。这一点和很多人直觉相反，务必记牢。

## 第五步：自定义工具——把你的 Python 函数变成 Claude 的能力

这是 SDK 相对 `-p` 最大的杀手锏。用 `@tool` 装饰器 + `create_sdk_mcp_server()`，一个普通 async 函数就变成 Claude 可调用的工具，而且**运行在同一个 Python 进程内**，没有子进程、没有 IPC 序列化开销：

```python
from claude_agent_sdk import tool, create_sdk_mcp_server, ClaudeAgentOptions, ClaudeSDKClient

@tool("query_wiki", "查询公司内部知识库", {"keyword": str})
async def query_wiki(args):
    # 这里可以直接读数据库、调内部 API——任何 Python 能做的都行
    db = {"sso": "单点登录，见 TS 33.501", "amf": "接入与移动性管理功能"}
    text = db.get(args["keyword"].lower(), f"未找到: {args['keyword']}")
    return {"content": [{"type": "text", "text": text}]}

@tool("post_comment", "在工单系统发表评论", {"ticket": str, "body": str})
async def post_comment(args):
    # 真实场景：调用 Jira/GitLab API
    print(f">>> 已在 {args['ticket']} 发表：{args['body']}")
    return {"content": [{"type": "text", "text": "ok"}]}

server = create_sdk_mcp_server(name="corp-tools", version="1.0.0",
                               tools=[query_wiki, post_comment])

options = ClaudeAgentOptions(
    mcp_servers={"corp": server},
    allowed_tools=["mcp__corp__query_wiki", "mcp__corp__post_comment", "Read"],
)
```

工具命名规则是 `mcp__<服务器名>__<工具名>`，在 `allowed_tools` 里预批准它，就能免去权限弹窗。返回值结构固定为 `{"content": [{"type":"text","text":...}]}`；要标记错误，加 `"is_error": True`。

**预期结果**：Claude 在回答"我们系统里 AMF 是什么"时，会主动调用 `query_wiki`，拿到你的内部定义后再组织答案——它终于能"看见"你的私有数据了。

## 第六步：Hooks——应用层的确定性拦截

注意区分两类"hook"：Claude Code CLI 用的 `settings.json` 钩子是配置文件式的；**SDK 的 Hooks 是 Python 函数**，由 Claude Code **应用**（而非模型）在 Agent 循环特定节点调用。最常用的是 `PreToolUse`，可以在工具执行**前**返回一个权限决策：

```python
from claude_agent_sdk import HookMatcher

DANGEROUS = ["rm -rf", ">", "| sh", "sudo", "git push --force"]

async def guard_bash(input_data, tool_use_id, context):
    if input_data.get("tool_name") != "Bash":
        return {}
    cmd = input_data["tool_input"].get("command", "")
    for pattern in DANGEROUS:
        if pattern in cmd:
            return {
                "hookSpecificOutput": {
                    "hookEventName": "PreToolUse",
                    "permissionDecision": "deny",
                    "permissionDecisionReason": f"命中危险模式: {pattern}",
                }
            }
    return {}

options = ClaudeAgentOptions(
    allowed_tools=["Bash"],
    hooks={"PreToolUse": [HookMatcher(matcher="Bash", hooks=[guard_bash])]},
)
```

`permissionDecision` 可取 `"deny"` / `"allow"` / `"ask"`。这套机制比正则过滤命令更可靠，因为它是**官方循环内置的拦截点**，无法被 prompt 绕过。

## 第七步：`can_use_tool`——人工审批网关

当 `permission_mode="default"` 且工具不在 `allowed_tools` 里时，SDK 会通过 `can_use_tool` 回调征求你的裁决。这就是构建"人工在环"审批的地方：

```python
async def approval_gate(tool_name, tool_input, context):
    """高危操作弹人工确认，其余放行。"""
    if tool_name == "Bash" and any(p in tool_input.get("command","") for p in ["deploy","release"]):
        ans = input(f"\n⚠ 允许执行 {tool_input['command']} ? (y/n): ")
        return {"behavior": "allow" if ans.strip()=="y" else "deny",
                "updatedInput": tool_input}
    return {"behavior": "allow"}

options = ClaudeAgentOptions(
    permission_mode="default",
    can_use_tool=approval_gate,
)
```

> 注意：当前版本 `can_use_tool` 依赖流式模式（`ClaudeSDKClient`）才会触发，纯 `query()` 单次调用下若发现回调未触发，需要切换到 `ClaudeSDKClient` 或挂一个占位 hook。

## 第八步：`ClaudeSDKClient` 多轮与并发

`query()` 是一次性的；要支持多轮对话、并发收发、会话保持，必须用 `ClaudeSDKClient`：

```python
async with ClaudeSDKClient(options=options) as client:
    # 第一轮
    await client.query("审计这个目录的安全问题")
    async for msg in client.receive_response():
        display(msg)
    # 第二轮追问——复用同一上下文
    await client.query("刚才你说的 SQL 注入，给出修复代码")
    async for msg in client.receive_response():
        display(msg)
```

更进一步，可以用 `receive_messages()` 起一个后台协程持续接收，主协程持续 `client.query()` 投喂——实现"边问边答"的并发流水线。配合 `asyncio.gather`，就能让多个 `ClaudeSDKClient` 实例并发处理多个仓库。

## 实战案例：SDK 驱动的批量代码审计流水线

把上面七步合起来，做一个完整可运行的项目：读取多个仓库，对每个仓库并发审计，注入内部 Wiki 工具与工单评论工具，用 Hook 拦截危险命令，输出结构化报告。

目录结构：

```
audit-pipeline/
├── pipeline.py        # 主程序
├── tools.py           # 自定义工具
├── repos/             # 待审计仓库
└── reports/           # 输出目录
```

**`tools.py`**（进程内工具集）：

```python
from claude_agent_sdk import tool, create_sdk_mcp_server
import json, pathlib

@tool("read_policy", "读取公司安全编码规范", {"topic": str})
async def read_policy(args):
    policies = {
        "secrets": "禁止硬编码密钥，必须走环境变量或密钥管理服务",
        "sqli": "所有 SQL 必须参数化，禁止字符串拼接",
        "auth": "认证接口必须校验权限，默认拒绝",
    }
    return {"content":[{"type":"text","text":policies.get(args["topic"],"无此条目")}]}

@tool("write_report", "把审计报告写入文件", {"repo": str, "findings": str})
async def write_report(args):
    path = pathlib.Path("reports") / f"{args['repo']}.json"
    path.parent.mkdir(exist_ok=True)
    path.write_text(json.dumps({"repo": args["repo"], "findings": args["findings"]},
                               ensure_ascii=False, indent=2), encoding="utf-8")
    return {"content":[{"type":"text","text":f"已写入 {path}"}]}

def build_server():
    return create_sdk_mcp_server(name="audit", version="1.0.0",
                                 tools=[read_policy, write_report])
```

**`pipeline.py`**（主编排）：

```python
import asyncio, pathlib
from claude_agent_sdk import (
    ClaudeAgentOptions, ClaudeSDKClient, HookMatcher,
    AssistantMessage, ResultMessage, TextBlock,
)
from tools import build_server

DANGEROUS = ["rm -rf", "git push", ">", "curl"]

async def guard(input_data, tool_use_id, context):
    if input_data.get("tool_name") != "Bash":
        return {}
    cmd = input_data["tool_input"].get("command", "")
    if any(p in cmd for p in DANGEROUS):
        return {"hookSpecificOutput": {"hookEventName":"PreToolUse",
                "permissionDecision":"deny",
                "permissionDecisionReason":f"审计场景禁止: {p}"}}
    return {}

def options_for(repo_path):
    server = build_server()
    return ClaudeAgentOptions(
        system_prompt=f"你是安全审计员。只读审计 {repo_path}，"
                      "按 secrets/sqli/auth 三类列出问题，最后调用 write_report 落盘。",
        cwd=str(repo_path),
        max_turns=12,
        allowed_tools=["Read","Grep","Glob",
                       "mcp__audit__read_policy","mcp__audit__write_report"],
        mcp_servers={"audit": server},
        hooks={"PreToolUse":[HookMatcher(matcher="Bash", hooks=[guard])]},
    )

async def audit_one(repo_path):
    findings = []
    async with ClaudeSDKClient(options=options_for(repo_path)) as client:
        await client.query("开始审计，先读取所有源码，再对照公司规范给出问题清单。")
        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for b in msg.content:
                    if isinstance(b, TextBlock): findings.append(b.text)
    return repo_path.name, findings

async def main():
    repos = [p for p in pathlib.Path("repos").iterdir() if p.is_dir()]
    # 并发审计所有仓库——一个进程内多个客户端实例
    results = await asyncio.gather(*[audit_one(r) for r in repos])
    for name, findings in results:
        print(f"\n===== {name} =====")
        print("\n".join(findings)[:500])

asyncio.run(main())
```

**运行与预期结果**：

```bash
python pipeline.py
```

你会看到多个仓库被并发审计，每个仓库结束时 `reports/<repo>.json` 自动落盘；若 Claude 尝试执行 `rm` 或 `git push`，会被 Hook 当场拒绝并在会话里看到拒绝理由。整条流水线**单进程、无子进程、无外部 MCP**，且工具直接访问了你的私有规范库。

下面是这条流水线的运行时架构：

```mermaid
sequenceDiagram
    participant U as pipeline.py
    participant C as ClaudeSDKClient (×N 并发)
    participant H as PreToolUse Hook
    participant T as 进程内工具 (audit server)
    participant M as Claude 模型
    U->>C: query("审计该仓库")
    C->>M: 投递 prompt + 工具表
    loop Agent 循环
        M->>C: 调用 Read/Grep
        C->>H: PreToolUse 拦截校验
        H-->>C: allow/deny
        C->>T: 执行工具(进程内)
        T-->>C: 结果
        C->>M: 回灌结果
    end
    M->>T: write_report 落盘
    T-->>C: ok
    C-->>U: ResultMessage(含成本)
```

## 常见问题与排错

- **回调不触发**：`can_use_tool` 当前需要流式模式；用 `ClaudeSDKClient` 而非裸 `query()`。
- **工具不执行**：检查 `allowed_tools` 里工具名是否为 `mcp__<server>__<tool>` 三段式。
- **想禁用某工具却还在用**：误把工具名写进 `allowed_tools` 以为能"声明使用"，实际要用 `disallowed_tools` 才能真正移除。
- **成本失控**：给 `max_turns` 设硬上限，并在 `ResultMessage` 里累加 `cost`，超过阈值 `break`。
- **升级报错**：0.1.0 起 `ClaudeCodeOptions` 已重命名为 `ClaudeAgentOptions`，旧脚本要改。

## 延伸阅读

- 官方 Agent SDK 总览：https://code.claude.com/docs/en/agent-sdk/overview
- Headless 编程式运行：https://code.claude.com/docs/en/headless
- Python API 参考：https://code.claude.com/docs/en/agent-sdk/python
- 用户输入与审批回调：https://code.claude.com/docs/en/agent-sdk/user-input
- 官方仓库与示例：https://github.com/anthropics/claude-agent-sdk-python

## 训练营适用性评估

- **适合类型**：中高级 AI 工程训练营、企业内部 Agent 平台开发课、"把 LLM 嵌入业务系统"专题工作坊。不适合零基础入门营（需异步编程与 CLI 前置经验）。
- **教学时长**：建议 4 学时。其中 1 学时讲架构差异与 `query()`、1 学时讲自定义工具与 Hooks、1 学时讲权限回调与多轮、1 学时做实战案例上机。
- **考核建议**：让学员扩展实战案例，新增一个"危险操作人工审批网关"工具，并对超过 5 个仓库的并发做成本熔断——能独立完成即视为掌握。

## 参考来源

1. [Agent SDK overview — Claude Code Docs](https://code.claude.com/docs/en/agent-sdk/overview) — 官方 SDK 总览，明确"以库形式提供与 Claude Code 相同的工具、循环与上下文管理"
2. [claude-agent-sdk-python README — GitHub](https://github.com/anthropics/claude-agent-sdk-python/blob/main/README.md) — 官方仓库 README，提供 query()/ClaudeSDKClient/自定义工具/Hooks 的权威代码示例
3. [Run Claude Code programmatically — Claude Code Docs](https://code.claude.com/docs/en/headless) — Headless 编程式调用与 SDK 内核捆绑 CLI 的说明
4. [query() Function — DeepWiki](https://deepwiki.com/anthropics/claude-agent-sdk-python/3.1-query()-function) — query() 异步生成器与消息流的深度解析
5. [Handle approvals and user input — Claude Code Docs](https://code.claude.com/docs/en/agent-sdk/user-input) — can_use_tool 权限回调与人工审批机制的官方文档

> [!tip] 51学通信知识星球
> **星主：爱卫生**（微信：gprshome201101）——公众号「51学通信」创办者，19 年通信行业运维与讲师经验，著有《GPRS网络信令实例详解》《EPC原理与实践》《5G核心网原理与实践》三本纸质书（各大电商平台有售）。
> **星球定位**：深耕 5G 核心网、IMS、NFV/AI、云原生等 ICT 领域，主打运维岗位与信令分析。
> **当前规模**：5200 篇主题 / 990 个回答，198 元/年。
> **加入福利**：加入星球后请加爱卫生微信，拉您进知识星球年度陪伴与答疑群——不限量回复您的各种问题，另有群内分享。
> **AI 加持**：51学通信知识星球已作为 skill 接入 WorkBuddy，使用 deepseek-4-flash 在海量星球帖子中精准找贴、找视频、找 FAQ 和案例，做总结、找文档和 PPT 等场景全覆盖。

![51学通信知识星球海报-扫码加入](https://wiki.500nian.com/images/zhixingqiu_qrcode.png)