☰
Claude Agent SDK Python 实战:用 async/await 搭一个可复现的 Agent 骨架
2026/10/1 20:43:17 网站建设 项目流程

1. 为什么你的第一个 Agent 总是跑不起来:从 Claude Agent SDK Python 的异步骨架说起

很多人第一次接触 Claude Agent SDK Python,卡住的地方往往不是模型能力,而是骨架没搭对。你可能已经看过官方文档里那句“围绕 async/await 模式设计的模块化架构”,也照着示例敲了几行,结果一运行就报RuntimeError: asyncio.run() cannot be called from a running event loop,或者工具调用返回了却不知道怎么把结果塞回对话。这不是你代码写得差,而是异步 Agent 的最小可运行骨架本身有几个必须踩准的点:事件循环、工具注册、消息回传、配置加载。

Claude Agent SDK Python 能做什么?简单说,它把“让大模型调用本地函数、读取文件、执行命令、再根据结果继续推理”这条链路封装成了异步接口。适合谁?适合已经会写 Python、想从“调一次 API 拿一段文本”升级到“让模型自己决定下一步做什么”的开发者。它和普通 Chat API 最大的区别是:普通 API 是你问一句它答一句;Agent SDK 是你给一个目标,它自己规划、调用工具、观察结果、再决定要不要继续。

我试过用最朴素的方式手写 while 循环加 function call,能跑,但一旦涉及流式输出和多轮工具调用,状态管理就会乱成一团。Claude Agent SDK Python 的价值在于它把 async/await 作为一等公民,工具调用天然就是 await 一个协程,流式事件也是异步迭代器。这意味着你不需要自己维护“等模型返回→解析工具名→执行→拼回消息”这套状态机,SDK 帮你管了。

但“帮你管了”不等于“你什么都不用配”。最小骨架需要三样东西:一个能加载配置的入口(settings.json 或 config.toml)、一个注册了至少一个工具的 Agent 实例、一个驱动 async 事件循环的 main 函数。缺任何一个,你都会在某个报错里卡住。下面我从配置骨架开始,一步步把这条链路跑通。

2. TaoToken 前置:把 Base URL、Key、Model ID 三件套配进 settings.json 与 config.toml

在写 Agent 代码之前,先把模型接入层配好。Claude Agent SDK Python 本身不绑定某一家服务,它通过 Base URL + API Key + Model ID 来定位模型。我用的接入地址是https://taotoken.net/api,控制台在https://taotoken.net/console,API Key 在https://taotoken.net/api-keys生成。这三个地址建议先记下来,后面配置里会反复用到。

配置骨架有两种常见形式:一种是settings.json,适合放在项目根目录让 SDK 自动读取;另一种是config.toml,适合你已经有 TOML 配置习惯的项目。两种我都给可复制片段,你选一种即可,不要混用。

先看settings.json。这个文件放在项目根目录,SDK 初始化时会按约定路径查找。注意base_url不要带末尾斜杠,model填你在控制台看到的模型 ID,api_key建议用环境变量引用而不是硬编码:

{ "agent": { "base_url": "https://taotoken.net/api", "api_key": "${TAOTOKEN_API_KEY}", "model": "claude-sonnet-4-20250514", "max_tokens": 4096, "timeout": 60 }, "tools": { "enabled": ["read_file", "write_file", "run_shell"], "workdir": "./workspace" } }

如果你更习惯 TOML,config.toml等价写法如下。注意 TOML 里字符串用双引号,布尔值小写,数组用方括号:

[agent] base_url = "https://taotoken.net/api" api_key = "${TAOTOKEN_API_KEY}" model = "claude-sonnet-4-20250514" max_tokens = 4096 timeout = 60 [tools] enabled = ["read_file", "write_file", "run_shell"] workdir = "./workspace"

