Agent/gateway

Gateway - Agent 消息路由服务

框架无关的 Agent 间即时通讯网关，提供 Agent 注册、消息路由（Webhook 推送）、在线状态管理。

概述

Gateway 是一个任务导向的 Agent 即时通讯系统，支持任何 Agent 框架使用。核心功能：

• Agent 注册和发现
• 消息路由（HTTP Webhook 推送）
• 在线状态管理（HTTP 心跳）
• 活跃协作者管理

独立性：Gateway 与 Agent Core 并列，可以独立部署。

设计原则

1. 框架无关：Gateway 核心不依赖任何特定的 Agent 框架
2. 纯 HTTP：所有交互均通过标准 HTTP API 完成，无 WebSocket 依赖
3. 易于集成：提供通用的 Python SDK 和 CLI 工具
4. 可扩展：支持不同框架的适配和集成

架构

gateway/
├── core/              # Gateway 核心服务（框架无关）
│   ├── registry.py    # Agent 注册和在线状态管理（含 webhook_url）
│   └── router.py      # HTTP API 端点和消息路由
│
├── client/            # 通用客户端工具
│   ├── python/        # Python SDK
│   │   ├── tools.py   # 通用工具函数
│   │   ├── client.py  # GatewayClient
│   │   └── cli.py     # CLI 工具
│   └── a2a_im.md      # Agent Skill 文档
│
├── docs/              # 详细文档
│   ├── architecture.md
│   ├── deployment.md
│   ├── api.md
│   ├── decisions.md
│   └── enterprise/
│
└── enterprise/        # 企业功能（可选）
    ├── auth.py
    ├── audit.py
    └── cost.py

examples/gateway_integration/  # 集成示例（项目根目录）

快速开始

1. 启动 Gateway 服务

python gateway_server.py

服务器将在 http://localhost:8001 启动。

2. 注册 Agent（含 Webhook 地址）

from gateway.client.python import GatewayClient, AgentCard

client = GatewayClient(
    gateway_url="http://localhost:8001",
    agent_card=AgentCard(
        agent_id="my-agent-001",
        agent_name="My Agent",
        capabilities=["search", "analyze"]
    ),
    webhook_url="http://my-agent.local:8080/webhook/a2a-messages"
)
await client.connect()  # 注册 + 启动心跳

3. 发送消息

await client.send_message(
    to_agent_id="target-agent",
    content=[{"type": "text", "text": "Hello"}],
    conversation_id="conv-123"
)

4. 接收消息（Webhook）

在 Agent 服务中添加 webhook 端点，Gateway 收到消息后会 POST 到此地址：

# api_server.py
@app.post("/webhook/a2a-messages")
async def receive_a2a_message(message: dict):
    message_queue.push(message)
    return {"status": "received"}

消息通过 Context Injection Hook 机制周期性提醒 LLM，由 LLM 调用 check_messages 工具查看详情。详见 消息通知机制。

5. 使用 CLI

gateway-cli send --from my-agent --to target-agent --message "Hello!"
gateway-cli list
gateway-cli status target-agent

详见 gateway/client/a2a_im.md（Agent Skill 文档）。

API 端点

方法	路径	说明
POST	/gateway/register	注册 Agent（含 webhook_url）
POST	/gateway/heartbeat	心跳（每 30 秒）
POST	/gateway/unregister	注销 Agent
POST	/gateway/send	发送消息
GET	/gateway/status/{agent_id}	查询 Agent 状态
GET	/gateway/agents	列出在线 Agent
GET	/gateway/health	Gateway 状态

详见 API 文档。

文档

• 架构设计：通信流程、消息通知机制、设计决策
• API 文档：完整的 API 参考
• 部署指南：部署方式和配置
• 设计决策：架构决策记录
• Enterprise 层：认证、审计、多租户
• A2A IM Skill：Agent Skill 使用文档

文档维护规范

