EventMesh A2A Protocol Architecture & Functional Specification

1. Overview

The EventMesh A2A (Agent-to-Agent) Protocol is a specialized, high-performance protocol plugin designed to enable asynchronous communication, collaboration, and task coordination between autonomous agents.

With the release of v2.0, A2A adopts the MCP (Model Context Protocol) architecture, transforming EventMesh into a robust Agent Collaboration Bus. It bridges the gap between synchronous LLM-based tool calls (JSON-RPC 2.0) and asynchronous Event-Driven Architectures (EDA), enabling scalable, distributed, and decoupled agent systems.

2. Core Philosophy

The architecture adheres to the principles outlined in the broader agent community (e.g., A2A Project, FIPA-ACL, and CloudEvents):

  1. JSON-RPC 2.0 as Lingua Franca: Uses standard JSON-RPC for payload semantics, ensuring compatibility with modern LLM ecosystems (LangChain, AutoGen).
  2. Transport Agnostic: Encapsulates all messages within CloudEvents, allowing transport over any EventMesh-supported protocol (HTTP, TCP, gRPC, Kafka).
  3. Async by Default: Maps synchronous Request/Response patterns to asynchronous Event streams using correlation IDs.
  4. Native Pub/Sub Semantics: Supports O(1) broadcast complexity, temporal decoupling (Late Join), and backpressure isolation, solving the scalability limits of traditional P2P webhook callbacks.

2.1 Native Pub/Sub Semantics

Traditional A2A implementations often rely on HTTP Webhooks (POST /inbox) for asynchronous callbacks. While functional, this Point-to-Point (P2P) model suffers from significant scaling issues:

  • Insufficient Fan-Out: A publisher must send $N$ requests to reach $N$ subscribers, leading to $O(N)$ complexity.
  • Temporal Coupling: Consumers must be online at the exact moment of publication.
  • Backpressure Propagation: A slow subscriber can block the publisher.

EventMesh A2A solves this by introducing Native Pub/Sub capabilities:

graph LR
    Publisher[Publisher Agent] -->|1. Publish (Once)| Bus[EventMesh Bus]
    
    subgraph Fanout_Layer [EventMesh Fanout Layer]
        Queue[Topic Queue]
    end
    
    Bus --> Queue
    
    Queue -->|Push| Sub1[Subscriber 1]
    Queue -->|Push| Sub2[Subscriber 2]
    Queue -->|Push| Sub3[Subscriber 3]
    
    style Bus fill:#f9f,stroke:#333
    style Fanout_Layer fill:#ccf,stroke:#333

3. Architecture Design

3.1 System Context

graph TD
    Client[Client Agent / LLM] -- "JSON-RPC Request" --> EM[EventMesh Runtime]
    EM -- "CloudEvent (Request)" --> Server[Server Agent / Tool]
    Server -- "CloudEvent (Response)" --> EM
    EM -- "JSON-RPC Response" --> Client
    
    subgraph Runtime [EventMesh Runtime]
        Plugin[A2A Protocol Plugin]
    end
    
    style EM fill:#f9f,stroke:#333,stroke-width:4px
    style Plugin fill:#ccf,stroke:#333,stroke-width:2px

3.2 Component Design (eventmesh-protocol-a2a)

The core protocol logic resides in the eventmesh-protocol-plugin module.

  • EnhancedA2AProtocolAdaptor: The central brain of the protocol.
    • Intelligent Parsing: Automatically detects message format (MCP vs. Raw CloudEvent).
    • Protocol Delegation: Delegates to CloudEvents or HTTP adaptors when necessary.
    • Semantic Mapping: Transforms JSON-RPC methods and IDs into CloudEvent attributes.
  • A2AProtocolConstants: Defines standard operations like task/get, message/sendStream.
  • JsonRpc* Models: Strictly typed POJOs for JSON-RPC 2.0 compliance.
  • AgentCard / AgentSkill / AgentInterface: Agent capability discovery models.
  • A2ATopicFactory: Topic naming and parsing utility for request/response/status topics.
  • A2AClient: Java SDK for agent developers — AgentCard registration, task submission (sync/async), task status query, heartbeat, and transport-based request handling. Returns typed TaskResult objects.
  • A2AMessageTransport: Transport-agnostic pub/sub interface (InMemory implementation for dev/testing).