这里有个容易踩的坑:api_key写成${TAOTOKEN_API_KEY}这种占位符,SDK 不一定自动展开,取决于你用的加载库。稳妥做法是在 Python 里用os.environ读出来再覆盖,或者直接用python-dotenv加载.env文件。我一般会在项目里放一个.env:

TAOTOKEN_API_KEY=sk-你的实际key

然后在代码入口处from dotenv import load_dotenv; load_dotenv()。这样settings.json里的占位符即使不展开,你也可以在构造 Agent 时手动传入api_key=os.environ["TAOTOKEN_API_KEY"]。

Model ID 怎么选?如果你只是跑通骨架,选一个响应快的即可;如果要做长链路工具调用,选上下文窗口大一点的。控制台的模型列表里会标注每个 ID 对应的能力,复制那个 ID 填进model字段就行。不要自己拼模型名,拼错了会报model not found。

三件套配好之后,先别急着写 Agent 逻辑,用一条 curl 验证接入层是通的:

curl -X POST https://taotoken.net/api/v1/messages \ -H "Content-Type: application/json" \ -H "x-api-key: $TAOTOKEN_API_KEY" \ -H "anthropic-version: 2023-06-01" \ -d '{ "model": "claude-sonnet-4-20250514", "max_tokens": 64, "messages": [{"role": "user", "content": "ping"}] }'

如果返回里有content字段和一段文本,说明 Base URL、Key、Model ID 三件套没问题。如果返回 401,先检查 Key 有没有复制全、有没有多余空格;如果返回 404,检查 Base URL 是不是写成了https://taotoken.net/api/v1而 SDK 又自动拼了一次/v1。这个细节后面排障章节会展开。

3. 可复制配置:async/await 驱动工具调用的最小 Agent 骨架

配置通了之后,进入代码骨架。Claude Agent SDK Python 的核心是异步的,所以你的入口函数必须是async def,最后用asyncio.run()启动。下面这个骨架我尽量压到最小,但保留了工具调用、流式输出、多轮循环三个关键点。

先装依赖。SDK 的包名以你实际安装为准,这里用通用写法:

pip install anthropic python-dotenv

然后建一个agent_skeleton.py。注意看注释里标出的四个关键位置:加载配置、注册工具、异步主循环、流式事件处理。

import asyncio import os import json from dotenv import load_dotenv load_dotenv() # 1. 从 settings.json 加载配置骨架 with open("settings.json", "r", encoding="utf-8") as f: settings = json.load(f) AGENT_CONF = settings["agent"] BASE_URL = AGENT_CONF["base_url"] MODEL = AGENT_CONF["model"] API_KEY = os.environ["TAOTOKEN_API_KEY"] # 2. 定义一个最小工具:读文件 def read_file(path: str) -> str: with open(path, "r", encoding="utf-8") as f: return f.read() # 工具描述,告诉模型这个工具能干什么、参数是什么 TOOLS = [ { "name": "read_file", "description": "读取指定路径的文本文件内容", "input_schema": { "type": "object", "properties": { "path": {"type": "string", "description": "文件相对路径"} }, "required": ["path"] } } ] # 3. 异步主循环:驱动工具调用 async def run_agent(user_input: str): from anthropic import AsyncAnthropic client = AsyncAnthropic( base_url=BASE_URL, api_key=API_KEY, ) messages = [{"role": "user", "content": user_input}] while True: # 流式请求 async with client.messages.stream( model=MODEL, max_tokens=AGENT_CONF["max_tokens"], tools=TOOLS, messages=messages, ) as stream: async for event in stream: # 4. 处理流式文本事件 if event.type == "content_block_delta": if hasattr(event.delta, "text"): print(event.delta.text, end="", flush=True) final = await stream.get_final_message() # 把模型回复追加进消息历史 messages.append({"role": "assistant", "content": final.content}) # 检查是否有工具调用 tool_uses = [b for b in final.content if b.type == "tool_use"] if not tool_uses: break # 执行工具,把结果拼回消息 tool_results = [] for tool_use in tool_uses: if tool_use.name == "read_file": try: result = read_file(tool_use.input["path"]) except Exception as e: result = f"读取失败: {e}" tool_results.append({ "type": "tool_result", "tool_use_id": tool_use.id, "content": result, }) messages.append({"role": "user", "content": tool_results}) return messages if __name__ == "__main__": asyncio.run(run_agent("读取 workspace/hello.txt 的内容并总结"))

