编程 MCP Mesh 的 A2A 桥接：协议只给信封，注册、故障转移和 DDDI 由 mesh 补

2026-09-24 00:04:41

MCP Mesh 的 A2A 桥接：协议只给信封，注册、故障转移和 DDDI 由 mesh 补

项目信息：MCP Mesh — GitHub｜官网｜A2A 文档。License MIT，语言 Python，创建 2025-06-02，约 41 stars。Topics：a2a, a2a-protocol, agentic-ai, ai-agents, distributed-ai, java, kubernetes, mcp, modelcontextprotocol, python, typescript。

1. A2A v1.0 本身只定义信封

A2A v1.0 spec 定义的是 HTTP + JSON-RPC envelope，用于跨 vendor 的 agent 调用。它停在协议层：能力发现按 card，没有 resolver，没有 failover，没有 DDDI。A2A v1.0 协议语义，包括 wire format、JSON-RPC envelopes、task lifecycle states，见 A2A v1.0 spec 和 JSON-RPC 2.0。

MCP Mesh 在 wire 两侧实现 A2A v1.0：mesh agent 可以 outbound 调外部 A2A endpoint（consumer 侧），也可以把自己的 mesh tool 暴露成 A2A skill（producer 侧）。mesh agent 之间的共享传输仍是 MCP；A2A 是桥到 mesh 外说同一协议的端点。

2. MCP Mesh 在 A2A 外套的层

注册发现：producer 入口根据声明的 metadata 自动构建 agent card，并把 /.well-known/agent.json 和 JSON-RPC entrypoint 挂到用户自己的 hosting framework 上：Python FastAPI 用 mesh.a2a.mount(app, ...)，Java Spring Boot bean 用 @MeshA2A，TypeScript Express 用 mesh.a2a.mount(app, ...)。支持 sync、long-running（task=True/task=true）和 SSE handler。
Consumer：一个 mesh capability，其 body 对别家 A2A 后端发 outbound tasks/send / tasks/sendSubscribe。Bridge 把上游 skill 重新发布成普通 mesh capability，下游 caller 不需要知道背后是 A2A。
Tag + 能力故障转移：两个 consumer 把同一逻辑 capability bridge 到不同 vendor（weather.com vs accuweather.com）时，会用 agent name 自动打 tag。下游可以按 tag pin 某个 provider，也可以让 resolver 挑一个健康的。
健康驱动重连：consumer 死掉时，registry 的 orphan-reset 会把新调用透明路由到 peer consumer，通常秒级完成。
DDDI：A2A consumer 是一等 @mesh.tool capability，其他 tool 通过同一套依赖注入管线消费它。
长任务：task=True consumer 把 A2A tasks/get polling（或 tasks/sendSubscribe SSE）镜像进 JobController，外部长任务在调用方看来是标准 MeshJob。

数据流：

External A2A backend (e.g. weather.com) --> tasks/send --> Consumer mesh agent (the bridge) --> register capability --> Mesh Registry --> resolve dependency --> Downstream caller (any mesh agent) --> tools/call --> Bridge --> outbound A2A --> External.

3. Producer / Consumer 对照与接入

Aspect	Producer	Consumer
Direction	Mesh tool exposed AS A2A	External A2A skill bridged INTO mesh
Decorator/marker	`mesh.a2a.mount(app, ...)` (Py/TS) / `@MeshA2A` (Java)	`@mesh.a2a_consumer` (Py) / `@A2AConsumer` (Java) / `a2aConfig` (TS)
Runtime support	Python, Java, TypeScript	Python, Java, TypeScript
Card / discovery	Auto-generated at `/.well-known/agent.json`	Card fetched once at scaffold time (or `--offline`)
Long-running	Return `JobProxy`; framework parks the task	`task=True` body submits + `bridge(JobController)`
SSE	`tasks/sendSubscribe` handler returns `JobProxy`	`A2AClient.subscribe(...)` + `stream.bridge(JobController)`
Auth	Bearer enforcement on the JSON-RPC route	`A2ABearer` / `authBearerEnv` / `tokenEnv`

一个 consumer 单行示例：把外部 A2A 的 get-date skill 重新发布成普通 mesh current-date capability。

import json
import mesh

@app.tool()
@mesh.a2a_consumer(
    capability="current-date",
    a2a_url="http://upstream.example.com/agents/date",
    a2a_skill_id="get-date",
)
async def current_date(_a2a: mesh.A2AClient = None) -> dict:
    response = await _a2a.send(
        message={"role": "user", "parts": [{"type": "text", "text": "now"}]},
    )
    return json.loads(response.artifact_text)

下游 mesh tool 之后像依赖任何普通 capability 一样依赖 current-date，不需要感知工作是在 A2A 上完成的。

Issue #972：每个 agent payload 还带两个自声明布尔值，a2a_producer 和 a2a_consumer。只要注册了 producer surface 或 consumer binding，SDK 就会把它们置为 true。它们会出现在 AgentInfo 上供下游工具使用。两者默认 false；registry 把 key 当 optional，旧 SDK 客户端仍能工作。

4. DDDI 与 @mesh.tool 依赖注入

