Claude Code 10：Team Protocols——结构化通信与握手协议

在上一篇中，我们构建了基于磁盘 Mailbox 的文件消息总线，使 Agent 团队成员之间能够互发消息。

然而，如果消息只是任意的自由自然语言文本，团队很快会在高风险场景中陷入混乱与失控：

  1. 强制关机导致数据损坏：Leader 突然发一句“请停止工作退出”，而 Coder 正在写入关键文件，强行退出会导致磁盘留下半截残缺代码；
  2. 高危行为缺乏前置审批：Worker 智能体在处理疑难依赖时，擅自调用 rm -rf 清空目录或重置 Git 分支；
  3. 分布式通信死锁：Agent A 等待 Agent B 确认接口定义，而 Agent B 也在等待 Agent A 产出模型结构，两个智能体在消息队列前无限期相互挂起。

为了解决这些现实工程痛点，Claude Code 在自由通信之上建立了一套严格的强类型协议层（Team Protocols）。通过唯一的 request_id 追踪状态，实现“请求 $\rightarrow$ 审批/拒绝 $\rightarrow$ 执行反馈”的可靠闭环。

本文深入剖析其结构化协议的设计细节与死锁防范策略。

生产源码探秘：SendMessageTool 的结构化协议定义

在 Claude Code 源码 src/tools/SendMessageTool/SendMessageTool.ts 中，message 字段并不是普通的字符串，而是一个复合联合体：支持普通纯文本，或者严格受控的 StructuredMessage。

结构化协议的类型声明

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
// SendMessageTool.ts 核心契约（基于 Zod Discriminated Union）
const StructuredMessage = z.discriminatedUnion('type', [
  // 1. 关机申请：Leader 向成员发起文明下线询问
  z.object({
    type: z.literal('shutdown_request'),
    reason: z.string().optional()
  }),
  
  // 2. 关机答复：成员基于当前工作状态决定是否同意下线
  z.object({
    type: z.literal('shutdown_response'),
    request_id: z.string(),
    approve: semanticBoolean(),       // 语义布尔值 true/false
    reason: z.string().optional()      // 拒绝时必须说明理由
  }),
  
  // 3. 规划审批答复：针对高危代码方案的批准或驳回反馈
  z.object({
    type: z.literal('plan_approval_response'),
    request_id: z.string(),
    approve: semanticBoolean(),
    feedback: z.string().optional()
  })
])

生产级硬校验约束（Input Validation）

源码在执行消息发送前，强制实施了三条铁律：

1
2
3
4
5
6
7
8
9
10
11
12
13
// validateInput 核心校验逻辑
if (input.message.type === 'shutdown_response' && input.to !== TEAM_LEAD_NAME) {
  return { result: false, message: '下线确认协议 (shutdown_response) 必须且只能发送给 "team-lead"！' }
}

if (input.message.type === 'shutdown_response' && !input.message.approve
    && (!input.message.reason || input.message.reason.trim().length === 0)) {
  return { result: false, message: '拒绝下线请求时，必须在 reason 字段中明确说明当前阻塞原因！' }
}

if (input.to === '*') {
  return { result: false, message: '结构化握手协议禁止执行广播 (to: "*")，必须点对点通信！' }
}

这三条规则直接从物理层面掐死了常见的协作漏洞：

  • 成员下线必须向 Leader 报备，防止越权指挥；
  • 拒绝不能沉默：成员如果拒绝 Leader 的关机命令，必须附带理由（例如：“当前正在执行单测写入，预计 10 秒后就绪”）；
  • 握手协议禁止广播：握手涉及状态机单向转移，广播会导致确认状态混乱（Split-Brain）。

请求-响应（Request-Response）握手状态机

每一次协议交互通过递增或随机生成的 request_id 强关联。接收方回填消息时必须携带原请求的 ID，任何没有匹配上下文的孤儿回执会被丢弃。

极简 Python 实现：支持握手的协议引擎