3.3 Gateway Runtime Architecture (eventmesh-runtime)

The Gateway runtime provides a standalone Netty HTTP server bridging external clients to the A2A event bus.

graph TD
    Client["Client / A2AClient SDK"] -- "HTTP REST" --> Server["A2AGatewayServer<br/>(Netty HTTP)"]
    Server --> Handler["A2AGatewayHttpHandler"]
    Handler --> GwService["A2AGatewayService"]
    GwService --> Registry["TaskRegistry<br/>(state machine + TTL)"]
    GwService --> Transport["InMemoryA2AMessageTransport"]
    GwService --> PubSub["A2APublishSubscribeService<br/>(AgentCard discovery)"]
    Transport -- "publish/subscribe" --> Agent["Target Agent"]
    Agent -- "response event" --> Transport
    Transport --> GwService

    style Server fill:#f9f,stroke:#333,stroke-width:2px
    style Registry fill:#cfc,stroke:#333
    style Transport fill:#ccf,stroke:#333

Core Components

ComponentModuleResponsibility
A2AGatewayServerruntimeStandalone Netty HTTP server entry point. Pre-registers mock agents, wires all components.
A2AGatewayHttpHandlerruntimeHTTP request router. Maps REST endpoints to service calls. Supports SSE streaming.
A2AGatewayServiceruntimeCore orchestration: task submission, response handling, status subscription, SSE push.
TaskRegistryruntimeIn-memory task lifecycle state machine with TTL auto-cleanup.
A2APublishSubscribeServiceruntimeAgentCard registration, discovery, and heartbeat management.
InMemoryA2AMessageTransportruntimeIn-memory pub/sub (replaceable by EventMesh broker).
A2ACardHttpHandlerruntimeAgentCard CRUD REST endpoints (/a2a/cards/*).
A2AClientprotocol-a2aJava SDK for agent developers (HTTP + transport).

Task Lifecycle State Machine

SUBMITTED → WORKING → COMPLETED
                    ↘ FAILED
                    ↘ CANCELLED
  • TTL Auto-Cleanup: Terminal-state tasks are automatically removed after a configurable TTL (default: 5 minutes). A daemon thread runs cleanup every 60 seconds.
  • Race Condition Prevention: pendingTasks.put(taskId, future) is called before transport.publish() to ensure the future is registered before any synchronous delivery could trigger handleResponse().

REST API

MethodPathDescription
POST/a2a/tasks?mode=syncSubmit task synchronously (wait for result)
POST/a2a/tasks?mode=asyncSubmit task asynchronously (return taskId immediately)
GET/a2a/tasks/{taskId}Get task status and result
DELETE/a2a/tasks/{taskId}Cancel a task
GET/a2a/tasks/{taskId}/waitLong-poll wait for task result
GET/a2a/tasks/{taskId}/streamSSE stream of task status updates
GET/a2a/agentsList all registered agents
POST/a2a/heartbeatAgent heartbeat
GET/a2a/cards/listList all AgentCards
POST/a2a/cards/card/{org}/{unit}/{agent}Register an AgentCard

3.4 Asynchronous RPC Mapping ( The “Async Bridge” )

To support MCP on an Event Bus, synchronous RPC concepts are mapped to asynchronous events:

ConceptMCP / JSON-RPCCloudEvent Mapping
Actionmethod (e.g., tools/call)Type: org.apache.eventmesh.a2a.tools.call.req
Extension: a2amethod
Correlationid (e.g., req-123)Extension: collaborationid (on Response)
ID: Preserved on Request
DirectionImplicit (Request vs Result)Extension: mcptype (request or response)
P2P Routingparams._agentIdExtension: targetagent
Pub/Sub Topicparams._topicSubject: The topic value (e.g. market.btc)
Streaming Seqparams._seqExtension: seq

4. Functional Specification

4.1 Message Processing Flow

  1. Ingestion: The adaptor receives a ProtocolTransportObject (byte array/string).
  2. Detection: Checks for jsonrpc: "2.0".
  3. Transformation (MCP Mode):
    • Request: Parses method.
      • If message/sendStream, sets type suffix to .stream and extracts _seq.
      • If _topic present, sets subject (Pub/Sub).
      • If _agentId present, sets targetagent (P2P).
    • Response: Parses result/error. Sets collaborationid = id.
  4. Batch Processing: Splits JSON Array into a List<CloudEvent>.

4.2 Key Features

A. Intelligent Routing Support

  • Mechanism: Promotes _agentId or _topic from JSON body to CloudEvent attributes.
  • Benefit: Enables EventMesh Router to perform content-based routing (CBR) efficiently.

B. Batching

  • Benefit: Significantly increases throughput for high-frequency interactions.

C. Streaming Support

  • Operation: message/sendStream
  • Mechanism: Maps to .stream event type and preserves sequence order via seq extension attribute.

D. SSE Task Streaming (Gateway)

  • Endpoint: GET /a2a/tasks/{taskId}/stream
  • Mechanism: Server-Sent Events (text/event-stream) pushes real-time task state transitions.
  • Flow: Initial state → WORKING updates → terminal state → connection close.
  • Implementation: Handler writes DefaultHttpContent chunks directly to the Netty channel, returning null to skip the standard FullHttpResponse path.

E. Task TTL Auto-Cleanup (Gateway)

  • Terminal-state tasks are automatically removed by a daemon scheduler after a configurable TTL (default: 5 minutes, cleanup interval: 60 seconds), preventing memory leaks.

F. AgentCard Discovery & Heartbeat (Gateway)

  • AgentCards expire after 60 seconds without heartbeat. POST /a2a/heartbeat refreshes the last-seen timestamp.

5. Usage Examples

5.1 Sending a Tool Call (Request)

Raw Payload:

{
  "jsonrpc": "2.0",
  "method": "tools/call",
  "params": {
    "name": "weather_service",
    "arguments": { "city": "New York" }
  },
  "id": "msg-101"
}

5.2 Pub/Sub Broadcast

Raw Payload:

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

Generated CloudEvent:

  • subject: market.crypto.btc
  • targetagent: (Empty)

5.3 Gateway REST API (HTTP)

The A2A Gateway provides a REST API for external clients and non-Java agents.

5.3.1 Submit Task (Sync)

curl -X POST 'http://localhost:10105/a2a/tasks?mode=sync' \
  -H 'Content-Type: application/json' \
  -d '{"targetAgent":"weather-agent","message":"Beijing"}'

Response:

{
  "taskId": "task-a1b2c3d4",
  "state": "COMPLETED",
  "data": "The weather in Beijing is sunny, 25°C"
}

5.3.2 Submit Task (Async)

curl -X POST 'http://localhost:10105/a2a/tasks?mode=async' \
  -H 'Content-Type: application/json' \
  -d '{"targetAgent":"weather-agent","message":"Shanghai"}'

5.3.3 SSE Stream

curl -N http://localhost:10105/a2a/tasks/{taskId}/stream

Response (text/event-stream):

data: {"taskId":"task-a1b2c3d4","state":"SUBMITTED"}

data: {"taskId":"task-a1b2c3d4","state":"WORKING","data":"processing..."}

data: {"taskId":"task-a1b2c3d4","state":"completed","data":"result..."}

5.3.4 List Agents

curl http://localhost:10105/a2a/agents

5.4 A2AClient SDK (Java)

A2AClient client = A2AClient.builder()
    .gatewayUrl("http://localhost:10105")
    .namespace("global")
    .agentName("my-agent")
    .agentCard(card)
    .heartbeatInterval(30_000)
    .build();

client.start();

// Synchronous task (returns typed TaskResult)
TaskResult result = client.sendTaskSync("weather-agent", "Beijing", null);

// Asynchronous task (returns taskId immediately)
String taskId = client.sendTaskAsync("weather-agent", "Shanghai", null);

// Poll status
TaskResult status = client.getTaskStatus(taskId);

// Cancel
boolean cancelled = client.cancelTask(taskId);

// List registered agents (typed List<String>)
List<String> agents = client.listAgents();

client.shutdown();

6. Future Roadmap

  • EventMesh Broker Integration: Replace InMemoryA2AMessageTransport with the real EventMesh broker for production deployment.
  • Schema Registry: Implement dynamic discovery of Agent capabilities via methods/list.
  • Sidecar Injection: Fully integrate the adaptor into the EventMesh Sidecar.
  • WebSocket Streaming: Extend SSE to bidirectional WebSocket for real-time agent dialogue.
  • Task Persistence: Persist TaskRegistry state to a durable store for crash recovery.
  • Authentication: Add API key / JWT authentication to the Gateway REST API.