1. 从一次工具调用失败说起Agent Loop 到底难在哪如果你正在自建 Agent大概率遇到过这种场景模型说要调用工具工具也执行了但下一轮模型却像失忆一样完全不知道刚才发生了什么。或者更糟用户中途按了 CtrlC整个会话历史直接错位后面每一轮都在报错。这不是模型笨而是 Agent Loop 的上下文管理没做对。Claude Code 的源码里Agent Loop 的核心其实就一句话用 for 循环跑任务max_rounds 设成 50每一轮判断有没有 tool call有就执行工具、把结果塞回 history没有就结束。听起来简单但真正难的是那些边界情况——工具报错怎么分类、中断怎么恢复、上下文太长怎么压缩、多个工具能不能并行、子 Agent 怎么隔离。这篇面向想自建 Agent 的开发者把 Claude Code 里 Agent Loop 与 Tool Call 的关键设计拆开讲并给出可复制的 settings.json 骨架和 TaoToken 统一 Key 配置最后演示一次完整的工具调用链路验证。你不需要读完整个源码跟着做就能理解上下文工程和 Subagent 拆分的思路。2. 前置准备用 TaoToken 统一管理模型 Key在动手写 Agent Loop 之前先把模型接入这层理顺。自建 Agent 最烦的就是 Key 管理——不同模型不同 Key环境变量散落各处换台机器就要重新配。我试过用 TaoToken 做统一入口一个 Key 走通对话、编码、工具调用配置集中在一个文件里迁移成本低很多。TaoToken 的定位是统一模型接入层官网在 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 端点是 https://taotoken.net/api 。它适合这几类人想快速验证 Agent 链路、不想在多个模型供应商之间反复切换、需要把 Key 配置和业务代码解耦的开发者。配置方式很简单把 Key 写进环境变量代码里只读环境变量。这样你的 Agent Loop 代码不绑定任何具体供应商换模型只改配置不改逻辑。下面这段是 settings.json 的骨架你可以直接拿去改{ model: { provider: taotoken, base_url: https://taotoken.net/api, api_key_env: TAOTOKEN_API_KEY, default_model: claude-sonnet-4-20250514, max_tokens: 8192, temperature: 0.7 }, agent: { max_rounds: 50, tool_timeout_seconds: 30, parallel_tools: true, max_parallel_workers: 4 }, context: { compress_threshold_tokens: 60000, keep_head_messages: 3, keep_tail_messages: 5 } }注意api_key_env 指向的是环境变量名不是 Key 本身。不要把真实 Key 写进配置文件提交到仓库。拿到 Key 的入口在控制台的 API Keys 页面地址是 https://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite 。生成后导出到环境变量export TAOTOKEN_API_KEY你的Key如果你更想先验证模型对话是否通可以直接用模型对话页面试一条消息地址是 https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewrite 。确认能正常返回后再进入 Agent Loop 的编码环节。3. 可复制配置Agent Loop 与 Tool Call 骨架3.1 Agent Loop 主循环Claude Code 的循环逻辑可以简化成下面这样。核心是每轮把 history 发给模型检查返回里有没有 tool_call有就执行、把结果追加回 history然后进入下一轮没有就直接返回文本。import os import json import requests API_URL https://taotoken.net/api/v1/chat/completions API_KEY os.environ[TAOTOKEN_API_KEY] MAX_ROUNDS 50 def agent_loop(user_input, tools, historyNone): history history or [] history.append({role: user, content: user_input}) for round_idx in range(MAX_ROUNDS): resp requests.post( API_URL, headers{Authorization: fBearer {API_KEY}}, json{ model: claude-sonnet-4-20250514, messages: history, tools: tools, tool_choice: auto }, timeout60 ) msg resp.json()[choices][0][message] history.append(msg) tool_calls msg.get(tool_calls) if not tool_calls: return msg[content], history for call in tool_calls: result execute_tool(call) history.append({ role: tool, tool_call_id: call[id], content: result }) return 达到最大轮次限制, history这段代码里有两个关键点。第一tool_call_id 必须和 tool 结果的 tool_call_id 严格对应否则模型无法把结果关联回它发起的调用。第二max_rounds 是硬上限防止模型陷入无限调用循环。3.2 Tool 定义与 function calling 格式工具定义遵循 OpenAI 的 function calling 格式包含 name、description、parameters 三部分。description 写得好不好直接决定模型会不会在正确时机调用它。def build_tool(name, description, parameters, executor): return { type: function, function: { name: name, description: description, parameters: parameters } }, executor read_file_tool, read_file_exec build_tool( nameread_file, description读取指定路径的文件内容用于查看代码或配置, parameters{ type: object, properties: { path: {type: string, description: 文件绝对路径} }, required: [path] }, executorlambda path: open(path).read() )3.3 工具报错的两类区分Claude Code 把工具错误分成两类这个分类很实用。第一类是参数传递错误——模型给的参数格式不对、缺字段、类型错这类错误应该在调用工具函数之前就拦截。第二类是工具执行错误——参数没问题但执行时抛异常比如文件不存在、命令返回非零。def execute_tool(call): name call[function][name] try: args json.loads(call[function][arguments]) except json.JSONDecodeError as e: return json.dumps({error: param_parse_error, detail: str(e)}) executor TOOL_REGISTRY.get(name) if not executor: return json.dumps({error: unknown_tool, name: name}) try: return str(executor(**args)) except Exception as e: return json.dumps({error: exec_error, detail: str(e)})把这两类错误分开返回模型在下一轮就能判断是重新组织参数还是换个思路。如果混在一起返回模型往往分不清该改参数还是改策略。3.4 中断恢复让历史重新合法用户按 CtrlC 时如果恰好截断在模型发起 tool_call 和工具执行之间history 里就会出现一个没有对应 tool 结果的 tool_call。对模型来说这个工具像是执行了但没内容整个会话历史就非法了。Claude Code 的处理方式是给这个位置补一个 content 为 interrupt 的 tool 消息让历史重新合法再把异常抛给上层。def handle_interrupt(history): last history[-1] if last.get(role) assistant and last.get(tool_calls): for call in last[tool_calls]: history.append({ role: tool, tool_call_id: call[id], content: interrupt }) return history3.5 上下文压缩的三种策略上下文工程是 Agent Loop 里最容易被低估的部分。Claude Code 用了三层压缩保障。第一层tool_call 的返回结果往往只在当时任务有用后续任务不再需要所以保留头尾、删除中间。第二层把历史消息压缩成摘要但保留结构化内容而不是背景文本。第三层如果前两层还不够只保留前几轮加摘要大幅删除。这里有个坑如果删除的位置恰好是 tool_call 的位置tool_call_id 和 tool 结果会被分割又回到中断那个问题。所以压缩时要移动分割边界保证 tool_call 和它的结果始终在一起。def compress_history(history, threshold60000): total sum(len(str(m)) for m in history) if total threshold: return history head history[:3] tail history[-5:] middle history[3:-5] summary summarize(middle) compressed head [{role: system, content: f历史摘要{summary}}] tail return fix_tool_boundary(compressed)3.6 Subagent 拆分思路Subagent 的价值在于让主 Agent 的上下文窗口保持干净。Claude Code 里没有显式定义子 Agent而是通过 AgentTool 来定义子 Agent 能共享主 Agent 的资源但单独开 context主 Agent 只需要子 Agent 的执行结果。关键约束是子 Agent 不能再调用 AgentTool否则会无限嵌套。子 Agent 像工具一样执行完就结束不保持持续状态。def agent_tool(sub_task, shared_resources): sub_history [{role: user, content: sub_task}] result, _ agent_loop( user_inputsub_task, toolsshared_resources[base_tools], historysub_history ) return result4. 验证请求跑通一次工具调用链路配置写完了得验证它真的能跑。下面这段脚本会发起一次请求让模型调用 read_file 工具然后检查返回的 tool_call_id 和 tool 结果是否对应。if __name__ __main__: tools [read_file_tool] TOOL_REGISTRY {read_file: read_file_exec} answer, final_history agent_loop( user_input帮我读取 /etc/hostname 的内容, toolstools ) print(最终回答, answer) print(--- 历史消息角色序列 ---) for m in final_history: role m.get(role) has_tool tool_calls in m print(f{role} | tool_call{has_tool})成功的结果应该长这样历史里先出现 user再出现带 tool_calls 的 assistant然后出现 role 为 tool 的消息最后是 assistant 的文本回答。tool 消息的 tool_call_id 必须和 assistant 里那个 call 的 id 完全一致。如果你在验证时想换个模型对比效果可以用模型对话页面手动发一条同样的请求观察不同模型的工具调用倾向。地址是 https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewrite 。5. 本篇常见错排查5.1 tool_call_id 对不上导致 400最常见的报错是模型返回 400提示 tool 消息的 tool_call_id 无效。原因通常是你在追加 tool 结果时用了自己生成的 id而不是 call[id]。检查你的 execute_tool 调用处确保用的是模型返回的那个 id。5.2 并行工具执行时的数据覆盖Claude Code 支持并行执行多个工具但并行不是肆无忌惮的。如果线程 1 在保存 A 目录线程 2 并行保存 B查看 A 的内容可能被覆盖。Python 里用 threading.local() 给每个线程一份独立副本线程之间互不可见。import threading _local threading.local() def parallel_execute(calls): results {} def worker(call): _local.workspace {} results[call[id]] execute_tool(call) threads [threading.Thread(targetworker, args(c,)) for c in calls] for t in threads: t.start() for t in threads: t.join() return results5.3 压缩后历史非法如果你实现了上下文压缩跑几轮后突然报错大概率是压缩把 tool_call 和它的结果分开了。检查 fix_tool_boundary 函数确保任何切割点都不落在 assistant(tool_calls) 和对应 tool 消息之间。5.4 子 Agent 无限嵌套子 Agent 如果也能调用 AgentTool就会无限递归。在传给子 Agent 的 tools 列表里把 AgentTool 排除掉。5.5 危险命令未拦截BashTool 执行前必须过一遍危险命令检测表rm、dd、mkfs 这类命令要拦截或要求确认。Claude Code 把命令执行构建成了有语义分类和权限约束的系统能力不是简单地把字符串丢给 shell。DANGEROUS [rm -rf, dd if, mkfs, :(){ :|: };:] def safe_bash(cmd): for d in DANGEROUS: if d in cmd: return json.dumps({error: blocked, cmd: cmd}) return subprocess.run(cmd, shellTrue, capture_outputTrue, textTrue).stdout6. 继续深入从看懂到验证Agent Loop 的骨架搭起来之后下一步是把它跑在真实任务上。如果你打算长期做编码类 Agent或者要接多个工具做复杂工作流建议用 Coding Plan 来管理模型额度和调用配额入口在 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite 。接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 里面有完整的 API 参数说明和工具调用示例。Claude Code 源码里还有几个值得深挖的点Hook 机制把扩展逻辑从 Agent Loop 里解耦出来让权限判断、日志打印、大输出处理可以独立注册Todo 和 Task 系统让 Agent 有规划能力和任务依赖管理Background Task 让耗时命令不阻塞主循环。这些都不是必须的但当你发现 Agent 开始变慢、变乱、上下文爆炸时回头看这些设计会有新的理解。验证代码能不能跑通比看懂代码更重要。把上面的 settings.json 和 agent_loop 复制下来换成你自己的 Key跑一次 read_file 调用看历史消息的角色序列对不对。这一步过了后面加工具、加压缩、加子 Agent 都是在这个骨架上长出来的。
从 Claude Code 源码拆解 Agent Loop：CC 中我们能学到什么
1. 从一次工具调用失败说起Agent Loop 到底难在哪如果你正在自建 Agent大概率遇到过这种场景模型说要调用工具工具也执行了但下一轮模型却像失忆一样完全不知道刚才发生了什么。或者更糟用户中途按了 CtrlC整个会话历史直接错位后面每一轮都在报错。这不是模型笨而是 Agent Loop 的上下文管理没做对。Claude Code 的源码里Agent Loop 的核心其实就一句话用 for 循环跑任务max_rounds 设成 50每一轮判断有没有 tool call有就执行工具、把结果塞回 history没有就结束。听起来简单但真正难的是那些边界情况——工具报错怎么分类、中断怎么恢复、上下文太长怎么压缩、多个工具能不能并行、子 Agent 怎么隔离。这篇面向想自建 Agent 的开发者把 Claude Code 里 Agent Loop 与 Tool Call 的关键设计拆开讲并给出可复制的 settings.json 骨架和 TaoToken 统一 Key 配置最后演示一次完整的工具调用链路验证。你不需要读完整个源码跟着做就能理解上下文工程和 Subagent 拆分的思路。2. 前置准备用 TaoToken 统一管理模型 Key在动手写 Agent Loop 之前先把模型接入这层理顺。自建 Agent 最烦的就是 Key 管理——不同模型不同 Key环境变量散落各处换台机器就要重新配。我试过用 TaoToken 做统一入口一个 Key 走通对话、编码、工具调用配置集中在一个文件里迁移成本低很多。TaoToken 的定位是统一模型接入层官网在 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 端点是 https://taotoken.net/api 。它适合这几类人想快速验证 Agent 链路、不想在多个模型供应商之间反复切换、需要把 Key 配置和业务代码解耦的开发者。配置方式很简单把 Key 写进环境变量代码里只读环境变量。这样你的 Agent Loop 代码不绑定任何具体供应商换模型只改配置不改逻辑。下面这段是 settings.json 的骨架你可以直接拿去改{ model: { provider: taotoken, base_url: https://taotoken.net/api, api_key_env: TAOTOKEN_API_KEY, default_model: claude-sonnet-4-20250514, max_tokens: 8192, temperature: 0.7 }, agent: { max_rounds: 50, tool_timeout_seconds: 30, parallel_tools: true, max_parallel_workers: 4 }, context: { compress_threshold_tokens: 60000, keep_head_messages: 3, keep_tail_messages: 5 } }注意api_key_env 指向的是环境变量名不是 Key 本身。不要把真实 Key 写进配置文件提交到仓库。拿到 Key 的入口在控制台的 API Keys 页面地址是 https://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite 。生成后导出到环境变量export TAOTOKEN_API_KEY你的Key如果你更想先验证模型对话是否通可以直接用模型对话页面试一条消息地址是 https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewrite 。确认能正常返回后再进入 Agent Loop 的编码环节。3. 可复制配置Agent Loop 与 Tool Call 骨架3.1 Agent Loop 主循环Claude Code 的循环逻辑可以简化成下面这样。核心是每轮把 history 发给模型检查返回里有没有 tool_call有就执行、把结果追加回 history然后进入下一轮没有就直接返回文本。import os import json import requests API_URL https://taotoken.net/api/v1/chat/completions API_KEY os.environ[TAOTOKEN_API_KEY] MAX_ROUNDS 50 def agent_loop(user_input, tools, historyNone): history history or [] history.append({role: user, content: user_input}) for round_idx in range(MAX_ROUNDS): resp requests.post( API_URL, headers{Authorization: fBearer {API_KEY}}, json{ model: claude-sonnet-4-20250514, messages: history, tools: tools, tool_choice: auto }, timeout60 ) msg resp.json()[choices][0][message] history.append(msg) tool_calls msg.get(tool_calls) if not tool_calls: return msg[content], history for call in tool_calls: result execute_tool(call) history.append({ role: tool, tool_call_id: call[id], content: result }) return 达到最大轮次限制, history这段代码里有两个关键点。第一tool_call_id 必须和 tool 结果的 tool_call_id 严格对应否则模型无法把结果关联回它发起的调用。第二max_rounds 是硬上限防止模型陷入无限调用循环。3.2 Tool 定义与 function calling 格式工具定义遵循 OpenAI 的 function calling 格式包含 name、description、parameters 三部分。description 写得好不好直接决定模型会不会在正确时机调用它。def build_tool(name, description, parameters, executor): return { type: function, function: { name: name, description: description, parameters: parameters } }, executor read_file_tool, read_file_exec build_tool( nameread_file, description读取指定路径的文件内容用于查看代码或配置, parameters{ type: object, properties: { path: {type: string, description: 文件绝对路径} }, required: [path] }, executorlambda path: open(path).read() )3.3 工具报错的两类区分Claude Code 把工具错误分成两类这个分类很实用。第一类是参数传递错误——模型给的参数格式不对、缺字段、类型错这类错误应该在调用工具函数之前就拦截。第二类是工具执行错误——参数没问题但执行时抛异常比如文件不存在、命令返回非零。def execute_tool(call): name call[function][name] try: args json.loads(call[function][arguments]) except json.JSONDecodeError as e: return json.dumps({error: param_parse_error, detail: str(e)}) executor TOOL_REGISTRY.get(name) if not executor: return json.dumps({error: unknown_tool, name: name}) try: return str(executor(**args)) except Exception as e: return json.dumps({error: exec_error, detail: str(e)})把这两类错误分开返回模型在下一轮就能判断是重新组织参数还是换个思路。如果混在一起返回模型往往分不清该改参数还是改策略。3.4 中断恢复让历史重新合法用户按 CtrlC 时如果恰好截断在模型发起 tool_call 和工具执行之间history 里就会出现一个没有对应 tool 结果的 tool_call。对模型来说这个工具像是执行了但没内容整个会话历史就非法了。Claude Code 的处理方式是给这个位置补一个 content 为 interrupt 的 tool 消息让历史重新合法再把异常抛给上层。def handle_interrupt(history): last history[-1] if last.get(role) assistant and last.get(tool_calls): for call in last[tool_calls]: history.append({ role: tool, tool_call_id: call[id], content: interrupt }) return history3.5 上下文压缩的三种策略上下文工程是 Agent Loop 里最容易被低估的部分。Claude Code 用了三层压缩保障。第一层tool_call 的返回结果往往只在当时任务有用后续任务不再需要所以保留头尾、删除中间。第二层把历史消息压缩成摘要但保留结构化内容而不是背景文本。第三层如果前两层还不够只保留前几轮加摘要大幅删除。这里有个坑如果删除的位置恰好是 tool_call 的位置tool_call_id 和 tool 结果会被分割又回到中断那个问题。所以压缩时要移动分割边界保证 tool_call 和它的结果始终在一起。def compress_history(history, threshold60000): total sum(len(str(m)) for m in history) if total threshold: return history head history[:3] tail history[-5:] middle history[3:-5] summary summarize(middle) compressed head [{role: system, content: f历史摘要{summary}}] tail return fix_tool_boundary(compressed)3.6 Subagent 拆分思路Subagent 的价值在于让主 Agent 的上下文窗口保持干净。Claude Code 里没有显式定义子 Agent而是通过 AgentTool 来定义子 Agent 能共享主 Agent 的资源但单独开 context主 Agent 只需要子 Agent 的执行结果。关键约束是子 Agent 不能再调用 AgentTool否则会无限嵌套。子 Agent 像工具一样执行完就结束不保持持续状态。def agent_tool(sub_task, shared_resources): sub_history [{role: user, content: sub_task}] result, _ agent_loop( user_inputsub_task, toolsshared_resources[base_tools], historysub_history ) return result4. 验证请求跑通一次工具调用链路配置写完了得验证它真的能跑。下面这段脚本会发起一次请求让模型调用 read_file 工具然后检查返回的 tool_call_id 和 tool 结果是否对应。if __name__ __main__: tools [read_file_tool] TOOL_REGISTRY {read_file: read_file_exec} answer, final_history agent_loop( user_input帮我读取 /etc/hostname 的内容, toolstools ) print(最终回答, answer) print(--- 历史消息角色序列 ---) for m in final_history: role m.get(role) has_tool tool_calls in m print(f{role} | tool_call{has_tool})成功的结果应该长这样历史里先出现 user再出现带 tool_calls 的 assistant然后出现 role 为 tool 的消息最后是 assistant 的文本回答。tool 消息的 tool_call_id 必须和 assistant 里那个 call 的 id 完全一致。如果你在验证时想换个模型对比效果可以用模型对话页面手动发一条同样的请求观察不同模型的工具调用倾向。地址是 https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewrite 。5. 本篇常见错排查5.1 tool_call_id 对不上导致 400最常见的报错是模型返回 400提示 tool 消息的 tool_call_id 无效。原因通常是你在追加 tool 结果时用了自己生成的 id而不是 call[id]。检查你的 execute_tool 调用处确保用的是模型返回的那个 id。5.2 并行工具执行时的数据覆盖Claude Code 支持并行执行多个工具但并行不是肆无忌惮的。如果线程 1 在保存 A 目录线程 2 并行保存 B查看 A 的内容可能被覆盖。Python 里用 threading.local() 给每个线程一份独立副本线程之间互不可见。import threading _local threading.local() def parallel_execute(calls): results {} def worker(call): _local.workspace {} results[call[id]] execute_tool(call) threads [threading.Thread(targetworker, args(c,)) for c in calls] for t in threads: t.start() for t in threads: t.join() return results5.3 压缩后历史非法如果你实现了上下文压缩跑几轮后突然报错大概率是压缩把 tool_call 和它的结果分开了。检查 fix_tool_boundary 函数确保任何切割点都不落在 assistant(tool_calls) 和对应 tool 消息之间。5.4 子 Agent 无限嵌套子 Agent 如果也能调用 AgentTool就会无限递归。在传给子 Agent 的 tools 列表里把 AgentTool 排除掉。5.5 危险命令未拦截BashTool 执行前必须过一遍危险命令检测表rm、dd、mkfs 这类命令要拦截或要求确认。Claude Code 把命令执行构建成了有语义分类和权限约束的系统能力不是简单地把字符串丢给 shell。DANGEROUS [rm -rf, dd if, mkfs, :(){ :|: };:] def safe_bash(cmd): for d in DANGEROUS: if d in cmd: return json.dumps({error: blocked, cmd: cmd}) return subprocess.run(cmd, shellTrue, capture_outputTrue, textTrue).stdout6. 继续深入从看懂到验证Agent Loop 的骨架搭起来之后下一步是把它跑在真实任务上。如果你打算长期做编码类 Agent或者要接多个工具做复杂工作流建议用 Coding Plan 来管理模型额度和调用配额入口在 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite 。接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 里面有完整的 API 参数说明和工具调用示例。Claude Code 源码里还有几个值得深挖的点Hook 机制把扩展逻辑从 Agent Loop 里解耦出来让权限判断、日志打印、大输出处理可以独立注册Todo 和 Task 系统让 Agent 有规划能力和任务依赖管理Background Task 让耗时命令不阻塞主循环。这些都不是必须的但当你发现 Agent 开始变慢、变乱、上下文爆炸时回头看这些设计会有新的理解。验证代码能不能跑通比看懂代码更重要。把上面的 settings.json 和 agent_loop 复制下来换成你自己的 Key跑一次 read_file 调用看历史消息的角色序列对不对。这一步过了后面加工具、加压缩、加子 Agent 都是在这个骨架上长出来的。
