Agent Communication Hub#

多智能体消息转发与上下文共享中间件 — v3.0.25

让两个或多个独立 AI 智能体之间实现 实时双向通信 和 上下文自动同步。基于 MCP 协议 + stdio 模式，消息本地持久化，延迟 < 50ms。

架构概览#

text

┌──────────────┐         ┌──────────────────────────────┐         ┌──────────────┐
│   Agent A    │  SSE    │   Agent Communication Hub    │  SSE    │   Agent B    │
│  (Hermes)    │◄───────►│  (stdio)                    │◄───────►│ (WorkBuddy)  │
│              │  MCP    │                              │  MCP    │              │
└──────────────┘◄───────►│  SQLite WAL + 30 表          │◄───────►└──────────────┘
                          │  58 MCP 工具 + RBAC 权限     │
                          │  上下文暂存 + 建议闭环       │
                          └──────────────┬──────────────┘
                                         │
                                    SQLite (WAL)

三层协议：

层协议用途延迟MCP 工具层stdio JSON-RPC结构化操作（发消息、分配任务、查状态）<50msSSE 推送层Server-Sent Events实时事件通知（新消息、新任务、建议确认）<50ms

快速上手 (5 分钟)#

从零到完成第一次 Agent 间通信的编号流程：

Step 1: 确认 Hub 运行状态#

确认 Agent Communication Hub 服务器正在运行。如果通过 stdio 模式接入，检查 MCP 配置是否正确加载：

text

调用: get_online_agents()
期望: 返回在线 Agent 列表（至少含自己）
失败: Hub 未运行 → 先启动 Hub 服务器

[检查点] 用户确认：如果 Hub 未运行，询问用户是否要启动 Hub 服务器。

Step 2: 注册或确认身份#

检查自己是否已在 Hub 注册，如果没有则注册：

text

1. 调用: query_agents(status='all') → 查看所有 Agent
2. 如果自己的 Agent ID 不在列表中
   → register_agent(invite_code, name, capabilities)
3. 如果已注册 → 记下自己的 agent_id 供后续使用

[检查点] 用户确认：注册新 Agent 需要 invite_code，先问用户是否有可用的邀请码。

Step 3: 维持在线状态#

启动心跳维持在线，确保能接收实时消息推送：

text

调用: heartbeat(agent_id='你的ID')
频率: 每 30 秒一次（超过 90 秒无心跳则自动标记为离线）

Step 4: 检查未读消息#

上线后第一时间检查是否有离线期间缓存的消息：

text

1. 调用: search_messages(query='你的ID', limit=20)
2. 筛选 status='unread' 的消息
3. 按时间顺序处理，先 acknowledge_message 确认收到，再回复

[检查点] 用户确认：找到未读消息后，逐条向用户摘要汇报，请用户确认如何处理。

Step 5: 发送第一条消息#

向另一个 Agent 发送消息，验证双向通信：

text

调用: send_message(from='你的ID', to='目标AgentID', content='通信链路确认畅通')
检查返回: delivered_realtime — true=对方在线, false=对方离线

[检查点] 用户确认：发送前向用户确认消息内容和目标 Agent。broadcast_message 必须逐条确认。

完整闭环示例#

text

场景：Hub 在线 → 检查 WorkBuddy 是否有未读消息 → 处理并回复

1. get_online_agents()                    # 确认自己和对方在线
2. search_messages(limit=10)              # 查最近消息
3. acknowledge_message(msg_id, agent_id)  # 标记已读
4. send_message(to='workbuddy', content='已收到，正在处理')  # 回复
5. mark_consumed(resource=msg_id, action='replied')  # 消费水位线

核心能力#

58 个 MCP 工具（当前版本）#

Identity 身份 (6)

工具功能register_agent注册新 Agent，需提供 HUB_AUTH_TOKEN 认证heartbeatAgent 心跳上报，维持在线状态，每 3 次连续心跳记录 +1query_agents查询 Agent 列表，支持状态/角色筛选get_online_agents获取当前在线 Agent 列表

