首页 › tutorials › Claude Agent SDK编程式自动化实战

概述

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

前置条件

与命令行 -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 
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() 调用

操作：

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

新建 hello.py：

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。

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：

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 开销。

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 拦截危险命令。

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()，每个有自己的工作目录和上下文。注意三件事：限流、隔离、聚合。

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，把成本写入日志。

# 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 每周执行。

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 提供了细粒度异常类型。
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 异常的基类，兜底
    ...

训练营适用性评估

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

延伸阅读

参考来源

  1. anthropics/claude-agent-sdk-python — GitHub README — 官方 SDK 源码与 query()/ClaudeSDKClient/@tool/Hooks 完整 API 与代码示例，本文所有代码均据此改写。
  2. Agent SDK overview — Claude Code 官方文档 — 官方对 SDK 能力全景（核心概念、Agent Loop、会话、流式、结构化输出、子代理、Hooks、OpenTelemetry）的总览。
  3. Building Powerful Agents with the Claude Agent SDK — 官方 Cookbook — 四 Notebook 进阶教程（研究 Agent、Chief of Staff、Observability、SRE 自愈），覆盖 MCP、Hooks 审计与多 Agent 编排。
  4. claude-agent-sdk — PyPI — 包发行信息，明确 query() 与 ClaudeSDKClient（自定义工具 + Hooks）的能力分界及 CLI 自动捆绑机制。
  5. Claude Agent SDK in Python: First Agent to Workflows — Augmentcode Guide — 第三方实战指南，补充 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学通信知识星球海报-扫码加入

📎 本分类最新文章