Claude Code 09：Agent Teams——文件消息总线与 Mailbox 协作

在第 4 篇中，我们引入了 Subagent（子智能体）。但那时的 Subagent 是一次性的临时工：任务派生出去，子 Agent 执行完毕返回纯文本结论后随即被销毁。它没有持久化的团队身份，不能与其他平级 Worker 交流，更无法支持长期跨会话的多方协同。

当面对庞大的工程重构——例如微服务拆分或全栈功能迭代时，我们需要一个常驻的 Agent 团队（Agent Teams）：

  • 每个团队成员有明确的持久身份（Leader、Frontend、Backend、Tester）；
  • 成员之间拥有私有收件箱，能够自主收发点对点消息；
  • 团队元数据与通信记录完全持久化在磁盘上。

很多团队构建多智能体系统时第一反应是引入 RabbitMQ、Redis 或 Kafka 等重型消息中间件。然而，Claude Code 展现了极致务实的工程取向：完全基于本地文件系统的 Mailbox（邮箱目录）架构构建高可靠消息总线。

本文深入剖析 Claude Code 团队创建的物理机制与文件消息总线实现。

团队的诞生：TeamCreateTool 生产源码实证

在 Claude Code 源码 src/tools/TeamCreateTool/TeamCreateTool.ts 中，创建一个多智能体团队有着极其严谨的约束逻辑：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
// src/tools/TeamCreateTool/TeamCreateTool.ts 核心流程
export class TeamCreateTool {
  async call(input, context) {
    const { setAppState, getAppState } = context

    // ① 单主导约束：一个 Leader 只能同时掌管一个活动 Team
    const existingTeam = appState.teamContext?.teamName
    if (existingTeam) {
      throw new Error(`当前已是指挥团队 "${existingTeam}" 的 Leader。请先调用 TeamDelete 解散旧团队。`)
    }

    // ② 冲突避碰：如果团队名冲突，自动附带随机哈希后缀
    const finalTeamName = generateUniqueTeamName(input.team_name)

    // ③ 确定性 Agent ID：规则为 "team-lead@{teamName}"
    const leadAgentId = formatAgentId(TEAM_LEAD_NAME, finalTeamName)

    // ④ 磁盘持久化 TeamFile 状态
    const teamFile: TeamFile = {
      name: finalTeamName,
      createdAt: Date.now(),
      leadAgentId,
      leadSessionId: getSessionId(),
      members: [{
        agentId: leadAgentId,
        name: TEAM_LEAD_NAME,
        agentType: 'lead',
        joinedAt: Date.now(),
        cwd: getCwd(),
        subscriptions: []
      }]
    }
    await writeTeamFileAsync(finalTeamName, teamFile)

    // ⑤ 团队清理钩子：会话结束时安全清理临时文件
    registerTeamForSessionCleanup(finalTeamName)

    // ⑥ 任务看板绑定：Team = Project = TaskList，每个新团队独立维护任务池
    await resetTaskList(sanitizeName(finalTeamName))
    
    // ...
  }
}

关键设计决策

  1. 确定性 Agent ID（team-lead@{teamName}）：任何后续加入的成员 Agent 都可以根据当前团队名直接推算出 Leader 的 ID，无需任何动态网络服务发现；
  2. 生命周期绑定与磁盘垃圾回收：注册 registerTeamForSessionCleanup，防止会话结束后在用户硬盘中遗留孤儿团队元数据；
  3. 团队与任务池一一绑定：新建团队自动重置专属的任务列表，确保团队上下文纯净。

基于文件系统的 Mailbox 消息总线

不需要配置任何外部网络服务，每个 Agent 实例在磁盘上分配一个专属收件箱目录：

消息投递的物理原子性：.tmp 重命名机制

在多进程同时读写文件系统时，最致命的隐患是：写入方正在往文件写了一半，接收方恰好执行读取，导致读出残缺的 JSON 引发解析崩溃。

标准的解决范式是利用 POSIX 文件系统的原子重命名特性：先将消息写入 msg_123.tmp，完全写入并同步磁盘（flush）后，通过 os.replace() 原子重命名为 msg_123.json。接收方只监听 .json 扩展名的文件，确保读到的内容百分之百完整。

极简 Python 实现：FileMessageBus

