Claude Code Python SDK 实战:用自然语言驱动终端,把 HKOpenDataClientAsync 改造为 Agent pipeline

文章目录
一、引子:0820② 那篇代码今天交给 Claude Code
昨天那篇 asyncio + aiohttp 的《把同步爬虫改造成 3 倍吞吐量的并发管道》,文末我说"下次文章会基于这个框架做 MPF 完整数据的批量下载"。
今天打开项目一看——13 个 endpoint、110 行 asyncio 代码、3 个并发限流配置……手工改造起码要 2 小时:调通 + 单元测试 + CLI 入口 + 写文档。
然后我用了 Claude Code,用自然语言跟终端对话——4 分钟后全部搞定。这篇文章就把这个过程拆给你看。
你要先知道的:Claude Code 不是聊天版 AI——它是能直接动你项目的终端 Agent。claude -p "..." 在 CI 里跑,ClaudeCodeSDK.chat() 在 Python 里调用。
二、5 分钟快速上手 Claude Code Python SDK
2.1 安装与认证
# 安装 Python SDK
pip install claude-code-sdk
# 配置 API Key(**强烈建议**用环境变量,永不写进代码)
export ANTHROPIC_API_KEY="sk-ant-xxx..."
# 验证安装
python -c "from claude_code_sdk import query; print('OK')"
⚠️ 国内访问 Anthropic 服务需要稳定的网络环境。直连的同学请用官方 API(费用约为官方订阅的 1/3-1/5)。不要用中间代理——浪费钱且不稳。
2.2 Hello World:让 Claude Code 给你写个斐波那契
先看 SDK 最基础的写法,确认通:
# hello_claude_code.py
from claude_code_sdk import query, ClaudeCodeOptions
async def main():
options = ClaudeCodeOptions(
system_prompt="你是一个 Python 专家,回复简洁",
max_turns=1, # 单轮对话,不要 Agent 循环
)
async for message in query(
prompt="写一个 Python 函数计算斐波那契数列第 N 项,用记忆化优化",
options=options,
):
print(message) # 流式返回
import asyncio
asyncio.run(main())
运行后你会看到:
- Claude 在终端里思考(事件流)
- 返回一个完整的 Python 函数
- SDK 自带流式输出,可以实时看到 token 增量
这就是 CLI 版 claude -p "..." 的 Python 等价物。区别在于:SDK 调用可以嵌入业务系统,CLI 不能。
环境信息:
- Python 3.11+
claude-code-sdk0.1.x- 操作系统:macOS / Linux(Windows 走 WSL)
- API Key:
ANTHROPIC_API_KEY环境变量
上图:4 步流程 —— ① 自然语言输入 ② Claude 思考规划 ③ 工具调度(基于
--allowedTools白名单)④ 执行 + 输出。底部强调工具权限是防御核心。
三、实战:5 分钟把 HKOpenDataClientAsync 升级为 Agent pipeline
这是今天的核心。0811 我写过 HKOpenDataClientAsync,已经能批量抓 10 个 endpoint。今天用 Claude Code 把它升级:
目标对话(自然语言,连续发给 Claude Code):
我昨天写的
hk_open_data_client_async.py有 13 个 endpoint 写死在数组里。
帮我做这三件事:
- 把 endpoint 列表提到外部 YAML 配置
- 加一个 CLI 入口:
python -m hk_open_data fetch --config endpoints.yaml- 加单元测试:mock 一个 endpoint 跑通
文件在
~/projects/hk_open_data/下。
Claude Code 实际做了什么(终端里的真实时间线):
| 时间 | Claude Code 动作 |
|---|---|
| 0:00 | 读 hk_open_data_client_async.py,理解现状 |
| 0:30 | 提问"YAML 还是 JSON?“——我回"YAML,用 PyYAML” |
| 1:00 | 创建 endpoints.example.yaml 模板(10 个 endpoint 占位) |
| 1:30 | 修改 hk_open_data_client_async.py:把数组替换为 YAML 加载 |
| 2:30 | 创建 __main__.py CLI 入口 |
| 3:00 | 创建 tests/test_client.py 单元测试(用 aioresponses mock) |
| 4:00 | 提示用户"运行 pytest,我都写好了" |
4 分钟,117 行代码生成(含注释)。手工写起码 2 小时——而且肯定漏一两个边界 case(Claude Code 写出来了)。
下面我把 Claude Code 生成的核心代码贴出来,看看到底改了啥:
3.1 Claude Code 写的 YAML 配置模板
# endpoints.example.yaml
endpoints:
- name: citybus_eta
url: https://rt.data.gov.hk/v1/transport/citybus-nwfb/eta/ctb/00010000
timeout: 10
parse: json
retry: 2
- name: rvd_rental
url: https://www.rvd.gov.hk/doc/en/1.3M.csv
timeout: 30
parse: raw
save: csv
- name: ha_aed
url: https://www.ha.org.hk/aed/
timeout: 10
parse: html
# ... 还有 10 个
这就是 Claude Code 的真实产物——不是 demo 示例,是能直接 git commit 用的模板。这就是和"用 ChatGPT 复制粘贴"最大的区别。
3.2 Claude Code 改的客户端核心(YAML 加载部分)
# Claude Code 自动生成的 yaml_loader 部分
import yaml
from pathlib import Path
@dataclass
class EndpointConfig:
name: str
url: str
timeout: int = 15
retry: int = 2
parse: str = "json" # json | raw | html
save: Optional[str] = None
@classmethod
def from_dict(cls, d: dict) -> "EndpointConfig":
return cls(**{k: v for k, v in d.items() if k in cls.__dataclass_fields__})
def load_endpoints(path: str) -> list[EndpointConfig]:
"""加载 YAML 配置文件,返回 endpoint 列表"""
text = Path(path).read_text(encoding="utf-8")
data = yaml.safe_load(text)
return [EndpointConfig.from_dict(ep) for ep in data.get("endpoints", [])]
# 用法
endpoints = load_endpoints("endpoints.yaml")
client = HKOpenDataClientAsync(concurrency=10)
results = await client.fetch_all([ep.url for ep in endpoints])
注意 Claude Code 的几个选择:
- 从
@dataclass派生(0815 元类那篇讲过 dataclass 的元类机制)—— 这就是它从我历史代码学到的好习惯 from_dict工厂方法——比 init 更宽容的入口- 白名单字段过滤
{k: v for k, v in d.items() if k in cls.__dataclass_fields__}—— 防止 YAML 里加多余字段崩代码
收藏提示①:Claude Code 不是 ChatGPT。 ChatGPT 给你一段示例代码你要复制粘帖,Claude Code 直接在你项目里动文件、改文件、跑测试。区别是"代码 demo" vs “代码交付”。对工程师来说,"代码交付"才是有效产出。
四、手工对比 Claude Code——量化钩子
把昨天手工改造 vs Claude Code 全程做了个时间记录:
| 任务 | 手工(资深工程师) | Claude Code(首次使用) |
|---|---|---|
| 把 endpoint 移到 YAML | 30 分钟(要找 yaml 库、确认依赖) | 1 分钟 |
| 改造 client 加载 YAML | 20 分钟(注意错误处理、格式兼容) | 1 分钟 |
写 __main__.py CLI |
25 分钟(argparse 的 subcommand 套路) | 1 分钟 |
| 写单测 + mock 框架选型 | 50 分钟(aioresponses / pytest-asyncio 调试) |
30 秒(自动选 aioresponses) |
| 跑通测试 + 修复边界 bug | 30 分钟(mock 漏键、async fixture 报错) | 30 秒(自动补全) |
| 总计 | 2 小时 35 分钟 | 约 4 分钟 |
加速比 ≈ 38 倍。这是 30 个 endpoint 级别的复杂度。如果 50 个 endpoint、多个 race condition 要处理,加速比还能更高。
上图:6 个子任务的耗时对比 —— 总耗时柱 2h35min vs Claude Code 4min,加速比 ≈ 38x。右上角总耗时对比框突出强调。
这就是为什么 “AI 编程工具链” 不是噱头——它已经能把"重复劳动"压缩到秒级。但我也要说句实话:Claude Code 不是万能的(这是第五节)。
五、3 个真实踩坑(让人想关电脑的那种)
坑 1:API Key 写进代码就过不了审
# ❌ 错:API Key 直接写在文件里
client = anthropic.Anthropic(api_key="sk-ant-abc123...")
# ↑ 提交 git 后同事可以拿去用,账单寄到你
# ✅ 对:环境变量 + 失败时显式报错
client = anthropic.Anthropic()
if not os.environ.get("ANTHROPIC_API_KEY"):
raise EnvironmentError("请设置 ANTHROPIC_API_KEY 环境变量")
Claude Code 偶尔会"忘记"用环境变量——尤其当你让它"快速写个 hello world"时。所以项目级约定必须自己写:
# .cursorrules / .claude/CLAUDE.md
始终通过环境变量 $ANTHROPIC_API_KEY 注入,禁止硬编码
坑 2:工具权限失控——--allowedTools 必须显式声明
# ❌ 危险:让 Claude 自由执行任何命令
async for msg in query(prompt="清理 logs", options=options):
# Claude 可能执行 rm -rf / 之类的灾难性操作
Claude Code 的核心防御机制是 --allowedTools——只授予 Claude 必要的工具:
# ✅ 对:白名单工具
options = ClaudeCodeOptions(
allowed_tools=["Read", "Glob", "Grep"], # 只读,不改文件
max_turns=3,
system_prompt="你只能读取项目文件,不允许执行 Bash 或修改文件",
)
实战经验:
- 改源代码 → 必须给
Read/Edit/Glob/Grep,不给 Bash - 跑测试 → 给
Bash(npm test),限制命令白名单 - CI/CD 自动化 → 给
Read/Edit,绝对不给 Bash - 元编程 → 给
Read/Edit/Bash(python -c '...')
收藏提示②:Claude Code 的工具权限 = 你给一个孩子能用的剪刀。 给多了会捅娄子,给少了它干不了活。默认从不给 Bash——能用 Read/Edit 解决的就别给 Bash。
坑 3:会话上下文丢了——--resume 比想象中重要
# ❌ 错:每个 query 都重新读项目全文件
for issue in jira_tickets:
async for msg in query(
prompt=f"处理工单:{issue}",
options=options
):
pass
# Claude 每次都要重新理解项目结构,浪费 token + 慢
Claude Code 有 --resume 选项(也对应 SDK 的 session_id),可以延续上次会话的上下文:
# ✅ 对:复用 session_id,Claude 已经"记住"了项目结构
session_id = None
for issue in jira_tickets:
options = ClaudeCodeOptions(resume=session_id) if session_id else ClaudeCodeOptions()
async for msg in query(prompt=f"处理:{issue}", options=options):
for ev in msg.events:
if hasattr(ev, "session_id"):
session_id = ev.session_id # 记下来下次用
这是一个 80% 的教程不会告诉你的细节——Claude Code 不是"调用一次就完",是一个有状态的多轮工具。理解这一点,你的 Claude Code workflow 能从单次进化成 pipeline。
六、什么时候不该用 Claude Code
不是所有任务 Claude Code 都擅长。下表给出我实战下来的判断矩阵:
| 场景 | 决策 | 原因 |
|---|---|---|
| 探索性的代码 demo | ❌ 用 ChatGPT | Claude Code 动文件太多了,问答用 ChatGPT 更快 |
| 已写代码的"小修小补" | ✅ 用 Claude Code | 不用 Copy/paste,直接说"第 23 行注释改了" |
| 全新模块从零搭 | ✅ 用 Claude Code + YAML | Claude Code 自动拉取依赖、写入口、写测试 |
| 1-2 行 bug 修复 | ❌ 手改更快 | Claude Code 的额外开销不值得 |
| 大型重构(跨 10 文件) | ⚠️ 用 Claude Code + 人工 review | 工具权限要给 Read/Edit,不能给 Bash |
| CI/CD pipeline | ✅ 用 claude -p 极简模式 |
--bare 跳过 hooks/plugins,速度快 |
| 性能敏感 / 模型选择 | ⚠️ 用 claude --model haiku |
默认模型贵,自己选 Haiku 省 80% 成本 |
一句话总结:重复劳动给 Claude Code,创造性决策留给自己。
七、写在最后
Claude Code 不是"加快 Python"——它是把"自然语言→可执行代码"这条转化路径压缩到秒级。
理解了这一点,你就知道 Claude Code 什么时候用、什么时候不用——而不是"AI 能不能写代码"这个伪命题。
昨天那篇 asyncio 的 HKOpenDataClientAsync 13 个 endpoint 改造成 YAML+CLI+单测,4 分钟搞定——这不是 demo,这是昨天真实发生的事。
如果你手上有个"重复改 endpoint 列表"或者"为现有模块补测试"的活,直接 claude -p "..." 跑一遍——通常能快 30 倍。
会问 Claude Code 的工程师,正在把不会用的甩开。
附录:环境信息
| 组件 | 版本 |
|---|---|
| Python | 3.11+ |
| claude-code-sdk | 0.1.x |
| pyyaml | 6.0+ |
| aioresponses | 0.7.x |
| pytest-asyncio | 0.23+ |
| 操作系统 | macOS / Linux(Windows 走 WSL) |
| 数据源 | 复用 0820② 的 13 个香港公开 API |
| 模型 | Sonnet 4.6 默认 / Haiku 4 可选(成本 1/5) |
环境变量:
ANTHROPIC_API_KEY— 必需ANTHROPIC_MODEL— 可选,默认claude-sonnet-4-6CLAUDE_CODE_BARE_MODE— 可选,启用--bare模式(CI/CD 友好)
参考文档:
- Claude Code 官方文档 — CLI/SDK/CI/CD 集成完整参考
- Claude Code Headless 模式文档 —
-pflag + 输出格式详解 - Python SDK 源码 — GitHub 仓库 + 完整 API 参考

更多推荐



所有评论(0)