我们在消息总线之上封装一层强类型协议调度器：

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
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
import time
import uuid
from typing import Dict, Optional

class ProtocolEngine:
    def __init__(self, message_bus, current_agent_id: str):
        self.bus = message_bus
        self.agent_id = current_agent_id
        # 跟踪当前等待对方回复的请求挂起表: request_id -> 请求元数据
        self.pending_requests: Dict[str, dict] = {}

    def send_shutdown_request(self, target_agent: str, reason: str = "") -> str:
        """向目标 Agent 发送文明关机请求"""
        req_id = f"req_{uuid.uuid4().hex[:6]}"
        payload = {
            "type": "shutdown_request",
            "request_id": req_id,
            "reason": reason
        }
        self.pending_requests[req_id] = {
            "target": target_agent,
            "type": "shutdown",
            "send_time": time.time(),
            "status": "waiting"
        }
        self.bus.send_message(
            sender_id=self.agent_id,
            recipient_id=target_agent,
            content=json.dumps(payload),
            msg_type="protocol"
        )
        return req_id

    def handle_protocol_message(self, raw_msg: dict) -> Optional[str]:
        """解析并响应收到的协议消息"""
        data = json.loads(raw_msg["content"])
        p_type = data.get("type")
        sender = raw_msg["sender"]

        if p_type == "shutdown_request":
            req_id = data["request_id"]
            # 业务安全检查：模拟检查当前是否有未完成工作
            has_unsaved_work = False  # 假设当前工作已保存
            
            if has_unsaved_work:
                reply = {
                    "type": "shutdown_response",
                    "request_id": req_id,
                    "approve": False,
                    "reason": "单测正在运行，预计 15 秒后完成"
                }
            else:
                reply = {
                    "type": "shutdown_response",
                    "request_id": req_id,
                    "approve": True
                }
            
            self.bus.send_message(
                sender_id=self.agent_id,
                recipient_id=sender,
                content=json.dumps(reply),
                msg_type="protocol"
            )
            return f"已响应来自 {sender} 的关机请求 [{req_id}]"

        elif p_type == "shutdown_response":
            req_id = data["request_id"]
            if req_id in self.pending_requests:
                approve = data["approve"]
                self.pending_requests[req_id]["status"] = "approved" if approve else "rejected"
                return f"收到 {sender} 的关机审批结果: approve={approve}, reason={data.get('reason', '')}"

        return None

防死锁与超时熔断保护

在分布式 Agent 系统中，通信死锁是最难定位的 Bug。Claude Code 制定了严格的层级与超时保护规程：

1. 严格层级命令流向（单向约束）

  • shutdown_request 只能由 Leader 下发给 Worker，Worker 严禁向 Leader 反向发关机申请；
  • 同级平级 Worker 之间仅允许共享只读的分析数据，严禁相互施加阻塞式审批指令。

2. 超时快速熔断（Circuit Breaker）

挂起表中的请求设置默认超时窗口（如 60 秒）。若目标 Worker 由于死循环未能在规定时间内返回结构化确认：

  • 系统自动判定为 TIMEOUT_REJECTED；
  • 解除 Leader 的等待阻塞状态；
  • 将故障子 Agent 标记为失联，触发强行回收与告警，防止整个团队长期卡死。

总结

Team Protocols 让多智能体协作从“随意的聊天群”蜕变为“严谨的生产流水线”：

  1. 强类型消息契约：区分普通文本与协议指令，杜绝非结构化文本的理解歧义；
  2. 状态显式握手：用唯一的 request_id 追踪关键决策流，确保下线与高危操作绝对可控；
  3. 超时熔断防护：规避分布式协作中的相互挂起与死锁风险。

但到目前为止，团队的所有任务依然依赖 Leader 手动拆解并点名派发给具体的 Worker。如果团队有 10 个智能体，Leader 就会成为吞吐瓶颈。

下一篇我们将探讨去中心化协同范式：Autonomous Agents 自组织团队与动态抢单机制。