Message 消息 (5)

工具功能send_messageAgent 间点对点消息，支持 Markdown，自动去重（sha256）broadcast_message(需逐条确认后发送)acknowledge_message确认已读消息，防止重复出现search_messages全文搜索消息历史batch_acknowledge_messages批量确认消息（1-500 条/次），用于清理消息积压

File 文件 (3)

工具功能upload_file发送文件附件（Base64，10MB 限制），关联到消息download_file接收附件，返回 Base64 编码内容list_attachments列出附件，支持按消息/Agent 筛选

Task 任务 (3)

工具功能assign_task创建并分配任务，支持上下文传递update_task_status更新任务状态（inbox→assigned→in_progress→completed/failed）get_task_status查询任务详情，含依赖、Pipeline、交接信息

Context 上下文暂存 (5)

工具功能store_memory临时暂存当前任务参考信息recall_memory检索已暂存的上下文list_memories列出当前 Agent 的暂存条目delete_memory删除暂存条目（仅 creator）search_memories检索当前 Agent 的暂存内容

经验记录

经验记录和策略管理需特定权限配置。

任务协同

工具功能add_dependency添加任务依赖关系（依赖检查）remove_dependency删除任务依赖关系get_task_dependencies查询任务上下游依赖create_parallel_group创建并行任务组（2-10 个任务）request_handoff请求任务交接accept_handoff接受任务交接reject_handoff拒绝任务交接（含理由）add_quality_gate在 Pipeline 中添加质量门evaluate_quality_gate评估质量门（passed/failed）recalculate_trust_scores按调度执行分数维护create_pipeline创建 Pipeline 流水线get_pipeline查询 Pipeline 详情list_pipelines列出 Pipelineadd_task_to_pipeline向 Pipeline 添加任务

运维工具 (4)

工具功能get_db_stats数据库统计信息（表行数、大小、Agent 数等）archive_data数据维护工具（其余 2 个内部工具）权限验证与控制（其余 2 个内部工具）权限验证与控制

消费水位线 (2)

工具功能mark_consumed标记任务/消息为已消费，防止重复处理check_consumed查询资源是否已被消费

所有工具内置 try-catch + 3 次指数退避重试（100ms → 200ms → 400ms）。v2.4.0 统一错误格式： HubError 错误码 + mcpError()/ mcpFail() 标准返回。 check_consumed 查询失败时降级返回 consumed=false（不阻塞业务）。

任务状态机#

text

inbox → assigned → [waiting] → in_progress → completed / failed / cancelled

用户确认检查点#

以下操作必须在执行前暂停，向用户摘要说明并等待确认：

#操作检查点说明风险1broadcast_message广播消息会发送给多个 Agent，逐条确认内容和接收列表高2assign_task分配任务前确认：描述是否清晰、目标 Agent 是否合适、Priority 正确中3batch_acknowledge_messages批量确认会一次性标记多条消息为已处理，确认不会遗漏重要信息中4create_pipeline创建流水线前确认任务顺序、质量门设置、参与 Agent中5add_quality_gate质量门失败会阻塞后续任务，确认评估标准合理高6request_handoff交接任务前确认目标 Agent 有能力接手、理由充分中7archive_data归档操作会移动数据到归档表，确认归档范围和天数高8store_memory(scope='group')写入组内共享记忆前确认内容适当，不会泄露敏感信息中9propose_strategy提议策略前确认内容准确、分类正确、有实际价值低10reject_handoff拒绝交接需提供理由，确认不会导致任务阻塞中

规则：LLM 遇到上表操作时，先向用户输出摘要说明，明确询问"是否继续？"，得到肯定答复后再执行。用户可随时跳过检查点。

数据隔离与安全边界#

边界实现方式接收方校验send_message/ assign_task 中的 to_agent 必须为已注册 Agent，未注册 Agent 被拒绝Per-Agent 数据隔离每个 Agent 仅可见自身消息、任务和暂存条目；跨 Agent 查询受 4 级权限控制暂存内容保护store_memory 创建的条目仅 creator 可检索和删除，不会自动暴露给其他 Agent经验记录审批share_experience 提交的记录需经 full 权限确认后才对其他 Agent 可见

