<!-- title: Claude Agent SDK编程式自动化实战 -->

## 概述

本教程面向已熟悉 `claude -p` 命令行模式、希望把 Claude Code 升级为**可编程、可观测、可批量**的自动化引擎的开发者。你将学会用 Python 的 `claude-agent-sdk`（原 Claude Code SDK）以代码方式驱动 Agent：从一行 `query()` 调用，到双向流式会话、进程内自定义工具、Hooks 护栏，最后落地一个并发处理多个仓库的批量任务编排器。读完即可把 Claude Code 织进任何 CI、数据处理或研发流水线。适用人群：Python 开发者、平台/DevOps 工程师、Agent 应用构建者。

## 前置条件

- **Python 3.10+**，会写基本 `async`/`await` 与 `asyncio.gather`
- 一枚 **Anthropic API Key**（`platform.claude.ai` 获取）或已通过 `claude` 登录的 Claude Code 订阅
- 终端能联网；Claude Code CLI **无需单独安装**——`claude-agent-sdk` 安装时会自动捆绑对应版本的 CLI
- 建议用 `uv` 或 `venv` 隔离环境

> 与命令行 `-p` 模式相比，SDK 把"一次调用"升级为"一段可调试的 Python 程序"：你能拿到结构化的消息对象、并发跑多个会话、用 Python 函数定义工具和护栏，而不是靠 `jq` 解析 stdout。

## 核心概念：为什么需要 SDK

`claude -p "..."` 适合"一锤子"任务，但它有三个天花板：**输出是文本流、状态无法跨调用、并发全靠你手动起进程**。SDK 用三条原语打通它们：

| 能力 | `-p` 模式 | Agent SDK |
|---|---|---|
| 取结构化消息 | `jq` 解析 stdout | `AssistantMessage` / `TextBlock` / `ToolUseBlock` 对象 |
| 跨轮交互 | 每次重新起进程 | `ClaudeSDKClient` 双向流式会话 |
| 并发批量 | shell 循环 + `&` | `asyncio.gather` + 多个 `query()` |
| 自定义工具 | MCP 子进程 | `@tool` 进程内 MCP，零 IPC 开销 |
| 护栏/审计 | settings hooks | Python 函数级 Hooks |

```mermaid
flowchart LR
    A[你的 Python 程序] --> B[claude_agent_sdk]
    B -->|捆绑| C[Claude Code CLI]
    C --> D[Agent Loop]
    D -->|调用| E[内置工具 Read/Write/Bash]
    D -->|调用| F[进程内自定义工具 @tool]
    D -->|触发| G[Python Hooks 护栏]
    D -->|流式| H[AssistantMessage 对象]
    H --> A
```

## 步骤一：安装与首个 query() 调用

**操作：**

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

新建 `hello.py`：

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

async def main():
    async for message in query(prompt="用一句话解释什么是 PDU 会话"):
        # 只打印助手输出的文本块
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    print(block.text)

anyio.run(main)
```

**预期结果：** 运行 `python hello.py`，控制台输出一句对 PDU 会话的解释。SDK 首次启动会拉起捆绑的 CLI 子进程，耗时约 2–4 秒，之后稳定在毫秒级流式输出。

**常见问题：**
- *`CLINotFoundError`*：网络受限环境下 CLI 没捆绑成功。按 `curl -fsSL https://claude.ai/install.sh | bash` 单独装 CLI，再用 `ClaudeAgentOptions(cli_path="/path/to/claude")` 显式指定路径。
- *首次调用报 401*：确认 `ANTHROPIC_API_KEY` 已导出，或本机 `claude` 已完成登录（SDK 会复用订阅凭证）。

## 步骤二：用 ClaudeAgentOptions 掌控权限与边界

`query()` 的第二参数 `options` 是控制 Agent 行为的总开关。注意包名迁移：旧版叫 `ClaudeCodeOptions`，新版统一为 `ClaudeAgentOptions`。

```python
from pathlib import Path
from claude_agent_sdk import query, ClaudeAgentOptions

options = ClaudeAgentOptions(
    system_prompt="你是 3GPP 核心网专家，回答须引用规范编号",
    cwd=Path("/home/me/free5gc"),        # 工作目录，Read/Grep 在此范围内
    allowed_tools=["Read", "Grep", "Glob"],  # 自动批准的工具
    disallowed_tools=["Bash", "Write"],   # 彻底禁用的工具
    permission_mode="acceptEdits",        # 兜底权限模式
    max_turns=10,                         # 限制 Agent 循环轮数，防爆 token
)

async for message in query(
    prompt="找出所有处理 PFCP 的 Go 文件并总结职责",
    options=options,
):
    print(message)
```

