首页 › tutorials › 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、内部平台或批量任务系统的中高级开发者。

前置条件

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

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

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 回调让你实现真正的人工审批网关。

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

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：

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

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

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 用量、耗时、成本

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

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 里唯一的配置对象，几乎所有控制都通过它完成：

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 序列化开销：

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，可以在工具执行前返回一个权限决策：

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 回调征求你的裁决。这就是构建"人工在环"审批的地方：

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：

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（进程内工具集）：

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（主编排）：

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())

运行与预期结果：

python pipeline.py

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

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

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(含成本)

常见问题与排错

延伸阅读

训练营适用性评估

参考来源

  1. Agent SDK overview — Claude Code Docs — 官方 SDK 总览，明确"以库形式提供与 Claude Code 相同的工具、循环与上下文管理"
  2. claude-agent-sdk-python README — GitHub — 官方仓库 README，提供 query()/ClaudeSDKClient/自定义工具/Hooks 的权威代码示例
  3. Run Claude Code programmatically — Claude Code Docs — Headless 编程式调用与 SDK 内核捆绑 CLI 的说明
  4. query() Function — DeepWiki — query() 异步生成器与消息流的深度解析
  5. Handle approvals and user input — Claude Code Docs — 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学通信知识星球海报-扫码加入

📎 本分类最新文章