接入配置（stdio 模式）#

在 MCP 配置文件中添加 Hub 为 stdio 服务器，提供 HUB_AUTH_TOKEN 环境变量进行认证。Hub 通过 stdio 传输 MCP 协议，Agent 的 LLM 可直接调用 Hub 工具。 stdio 模式必须设置 HUB_AUTH_TOKEN，缺失将拒绝启动。

json

{
  "mcpServers": {
    "agent-comm-hub": {
      "command": "node",
      "args": ["<hub-install-path>/stdio.js"],
      "env": {
        "HUB_AUTH_TOKEN": "your-connection-key"
      }
    }
  }
}

资源索引#

此 skill 目录下已有 Hub 完整源码，可直接参考：

本地源文件#

文件用途src/server.ts服务端入口，Express + MCP/SSE 双通道src/tools.ts全部 MCP 工具的 TypeScript 实现src/db.tsSQLite WAL 数据库初始化与连接src/identity.tsAgent 注册、认证、4 级权限控制src/dedup.tsSHA256 消息去重实现src/errors.tsHubError 统一错误码（v2.4.0+）src/stdio.tsStdio 模式传输层src/sse.tsSSE 推送通道src/orchestrator.ts任务编排、Pipeline、质量门src/evolution.tsEvolution Engine：策略/经验/信任分src/memory.ts上下文暂存与管理src/security.ts安全验证、CORS、Token 管理src/metrics.ts统计指标收集src/repo/数据访问层（repository pattern）package.jsonNode.js 依赖与版本定义

参考链接#

• GitHub 仓库: https://github.com/liuboacean/agent-comm-hub
• MCP 协议规范: https://spec.modelcontextprotocol.io

权限说明（4 级）#

级别说明可用工具范围authenticated已认证（HUB_AUTH_TOKEN）register_agent（初始注册）member已注册 Agent消息 send_message/ acknowledge_message + 任务 assign_task/ get_task_statusgroup_manager并行组管理任务协同 + Pipeline 工具（不含暂存/经验）full完整权限全部工具（含运维与建议管理）

部分管理类工具仅特定权限可调用，具体以实际角色配置为准。

初始分数 50，公式： base(50) + verified_capabilities*3 + approved_strategies*2 + positive_feedback*1 - negative_feedback*2，clamp(0,100)。

版本历史#

v3.0.24 — 宿主执行器闭环收口（HostExecutor 注入）#

类别内容说明宿主执行HostExecutor 契约 + 参考实现新增 client-sdk/adapters/host-executor.ts（ LlmHostExecutor / HttpHostExecutor / defaultHostExecutor()）； AbstractHostTaskBridge 可注入 executor闭环消灭 setTimeout 占位WorkBuddy / Hermes 桥 runTask() 委托 this.executor.execute()，任务到达即触发宿主真实能力

v3.0.25 — 跨进程 SSE 桥接 + watcher 实时解析修复#

类别内容说明推送跨进程 SSE 桥接（方案 B）pushToAgent 在当前进程无目标 SSE 连接且配置 SSE_BRIDGE_URL 时，转发事件至桥接进程 /internal/sse/push，打通 stdio MCP 派单进程 → HTTP SSE 服务进程的实时送达服务端桥接接收端点server.ts 新增 POST /internal/sse/push（鉴权保护），由持有 SSE 连接的进程本地投递客户端(watcher)SSE 实时解析修复hub_watcher.py 由 resp.read(4096) 缓冲读改为逐行 readline()，消除低吞吐 SSE 流缓冲阻塞；修正事件分发（ payload.event 取真实类型）与受保护 REST API 的 Bearer 鉴权头效果派单→自动执行闭环接收方持有 SSE 长连接（如 WorkBuddy 本地 watcher）即可实时收到 task_assigned 并触发宿主自动化执行