**关键点：** `allowed_tools` 只是**自动批准**白名单，不改变工具集合；要"删掉"某工具，必须用 `disallowed_tools`。`max_turns` 是无人值守任务最重要的省钱旋钮——Agent 迷路时它强制终止。

**常见问题：**
- *Agent 一直转不停烧 token*：把 `max_turns` 降到 6–8，并在 `system_prompt` 里写明"完成后立即停止"。
- *误删了文件*：默认就应把 `Write`/`Edit`/`Bash` 放进 `disallowed_tools`，只读分析类任务零风险。

## 步骤三：ClaudeSDKClient 双向流式会话

`query()` 是"单向、跑完即弃"。若要多轮对话、动态追加指令，用 `ClaudeSDKClient`：

```python
import anyio
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions

async def main():
    options = ClaudeAgentOptions(allowed_tools=["Read"], cwd="/home/me/repo")
    async with ClaudeSDKClient(options=options) as client:
        await client.query("列出 src 下的顶层模块")
        async for msg in client.receive_response():
            print(msg)

        # 同一上下文里继续追问，状态保留
        await client.query("其中哪个模块负责鉴权？深入读它的入口文件")
        async for msg in client.receive_response():
            print(msg)

anyio.run(main)
```

**预期结果：** 第二个问题不会重新读一遍目录——`ClaudeSDKClient` 在两次 `query()` 间保持会话状态，省 token、更连贯。

## 步骤四：用 @tool 定义进程内自定义工具

这是 SDK 相对命令行模式最大的红利：把 Python 函数直接注册成 Claude 可调用的工具，**无需起独立 MCP 进程、零 IPC 开销**。

```python
import anyio
from claude_agent_sdk import (
    tool, create_sdk_mcp_server,
    ClaudeAgentOptions, ClaudeSDKClient,
)

@tool("count_files", "统计某扩展名的文件数量",
      {"directory": str, "extension": str})
async def count_files(args):
    from pathlib import Path
    d = Path(args["directory"])
    n = len(list(d.rglob(f"*.{args['extension']}")))
    return {"content": [{"type": "text", "text": f"共 {n} 个 .{args['extension']} 文件"}]}

server = create_sdk_mcp_server(name="stats", version="1.0.0", tools=[count_files])

options = ClaudeAgentOptions(
    mcp_servers={"stats": server},
    allowed_tools=["mcp__stats__count_files"],  # 注意 mcp__服务名__工具名 的命名规则
    permission_mode="acceptEdits",
)

async def main():
    async with ClaudeSDKClient(options=options) as client:
        await client.query("帮我统计 /home/me/free5gc 下有多少个 go 文件")
        async for msg in client.receive_response():
            print(msg)

anyio.run(main)
```

**关键点：** 工具名前缀必须是 `mcp__<服务名>__<工具名>`，把它写进 `allowed_tools` 才能免确认直接调用。进程内 MCP 的优势是部署为单进程、调试断点直接命中、类型安全。

## 步骤五：用 Python Hooks 织护栏与审计

Hooks 不是给 Claude 的提示，而是**应用层**在 Agent 循环关键点插入的确定性逻辑。最典型的是 `PreToolUse` 拦截危险命令。

```python
import anyio
from claude_agent_sdk import (
    ClaudeAgentOptions, ClaudeSDKClient, HookMatcher,
)

async def guard_bash(input_data, tool_use_id, context):
    if input_data["tool_name"] != "Bash":
        return {}
    cmd = input_data["tool_input"].get("command", "")
    forbidden = ["rm -rf", "drop table", "> /dev/sda", "sudo"]
    for bad in forbidden:
        if bad in cmd:
            return {
                "hookSpecificOutput": {
                    "hookEventName": "PreToolUse",
                    "permissionDecision": "deny",
                    "permissionDecisionReason": f"命中禁止模式: {bad}",
                }
            }
    return {}  # 放行

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

async def main():
    async with ClaudeSDKClient(options=options) as client:
        await client.query("运行 rm -rf /tmp/old_build 清理")
        async for msg in client.receive_response():
            print(msg)  # Agent 会被告知命令被拒，转而用更安全的方式

anyio.run(main)
```

**预期结果：** Agent 尝试执行 `rm -rf` 时被 Hook 拒绝，收到 `permissionDecisionReason`，随后主动改用 `rm -r` 或先确认目录。这比纯 `system_prompt` 劝说可靠得多，因为它是硬性阻断。

## 步骤六：批量任务的并发编排（fan-out）

