生命不息,折腾不止。前面三课我们都在终端里敲 claude,今天换个玩法——把 Claude Code 变成你 Python / TypeScript 程序里的一个函数,随叫随到,想造啥自动化流水线都行。

前面我们讲过无头模式(claude -p),那已经能让 CI 里跑 Claude 了。但 -p 有个别扭的地方:它是「一个命令起一个进程、跑完就退」,消息是拼在 stdout 里当文本处理的,真要做到「程序里调用 Claude、拿到结构化的结果、还能传自定义工具」就得费劲 parse。Claude Agent SDK 就是官方把这件事做干净的答案——它把 Claude Code 那套 Agent 引擎(能读文件、跑命令、用工具、多轮思考)包装成了普通函数调用,一条 async for 就能流式拿到结构化的消息。

这一课我们就把它接起来,看完你就能写「程序化 Agent 流水线」。

一、先搞清 SDK 到底是个啥

Claude Agent SDK 是 Anthropic 官方出的库,就干一件事:用代码驱动 Claude Code。它有两个语言版本、两个包名:

1
2
3
4
5
# Python 版
pip install claude-agent-sdk

# TypeScript / Node 版
npm install @anthropic-ai/claude-agent-sdk

几个关键点先划好(省得后面踩坑):

  • 自带 Claude Code:SDK 包里把 Claude Code CLI 的二进制捆绑进来了,绝大多数情况你不用单独装 Claude Code。它默认用捆绑的那份;如果你想用系统里装好的特定版本,可以传 cli_path 指定路径(Python 里的 ClaudeAgentOptions(cli_path="/path/to/claude"))。
  • Python 版要求 3.10+,底层用 anyio 做异步,所以示例里多用 anyio.run 或 asyncio.run 包一下。
  • 两个入口:Python 侧有 query() 和 ClaudeSDKClient,TS 侧主要是 query()。区别后面细说,记住一句话——一次性任务用 query(),要多轮聊、要传自定义工具用 ClaudeSDKClient。

二、五分钟上手:Python 版 query()

先来个最小例子,感受下「调用一个 Agent」有多轻:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
import anyio
from claude_agent_sdk import query, ClaudeAgentOptions, AssistantMessage, TextBlock