我们实现一个具备点对点发送与收件箱轮询的轻量文件消息总线：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
import os
import json
import time
import uuid
from pathlib import Path
from typing import List, Optional

class FileMessageBus:
    def __init__(self, team_dir: str = "./.team"):
        self.team_dir = Path(team_dir)
        self.inbox_dir = self.team_dir / "inbox"
        self.inbox_dir.mkdir(parents=True, exist_ok=True)

    def _get_agent_inbox(self, agent_id: str) -> Path:
        p = self.inbox_dir / agent_id.replace("@", "_")
        p.mkdir(exist_ok=True)
        return p

    def send_message(self, sender_id: str, recipient_id: str, content: str, msg_type: str = "text") -> str:
        """点对点原子投递消息"""
        msg_id = f"m_{int(time.time()*1000)}_{uuid.uuid4().hex[:6]}"
        payload = {
            "msg_id": msg_id,
            "sender": sender_id,
            "recipient": recipient_id,
            "type": msg_type,
            "content": content,
            "timestamp": time.time()
        }

        inbox = self._get_agent_inbox(recipient_id)
        tmp_file = inbox / f"{msg_id}.tmp"
        final_file = inbox / f"{msg_id}.json"

        # 1. 完整写入临时文件
        with open(tmp_file, "w", encoding="utf-8") as f:
            json.dump(payload, f, ensure_ascii=False, indent=2)
            f.flush()
            os.fsync(f.fileno())

        # 2. 原子重命名，对接收方立即可见且无碎片风险
        os.replace(tmp_file, final_file)
        return msg_id

    def read_inbox(self, agent_id: str) -> List[dict]:
        """读取并清空当前 Agent 的收件箱"""
        inbox = self._get_agent_inbox(agent_id)
        messages = []
        for file in sorted(inbox.glob("*.json")):
            try:
                with open(file, "r", encoding="utf-8") as f:
                    msg = json.load(f)
                messages.append(msg)
                # 读取完毕即安全删除，保证消息不被重复消费
                file.unlink()
            except Exception:
                continue
        return messages

多智能体协作链路验证

  1. Leader 组建团队：创建 lead_id = "team-lead@auth-refactor"，派生专职智能体 coder_id = "coder@auth-refactor"；
  2. 派发子目标：Leader 调用 send_message(sender=lead_id, recipient=coder_id, content="请编写 src/auth/jwt.ts")；
  3. Coder 独立循环：Coder 在其专属上下文的每轮循环前调用 read_inbox()，拾取到该指令，自主执行编码；
  4. 进度闭环：编码完成后，Coder 调用 send_message(sender=coder_id, recipient=lead_id, content="代码已编写完毕并通过静态语法检查")，Leader 收到回执，驱动下一阶段单测。

架构对比：重型消息中间件 vs 磁盘 Mailbox

评估维度 传统消息中间件（RabbitMQ / Redis） 磁盘 Mailbox 架构（Claude Code）
外部依赖 需要安装 Docker 容器、配置网络端口与认证 零外部依赖，操作系统文件系统原生支持
持久化与调试 需专用控制台或命令行工具探查队列消息 极度透明，直接用 VSCode 打开查看 JSON
单机并发性能 高吞吐（万级 QPS），但存在网络序列化损耗 毫秒级 IO 延迟，完全契合开发机几十个智能体并发
崩溃恢复 依赖持久化日志与消费确认 ACK 复杂配置 未处理文件天然驻留在磁盘目录，重启自动续读

总结

Agent Teams 迈出了智能体协作的最关键一步：

  1. 持久化团队身份：每个 Agent 具备确定性命名与专有生命周期；
  2. 轻量文件消息总线：摒弃沉重的网络依赖，用原子写入保障多进程通信可靠性；
  3. 彻底解耦执行环境：每个队友可以拥有完全不同的 Prompt、工具权限和模型档次，只通过纯文本消息协议握手。

但在团队成员开始频繁发消息时，新的系统隐患随之浮现：如果一个成员给另一个成员发指令，另一个成员拒绝执行怎么办？如果通信双方相互等待，导致系统死锁怎么办？

下一篇我们将探讨通信规范与容错闭环：Team Protocols 结构化握手协议与死锁熔断。