"批量任务"的真正价值是**并发**。Python 的 `asyncio.gather` 让你同时跑 N 个独立 `query()`，每个有自己的工作目录和上下文。注意三件事：**限流、隔离、聚合**。

```python
import asyncio, json
from pathlib import Path
from claude_agent_sdk import query, ClaudeAgentOptions, AssistantMessage, TextBlock

MAX_CONCURRENCY = 4  # 并发上限，兼顾 Anthropic 速率限制与 token 预算

async def summarize_one(project_dir: Path, sem: asyncio.Semaphore):
    async with sem:  # 控制并发数
        options = ClaudeAgentOptions(
            cwd=project_dir,
            allowed_tools=["Read", "Grep", "Glob"],
            max_turns=6,
            system_prompt="你是技术文档作者，读完项目后输出 JSON："
                          '{"name":"","summary":"80字内","entry":["入口文件路径"]}',
        )
        chunks = []
        try:
            async for message in query(
                prompt="阅读本仓库结构，输出上述 JSON，不要任何额外文字",
                options=options,
            ):
                if isinstance(message, AssistantMessage):
                    for b in message.content:
                        if isinstance(b, TextBlock):
                            chunks.append(b.text)
        except Exception as e:
            return {"dir": str(project_dir), "error": str(e)}
        return {"dir": str(project_dir), "raw": "".join(chunks)}

async def main():
    dirs = [p for p in Path("/home/me/repos").iterdir() if p.is_dir()]
    sem = asyncio.Semaphore(MAX_CONCURRENCY)
    results = await asyncio.gather(*[summarize_one(d, sem) for d in dirs])
    Path("summaries.json").write_text(
        json.dumps(results, ensure_ascii=False, indent=2), encoding="utf-8")
    print(f"完成 {len(results)} 个仓库")

asyncio.run(main())
```

**关键点：**
- **限流**：用 `Semaphore` 把并发钉在 4–8，超出会撞 Anthropic API 的 RPM/TPM 限制报 429。
- **隔离**：每个任务独立 `cwd` + 独立上下文，互不污染，这是子进程 `-p` 循环做不到的干净隔离。
- **聚合**：`gather` 的返回顺序与输入一致，可直接 `json.dumps` 落盘。
- **健壮性**：单个任务 `try/except` 包住，一个仓库失败不会让整批崩溃。

**常见问题：**
- *批量 429*：把 `MAX_CONCURRENCY` 减半，或给每个任务加 `await asyncio.sleep(0.5)` 错峰。
- *JSON 不规范*：`max_turns` 调高，或在 prompt 里追加"只输出 JSON，以 `{` 开头 `}` 结尾"。

## 实战案例：多仓库文档巡检机器人

把前面六步组装成一个可复用工具：扫描若干 Git 仓库，并发生成/更新每个仓库的中文 `README.md`，并用 Hook 拦截任何 `git push`，把成本写入日志。

```python
# agent_doc_robot.py
import asyncio, json, time
from pathlib import Path
from claude_agent_sdk import (
    query, ClaudeAgentOptions, ClaudeSDKClient, HookMatcher,
    AssistantMessage, TextBlock,
)

async def cost_logger(input_data, tool_use_id, context):
    # PostToolUse 钩子：每次工具调用后记录，用于成本审计
    return {}

async def update_readme(repo: Path, sem, log):
    async with sem:
        options = ClaudeAgentOptions(
            cwd=repo,
            allowed_tools=["Read", "Grep", "Glob", "Write"],
            disallowed_tools=["Bash"],          # 不允许执行任何命令
            max_turns=8,
            hooks={"PostToolUse": [
                HookMatcher(matcher="Write", hooks=[cost_logger])]},
        )
        t0 = time.monotonic()
        try:
            async for message in query(
                prompt="阅读本仓库，若没有 README.md 则生成一份中文技术说明文档写回；"
                       "若已存在，则核对是否与现状一致并补充缺失章节。",
                options=options,
            ):
                pass  # 流式消费即可，写盘由 Agent 用 Write 完成
            log.append({"repo": str(repo), "ok": True,
                        "sec": round(time.monotonic() - t0, 1)})
        except Exception as e:
            log.append({"repo": str(repo), "ok": False, "error": str(e)})

async def main():
    repos = [p for p in Path("/home/me/repos").iterdir() if (p / ".git").exists()]
    sem = asyncio.Semaphore(5)
    log = []
    await asyncio.gather(*[update_readme(r, sem, log) for r in repos])
    Path("doc_robot_log.json").write_text(
        json.dumps(log, ensure_ascii=False, indent=2), encoding="utf-8")

asyncio.run(main())
```

