Apache EventMesh A2A 插件 v2.0：MCP over CloudEvents 的映射、双模式与 10105 Gateway

项目信息

• GitHub：https://github.com/apache/eventmesh
• 官网：https://eventmesh.apache.org/
• A2A 文档：https://github.com/apache/eventmesh/blob/develop/docs/a2a-protocol/README_EN.md

为什么把 MCP 工具调用搬进事件总线

MCP / JSON-RPC 2.0 的工具调用原本是同步的：客户端发 tools/call，服务端 Webhook 回调。点对点回调在 agent 数量上来后不好扩——调用方要维护每个 agent 的地址，回调链路要自己处理重试、限流和反压。EventMesh A2A v2.0 的做法是把同步 RPC 映射成异步请求/响应事件流，走 CloudEvents 信封，底层传输可以是 HTTP、TCP、gRPC 或 Kafka。

改成事件总线后有几个直接收益：

• O(1) 广播：发布者往 Topic 发一次，EventMesh 扇出到所有订阅者。
• 解耦：发布者不需要知道消费者 agent 的地址。
• 背压隔离：发布者和订阅者之间由 broker 隔开，慢消费者不会把快发布者拖死。

MCP → CloudEvent 映射

A2A 的异步 RPC 模式用一张映射表把 MCP 概念落到 CloudEvent：

路由提示从 MCP 参数里抽：_agentId 用于点对点，_topic 用于发布订阅。抽出来后注入 CloudEvents attributes，做 zero-decoding routing——broker 侧不用解 payload 就能决定往哪投。

双模式：MCP 翻译还是原生 CloudEvent

EnhancedA2AProtocolAdaptor 负责判定：

• 如果 payload 里带 jsonrpc: "2.0"，走 MCP 翻译引擎，把 JSON-RPC 对象包成 CloudEvent。目标是 LLM、Python/JS 脚本、LangChain 这类调用方。
• 没有 jsonrpc 字段，就按标准 CloudEvent 处理。目标是 EventMesh 原生应用、Knative、Serverless 函数，保留完整事件元数据和自定义/二进制数据透传。

同一套 adaptor 同时吃两种输入，客户端不用改协议。

报文示例

P2P 调用，_agentId 指到 weather-service：

{
"jsonrpc": "2.0",
"method": "tools/call",
"params": { "name": "get_weather", "_agentId": "weather-service" },
"id": "req-123"
}

发布订阅，_topic 指到 market.btc：

{
"jsonrpc": "2.0",
"method": "market/update",
"params": { "price": 50000, "_topic": "market.btc" }
}

批量 JSON-RPC 请求也支持，会拆成并行事件流；流式操作（message/sendStream）用 sequence ID 保序。

A2A Gateway：REST / SSE

Gateway 跑在 10105 端口，对外暴露这组 REST 端点：

# 同步任务
curl -X POST 'http://localhost:10105/a2a/tasks?mode=sync' -H 'Content-Type: application/json' -d '{"targetAgent":"weather-agent","message":"Beijing"}'
# 异步任务
curl -X POST 'http://localhost:10105/a2a/tasks?mode=async' -H 'Content-Type: application/json' -d '{"targetAgent":"weather-agent","message":"Shanghai"}'
# 查询状态
curl http://localhost:10105/a2a/tasks/{taskId}
# SSE 流
curl -N http://localhost:10105/a2a/tasks/{taskId}/stream
# Agent 列表
curl http://localhost:10105/a2a/agents

端点对应：POST /a2a/tasks?mode=sync|async 提交任务，GET /a2a/tasks/{taskId} 查状态，GET /a2a/tasks/{taskId}/stream 是 SSE，GET /a2a/agents 列 agent。

Gateway 侧工程决策（commit #5260）：任务超时后自动失败非终态任务，阈值可配；SSE 加 heartbeat，长连接用 socketTimeout(0)；并发容器用 CopyOnWriteArrayList + ConcurrentHashMap；所有 A2A 端点带 CORS Access-Control-Allow-Origin；Agent 校验会拒绝未注册 agent 的任务。

TaskRegistry 是内存任务状态机，TTL 5 分钟自动清理。测试覆盖 A2AGatewayServiceTest 15+ 用例（超时、agent 校验、取消、SSE、分页）、TaskRegistryTest（TTL 与生命周期）、A2AClientServerIntegrationTest、A2AGatewayEndToEndTest。

模块与存储

模块划分：

• protocol-a2a：A2AClient、A2AMessageTransport、A2ATopicFactory。
• runtime：A2AGatewayServer、A2AGatewayHttpHandler、A2AGatewayService、TaskRegistry、A2APublishSubscribeService、InMemoryA2AMessageTransport。

默认 transport 是 InMemoryA2AMessageTransport，可以换成 EventMesh broker：RocketMQ、Kafka、Pulsar、Redis。Pub/Sub 侧走 A2APublishSubscribeService，基于 EventMeshProducer/Consumer，topic 约定为 a2a.tasks.*、a2a.results、a2a.status。AgentRegistry 做基于能力的发现和心跳监控，CollaborationManager 做多 agent 工作流编排、会话管理和失败重试/恢复。任务生命周期是 Request → Message → Processing → Result，支持指数退避重试、超时/取消、correlation ID 和优先级。

Java SDK 侧提交与订阅：

A2ATaskRequest taskRequest = A2ATaskRequest.builder()
.taskType("data-processing")
.payload(Map.of("data", "user-behavior"))
.requiredCapabilities(List.of("data-processing"))
.priority(A2ATaskPriority.HIGH)
.build();
pubSubService.publishTask(taskRequest);
pubSubService.subscribeToTaskType("agent-001", "data-processing",
List.of("data-processing", "analytics"), taskHandler);

MCP 模式也可以用 Java 发：

String mcpRequest = "{" +
"\"jsonrpc\": \"2.0\"," +
"\"method\": \"tools/call\"," +
"\"params\": { \"name\": \"weather\", \"_agentId\": \"weather-agent\" }," +
"\"id\": \"req-001\"" +
"}";
eventMeshProducer.publish(new A2AProtocolTransportObject(mcpRequest));

服务端订阅 org.apache.eventmesh.a2a.tools.call.req，处理后用匹配的 id 回响应。

边界与坑

• 实现仍在推进：issue #5202 “[Feature] implement A2A protocol” 由 qqeasonchen 于 2025-08-14 打开，已关闭，标签 feature、Stale；commit #5260、#5214 是分阶段落地。
• 默认 transport 是 InMemoryA2AMessageTransport，生产要换成 EventMesh broker。
• Java 编译口径有 8/21 两条线，选版本时注意。
• 文档口径为 v2.0，实际行为需自行验证。

EventMesh 本身的架构是 CloudEvents-over-MQ：MQ 当纯 write-ahead log，没有 consumer group、没有 tag；无状态 Runtime 通过 SubscriptionManager 做投递逻辑（load-balance/broadcast/multicast），HTTP + CloudEvents 1.0 SDK 提供 publish/subscribe/unsubscribe，连接器走 SPI。存储插件有 RocketMQ、Kafka、Pulsar、RabbitMQ、Redis；meta service 支持 Consul、Nacos、ETCD、ZooKeeper。A2A 插件就是在这一层上面把同步 MCP 工具调用接到异步事件流。