我用 600 行代码让本机的 4 个 AI 编程工具真正协作
一、问题多 agent 不是「更多窗口」是「更多孤岛」我本机装了 Claude Code、Codex、Reasonix、ZCode 四个 AI 编程工具。它们各自都能干活但彼此之间是完全隔离的进程——没有共享状态不知道对方在做什么。我想让「擅长架构推理的那个」把实现细节甩给「擅长写代码的那个」时只能自己当人肉消息队列复制上下文粘到另一个窗口。要解决的本质问题其实是经典的多进程协作只不过参与方换成了 LLM agent一个共享的、并发安全的任务状态谁在做什么、做到哪了一套所有 agent 都能调用的通信原语派发 / 认领 / 完成 / 提问 / review一个让异构客户端都能接入的传输层一点边界隔离别让 A 项目的 agent 翻到 B 项目下面逐个拆。二、并发模型为什么是flockr而不是数据库任务板就是一个 JSON 文件。多个 agent 进程会并发读写它所以核心是read-modify-write 必须原子。常见的错误写法是「读出来 → 改 → 写回去」分三次开文件中间任何一个进程插进来就丢更新。正确做法是在同一把排他锁下完成整个读改写defatomic_update_board(path,update_fn):Hold exclusive lock across read-modify-write. update_fn(board) - board.ifnotpath.exists():withopen(path,w)asf:json.dump({version:BOARD_VERSION,tasks:[]},f)# 关键用 r 打开一把锁里既读又写withopen(path,r)asf:fcntl.flock(f,fcntl.LOCK_EX)# 排他锁其他写者阻塞try:f.seek(0)boardjson.load(f)# 读boardupdate_fn(board)# 改调用方传进来的纯函数board[version]BOARD_VERSION f.seek(0);f.truncate()# 清空再写避免残留旧内容尾巴json.dump(board,f,indent2)# 写f.flush();os.fsync(f.fileno())# 落盘finally:fcntl.flock(f,fcntl.LOCK_UN)几个容易栽的细节r而不是ww一打开就把文件清空了还没拿到锁内容就没了。r保留内容等拿到锁再truncate。seek(0) truncate()新内容比旧的短时不 truncate 会留下旧数据的尾巴JSON 直接解析失败。fsyncflush只到内核缓冲fsync才真正落盘防止崩溃丢任务。所有写操作都走这一个函数改动以「传一个update_fn(board)-board闭包」的形式表达。claim、done、question、review……全是这一个原子原语的不同闭包defcmd_claim(args):def_claim(board):fortinboard[tasks]:ift[id]args.task_id:ift[to]!name:# 只能认领派给自己的raiseSystemExit(not assigned to you)ift[status]notin(pending,input_required,changes_requested):raiseSystemExit(falready{t[status]})# 防重复认领t[status]workingreturnboardraiseSystemExit(not found)atomic_update_board(bp,_claim)读操作则用共享锁LOCK_SH多个读者可以并发只和写者互斥defread_board(path):withopen(path,r)asf:fcntl.flock(f,fcntl.LOCK_SH)try:returnjson.load(f)finally:fcntl.flock(f,fcntl.LOCK_UN)为什么不上 SQLite单机、agent 交互频率每次人类回合一两次调用JSON flock 完全够而且人能直接cat board.json看状态、出问题手动改。引入数据库是用一个常驻复杂度换一个我根本不会遇到的吞吐瓶颈。活动日志activity.jsonl同理追加写 flock超过 1 万行砍掉前一半轮转避免 hook 读它的时候越来越慢。三、收件箱是个状态机不是「派给我的就显示」最容易写错的地方什么任务该出现在我的收件箱里直觉是「to 我的」但这是错的。一个任务在生命周期里会在两个 agent 之间来回弹我被派了活 → 在我的箱子里我做的时候有疑问question打回去 → 应该在派活人的箱子里等他回答不在我这他answer了 → 又回到我箱子里我做完请 review → 在reviewer也就是原派活人箱子里他说changes要返工 → 又回我箱子所以收件箱是按状态 角色双重判断的过滤器def_inbox_filter(task,me):statustask.get(status)# 提问/请求 review回到原发起人from那里去处理ifstatusin(input_required,review_requested):returntask.get(from)me# 否则只看派给我的iftask.get(to)!me:returnFalse# 作为受派人要干的新任务或被打回返工的returnstatusin(pending,changes_requested)完整状态集任务字段statuspending → working → completed ↑ ↓ │ input_required (提问等 answer) │ review_requested → review_approved └──── changes_requested (返工)这套设计的好处一个任务永远只在一个 agent 的「待办」里不会两边同时亮、也不会两边都不管。每个 agent 每个回合开头调一次bridge_status只看到「这回合轮到我动」的东西。四、传输层一个零依赖的 MCP stdio serverCLIbridge.py解决了「人和脚本怎么用」但桌面 App 形态的 agent 需要的是MCPModel Context Protocol——它是 Claude / Codex / Reasonix / ZCode 的最大公约数。我没有引入任何 MCP SDK直接按 JSON-RPC 2.0 over stdio 手写了一个整个 server 是 bridge.py 的薄包装defhandle(req,identity):methodreq.get(method)ifmethodinitialize:respond(rid,{protocolVersion:2024-11-05,capabilities:{tools:{}},serverInfo:{name:agent-bridge,version:0.1.0}})elifmethodtools/list:respond(rid,{tools:[{name:n,description:t[description],inputSchema:t[schema]}forn,tinTOOLS.items()]})elifmethodtools/call:specTOOLS[params[name]]# 把 MCP 调用翻译成一次 bridge.py 子进程调用rsubprocess.run(build_argv(spec,args,identity),capture_outputTrue,textTrue,timeout40)out(r.stdoutr.stderr).strip()or(no output)respond(rid,{content:[{type:text,text:out}],isError:r.returncode!0})每个工具就是一张「MCP 工具名 → CLI 子命令 参数」的映射表TOOLS{bridge_send:{sub:send,pos:[],flags:[to,skill,subject,body,files,project],schema:{type:object,required:[subject],properties:{...}},},# ... 共 14 个}为什么是子进程而不是把逻辑 import 进来因为 agent 交互速率下「一次调用 一个进程」的开销完全无所谓但换来的是CLI 和 MCP 共用一套逻辑、永不漂移——不会出现 CLI 修了个 bug 而 MCP 路径还是老的。这是典型的「用一点性能换一份正确性和零重复」。踩坑实录模型调store_true这种布尔开关时有的传真正的true有的传字符串true。直接if v:会把字符串false也判成真。所以加了一层def_truthy(v):returnvisTrueor(isinstance(v,str)andv.strip().lower()in(true,1,yes))五、安全边界用 cwd 做项目隔离而不是搞鉴权单机单用户没必要上 token / 鉴权。我要的只是作用域隔离A 仓库里的 agent 别看到 B 仓库的任务板。规则借鉴 git 发现仓库的方式——项目绑定到一个 workspace 目录从当前 cwd 反查所属项目取最长前缀匹配defresolve_project(explicit):ifexplicit:returnexplicit cwdstr(Path.cwd().resolve())bestNoneforpjinPROJECTS_DIR.glob(*/project.json):wsjson.load(open(pj)).get(workspace,)wsrstr(Path(ws).resolve())if_under(cwd,wsr)and(bestisNoneorlen(wsr)best[1]):best(pj.parent.name,len(wsr))# 最长前缀胜出支持嵌套项目returnbest[0]ifbestelsedefaultdefenforce_workspace(pid):绑定了 workspace 的项目从目录外访问直接拒绝wsproject_workspace(pid)ifnotws:return# 未绑定的如 default开放ifnot_under(cwd,str(Path(ws).resolve())):sys.exit( refusing cross-project access)一句话概括这条安全模型两个 agent 能协作当且仅当它们的 cwd 落在同一个 workspace 下。这是「划范围」不是「防黑客」——威胁模型就是我自己一台机器够用就行。过度设计一套鉴权反而是给自己半夜 3 点的 oncall 埋雷。六、Pull 为主、Push 兜底怎么叫醒一个没开着的 agent整套是pull 模型每个 agent 每回合开头自己查收件箱。这对「正开着的」agent 没问题但派给一个当前没运行的 agent 怎么办加了两个尽力而为best-effort失败绝不影响主流程的推送def_wake_agent(name):跑目标 agent 注册的 headless 唤醒命令如 Reasonix 的 reasonix runwakejson.load(open(agent_dir(name)/agent.json)).get(wake)ifnotwake:returnFalsepromptRun bridge inbox; if a task is pending, claim and complete it.subprocess.Popen(wake.split()[prompt],stdoutDEVNULL,stderrDEVNULL,start_new_sessionTrue)returnTrue外加一个osascript/notify-send的桌面通知提醒人类切过去。最爽的一次端到端验证我在 Claude 这边bridge send --to reasonix --wakeReasonix跑 DeepSeek被 headless 拉起来自己bridge show读任务、claim、算完、done写回结果全程我没碰键盘。那一刻才算真正「两个 AI 在协作」而不是我在中间搬运。七、路由不是写死的表是让协调者「读简历自己判断」我没维护「这类任务派给这个工具」的静态映射。每个 agent 注册时带一段自由文本strengths比如「hard reasoning, architecture (GPT-5.5)」项目里第一个动手的 agent 成为协调者它读一眼bridge agents的能力矩阵 项目的CONTEXT.md自己决定这个项目怎么分工defcmd_send(args):set_coordinator(pid,name)# 第一个发任务的人成为协调者# --skill 只是个可选的兜底自动路由不是硬规则# 真正的路由是协调者「模型」的判断为什么不写死因为「谁擅长什么」是会变的模型升级、换后端把它编码进 if-else 就是给未来的自己留维护债。让模型读文本自己判断反而少写一堆代码。八、那些只有真机安装才会暴露的坑纸面设计跑 demo 都顺真往四个工具上装才掉链子。挑几个有代表性的心跳把简历冲了每次status调用要刷新last_seen心跳最初实现是整个agent.json重写结果把strengths/skills覆盖没了。修法是读出来只改时间戳字段再写回def_touch_heartbeat(name):datajson.load(open(af))ifaf.exists()else{}data[name]name data[last_seen]utcnow()# 只动这个别的字段原样保留时区算错doctor 检查心跳新鲜度用time.mktime把 UTC 字符串当本地时间解析差了 8 小时。改用calendar.timegm明确按 UTC。Python 3.9 不认str | None得在文件头加from __future__ import annotations。最致命的一个agent 能在收件箱看到「有个任务」但看不到任务正文——等于知道有活却不知道干啥。原因是 inbox / board 只打印了 subject。补上 body / files / question / answer 的展示并加了个bridge show id看全量字段后协作才真正跑通。这个 bug 提醒我多 agent 系统里「能感知到消息」和「能读懂消息内容」是两件事少了后者整个系统看着在转其实空转。九、小结整套东西的设计取向就一句话能用标准库就别加依赖能用一个文件就别起服务能让模型判断就别写死规则。并发安全fcntl.flockr原子读改写一个atomic_update_board收敛所有写操作协作语义一个按「状态 角色」过滤的收件箱状态机互通零依赖手写 MCP stdio server薄包装 CLI 保证两条路径不漂移隔离cwd 最长前缀匹配的项目边界异步pull 为主 headless wake 兜底代码全在仓库里四个工具的接入脚本install.sh --auto也在https://github.com/xyva-yuangui/roundtable接下来想做跨机同步、Linux/Windows 的应用检测和 CI。欢迎来提 issue / PR尤其欢迎把更多 agent 接进来一起坐这张桌子。如果你也被一桌各干各的 AI 工具烦过顺手点个 star。