v3.0.x（安全加固）#

类别内容说明权限模型fail-closed 权限矩阵checkPermission 未注册工具默认拒绝； TOOL_PERMISSIONS 全量登记，杜绝 fail-open认证stdio 强制认证缺失 HUB_AUTH_TOKEN 直接 process.exit(1)，移除 glama-ci admin 兜底角色护栏admin 校验9 个 admin 类工具 handler 首行 requireAdmin(ctx)HTTP 中间件分级放行/health、 /metrics 仅内网/loopback 或 token 放行； /dashboard、 /api/* 需 token + admin审计WORM 不可篡改审计日志仅归档不删源，保留 no_delete / no_modify 触发器数据归属域隔离search_messages 强制本人收发域；记忆统计按 agent 隔离，admin 才指定他人对象级授权assertOwns 中间件message/attachment/task 三族工具插入 assertOwns() 归属校验（HUB_2004）； /health 收敛（删除内网 IP/路径泄露）sender 校验send_message 身份守卫broadcast_message + send_message 均强制 from === ctx.agentId，杜绝身份伪造memory 降级search_memories 补 agent_id 过滤FTS5 离线回退 SQL 增加 agent_id = ?，防止越权泄漏内部函数setAgentRole/updateAgentTrustScore 加 admin 校验底层函数不再信任 operatorId，显式查库验证 admin 身份SQL 修复registerCapability 占位符6→7 个占位符匹配实际 7 个字段值，修复运行时崩溃triggers 收窄SKILL.md 触发词移除 Hub/ 通信/ 消息 过宽泛触发词，改用具体标识

v2.4.0#

Phase内容变更Atools.ts 拆分2687 行 → 8 模块 + 30 行入口 + utils.tsB单元测试100 用例，role-control >= 70% / dedup branches>=60, functions>=70 / utils 100%CCI/CDGitHub Actions：typecheck + test + coverage 3 JobsD类型健壮any 归零 + HubError 统一错误码 + MCP 返回格式标准化

踩坑经验速查#

#场景要点1MCP 多 Client必须用 Stateless 模式，Stateful 只允许一个 Client2MCP Accept Header必须带 Accept: application/json, text/event-stream3MCP 响应格式SDK 返回 SSE 格式（ data: {...}），不是纯 JSON4ESM 兼容不能用 require()，用 import() 动态导入5UTF-8 块读取httpx resp.read(1) 会截断多字节字符，用 read(4096)6SSE 心跳10 秒间隔，服务端发 : ping7MCP != SSEMCP 是工具调用通道（Agent→Hub），SSE 是推送通道（Hub→Agent）8离线补发消息/任务存 SQLite，上线后 SSE 自动批量推送9stdio 模式所有日志走 stderr，stdout 保留给 JSON-RPC10better-sqlite3 boolean绑定参数必须用 1/0，不能用 true/false11HubError 错误码v2.4.0 统一用 mcpError()/mcpFail()，不要手动构造错误响应

安全配置#

配置项说明HUB_AUTH_TOKENstdio / REST 模式认证 Token，所有 Agent 接入必须提供，用于身份认证与消息完整性校验4 级权限模型authenticated → member → group_manager → full，逐级授权CORS 白名单默认拒绝跨域，通过 CORS_LIST 显式配置允许的来源

环境变量#

变量默认值说明HUB_AUTH_TOKEN—stdio / REST 认证 Token（必填）DB_PATH./comm_hub.dbSQLite 数据库路径LOG_LEVELinfo日志级别：debug / info / warn / errorCORS_LIST(空)CORS 白名单（逗号分隔），空=拒绝所有跨域

技术依赖#

Hub 服务器：

• Node.js 18+
• @modelcontextprotocol/sdk ^1.10.2（支持 StdioServerTransport）
• express ^4.19
• better-sqlite3 ^11.9
• zod ^3.23

Python 客户端（零外部依赖）：

• Python 3.9+（纯标准库：http.client / json / asyncio）