这段代码里有几个设计点值得说清楚。第一,AsyncAnthropic的base_url直接传https://taotoken.net/api,SDK 内部会拼/v1/messages,所以你不要在 base_url 里再写/v1。第二,messages.stream()返回的是异步上下文管理器,必须用async with,否则流不会关闭。第三,工具结果回传时role是user,content是一个数组,每个元素带tool_use_id,这个 ID 必须和模型返回的tool_use.id对上,对不上模型会认为工具没执行。

如果你用config.toml,把加载部分换成tomllib(Python 3.11+)即可:

import tomllib with open("config.toml", "rb") as f: settings = tomllib.load(f)

其余逻辑完全一样。这样你就有了一个可复现的骨架:配置从文件来,工具是普通 Python 函数,主循环是 async/await,流式输出实时打印。

4. 验证请求:一次端到端 Agent 任务的成功结果长什么样

骨架写好了,现在跑一次端到端验证。先准备一个测试文件:

mkdir -p workspace echo "Claude Agent SDK Python 的最小骨架已经跑通。" > workspace/hello.txt

然后运行:

python agent_skeleton.py

预期你会看到类似这样的输出(流式打印,不是一次性出来):

正在读取文件... 文件内容是:Claude Agent SDK Python 的最小骨架已经跑通。 总结:这是一个验证 Agent 工具调用链路的测试文件。

如果看到这段输出,说明整条链路通了:模型收到用户指令 → 决定调用read_file→ SDK 把工具调用事件流式返回 → 你的代码执行read_file→ 结果回传 → 模型基于结果生成总结。这就是 async/await 驱动工具调用的完整闭环。

但“跑通”和“跑稳”是两回事。我建议你在验证阶段加一个日志,把每一轮的消息角色和工具调用名打出来,方便观察循环次数:

print(f"\n[round] messages={len(messages)} tool_uses={[t.name for t in tool_uses]}")

正常情况下,这个任务应该只循环两轮:第一轮模型返回工具调用,第二轮模型返回最终文本。如果循环超过三轮还没结束,可能是工具结果格式不对,模型一直在重试。

还有一个验证点是流式输出是否真的流式。你可以在content_block_delta分支里加时间戳:

import time print(f"[{time.time():.2f}] {event.delta.text}", end="", flush=True)

如果时间戳是递增的、文本是逐段出现的,说明流式生效。如果所有文本在同一时间戳出现,说明你用的不是 stream 接口,或者被缓冲了。

端到端验证通过后,你可以把read_file换成更复杂的工具,比如run_shell执行命令、write_file写文件。骨架不用改,只需要在TOOLS里加描述、在工具执行分支里加处理逻辑。这就是这个骨架的价值:工具可插拔,主循环不动。

5. 本篇常见错排查:401、local proxy failed、reading choices、OAuth 报错对照

跑骨架的过程中,报错基本集中在接入层和异步层。我把几个高频报错和对应排查方法列出来,你对照自己的终端输出定位。

401 Unauthorized。最常见的原因是 API Key 没读到。先确认.env文件在项目根目录,且load_dotenv()在读取os.environ之前调用。然后确认 Key 没有多余空格或换行。如果用的是settings.json里的${TAOTOKEN_API_KEY}占位符,确认你的加载逻辑真的展开了它。快速验证:在代码里print(API_KEY[:8]),看前几位是不是sk-开头。