async def main():
options = ClaudeAgentOptions(
system_prompt="你是一个言简意赅的助手",
max_turns=1, # 最多跑 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)

跑起来它会流式往外吐消息——不是等你一次拿全,而是一条一条 yield 出来,体验跟终端里看 Claude 打字一模一样。这里有个所有示例都会处理的细节:返回的是消息对象,不是纯文本。Assistant 的回复是 AssistantMessage,里面 content 是一堆 block(文本是 TextBlock,工具调用是 ToolUseBlock,工具结果另有类型),所以一般要先 isinstance 过滤一下再取 .text。

query() 默认每次调用新建一个 session,上一句的上下文不保留。想续聊有两个办法:给 options 传 continue_conversation=True(复用上一次的会话),或者干脆用下一节的 ClaudeSDKClient,它天生就是「同一个会话多轮」。

文末我会给一个「程序化流水线」的完整例子,先把配置项补上。

三、给 Agent 上权限:ClaudeAgentOptions

在程序里跑 Agent,最怕它闷头乱改文件、乱跑命令。ClaudeAgentOptions 就是控制面板,几个最实用的:

1
2
3
4
5
6
7
options = ClaudeAgentOptions(
system_prompt="你是资深 Python 后端,动手前先说思路",
allowed_tools=["Read", "Write", "Bash"], # 这几个工具不用再问,直接放行
permission_mode="acceptEdits", # 自动接受文件编辑
max_turns=10, # 最多轮数,防死循环
cwd="/path/to/your/project", # 工作目录,Claude 在这个目录里操作
)

逐条解释:

  • allowed_tools:自动放行(auto-approve)的工具白名单,值就是 Claude Code 里的工具名(Read / Write / Bash / Grep 等)。注意它管的是「要不要弹权限确认」,不是「工具能不能用」——没列进去的工具 Claude 依然能调,只是会触发权限提示。程序里没人给你点「允许」,所以无头场景务必把会用到的工具都加进来。
  • permission_mode:预设的权限策略。常用值 acceptEdits(自动接受文件编辑)、bypassPermissions(跳过确认,最省心但要慎用)。
  • max_turns:限制 Agent 最多转几轮(一轮 = 一次工具调用 + 一次思考)。无头跑长任务时这是最重要的「熔断开关」,防止 Claude 陷入无尽的 self-loop。
  • cwd:指定工作目录,Claude 的文件读写、命令执行都锚定在这里。

一个小坑提前说:在老版本 Python SDK(0.1.59 及更早)里,setting_sources=[] 这个写法是想「不读磁盘上的配置文件」,但当时空列表被当成「没传」处理、失效了,想彻底不用磁盘配置得升到新版。写自动化脚本时注意下 SDK 版本,别被旧行为坑了。

四、进阶:ClaudeSDKClient 带自定义工具 + 多轮续聊

query() 够了八成场景,但有两个它搞不定的硬需求:

  1. 想给 Claude 塞你自己的工具(查数据库、调自己接口、发通知……);
  2. 要在一个上下文里多轮对话(query() 每次新 session 得手动续,麻烦)。

ClaudeSDKClient 一次性解决。它支持用 Python 函数直接当工具(通过 @tool 装饰器),还能把会话一直保持住:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
from claude_agent_sdk import tool, create_sdk_mcp_server, ClaudeAgentOptions, ClaudeSDKClient

# 1) 用装饰器定义一个工具,参数类型用个 dict 描述
@tool("greet", "跟指定名字的人打个招呼", {"name": str})
async def greet_user(args):
return {
"content": [{"type": "text", "text": f"你好,{args['name']}!"}]
}

# 2) 把工具打包成一个 SDK 内置的 MCP server
server = create_sdk_mcp_server(name="my-tools", version="1.0.0", tools=[greet_user])

options = ClaudeAgentOptions(
mcp_servers={"tools": server},
allowed_tools=["mcp__tools__greet"], # 放行这个自定义工具,命名规则 mcp__<server名>__<工具名>
)

async with ClaudeSDKClient(options=options) as client:
await client.query("用 greet 工具跟 Alice 打个招呼")
async for msg in client.receive_response():
print(msg)

这里的自定义工具走的是 MCP 协议,所以工具名要按 mcp__<server 名>__<工具名> 的格式写进 allowed_tools。这跟 Claude Code 里接 MCP server 是同一套机制,只是 SDK 让你在代码里原地造 server,不用额外起一个 MCP 进程。

多轮续聊更简单,ClaudeSDKClient 天生保持会话:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
from claude_agent_sdk import ClaudeSDKClient, AssistantMessage, TextBlock

async def main():
async with ClaudeSDKClient() as client:
await client.query("法国首都是哪?")
async for m in client.receive_response():
if isinstance(m, AssistantMessage):
for b in m.content:
if isinstance(b, TextBlock):
print("Claude:", b.text)

# 接着问,上下文还在
await client.query("那这个城市的人口有多少?")
async for m in client.receive_response():
if isinstance(m, AssistantMessage):
for b in m.content:
if isinstance(b, TextBlock):
print("Claude:", b.text)

一个小提醒:遍历 receive_response() 时别用 break 提前跳出,可能触发 anyio/asyncio 的清理问题。要么自然遍历完,要么用个标志位记录「已经找到想要的」然后让它继续走完。

它还有一堆方法值得知道:interrupt() 中断当前生成、set_model() 运行时换模型、get_context_usage() 查上下文用量,多轮交互式应用(比如聊天界面)基本都齐了。

五、接中转站 + 常见坑 + TS 一句话

接中转站:SDK 驱动的是 Claude Code,所以接中转站跟 CLI 一模一样——靠两个环境变量,不需要在代码里写死:

1
2
export ANTHROPIC_BASE_URL="https://ai.aklibk.com"     # API 网关地址
export ANTHROPIC_AUTH_TOKEN="sk-xxxx" # 你的中转站密钥

这样 SDK 内部的 Claude Code 进程就会把请求打到你的中转站。Claude 这类国外模型,国内直接调官方麻烦、还得绑卡,走中转站就舒服了:国内直连、免绑卡、人民币按量付费,省心还便宜。SDK 本身不管「模型从哪来」,你把这两个变量一设,剩下的跟官方体验一致。

常见坑(错一次记一次):

1
2
3
4
5
6
7
8
9
10
11
12
13
from claude_agent_sdk import CLINotFoundError, ProcessError, CLIJSONDecodeError

try:
async for message in query(prompt="Hello"):
pass
except CLINotFoundError:
# 找不到捆绑的 CLI(极少见),手动装一份:
# curl -fsSL https://claude.ai/install.sh | bash
print("没找到 Claude Code,先装一下")
except ProcessError as e:
print(f"进程退出码:{e.exit_code}") # Claude 那边崩了
except CLIJSONDecodeError as e:
print(f"响应解析失败:{e}") # 输出不是合法 JSON,多半是版本不匹配
  • 报错了先看是上面哪一种,别盲目重试。ProcessError 带 exit code,能判断是网络问题还是权限问题。
  • Python 侧如果 asyncio.run 报和 anyio 相关的错,就换成 anyio.run(main),两者在 asyncio 后端下等价。

TypeScript 版一句话:装 @anthropic-ai/claude-agent-sdk 后,玩法完全对应——

1
2
3
4
5
import { query } from "@anthropic-ai/claude-agent-sdk";

for await (const message of query({ prompt: "帮我看看这个项目", options: { allowedTools: ["Read", "Grep"] } })) {
console.log(message);
}

TS 版的 query() 本身就更接近 Python 的 ClaudeSDKClient(内部维护连接、可多轮),还出了一个 V2 预览接口(send() / stream())把多轮对话写得更顺——具体以官方文档为准。

六、收个尾:一个程序化 Agent 流水线的骨架

把前面的串起来,一个「读完 README 自动总结 + 写个变更说明」的流水线大概长这样:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
import anyio
from claude_agent_sdk import query, ClaudeAgentOptions, AssistantMessage, TextBlock

async def run_agent(prompt: str) -> str:
out = []
options = ClaudeAgentOptions(
system_prompt="你是项目维护者,总结要简短、用中文",
allowed_tools=["Read", "Grep"],
permission_mode="acceptEdits",
max_turns=8,
cwd="/path/to/your/repo",
)
async for message in query(prompt=prompt, options=options):
if isinstance(message, AssistantMessage):
for block in message.content:
if isinstance(block, TextBlock):
out.append(block.text)
return "".join(out)

async def main():
summary = await run_agent("读一下 README 和最近 10 条提交,写三句话的变更说明")
print(summary)

anyio.run(main)

有了这个 run_agent,后面你想接 FastAPI 出接口、接定时任务自动跑、接 GitHook 自动 review,都是往上面堆的事。Claude Code 从「你在终端里指挥的小工」,正式变成了「你代码里能编排的一个组件」。

生命不息,折腾不止。下一课(Claude Code 第五课)我们把这个 SDK 挂到 FastAPI 上,起一个团队共用的 Agent HTTP 服务:POST 一个任务进来、流式把结果推回去,配上队列和限流,让它从「个人脚本」升级成「可对外服务的 Agent 后端」。到时候见。