1. 先改文档，再动代码
2. 文档分层，链接代码 - 格式：module/file.py:function_name
3. 简洁快照，日志分离 - 决策依据在 docs/decisions.md 记录

相关项目

• Agent Core：Agent 核心框架

概述

Gateway 是一个任务导向的 Agent 即时通讯系统，支持任何 Agent 框架使用。核心功能：

• Agent 注册和发现
• 消息路由（WebSocket/HTTP）
• 在线状态管理
• 活跃协作者管理
• 联系人管理

独立性：Gateway 与 Agent Core 并列，可以独立部署。

依赖关系：Gateway 依赖 Agent Core（使用 ToolContext、TraceStore 等组件）。

设计原则

1. 框架无关：Gateway 核心不依赖任何特定的 Agent 框架
2. 通用协议：使用 MAMP（Minimal Agent Message Protocol）
3. 易于集成：提供通用的 Python SDK 和 CLI 工具
4. 可扩展：支持不同框架的适配和集成

架构

gateway/
├── core/              # Gateway 核心服务（框架无关）
│   ├── registry.py    # Agent 注册和在线状态管理
│   ├── router.py      # 消息路由
│   └── client.py      # Gateway 客户端
│
├── client/            # 通用客户端工具
│   ├── python/        # Python SDK
│   │   ├── tools.py   # 通用工具函数
│   │   ├── client.py  # GatewayClient
│   │   └── cli.py     # CLI 工具
│   └── a2a_im.md      # Agent Skill 文档
│
├── docs/              # 详细文档
│   ├── architecture.md
│   ├── deployment.md
│   ├── api.md
│   ├── decisions.md
│   └── enterprise/
│
└── enterprise/        # 企业功能（可选）
    ├── auth.py        # 认证和授权
    ├── audit.py       # 审计和监控
    └── cost.py        # 成本管理

examples/gateway_integration/  # 集成示例（项目根目录）
├── README.md
├── python_client_example.py      # 通用 SDK 使用示例
├── agent_tools_example.py        # Agent 框架集成示例
└── agent_skill_example.md        # Agent 框架 Skill 示例

快速开始

1. 启动 Gateway 服务

python gateway_server.py

服务器将在 http://localhost:8001 启动（默认端口）。

2. 使用 CLI 工具

# 发送消息
gateway-cli send --from my-agent --to target-agent --message "Hello!"

# 查询在线 Agent
gateway-cli list

# 查询 Agent 状态
gateway-cli status target-agent

详见 gateway/client/a2a_im.md（Agent Skill 文档）。

3. 使用 Python SDK

from gateway.client.python import tools

# 发送消息
result = await tools.send_message(
    gateway_url="http://localhost:8001",
    from_agent_id="my-agent",
    to_agent_id="target-agent",
    message="Hello!"
)

# 查询在线 Agent
agents = await tools.list_online_agents(
    gateway_url="http://localhost:8001"
)

# 查询 Agent 状态
status = await tools.get_agent_status(
    gateway_url="http://localhost:8001",
    agent_id="target-agent"
)

4. 集成到 Agent 框架

参考 examples/gateway_integration/ 中的示例：

# 在 agent 启动时注册 Gateway 工具
from examples.gateway_integration.agent_tools_example import (
    send_to_agent,
    get_active_collaborators,
    get_online_agents,
    get_agent_status
)

# 注册工具
tool_registry.register(send_to_agent)
tool_registry.register(get_active_collaborators)
tool_registry.register(get_online_agents)
tool_registry.register(get_agent_status)

# 加载 skill
skill_loader.load("gateway/client/a2a_im.md")

核心功能

Agent 注册

PC Agent 启动时通过 WebSocket 连接到 Gateway 并注册。

实现位置：gateway/core/registry.py:AgentRegistry

from gateway.client.python import GatewayClient, AgentCard

# 创建客户端并注册
async with GatewayClient(
    gateway_url="http://localhost:8001",
    agent_card=AgentCard(
        agent_id="my-agent-001",
        agent_name="My Agent",
        capabilities=["search", "analyze"]
    )
) as client:
    # 发送消息
    await client.send_message(
        to_agent_id="target-agent",
        content=[{"type": "text", "text": "Hello"}],
        conversation_id="conv-123"
    )

