1. 从单工具到多工具agent_loop 里 tool_use 到底卡在哪Claude Code 的 tool_use 机制说白了就是让模型在对话里“伸手”去调用你预先定义好的函数。第一章我们只给了它一个 bash模型想读文件得自己拼cat想改文件得自己写sed每一步都要输出一堆 token 去描述命令。到了第二章我们要把 bash 拆成 read_file、write_file、edit_file、glob 这些细粒度工具让模型直接说“我要调 edit_file参数是这些”而不是让它自己拼 shell。但工具一多agent_loop 里就冒出三个新问题第一模型返回的tool_useblock 怎么分发到对应函数第二模型传进来的路径可能是../../etc/passwd怎么保证它不跑出工作目录第三模型一口气返回多个tool_useblock 时是串行跑还是并发跑。这三个问题不解决多工具就是给自己挖坑。这篇就按“工具函数 → 分发表 → agent_loop 改造 → 沙箱校验 → 并发执行 → 用 TaoToken 统一通道验证”的顺序走一遍。适合已经跑通过第一章单工具版本、想把手搓 Claude Code 推进到多工具并发的读者。代码是 Python配置片段是 JSON 和 TOML都能直接复制。2. 前置用 TaoToken 统一 Key 和 API 通道在改 agent_loop 之前先把模型调用这条链路固定下来。手搓 Claude Code 最烦的一点是你调试的是工具调用逻辑结果每次报错都怀疑是不是 Key 过期、通道不稳、模型名写错。我试过把模型通道单独抽出来用 TaoToken 统一管 Key 和 API 地址agent_loop 里只留一个 client排障时能少一半干扰。TaoToken 在这里的角色是“统一入口”你拿一个 Key通过它的 API 地址去调模型模型对话、coding plan、API Keys 管理都在一个控制台里。对第二章来说重点是让 agent_loop 的模型调用部分不成为变量。你需要准备的东西一个 TaoToken 的 API Key在控制台的 API Keys 页面创建确认你要用的模型名在模型对话页面可以先手动聊一句验证通道通不通把 API 地址记成https://taotoken.net/api注意这个地址不带 UTM 参数配置里写干净的。如果你后面要长期跑编码类 Agent可以看下 Coding Plan 页面它面向的就是这种持续调用的场景。但第二章我们先把单次 agent_loop 跑通不急着上长期方案。3. 可复制配置settings.json 与 config.toml 片段先把配置落地再改代码。下面两个片段分别对应“Claude Code 侧读取的 settings.json”和“你本地 agent 项目读取的 config.toml”。字段名按你项目实际调整但结构可以直接抄。3.1 settings.json把模型通道和工具白名单写死{ model: { provider: taotoken, api_base: https://taotoken.net/api, api_key_env: TAOTOKEN_API_KEY, model_name: claude-sonnet-4-20250514, max_tokens: 4096, timeout_seconds: 60 }, tools: { enabled: [bash, read_file, write_file, edit_file, glob], workdir: ./workspace, max_concurrency: 4, sandbox: { allow_symlink_escape: false, blocked_commands: [rm -rf /, sudo, shutdown, reboot, /dev/] } } }这里api_key_env指向环境变量不要把 Key 明文写进 JSON。workdir是沙箱根目录后面safe_path会用它做边界校验。max_concurrency先给 4后面并发那节会解释为什么不是越大越好。3.2 config.tomlagent 运行时的并发与沙箱参数[agent] name learn-claude-code-s02 workdir ./workspace max_turns 30 [agent.concurrency] enabled true max_workers 4 tool_timeout_seconds 30 [agent.sandbox] root ./workspace resolve_symlinks true deny_outside_root true [agent.tools.bash] timeout_seconds 20 blocked_patterns [rm -rf /, sudo, shutdown, reboot, /dev/] [agent.tools.read_file] default_limit 200 [agent.tools.write_file] create_parents trueresolve_symlinks true配合deny_outside_root true就是防沙箱逃逸的核心。模型传link/../../etc/hosts这种路径时resolve()会先把符号链接和..全部展开再判断是否还在 root 下。4. 工具函数与分发表read/write/edit/glob 的沙箱实现配置有了回到代码。先把os.getcwd()换成Path.cwd()用WORKDIR接住这样后面所有工具都能引用同一个根。from pathlib import Path WORKDIR Path.cwd() / workspace WORKDIR.mkdir(parentsTrue, exist_okTrue)4.1 safe_path所有文件操作的唯一入口def safe_path(p: str) - Path: path (WORKDIR / p).resolve() if not path.is_relative_to(WORKDIR): raise ValueError(fPath escapes workspace: {p}) return pathresolve()会把..和符号链接都算成真实绝对路径is_relative_to再判断它有没有跑出WORKDIR。这一步不做后面 read/write/edit 全是漏洞。4.2 glob、read、write、edit 四个函数import glob as g def run_glob(pattern: str) - str: try: results [] for match in g.glob(pattern, root_dirWORKDIR): if (WORKDIR / match).resolve().is_relative_to(WORKDIR): results.append(match) return \n.join(results) if results else (no matches) except Exception as e: return fError: {e} def run_read(path: str, limit: int | None None) - str: try: lines safe_path(path).read_text().splitlines() if limit and limit len(lines): lines lines[:limit] [f... ({len(lines) - limit} more lines)] return \n.join(lines) except Exception as e: return fError: {e} def run_write(path: str, content: str) - str: try: file_path safe_path(path) file_path.parent.mkdir(parentsTrue, exist_okTrue) file_path.write_text(content) return fWrote {len(content)} bytes to {path} except Exception as e: return fError: {e} def run_edit(path: str, old_text: str, new_text: str) - str: try: file_path safe_path(path) text file_path.read_text() if old_text not in text: return fError: text not found in {path} file_path.write_text(text.replace(old_text, new_text, 1)) return fEdited {path} except Exception as e: return fError: {e}run_edit只替换第一个匹配这是故意的模型如果一次要改多处应该发多个 edit_file 调用而不是让你在函数里做全局替换。这样每次改动都可追溯。4.3 分发表一个字典把 tool_use 映射到函数TOOL_HANDLERS { bash: run_bash, read_file: run_read, write_file: run_write, edit_file: run_edit, glob: run_glob, }run_bash沿用第一章的实现但把危险命令检查加进去def run_bash(command: str) - str: dangerous [rm -rf /, sudo, shutdown, reboot, /dev/] if any(d in command for d in dangerous): return Error: Dangerous command blocked import subprocess try: result subprocess.run( command, shellTrue, cwdWORKDIR, capture_outputTrue, textTrue, timeout20 ) return (result.stdout result.stderr)[:4000] or (no output) except subprocess.TimeoutExpired: return Error: command timeout except Exception as e: return fError: {e}注意cwdWORKDIR这样 bash 的默认工作目录也被锁在沙箱里。5. 改造 agent_loop串行分发与并发执行现在改 agent_loop 里处理tool_use的那段。先看串行版本逻辑最清楚if block.type tool_use: print(f\033[33m$ {block.name}\033[0m) handler TOOL_HANDLERS.get(block.name) output handler(**block.input) if handler else fUnknown: {block.name} print(str(output)[:200]) results.append({ type: tool_result, tool_use_id: block.id, content: output, })这段跑通后模型一次只调一个工具结果按顺序回填。但真实场景里模型经常一口气返回多个tool_useblock比如同时读三个文件。串行跑就是一个个等并发跑能省时间。5.1 并发执行骨架from concurrent.futures import ThreadPoolExecutor, as_completed def execute_tool_blocks(blocks, max_workers4): results [] with ThreadPoolExecutor(max_workersmax_workers) as pool: future_map {} for block in blocks: handler TOOL_HANDLERS.get(block.name) if not handler: results.append({ type: tool_result, tool_use_id: block.id, content: fUnknown: {block.name}, }) continue future pool.submit(handler, **block.input) future_map[future] block for future in as_completed(future_map): block future_map[future] try: output future.result(timeout30) except Exception as e: output fError: {e} results.append({ type: tool_result, tool_use_id: block.id, content: str(output), }) return results这里有两个坑。第一as_completed返回顺序和提交顺序不一致但tool_result必须带tool_use_id所以顺序不影响模型理解。第二max_workers不要设太大文件读写是 IO 密集4 到 8 够用设成 32 反而会因为磁盘争抢变慢。5.2 并发下的沙箱一致性并发跑多个 write_file 时如果两个调用写同一个文件后写的会覆盖先写的。这不是沙箱问题是业务逻辑问题。我的做法是在run_write里加一个简单的文件锁import threading _file_locks {} _lock_guard threading.Lock() def _get_lock(path: Path): with _lock_guard: if path not in _file_locks: _file_locks[path] threading.Lock() return _file_locks[path]然后在run_write和run_edit里用with _get_lock(file_path):包住写操作。这样并发调用同一文件时会排队不同文件仍然并行。6. 验证请求跑一次多工具并发调用配置和代码都齐了来跑一次验证。先确认环境变量export TAOTOKEN_API_KEY你的Key然后启动 agent输入将 workspace 下所有 .py 文件里的 TODO 替换成 DONE并告诉我改了哪些文件预期行为模型先调glob找*.py拿到文件列表后并发调多个read_file再并发调多个edit_file。你会在终端看到类似$ glob main.py utils.py $ read_file $ read_file $ edit_file $ edit_file Edited main.py Edited utils.py如果模型只调了一次glob就停下来说明你的tool_result没回填对检查tool_use_id是否和请求里的id一致。如果并发没生效检查execute_tool_blocks是不是真的被调用了而不是还在走串行分支。验证通道是否正常可以先用模型对话页面手动发一句“你好”确认 Key 和 API 地址没问题。排障时优先看 API Keys 页面里 Key 的状态再看接入文档里的请求格式。7. 本篇常见错排查报错一Path escapes workspace。模型传了../或者绝对路径。先看WORKDIR是不是你预期的目录再看safe_path里resolve()有没有被跳过。如果模型频繁传越界路径在 system prompt 里明确写“所有路径必须是相对于 workspace 的相对路径”。报错二Unknown: xxx。模型调了一个不在TOOL_HANDLERS里的工具名。检查你传给模型的TOOLS列表和TOOL_HANDLERS的 key 是否完全一致大小写、下划线都要对。报错三并发时结果错乱。tool_result的tool_use_id对不上。并发版本里每个 future 都要绑定它对应的 block不能只按返回顺序 append。报错四bash 超时。subprocess.run的timeout设太短或者命令本身在等输入。把timeout_seconds调到 30并在 system prompt 里提醒模型不要跑交互式命令。报错五模型不调工具直接编答案。检查TOOLS的input_schema是不是合法 JSON Schema字段类型写错会导致模型无法生成合法参数。另外确认tool_choice没被设成none。8. 下一步把通道和工具链固定下来第二章跑通后你手里应该有一个能并发调多工具、带沙箱校验的 agent_loop。接下来要做的是把模型通道固定成长期可用的形态不然每次调试工具逻辑都要重新配 Key。如果你只是偶尔验证模型行为用模型对话页面就够了。如果你要长期跑编码类 Agent比如让它持续读写项目文件、跑测试、改配置那 Coding Plan 更合适它面向的就是这种高频调用场景。API Key 的管理在 API Keys 页面请求格式和参数说明在接入文档里遇到 401 或 429 先查这两处。工具链这边下一步可以加grep和list_dir把 glob 的模糊匹配补全。并发那块如果工具调用开始涉及网络请求记得把max_workers和超时分开配别让一个慢请求拖住整个批次。沙箱这块safe_path是底线但 bash 里的命令注入还得靠blocked_patterns和cwd双重限制别只依赖一个。