跑完后你会得到：每个仓库根目录多出/更新了 `README.md`，外加一份 `doc_robot_log.json` 记录每个仓库耗时与成败。整个流程无人值守、可放进 `cron` 每周执行。

```mermaid
sequenceDiagram
    participant P as Python 编排器
    participant S as claude_agent_sdk
    participant A as Agent Loop (×N 并发)
    participant F as 文件系统/仓库
    P->>S: asyncio.gather(query×N)
    S->>A: 各自 cwd=repo_i
    A->>F: Read/Grep 扫描
    A->>A: Hooks 校验 (禁 Bash)
    A->>F: Write README.md
    A-->>S: AssistantMessage 流
    S-->>P: 结果聚合 + 日志落盘
```

## 进阶：结构化输出与错误处理

当你需要把 Agent 输出喂给下游程序，**结构化提取**比正则靠谱。两类技巧：

1. **prompt 强约束 + 类型断言**：要求 Agent 只输出 JSON，拿到 `TextBlock` 后 `json.loads`，失败则重试。
2. **错误分级捕获**：SDK 提供了细粒度异常类型。

```python
from claude_agent_sdk import (
    ClaudeSDKError, CLINotFoundError, CLIConnectionError,
    ProcessError, CLIJSONDecodeError,
)

try:
    async for message in query(prompt="..."):
        ...
except CLINotFoundError:
    # 指引安装 CLI 或设 cli_path
    ...
except CLIJSONDecodeError as e:
    # CLI 返回了无法解析的 JSON，多为 CLI 版本不匹配
    ...
except ProcessError as e:
    print(f"子进程退出码 {e.exit_code}")
except ClaudeSDKError:
    # 所有 SDK 异常的基类，兜底
    ...
```

## 训练营适用性评估

- **Python 自动化训练营（中级，2–3 课时）**：★★★★★ 极契合。从 `query()` 到并发批量，刚好覆盖一节完整进阶课。
- **DevOps/SRE 训练营（2 课时）**：★★★★☆ Hooks 护栏 + MCP 集成 + 巡检机器人案例直击运维痛点。
- **通信/网络工程训练营（1.5 课时）**：★★★☆☆ 可把 `cwd` 指向 free5GC/UERANSIM 源码，让 Agent 做规范溯源与代码导览。
- **零基础入门营**：★☆☆☆☆ 不适合，需要 Python `async` 与包管理基础，建议先学入门篇。

预计教学时长：理论 + 五步演示约 90 分钟；完整批量案例动手约 60 分钟；共约 2.5 小时含答疑。

## 延伸阅读

- 官方四 Notebook 进阶系列（`anthropic-cookbook/claude_agent_sdk`）：从一行研究 Agent 到多 Agent 编排与 SRE 自愈，是本教程的自然延伸。
- `ClaudeSDKClient` 的会话持久化与 Session Forking：新版 SDK 支持把会话存到外部存储并分叉，适合做"检查点 + 重放"。
- OpenTelemetry 可观测性：SDK 原生支持导出 span/metric/usage，配合步骤五的 Hooks 可构建完整审计链。
- TypeScript 版 `@anthropic-ai/claude-agent-sdk`：API 与 Python 版对称，前端/全栈团队可无缝迁移。

## 参考来源

1. [anthropics/claude-agent-sdk-python — GitHub README](https://github.com/anthropics/claude-agent-sdk-python) — 官方 SDK 源码与 `query()`/`ClaudeSDKClient`/`@tool`/Hooks 完整 API 与代码示例，本文所有代码均据此改写。
2. [Agent SDK overview — Claude Code 官方文档](https://code.claude.com/docs/en/agent-sdk/overview) — 官方对 SDK 能力全景（核心概念、Agent Loop、会话、流式、结构化输出、子代理、Hooks、OpenTelemetry）的总览。
3. [Building Powerful Agents with the Claude Agent SDK — 官方 Cookbook](https://github.com/anthropics/claude-cookbooks/tree/main/claude_agent_sdk) — 四 Notebook 进阶教程（研究 Agent、Chief of Staff、Observability、SRE 自愈），覆盖 MCP、Hooks 审计与多 Agent 编排。
4. [claude-agent-sdk — PyPI](https://pypi.org/project/claude-agent-sdk/) — 包发行信息，明确 `query()` 与 `ClaudeSDKClient`（自定义工具 + Hooks）的能力分界及 CLI 自动捆绑机制。
5. [Claude Agent SDK in Python: First Agent to Workflows — Augmentcode Guide](https://www.augmentcode.com/guides/claude-agent-sdk-python) — 第三方实战指南，补充 `query()`、`ClaudeSDKClient`、自定义 MCP 工具与工作流编排的工程实践。

> [!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)