用 Claude Code 写了几个小项目之后我最大的感受是这个工具本身很强但它只给了你一个交互界面并没有帮你解决“规模化使用”的问题。你可以在终端里问它改一个函数、修一个测试可一旦你想让它批量处理几十个文件、自动跑完整轮迭代或者把它嵌进自己的自动化流水线里你马上就会需要一层“壳”——也就是标题里说的 Agent Harness。这篇文章要聊的就是这个东西它到底是什么、为什么要自己搭、核心部件怎么设计以及我是怎么从零写了一个最小可用的 harness 并跑通实测的。适合对 Claude Code 有基础了解、想把它推向批量和自动化场景的开发者参考。1. 项目概述与整体设计思路1.1 先搞清楚 Agent Harness 到底在解决什么问题很多人第一次听说“Agent Harness”会觉得是一个很玄的概念其实它没那么神秘。按我个人的理解Harness 的直译是“马具”放在 AI Agent 的场景里就是一套用来约束、驱动、监控 Agent 运行的骨架代码。你可以把 Claude Code 想象成一个能力很强但不太听话的外包员工你直接给他派活他可能干得不错也可能跑偏而 Agent Harness 就是那个项目经理兼监工负责拆任务、派工、盯进度、收结果、处理异常。在实际开发里Claude Code 本身提供了 REPL交互式终端模式和命令行模式你用claude -p some prompt就能单次调用它。但这里有三个痛点第一单次调用是阻塞的你要等它跑完才能拿到结果没法按时做进度反馈或中途干预。第二输出是给人看的文本不是结构化的程序数据你很难在代码里直接解析“它到底改了什么文件、用了哪些工具、花了多少时间”。第三没有容错机制网络波动、token 超限、工具执行失败都会让整个调用直接崩掉而 harness 可以把这些异常接住并做重试或降级处理。所以我搭这个 harness 的核心目标很明确把 Claude Code 变成一个可控的、可编程的子进程通过标准输入输出和事件流跟它通信然后用你自己的代码来管理任务的全生命周期。能做到这一点你才有资格谈“AI 编程助手自动化”这件事。1.2 设计取舍为什么不用现成框架偏要自己写一套在开始动手之前我其实调研过一些基础设施工具包括某些开源的 agent orchestration 框架。它们确实很强大但对我来说有一个共同的问题太重了。光是要理解它的配置体系、插件机制、数据模型就已经消耗掉大半天精力而且一旦出问题排查的链路非常长。而我的核心诉求其实只有四条要能捕获 Claude Code 的结构化输出、要能控制它的运行生命周期、要能执行任务队列、要能在失败时重试。这四条用标准库加少量胶水代码完全能实现不需要引入一整套框架。这也是我最终决定从零写 harness 的根本原因——不是那些框架不好而是对“只想把 Claude Code 用起来”的场景来说自研壳的性价比最高。另外我想强调一个设计原则就是“薄壳原则”。Harness 不要试图替代 Claude Code 本身的推理能力它只做控制流的事发指令、收事件、存结果。Agent 内部的思考、规划、工具调用全部交给 Claude Code 自己完成。这样你的 harness 代码始终是简单、可维护的不会因为大模型行为的不确定性而变得不可控。1.3 整体架构四个核心部件一张图说清先给一个宏观视图方便后续对照。整个 harness 由四个核心部件组成调度器Scheduler负责从任务队列里取任务维护每个任务的状态决定哪个任务该执行、哪个该暂停、哪个该重试。子进程管理器Process Manager负责启动和守护 Claude Code 的 CLI 子进程写入 prompt读取 stdout/stderr。事件解析器Event Parser负责解析 Claude Code 输出的 stream-json 事件流把文本块、工具调用、错误码全部转成 Python 对象。结果收集器Result Collector负责把多轮事件聚合成结构化的任务结果落盘保存并支持回调钩子。为什么要分层因为每一层的变化频率不一样。比如事件解析器会跟着 Claude Code 的输出格式升级而改而调度器的逻辑基本稳定。分层之后任何一层的改动都不会扩散到整个项目这也是一个工程经验把变化点隔离。2. 环境准备与基础概念2.1 你需要准备什么Node 环境与 Claude Code 安装在真正写 harness 之前环境准备工作不能省。Claude Code 本身是一个 npm 包所以机器上要先有 Node.js 环境。我建议装 LTS 版本因为 agent 类的 CLI 工具对运行时的稳定性要求比较高追新版本反而容易踩到兼容性坑。安装 Claude Code 只需要一条命令npm install -g anthropic-ai/claude-code装完之后在终端敲一下claude --version能正常输出版本号就说明安装成功。这里我想提醒一句很多人会忽略 API Key 的环境变量配置Claude Code 底层要调用大模型接口你必须确保ANTHROPIC_API_KEY已经导出或者在它自己的配置文件里登录过。否则后面跑子进程的时候你会看到一长串鉴权错误而且经常是隐藏在 stderr 里的比较容易忽略。顺便说明一下我在下面的实操里会用 Python 来写 harness 代码因为 Python 的subprocess模块对子进程的管理很顺手而且做任务队列、数据落盘、重试调度都不需要额外依赖。2.2 理解 Agent Loop为什么需要一个“壳”来管理在动手写代码之前我想先帮大家把“Agent 是怎么工作的”这个底层机制理清楚。Claude Code 的内部运行是一个循环它接收一个任务描述然后自己规划要做什么调用工具比如读文件、改文件、执行命令观察工具返回的结果再决定下一步动作直到它认为任务完成。这个循环在交互模式下你感知不到因为每一步都实时显示在屏幕上。但一旦你想要自动化就必须明白Claude Code 是一个“自主性很强的黑盒”它的每一步输出都值得你关注甚至需要你中途纠正方向。这也是 harness 最重要的一个作用——把黑盒的每个关键动作暴露成结构化事件让你有办法在循环中途介入而不是只能傻等最终结果。我再打个比方。你用交互模式跟 Claude Code 聊天就像你在工位上跟一个同事当面协作对方每个动作你都能看见不对可以及时打断。而 harness 相当于你把这位同事派到远程你没法一直盯着他只有“工作汇报事件流”。所以你会更依赖汇报的完整性和格式这也是为什么事件流解析是整个 harness 里最需要仔细处理的部分。2.3 认识 stream-jsonClaude Code 给外部程序留的门既然要写 harness最核心的接口就是 Claude Code 的命令行标志位。先说几个我实测过比较重要的-p, --print非交互式模式直接让 Claude Code 执行一次 prompt然后把结果输出到 stdout。适合单次调用。--output-format json配合 print 使用时输出不再是纯文本而是 JSON 结构这就能被程序解析了。--verbose输出详细日志适合排查问题。--max-turns限定 agent 推理循环的最大轮数防止它陷入死循环。--permission-mode控制工具权限模式比如是否允许它自动执行 bash、读文件还是写文件。以--output-format json为例输出的内容实际上是一行一条的 JSON 事件这就是 stream-json 的核心。每条事件都有type字段常见的包括system、assistant、user、result等。你要做的就是在 Python 里逐行读取子进程的 stdout把每个 JSON 对象解析出来按类型分发处理。这里有个非常容易踩的坑非交互模式下Claude Code 的输出流跟交互模式完全不一样如果你在终端直接跑claude -p hello看到的是一大段带格式的文本但如果你接住 stdout 再打印你会发现有大量\r回车符和颜色控制码直接解析文本会非常痛苦。所以我的建议是永远优先用--output-format json只把 JSON 事件作为事实来源其他格式只用于人工调试。3. 核心部件设计与关键技术解析3.1 状态机设计给任务一个清晰的生命周期Harness 里第一个要设计的组件是任务状态机。为什么需要状态机因为 AI 编程任务不是“提交-返回”的一锤子买卖它中间有多个阶段排队、启动、运行、等待工具结果、完成、失败。如果没有显式的状态管理你的代码会在各种分支判断里越来越乱。我设计的状态集合如下刻意保持精简状态含义PENDING任务已入队等待调度器启动RUNNINGClaude Code 子进程正在运行中WAITING_INPUTAgent 在等待外部输入或确认COMPLETED任务正常完成结果已收集FAILED任务运行出错或达到最大重试次数CANCELLED任务被外部取消状态机的好处在于每个状态只允许发生有限的合法迁移。比如只有RUNNING能变到COMPLETED或FAILEDPENDING应该永远变不到COMPLETED直接成功除非你写了荒谬的 bug。这种约束让整个调度器变得非常容易推理也给后续扩展并发执行打了好底子。3.2 子进程管理控制 Claude Code 从启动到退出Claude Code 本质上就是一个命令行程序所以 harness 最核心的技术动作就是操作子进程。我用 Python 的subprocess.Popen来做这件事原因很简单它给了你三个通信管道——stdin、stdout、stderr你可以写入 prompt、读取事件流、捕获错误日志。我重点想讲几个细节。第一启动参数里必须加上--output-format json并且把--verbose关掉否则事件流里混杂太多调试信息解析会很费劲。第二要用bufsize1开启行缓冲这样才能逐行读取输出实时处理事件。第三给子进程设置超时是必须的因为大模型推理的耗时不可预测一个复杂任务跑十几分钟都是正常的但也可能会有网络卡死导致的无限挂起。下面是我实际用的启动逻辑的核心代码import subprocess cmd [ claude, -p, prompt_text, --output-format, json, --max-turns, 50, --permission-mode, acceptEdits, ] proc subprocess.Popen( cmd, stdinsubprocess.PIPE, stdoutsubprocess.PIPE, stderrsubprocess.PIPE, textTrue, bufsize1, envenv_with_keys, )注意--permission-mode这里我用的是acceptEdits意思是允许 agent 直接修改文件而不需要人工逐次确认。这是一个很大的权限决定你必须在真实项目里评估风险。我自己通常在隔离的临时工作目录里跑这样即使它把文件改坏了也不会伤到主要代码。3.3 事件流解析把黑盒变成白盒的关键环节很多人问harness 里最值得花时间打磨的部分是什么我的答案永远是事件解析。因为 Claude Code 的行为是否可控完全取决于你能不能及时、准确地理解它的动作。stream-json 事件里的核心类型我用一个表格总结一下type含义关键字段system子系统状态比如模型初始化完成subtype比如initassistant模型发给用户的内容content数组含文本或工具调用user工具执行结果或用户输入content数组通常是工具返回内容result任务完结时的汇总result包含最终消息和耗时error运行错误error对象包含类型和消息解析逻辑其实很简单逐行读取json.loads然后按type分发。但有一个很重要的细节assistant事件里的content是一个数组里面可以同时包含text文本块和tool_use工具调用块。如果只接住text你会漏掉它“准备调用什么工具”的关键信息。我的做法是把事件流原封不动地追加到一个 JSONL 日志文件里。任务结束后这个文件就是整个 Agent 运行的完整回放数据再写任何分析、测试断言、性能统计都从这里取数。这也是我在反复踩坑之后总结出来的最佳实践先记录再解析永远不要只保留解析后的抽象结果。3.4 工具权限与沙箱隔离给 AI 一个“可以破坏”的环境讲工具权限管理之前先说一个我亲眼见过的教训。某个开发者第一次搭 harness直接用了完全放开的自动权限让 agent 在他真实的项目仓库里跑重构任务。结果 agent 一顿操作下来在十几处文件里产生了不一致的修改有些甚至是它自己生成后又误删的。想在 git 里恢复都费半天劲。所以我的忠告是harness 的默认环境必须是沙箱。我实操时通常准备两层隔离。第一层是工作目录隔离每个任务都在一个新的临时目录里执行任务结束后把结果文件复制出来整个目录直接删掉。第二层是工具权限约束Claude Code 本身支持不同的 permission mode比如只允许读文件、只读加写文件、完全信任。我建议权限从小到大逐步放开先跑通流程再按需放宽。很多人在这一步会有疑问权限放宽了跑得快我为什么要为了安全性拖慢效率我的看法是AI 编程助手的价值在于它能独立完成重复性高的重构和修补工作但如果它把代码改坏了你返工的成本会远远超过那点加速收益。沙箱不是限制而是安全带真正上赛道的时候你才会感谢它。4. 实操过程从零构建一个最小可用的 Agent Harness4.1 初始化项目先把骨架搭出来在写代码之前我习惯先建一个干净的目录结构这样后续扩展模块时不需要反复挪动文件。这里给出我用的目录布局你可以按需删减agent-harness/ ├── harness/ │ ├── __init__.py │ ├── scheduler.py # 任务调度器 │ ├── process_manager.py # 子进程管理器 │ ├── event_parser.py # 事件解析器 │ └── models.py # 任务和事件的数据结构 ├── tasks/ │ └── sample_task.py # 示例任务定义 ├── output/ │ └── .gitkeep # 结果输出目录 └── main.py # 入口脚本为什么要把任务定义独立在tasks/目录因为 harness 本身不应该关心具体业务它只负责“跑任务”。而任务描述、输入参数、验收条件是会频繁变化的抽出来之后你可以像配实验参数一样自由调整不需要动 harness 核心代码。这也是我把“薄壳原则”落实到了目录结构上。4.2 子进程管理器实现让 Python 稳稳接住子进程先把最底下跟 Claude Code 直接打交道的模块写好。这个模块有三个职责启动子进程、喂入 prompt、关闭进程。我保证它只有这四五十行代码后面的模块都会建立在它之上。import os import subprocess class ClaudeProcessManager: def __init__(self, work_dir: str, max_turns: int 50): self.work_dir work_dir self.max_turns max_turns self.proc None def start(self, prompt_text: str): cmd [ claude, -p, prompt_text, --output-format, json, --max-turns, str(self.max_turns), --permission-mode, acceptEdits, ] env os.environ.copy() env[CLAUDE_CODE_WORKING_DIR] self.work_dir self.proc subprocess.Popen( cmd, stdinsubprocess.PIPE, stdoutsubprocess.PIPE, stderrsubprocess.PIPE, textTrue, bufsize1, cwdself.work_dir, envenv, ) def read_line(self): if self.proc is None: raise RuntimeError(process not started) return self.proc.stdout.readline() def wait(self, timeout: float): return self.proc.wait(timeouttimeout) def terminate(self): if self.proc is not None: self.proc.terminate()有一个小细节我想强调一下env[CLAUDE_CODE_WORKING_DIR]这行可能在旧版 CLI 上不生效但保留它是无害的。如果你需要确保 agent 的“工作目录”就在你所指定的沙箱里可以在cwd参数里直接指定这也是我上面的做法。两个都在双保险。4.3 事件解析器实现逐行解析 JSONL 事件流接下来是事件解析器。核心任务就是给管理器的每一行输出装上一个“解码器”让后续代码只跟结构化对象打交道。import json from typing import Iterator def iter_events(proc) - Iterator[dict]: while True: line proc.stdout.readline() if not line: break line line.strip() if not line: continue try: event json.loads(line) except json.JSONDecodeError: # 这里是关键: 非 JSON 行不能吞, 要记到日志, 但可以忽略 print(f[parse-warning] non-json line: {line[:120]}) continue yield event这里的高亮点是JSONDecodeError的处理。实战中你偶尔会遇到一些以非 JSON 形式打印的警告行比如 CLI 在初始化阶段的日志。如果解析器一遇到非 JSON 就崩溃那 harness 就太脆弱了。我的做法是打印一个 warning 然后跳过同时把原始行记录下来事后排查再分析。宁可多留一点原始信息也不要让一次解析失败毁掉整个任务。4.4 调度器实现把任务队列串起来调度器是整条链路的“总指挥”。它要做的事情听起来很直接按顺序取任务、交给子进程管理器执行、接收事件流、更新状态、落盘结果。但我在实现的时候追加了两个关键机制重试和降级。import time from harness.process_manager import ClaudeProcessManager from harness.event_parser import iter_events class HarnessScheduler: def __init__(self, work_dir: str, max_retries: int 2, timeout: float 600.0): self.work_dir work_dir self.max_retries max_retries self.timeout timeout def run_task(self, task: dict) - dict: task_id task[id] prompt_text task[prompt] retries_left self.max_retries while retries_left 0: manager ClaudeProcessManager(work_dirself.work_dir) try: manager.start(prompt_text) events [] for event in iter_events(manager.proc): events.append(event) if event.get(type) result: break manager.wait(timeoutself.timeout) return { task_id: task_id, status: completed, events: events, retries_used: self.max_retries - retries_left, } except Exception as exc: retries_left - 1 print(f[scheduler] task {task_id} failed: {exc}, retries left: {retries_left}) manager.terminate() time.sleep(2) return {task_id: task_id, status: failed, events: []}我觉得这个实现已经能体现 harness 的核心价值了。你看整个循环它并没有干预 Claude Code 内部的推理但它给了任务“再来一次”的机会。真实场景里大模型的调用可能会因为临时超时、网络抖动、上下文被污染而失败简单重试往往就能救回来。这个增量代码只有十几行带来的稳定性提升却非常明显。4.5 主程序与入口把任务喂给 Harness入口脚本的任务是定义具体的任务清单然后调用调度器。我这里会完整展示怎么定义任务这样你可以直接替换成自己的需求。from harness.scheduler import HarnessScheduler def main(): tasks [ { id: task-001, prompt: 请分析当前目录下 src/main.py 的代码质量。 找出可以优化的函数给出具体的修改建议。 如果发现有 bug请直接修复。, }, { id: task-002, prompt: 请为项目追加一个 README.md 文档。 内容包括项目简介、安装方式、使用示例三个部分。 整体风格保持简洁。, }, ] scheduler HarnessScheduler(work_dir./sandbox_work, timeout900) results [] for task in tasks: print(f[main] starting task {task[id]}) result scheduler.run_task(task) results.append(result) print(f[main] task {task[id]} finished: {result[status]}) for r in results: if r[status] completed: result_event [e for e in r[events] if e.get(type) result] print(r[task_id], result_event[0][result] if result_event else no result) if __name__ __main__: main()跑这个入口之前必须确认两件事一是sandbox_work目录已经存在二是这个目录下确实有src/main.py否则第一个任务没东西可分析。我建议第一次跑的时候先用一条问候类的 prompt 试水比如“请你创建一个 hello.txt 文件内容为 hello”确认整个链路的子进程启动、事件解析、结果收集都正常再去跑真实代码分析任务。渐进式验证比一口吃成胖子靠谱得多。4.6 实操心得三个参数的计算与调整逻辑参数选择往往比代码本身更影响效果。这里我把自己实际调试中发现比较敏感的三个参数讲透。第一个是超时时间。我最初设置 300 秒结果复杂任务频繁失败。后来统计了一下中等难度的代码分析任务通常在 2 到 5 分钟内完成而涉及大量文件读写或长上下文的文档生成任务偶尔会超过 10 分钟。所以我建议先跑一轮任务统计实际耗时分布再把超时设成中位数的 1.5 到 2 倍这样既不会让任务无限挂起也不会误杀正常慢任务。第二个是max_turns。这个参数控制 agent 最多执行多少轮工具调用。如果太小复杂的修改任务会在做完一半时被截断结果文件处于中间状态。如果太大一些跑偏的 agent 可能会反复尝试错误方案白白消耗大量 token。我的建议是从 30 开始完成任务后看事件流里的实际轮次再往上调余量。第三个是重试次数。从成本角度考虑重试意味着重新消耗一轮 token所以我只设了 2 次上限。但它要根据任务性质调整如果是容错性高的任务比如生成临时文件、做代码格式化重试成本低、收益高如果是会污染共享资源的任务重试前必须做好清理。5. 常见问题与排查技巧实录5.1 启动即失败子进程根本没跑起来第一个高频问题是“harness 跑起来后什么输出都没有但是 Claude Code 在终端手工能用”。我排查过几次绝大多数情况是环境变量问题。手工操作时你终端里可能已经加载了密钥但subprocess.Popen是在 Python 的os.environ基础上启动的如果你的密钥是写在 shell 的 rc 文件里、而没有真正 export 出去Python 子进程是拿不到的。我的排查步骤是先在 Python 里打印os.environ.get(ANTHROPIC_API_KEY)确认能看到值如果看不到就在启动 harness 的脚本外层显式设置好环境变量或者用一个.env文件统一管理。5.2 事件流解析乱掉非 JSON 行穿插在日志里第二个高频问题是“解析到一半报 JSONDecodeError最后结果丢了”。前面我提到过CLI 初始化阶段偶尔会把非结构化日志打到 stdout这在非交互模式下尤其容易发生。这里我想补充一个更隐蔽的情况当 agent 调用某个工具时工具自身的输出如果含有异常字符可能被直接混入 stdout。我的终极解法是永远不要尝试把 stdout“净化”成纯 JSON而是把每一行都追加进 JSONL 日志文件只有在日志完整记录之后才做解析。出了问题直接翻日志文件恢复现场。这个思路跟“[先记录再解析]”完全一致我强烈建议其他开发者照做。5.3 任务明明完成了却没有 result 事件第三个问题很令人困惑子进程跑完、退出码是 0、事件流里一路都很正常可就是没有result类型的最终事件。我后来发现这往往是因为事件流并没有在任务结束时以标准事件形式发送result尤其在一些老版本或者特殊情况比如任务被 max_turns 截断下最终状态会以普通assistant文本输出收尾。所以我在解析器里给了一个兜底逻辑事件流结束后如果没碰到result事件就把最后几条assistant事件的文本拼接成final_text作为次优结果返回。这在评估和后续处理时很有用至少你不会白跑一次任务。5.4 上下文被污染同一目录里连续跑多个任务会互相干扰最后一个问题是工程性的如果你所有任务都在同一个工作目录里跑第二个任务启动时会看到第一个任务生成的文件、镜像、中间产物它的行为会被这些残留内容影响。我第一次跑批量重构任务时产物目录里全是它新生成的和旧文件混在一起的东西非常难清理。解决方案是每个任务独享一个子目录。调度器每次从队列里取任务时动态创建一个新的临时目录任务结束后把它整体挪走或删除。这是沙箱设计里面最容易实现也最有效的一环它保证每次任务都从“干净的画布”开始而不是在一堆杂乱的旧状态上继续涂鸦。5.5 常见问题速查表问题现象可能原因排查与解决子进程无输出任务秒失败环境变量缺失、CLI 不可执行检查ANTHROPIC_API_KEY用which claude确认路径stdout 混入非 JSON 行Claude Code 初始化日志解析时跳过非 JSON 行原始行写入警告日志任务结束后没有 result 事件max_turns 截断或老版本行为兜底拼接最后几段 assistant 文本任务超时频繁超时设置过短或 prompt 过于庞大统计历史耗时分布设置中位数 1.5 到 2 倍超时任务输出互相干扰共享工作目录残留每个任务独立临时子目录结束后清理权限错误导致无法写文件沙箱目录权限不足检查运行用户对工作目录的写权限6. 扩展与进阶让 Harness 更接近生产可用到这里为止最小可用的 harness 已经能跑通任务了。但如果目标是把它用在真实的开发流程或 CI/CD 流水线里还有几个方向值得扩展。这里我挑三个我认为性价比最高的来聊。第一个是“事件驱动回调”。你现在是任务跑完才统一处理结果但很多场景想在中间动作发生时就立刻响应。比如 agent 每完成一个文件的修改就自动跑一遍对应的单元测试。这个能力靠的是事件解析器里的钩子hook机制在事件循环里检测到tool_use类型且工具名是文件写入操作时回调你注册的函数。我实测之后发现这种“流式反馈”能把很多本来要等最终产物出来才能发现的错误提前好几步拦截掉。第二个是“人工审批门”。在一些高风险操作比如大规模删除文件、修改关键配置文件前你可能希望 agent 停一停等人工确认再继续。Claude Code 的事件流里可以插入用户确认事件但你需要在 harness 层维护一个“暂停继续”的信号。我目前的实现是让调度器在检测到特定工具名时把任务状态置为WAITING_INPUT并通过一个小 API 暴露确认接口。这样整体链条就从“自动”变成了“人在环上”的半自动模式稳定性会明显提升。第三个是“批量并发执行”。当前调度器是串行处理任务的效率偏低。在我的批量代码审查场景里我把它升级成了固定线程池并发执行最多同时跑四个 Claude Code 子进程。当然这一步会带来两个新问题并发数受 API 速率限制约束、多个子进程同时读文件可能导致资源竞争。所以并发扩展的前提是你要先给自己的任务和 API 配额估算出合理的并发上限。我的建议是从2开始看错误率没有上升再缓慢加。最后再分享一个从实际测试中总结出来的体会harness 的复杂度要跟着你的使用方式成长不要在第一天就把并发、回调、审批门全部装上。先把最基础的链路跑稳拿到几次完整的运行日志再去动并发和自动化这些更高级的配置。我现在这个 harness 已经稳定跑了几百个任务但它的核心代码依然只有几百行因为每一次扩展都是被真实的运行数据推动的而不是为了“技术看起来高级”而堆上去的。对一个工具型的项目来说能持续稳定地解决具体问题比功能列表长不长重要得多。