消息路由

Gateway 根据目标 Agent ID 路由消息。

实现位置：gateway/core/router.py:GatewayRouter

# 发送消息
POST /gateway/send
{
    "from_agent_id": "my-agent",
    "to_agent_id": "target-agent",
    "content": [{"type": "text", "text": "帮我分析代码"}]
}

在线状态管理

通过心跳机制跟踪 Agent 在线状态。

实现位置：gateway/core/registry.py:AgentRegistry.heartbeat

# 查询在线状态
GET /gateway/status/{agent_id}

API 端点

方法	路径	说明
WS	/gateway/connect	Agent 注册和保持连接
POST	/gateway/send	发送消息到其他 Agent
GET	/gateway/status/{agent_id}	查询 Agent 在线状态
GET	/gateway/agents	列出所有在线 Agent

实现位置：gateway/core/router.py:GatewayRouter

详见 API 文档。

部署方式

方式 1：单体部署

# 一个进程运行 Agent + Gateway
python main.py

方式 2：分离部署

# 两个独立进程
python api_server.py      # Agent Core
python gateway_server.py  # Gateway

方式 3：分层部署

# 三个独立进程
python api_server.py              # Agent Core
python gateway_server.py          # Gateway Core
python enterprise_gateway.py      # Enterprise Gateway

详见 部署指南。

为其他框架创建集成

参考 examples/gateway_integration/ 中的示例创建你自己的集成：

1. 使用 gateway/client/python/ 的通用 SDK 或 CLI 工具
2. 创建适配你框架的工具/函数
3. 根据需要适配框架特有的功能

详见 集成示例文档。

文档

模块文档（gateway/docs/）

• 架构设计：Gateway 架构和设计决策
• 部署指南：部署方式和配置
• API 文档：完整的 API 参考
• 设计决策：架构决策记录
• Enterprise 层：组织级功能（认证、审计、多租户）

项目级文档（../docs/）

• A2A IM 系统：完整的 A2A IM 文档
• MAMP 协议：消息格式和传输协议

客户端文档

• A2A IM Skill：Agent Skill 使用文档
• Client README：客户端安装和配置

文档维护规范

1. 先改文档，再动代码 - 新功能或重大修改需先完成文档更新、并完成审阅后，再进行代码实现；除非改动较小、不被文档涵盖
2. 文档分层，链接代码 - 重要或复杂设计可以另有详细文档；关键实现需标注代码文件路径；格式：module/file.py:function_name
3. 简洁快照，日志分离 - 只记录最重要的、与代码准确对应的或者明确的已完成的设计的信息，避免推测、建议，或大量代码；决策依据或修改日志若有必要，可在docs/decisions.md另行记录

是否应该拆分到独立 Git 项目？

当前建议：暂时不拆分

理由：

• ✅ Gateway 和 Agent 框架还在快速迭代
• ✅ 保持在 monorepo 便于协同开发
• ✅ 避免版本兼容性管理的复杂度

未来拆分时机（满足以下条件时考虑）：

• Gateway API 稳定（v1.0）
• 有其他团队/项目开始使用 Gateway
• Gateway 和 Agent 的发布周期明显不同

现在的准备：

• ✅ Gateway 核心完全独立（不依赖 agent 模块）
• ✅ 通用的 Gateway 客户端 SDK 和 CLI 工具
• ✅ 集成示例在 examples/ 目录
• ✅ 独立的文档结构
• ⚠️ 独立的测试（待完善）

当需要拆分时，只需：

1. 将 gateway/ 目录移到新仓库
2. 在 Agent 项目中通过 pip 安装 Gateway SDK
3. 保持 examples/gateway_integration/ 在 Agent 项目中作为集成示例

相关项目

• Agent Core：Agent 核心框架

License

MIT