DDDI 是 Distributed Dynamic Dependency Injection：

Distributed：依赖跨机器、云、Python/TypeScript/Java runtime。
Dynamic：服务在运行时发现并注入，不是编译期。
Hot-swappable：依赖通过 heartbeat-driven re-resolution 更新，不用重启。
Pull-based discovery with runtime function injection；Smart resolution 支持 version constraints、capability matching、tag scoring；LLM 也可以作为依赖。

Python 示例里，四个分布式调用按本地函数方式组合。每个依赖可以是普通 tool，也可以是 LLM agent，代码看不出区别。

from fastmcp import FastMCP
import mesh

app = FastMCP("TripPlanner")

@app.tool()
@mesh.tool(
    capability="plan_trip",
    dependencies=[
        {"capability": "weather", "tags": ["+claude"]},
        {"capability": "hotels",  "tags": ["+gpt"]},
        {"capability": "flights"},
        {"capability": "budget",  "tags": ["+claude"]},
    ],
)
async def plan_trip(
    destination: str,
    dates: str,
    weather: mesh.McpMeshTool = None,
    hotels:  mesh.McpMeshTool = None,
    flights: mesh.McpMeshTool = None,
    budget:  mesh.McpMeshTool = None,
) -> TripPlan:
    forecast = await weather(destination=destination, dates=dates)
    options  = await hotels(destination=destination, dates=dates)
    routes   = await flights(destination=destination, dates=dates)
    cost     = await budget(routes=routes, options=options)
    return TripPlan(forecast, options, routes, cost)

@mesh.agent(name="trip-planner", auto_run=True)
class TripAgent: pass

weather 可以是 REST API，也可以是由 Claude 驱动的 reasoning agent，返回 typed pydantic forecast。+claude 表示优先 reasoning agent；它挂掉时 mesh 自动改接到 API；Claude 恢复后再切回来。不需要 deploy、config 或 code change。路由留在 Python 里，不写在 YAML。

Claude-powered weather agent 示例：

from fastmcp import FastMCP
import mesh

app = FastMCP("ClaudeWeather")

@app.tool()
@mesh.llm(
    system_prompt="file://prompts/weather.j2",
    provider={"capability": "llm", "tags": ["+claude"]},
)
@mesh.tool(capability="weather", tags=["+claude"])
async def weather(destination: str, dates: str,
                  llm: mesh.MeshLlmAgent = None) -> Forecast:
    return await llm(f"Forecast for {destination} on {dates}")

@mesh.agent(name="claude-weather", auto_run=True)
class Agent: pass

5. meshctl CLI 与 K8s/Helm 部署

安装与 CLI：

npm install -g @mcpmesh/cli
meshctl --help
meshctl man

Requirements：Python 3.11+、Java 17+、TypeScript 5.0+、Go 1.23+、Rust stable。

Ops 侧：面向分布式网络的 near-complete MCP protocol support；增强 proxy system 用 kwargs 自动配置 timeouts/retries/streaming；提供 meshctl CLI；Kubernetes native，带 Helm charts、scaling、health checks。

Resilience：Registry 作为 facilitator，agent 之间直接通信并带 fault tolerance；self-healing；graceful degradation；background orchestration。

Observability：Grafana dashboards、Tempo tracing、Redis session management、OTLP export、cross-agent context propagation、Redis-backed stickiness across pod replicas。

6. 能力边界与坑

认证只覆盖 bearer。OAuth/mTLS 是 future work，Phase 1 仅发 bearer。
每个 card 一个 skill：当前每个 mesh.a2a.mount(...) / @MeshA2A surface 只发一个 skill。单 card 下多 skill 分组属于 v2 scope。
a2a_producer / a2a_consumer 是自声明布尔值，默认 false；registry 把它们当 optional。
A2A v1.0 protocol semantics 不在范围内：wire format、JSON-RPC envelopes、task lifecycle states 见 A2A v1.0 spec 和 JSON-RPC 2.0。
仓库创建于 2025-06-02，约 41 stars，很新；用于生产前需自行验证。

7. 与其他 agent mesh 方案对照

Solace Agent Mesh：https://solace.com/cn/products/agent-mesh/。开源 agent mesh，围绕 event broker 构建，异步编排 A2A-compliant agents；event mesh 负责异步 inter-agent communication；原生 Kubernetes deployment。
Apache EventMesh A2A protocol：https://github.com/apache/eventmesh/blob/develop/docs/a2a-protocol/README_EN.md。v2.0 采用 MCP 架构，把 EventMesh 变成 Agent Collaboration Bus。MCP over CloudEvents 把同步 JSON-RPC tools/call 和 resources/read 映射成异步 request/response event streams，封装在 CloudEvents envelopes 里，跑在 HTTP/TCP/gRPC/Kafka 上。双模式：JSON-RPC 2.0 (MCP mode) 和 native CloudEvents (power mode)；EnhancedA2AProtocolAdaptor 检测 jsonrpc: "2.0" 后启用 MCP translation engine。Routing hints：_agentId 用于 P2P，_topic 用于 pub/sub，注入 CloudEvents attributes 做 zero-decoding routing。