local proxy failed / connection refused。这个报错通常出现在你本地配了某些网络工具,导致请求没走到https://taotoken.net/api。排查方法:先用第 2 节的 curl 命令单独测,如果 curl 通而 Python 不通,检查 Python 环境变量里有没有HTTP_PROXY、HTTPS_PROXY之类的设置。在代码开头加os.environ.pop("HTTP_PROXY", None)和os.environ.pop("HTTPS_PROXY", None)再试。

Error reading choices / reading choices 相关报错。这个一般出现在你混用了 OpenAI 风格的响应解析。Claude 的响应结构是content数组,不是choices。如果你在代码里写了response.choices[0],就会报这个。检查你的解析逻辑,统一用final.content遍历content_block。

OAuth / authentication_error。如果你在配置里同时写了api_key和某些 OAuth 相关字段,SDK 可能优先走 OAuth 流程导致失败。确保settings.json里只有api_key,没有auth_token、oauth之类的字段。另外,anthropic-version请求头必须是2023-06-01,写错版本也会报认证类错误。

model not found。Model ID 拼写错误,或者你选的模型在当前接入点不可用。去控制台复制准确的 ID,不要手打。注意有些模型 ID 带日期后缀,比如-20250514,漏掉就找不到。

asyncio.run() cannot be called from a running event loop。这个报错说明你在 Jupyter Notebook 或某个已经运行事件循环的环境里调用了asyncio.run()。解决办法是用await run_agent(...)直接在当前循环里跑,或者用nest_asyncio。在纯.py脚本里不会遇到这个问题。

工具结果回传后模型重复调用同一个工具。检查tool_result里的tool_use_id是否和模型返回的tool_use.id完全一致。大小写、空格都不能差。另外确认content字段是字符串,不是嵌套对象。

把这几条对照一遍,基本能覆盖 90% 的首次运行报错。如果还有别的报错,先把完整 traceback 贴出来,重点看最后一行和倒数第二行的文件路径,通常能定位到是配置加载、请求发送还是响应解析阶段的问题。

6. 语义一致 CTA:从骨架到长期编码 Agent 的下一步

骨架跑通之后,你手里就有了一个可复现的 async/await Agent 最小闭环。接下来往哪个方向走,取决于你的使用场景。

如果你只是想验证模型对话效果,可以直接在模型对话页面里试不同 prompt 和模型 ID,不用改代码:https://taotoken.net/models?utm_source=taotoken_aicg_blog_end&utm_content=agent_sdk_python&utm_campaign=rewrite。这个页面适合快速对比不同模型在同一个任务上的表现,省去反复改配置的麻烦。

如果你准备把这个骨架用到日常编码任务里,比如让 Agent 读代码、改文件、跑测试,那建议把 Key 和额度单独规划一下。长期跑 Agent 任务和偶尔调一次 API 的消耗模式不一样,Coding Plan 更适合这种持续调用的场景:https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=agent_sdk_python&utm_campaign=rewrite。它解决的是“我不想每次跑任务都担心额度”的问题。

如果你需要管理多个项目的 Key,或者给不同环境分配不同的 Key,控制台里的 API Keys 页面可以按项目建 Key:https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=agent_sdk_python&utm_campaign=rewrite。我一般会给本地开发、CI、生产各建一个,方便排查问题时快速定位是哪个环境出的错。

接入文档里有更完整的参数说明和错误码对照,遇到本文没覆盖的报错可以去查:https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=agent_sdk_python&utm_campaign=rewrite。文档里对max_tokens、timeout、流式事件类型的说明比本文更细,适合你往骨架里加新工具时参考。

最后说一个实用技巧:把settings.json里的workdir和工具的白名单路径绑定,不要让 Agent 有权限读写项目目录之外的文件。骨架阶段无所谓,但一旦接入真实项目,这个边界必须划清楚。你可以在read_file和write_file里加一层路径校验,用os.path.abspath判断目标路径是否在workdir之下,不在就返回错误。这样即使模型被诱导去读敏感文件,也会被工具层拦住。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询