1. 为什么 Agent Loop 的骨架比模型选型更值得先动手如果你正在用 Python 驱动 Claude Code 这类工具大概率会遇到一个尴尬模型能力明明够但任务链一长就散架。读文件、跑命令、改代码、再验证中间任何一步的上下文丢了整个循环就退化成“单轮问答 手动复制粘贴”。这不是模型的问题是 Harness 的问题。Harness Engineering 这个词听起来抽象落到工程上其实就三件事给模型一双能操作环境的手工具、一份按需查阅的知识上下文、一套不会越界的规则安全边界。而 Agent Loop 是承载这三件事的最小运行时——一个 while 循环反复“请求模型 → 解析工具调用 → 执行 → 把结果塞回消息列表”直到模型不再要求调用工具为止。这篇面向的是已经会用 Python 调 Anthropic SDK、想把这套循环真正跑起来的开发者。我会给出可直接复制的config.toml与settings.json骨架演示如何通过 TaoToken 统一 Key 和 API 通道接入最后附一条 Agent Loop 任务链的验证动作和预期输出。目标很明确让你在本地跑通一个最小闭环而不是停在“看懂原理”的层面。适合谁写过脚本调 API、但没系统搭过 Agent 循环的人或者已经用上 Claude Code、想理解它底层 Harness 长什么样的人。不适合完全没碰过 Python 请求库的纯小白——但我会把每一步写清楚你照着敲也能跑。2. TaoToken 前置统一 Key 与 API 通道在写循环之前先把接入层理清楚。Agent Loop 会频繁发请求如果 Key 散落在环境变量、配置文件、代码硬编码里调试时你会花大量时间在“到底哪个 Key 生效了”上。TaoToken 在这里的作用是提供一个统一的 API 通道把模型调用收敛到一个 base_url 和一把 Key 上。你需要先拿到 API Key。登录后进入控制台在 API Keys 页面创建一个新 Key复制保存——它只显示一次。地址是https://taotoken.net/api注意这个地址不带任何查询参数直接作为 SDK 的base_url使用。关于模型名TaoToken 的通道兼容 Anthropic 的 messages 接口格式所以你在代码里仍然用client.messages.create(...)这套调用方式只是把base_url指过来、api_key换成 TaoToken 的 Key。这样你的 Agent Loop 代码不需要为接入层做任何特殊适配换通道就是换两个参数的事。注意Key 不要写进会提交到 Git 的文件。下面配置里我用环境变量占位实际运行时通过 shell 注入。如果你还没创建 Key可以先打开 API Keys 页面https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite3. 可复制配置config.toml 与 settings.json 骨架工程落地的第一步是把配置和代码分离。我习惯用config.toml管运行时参数模型、超时、循环上限用settings.json管工具白名单和安全边界。这样调参不用改代码改边界不用碰逻辑。先建目录结构mkdir -p agent-harness/{config,tools,transcripts} cd agent-harnessconfig/config.toml[api] base_url https://taotoken.net/api api_key_env TAOTOKEN_API_KEY model claude-sonnet-4-20250514 max_tokens 4096 timeout_seconds 60 [loop] max_iterations 12 max_tool_calls_per_turn 8 transcript_dir transcripts [tools] enabled [read_file, write_file, run_bash, list_dir]config/settings.json{ safety: { allowed_paths: [./workspace], bash_denylist: [rm -rf /, shutdown, reboot, mkfs], max_output_chars: 8000 }, context: { compress_threshold_tokens: 60000, keep_recent_messages: 6 }, logging: { events_file: transcripts/events.jsonl, save_full_transcript: true } }这两个文件的分工要清楚config.toml里的东西是“怎么连、连几次”settings.json里的东西是“能碰什么、碰多少”。allowed_paths限定文件工具只能操作./workspace目录bash_denylist是命令黑名单的第一道闸max_output_chars防止某条命令输出几万行把上下文撑爆。读取配置的代码骨架import json import os import tomllib from pathlib import Path def load_config(root: Path Path(.)): with open(root / config / config.toml, rb) as f: cfg tomllib.load(f) with open(root / config / settings.json, r, encodingutf-8) as f: settings json.load(f) cfg[api][api_key] os.environ[cfg[api][api_key_env]] return cfg, settingstomllib是 Python 3.11 起内置的如果你用更早版本装tomli并把 import 换掉即可。4. Agent Loop 主体与工具分发配置就位后核心循环本身很短。它的结构就是把消息列表发给模型检查stop_reason如果是tool_use就逐个执行工具、把结果作为tool_result追加回消息然后继续下一轮否则返回。import anthropic def build_client(cfg): return anthropic.Anthropic( api_keycfg[api][api_key], base_urlcfg[api][base_url], timeoutcfg[api][timeout_seconds], ) def agent_loop(client, cfg, settings, messages, tools, handlers): for step in range(cfg[loop][max_iterations]): response client.messages.create( modelcfg[api][model], max_tokenscfg[api][max_tokens], messagesmessages, toolstools, ) messages.append({role: assistant, content: response.content}) if response.stop_reason ! tool_use: return response, messages results [] for block in response.content: if block.type ! tool_use: continue handler handlers.get(block.name) if handler is None: output ferror: unknown tool {block.name} else: output handler(**block.input, _settingssettings) results.append({ type: tool_result, tool_use_id: block.id, content: str(output)[: settings[safety][max_output_chars]], }) messages.append({role: user, content: results}) raise RuntimeError(loop exceeded max_iterations without finishing)工具分发表用字典把工具名映射到 Python 函数这就是 Harness 里“工具腰带”的落地方式。加一个新工具只需要在TOOLS定义里加 schema、在handlers里加函数循环本身一行都不用改。from pathlib import Path def make_handlers(settings): root Path(settings[safety][allowed_paths][0]).resolve() def _safe(p): target (root / p).resolve() if not str(target).startswith(str(root)): raise ValueError(fpath outside workspace: {p}) return target def read_file(path, _settingsNone): return _safe(path).read_text(encodingutf-8) def write_file(path, content, _settingsNone): t _safe(path) t.parent.mkdir(parentsTrue, exist_okTrue) t.write_text(content, encodingutf-8) return fwrote {len(content)} chars to {path} def list_dir(path., _settingsNone): return \n.join(sorted(p.name for p in _safe(path).iterdir())) def run_bash(command, _settingsNone): deny _settings[safety][bash_denylist] if any(d in command for d in deny): return blocked by denylist import subprocess r subprocess.run(command, shellTrue, capture_outputTrue, textTrue, timeout30) return fexit{r.returncode}\n{r.stdout}\n{r.stderr} return { read_file: read_file, write_file: write_file, list_dir: list_dir, run_bash: run_bash, }工具 schema 用 Anthropic 的标准格式声明这里给两个示例其余照抄结构TOOLS [ { name: read_file, description: Read a UTF-8 text file inside the workspace., input_schema: { type: object, properties: {path: {type: string}}, required: [path], }, }, { name: run_bash, description: Run a shell command and return stdout/stderr., input_schema: { type: object, properties: {command: {type: string}}, required: [command], }, }, ]5. 验证请求跑通一条任务链并看预期输出现在把上面拼起来跑一条真实的任务链。我设计的验证任务是让 Agent 在 workspace 里创建一个 Python 文件、写入一段代码、然后运行它并报告结果。这条链会依次触发write_file、run_bash两个工具正好覆盖“写 执行 观察”的完整循环。if __name__ __main__: cfg, settings load_config() client build_client(cfg) handlers make_handlers(settings) Path(workspace).mkdir(exist_okTrue) messages [{ role: user, content: ( 在 workspace 下创建 hello.py内容为打印 1 到 5 的平方 然后用 python 运行它把输出告诉我。 ), }] response, messages agent_loop(client, cfg, settings, messages, TOOLS, handlers) print( final ) for block in response.content: if hasattr(block, text): print(block.text)运行前注入 Keyexport TAOTOKEN_API_KEY你的Key python main.py预期你会看到类似这样的过程第一轮模型返回一个write_file的 tool_use循环执行后把结果塞回第二轮模型返回run_bash执行python workspace/hello.py第三轮stop_reason变成end_turn模型用自然语言汇报输出。最终打印大致是 final 已创建 workspace/hello.py 并运行输出为 1 4 9 16 25同时transcripts/events.jsonl里应该能看到每一轮的事件记录。如果你把save_full_transcript打开完整的消息列表也会落盘——这就是后面做 Task-Process Data 收集的原始素材模型看到了什么、决定调什么工具、实际执行结果如何全都有迹可循。6. 本篇常见错排查报 401 或 authentication_error八成是TAOTOKEN_API_KEY没注入到当前 shell或者 Key 复制时带了空格。先echo $TAOTOKEN_API_KEY确认非空再检查base_url是不是写成了带路径的形式——它应该就是https://taotoken.net/api。循环跑满 max_iterations 还没结束通常是工具返回的内容让模型误以为任务没完成。检查run_bash的返回里有没有把 stderr 也带上模型看到报错会反复重试。把max_iterations设成 12 是保守值调试期可以调小到 5 快速暴露问题。path outside workspace 异常_safe函数用resolve()做了前缀校验如果模型传了../开头的路径会被拦下。这是预期行为不是 bug。如果你确实需要访问 workspace 外的文件改allowed_paths别绕过校验。tool_result 的 tool_use_id 对不上每个tool_result必须和对应的tool_use的id严格匹配顺序也要一致。如果你在循环里过滤了某些 block记得 results 列表也要同步过滤否则下一轮请求会被服务端拒绝。上下文突然变长导致响应变慢run_bash输出大文件时最容易触发。max_output_chars的截断是在追加前做的确认你截断的是content字段而不是整个 result 对象。压缩阈值compress_threshold_tokens到 60000 才触发前期不用管。7. 下一步把骨架接进你的工作流跑通这条最小闭环之后骨架的扩展方向很清晰。想加知识注入就在TOOLS里加一个load_skill工具按需把文档片段塞进上下文想加持久化任务就把消息列表定期写进 JSONL重启后从文件恢复想做多 Agent 协作每个 Agent 一个独立的消息列表和工具集用一个邮箱文件在它们之间传消息。如果你打算长期用这套骨架驱动编码任务可以了解下 Coding Plan它把这类循环的调用额度做了打包适合高频跑 Agent Loop 的场景https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite想先验证模型在具体任务上的表现可以直接在模型对话里试https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewrite接入细节和参数说明都在文档里https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite我自己的习惯是每次改完工具定义先跑一遍这条“写文件 执行 汇报”的验证链确认循环没退化再去接真实任务。骨架稳了后面加多少工具都只是往handlers字典里塞函数的事。
TaoToken 实战：用 Harness Engineering 搭建 Agent Loop 的 Python 配置骨架
1. 为什么 Agent Loop 的骨架比模型选型更值得先动手如果你正在用 Python 驱动 Claude Code 这类工具大概率会遇到一个尴尬模型能力明明够但任务链一长就散架。读文件、跑命令、改代码、再验证中间任何一步的上下文丢了整个循环就退化成“单轮问答 手动复制粘贴”。这不是模型的问题是 Harness 的问题。Harness Engineering 这个词听起来抽象落到工程上其实就三件事给模型一双能操作环境的手工具、一份按需查阅的知识上下文、一套不会越界的规则安全边界。而 Agent Loop 是承载这三件事的最小运行时——一个 while 循环反复“请求模型 → 解析工具调用 → 执行 → 把结果塞回消息列表”直到模型不再要求调用工具为止。这篇面向的是已经会用 Python 调 Anthropic SDK、想把这套循环真正跑起来的开发者。我会给出可直接复制的config.toml与settings.json骨架演示如何通过 TaoToken 统一 Key 和 API 通道接入最后附一条 Agent Loop 任务链的验证动作和预期输出。目标很明确让你在本地跑通一个最小闭环而不是停在“看懂原理”的层面。适合谁写过脚本调 API、但没系统搭过 Agent 循环的人或者已经用上 Claude Code、想理解它底层 Harness 长什么样的人。不适合完全没碰过 Python 请求库的纯小白——但我会把每一步写清楚你照着敲也能跑。2. TaoToken 前置统一 Key 与 API 通道在写循环之前先把接入层理清楚。Agent Loop 会频繁发请求如果 Key 散落在环境变量、配置文件、代码硬编码里调试时你会花大量时间在“到底哪个 Key 生效了”上。TaoToken 在这里的作用是提供一个统一的 API 通道把模型调用收敛到一个 base_url 和一把 Key 上。你需要先拿到 API Key。登录后进入控制台在 API Keys 页面创建一个新 Key复制保存——它只显示一次。地址是https://taotoken.net/api注意这个地址不带任何查询参数直接作为 SDK 的base_url使用。关于模型名TaoToken 的通道兼容 Anthropic 的 messages 接口格式所以你在代码里仍然用client.messages.create(...)这套调用方式只是把base_url指过来、api_key换成 TaoToken 的 Key。这样你的 Agent Loop 代码不需要为接入层做任何特殊适配换通道就是换两个参数的事。注意Key 不要写进会提交到 Git 的文件。下面配置里我用环境变量占位实际运行时通过 shell 注入。如果你还没创建 Key可以先打开 API Keys 页面https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite3. 可复制配置config.toml 与 settings.json 骨架工程落地的第一步是把配置和代码分离。我习惯用config.toml管运行时参数模型、超时、循环上限用settings.json管工具白名单和安全边界。这样调参不用改代码改边界不用碰逻辑。先建目录结构mkdir -p agent-harness/{config,tools,transcripts} cd agent-harnessconfig/config.toml[api] base_url https://taotoken.net/api api_key_env TAOTOKEN_API_KEY model claude-sonnet-4-20250514 max_tokens 4096 timeout_seconds 60 [loop] max_iterations 12 max_tool_calls_per_turn 8 transcript_dir transcripts [tools] enabled [read_file, write_file, run_bash, list_dir]config/settings.json{ safety: { allowed_paths: [./workspace], bash_denylist: [rm -rf /, shutdown, reboot, mkfs], max_output_chars: 8000 }, context: { compress_threshold_tokens: 60000, keep_recent_messages: 6 }, logging: { events_file: transcripts/events.jsonl, save_full_transcript: true } }这两个文件的分工要清楚config.toml里的东西是“怎么连、连几次”settings.json里的东西是“能碰什么、碰多少”。allowed_paths限定文件工具只能操作./workspace目录bash_denylist是命令黑名单的第一道闸max_output_chars防止某条命令输出几万行把上下文撑爆。读取配置的代码骨架import json import os import tomllib from pathlib import Path def load_config(root: Path Path(.)): with open(root / config / config.toml, rb) as f: cfg tomllib.load(f) with open(root / config / settings.json, r, encodingutf-8) as f: settings json.load(f) cfg[api][api_key] os.environ[cfg[api][api_key_env]] return cfg, settingstomllib是 Python 3.11 起内置的如果你用更早版本装tomli并把 import 换掉即可。4. Agent Loop 主体与工具分发配置就位后核心循环本身很短。它的结构就是把消息列表发给模型检查stop_reason如果是tool_use就逐个执行工具、把结果作为tool_result追加回消息然后继续下一轮否则返回。import anthropic def build_client(cfg): return anthropic.Anthropic( api_keycfg[api][api_key], base_urlcfg[api][base_url], timeoutcfg[api][timeout_seconds], ) def agent_loop(client, cfg, settings, messages, tools, handlers): for step in range(cfg[loop][max_iterations]): response client.messages.create( modelcfg[api][model], max_tokenscfg[api][max_tokens], messagesmessages, toolstools, ) messages.append({role: assistant, content: response.content}) if response.stop_reason ! tool_use: return response, messages results [] for block in response.content: if block.type ! tool_use: continue handler handlers.get(block.name) if handler is None: output ferror: unknown tool {block.name} else: output handler(**block.input, _settingssettings) results.append({ type: tool_result, tool_use_id: block.id, content: str(output)[: settings[safety][max_output_chars]], }) messages.append({role: user, content: results}) raise RuntimeError(loop exceeded max_iterations without finishing)工具分发表用字典把工具名映射到 Python 函数这就是 Harness 里“工具腰带”的落地方式。加一个新工具只需要在TOOLS定义里加 schema、在handlers里加函数循环本身一行都不用改。from pathlib import Path def make_handlers(settings): root Path(settings[safety][allowed_paths][0]).resolve() def _safe(p): target (root / p).resolve() if not str(target).startswith(str(root)): raise ValueError(fpath outside workspace: {p}) return target def read_file(path, _settingsNone): return _safe(path).read_text(encodingutf-8) def write_file(path, content, _settingsNone): t _safe(path) t.parent.mkdir(parentsTrue, exist_okTrue) t.write_text(content, encodingutf-8) return fwrote {len(content)} chars to {path} def list_dir(path., _settingsNone): return \n.join(sorted(p.name for p in _safe(path).iterdir())) def run_bash(command, _settingsNone): deny _settings[safety][bash_denylist] if any(d in command for d in deny): return blocked by denylist import subprocess r subprocess.run(command, shellTrue, capture_outputTrue, textTrue, timeout30) return fexit{r.returncode}\n{r.stdout}\n{r.stderr} return { read_file: read_file, write_file: write_file, list_dir: list_dir, run_bash: run_bash, }工具 schema 用 Anthropic 的标准格式声明这里给两个示例其余照抄结构TOOLS [ { name: read_file, description: Read a UTF-8 text file inside the workspace., input_schema: { type: object, properties: {path: {type: string}}, required: [path], }, }, { name: run_bash, description: Run a shell command and return stdout/stderr., input_schema: { type: object, properties: {command: {type: string}}, required: [command], }, }, ]5. 验证请求跑通一条任务链并看预期输出现在把上面拼起来跑一条真实的任务链。我设计的验证任务是让 Agent 在 workspace 里创建一个 Python 文件、写入一段代码、然后运行它并报告结果。这条链会依次触发write_file、run_bash两个工具正好覆盖“写 执行 观察”的完整循环。if __name__ __main__: cfg, settings load_config() client build_client(cfg) handlers make_handlers(settings) Path(workspace).mkdir(exist_okTrue) messages [{ role: user, content: ( 在 workspace 下创建 hello.py内容为打印 1 到 5 的平方 然后用 python 运行它把输出告诉我。 ), }] response, messages agent_loop(client, cfg, settings, messages, TOOLS, handlers) print( final ) for block in response.content: if hasattr(block, text): print(block.text)运行前注入 Keyexport TAOTOKEN_API_KEY你的Key python main.py预期你会看到类似这样的过程第一轮模型返回一个write_file的 tool_use循环执行后把结果塞回第二轮模型返回run_bash执行python workspace/hello.py第三轮stop_reason变成end_turn模型用自然语言汇报输出。最终打印大致是 final 已创建 workspace/hello.py 并运行输出为 1 4 9 16 25同时transcripts/events.jsonl里应该能看到每一轮的事件记录。如果你把save_full_transcript打开完整的消息列表也会落盘——这就是后面做 Task-Process Data 收集的原始素材模型看到了什么、决定调什么工具、实际执行结果如何全都有迹可循。6. 本篇常见错排查报 401 或 authentication_error八成是TAOTOKEN_API_KEY没注入到当前 shell或者 Key 复制时带了空格。先echo $TAOTOKEN_API_KEY确认非空再检查base_url是不是写成了带路径的形式——它应该就是https://taotoken.net/api。循环跑满 max_iterations 还没结束通常是工具返回的内容让模型误以为任务没完成。检查run_bash的返回里有没有把 stderr 也带上模型看到报错会反复重试。把max_iterations设成 12 是保守值调试期可以调小到 5 快速暴露问题。path outside workspace 异常_safe函数用resolve()做了前缀校验如果模型传了../开头的路径会被拦下。这是预期行为不是 bug。如果你确实需要访问 workspace 外的文件改allowed_paths别绕过校验。tool_result 的 tool_use_id 对不上每个tool_result必须和对应的tool_use的id严格匹配顺序也要一致。如果你在循环里过滤了某些 block记得 results 列表也要同步过滤否则下一轮请求会被服务端拒绝。上下文突然变长导致响应变慢run_bash输出大文件时最容易触发。max_output_chars的截断是在追加前做的确认你截断的是content字段而不是整个 result 对象。压缩阈值compress_threshold_tokens到 60000 才触发前期不用管。7. 下一步把骨架接进你的工作流跑通这条最小闭环之后骨架的扩展方向很清晰。想加知识注入就在TOOLS里加一个load_skill工具按需把文档片段塞进上下文想加持久化任务就把消息列表定期写进 JSONL重启后从文件恢复想做多 Agent 协作每个 Agent 一个独立的消息列表和工具集用一个邮箱文件在它们之间传消息。如果你打算长期用这套骨架驱动编码任务可以了解下 Coding Plan它把这类循环的调用额度做了打包适合高频跑 Agent Loop 的场景https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite想先验证模型在具体任务上的表现可以直接在模型对话里试https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewrite接入细节和参数说明都在文档里https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite我自己的习惯是每次改完工具定义先跑一遍这条“写文件 执行 汇报”的验证链确认循环没退化再去接真实任务。骨架稳了后面加多少工具都只是往handlers字典里塞函数的事。
