How to Implement the A2A v0.3 Protocol with JSON-RPC 2.0 and SSE for Agent-to-Agent Communication

OmniRoute provides a complete A2A server implementation that uses JSON-RPC 2.0 over HTTP POST for requests and Server-Sent Events (SSE) for streaming responses, enabling standardized agent-to-agent communication through a stateless HTTP layer with an in-memory task manager.

The OmniRoute repository ships with a production-ready A2A (Agent-to-Agent) server that implements the v0.3 protocol specification. This implementation combines JSON-RPC 2.0 for structured request-response patterns with Server-Sent Events (SSE) for real-time streaming, allowing AI agents to communicate through a centralized yet stateless routing layer.

Core Architecture of the A2A Implementation

Request Entry Point and JSON-RPC Router

Located in src/app/a2a/route.ts, the main entry point handles HTTP POST requests and implements the JSON-RPC 2.0 envelope validation. The route performs several steps in sequence:

1. Bearer token authentication – returns jsonRpcError(-32600) on failure.
2. JSON body parsing – returns jsonRpcError(-32700) on parse errors.
3. Envelope verification – validates required fields and returns jsonRpcError(-32600) if malformed.
4. Settings check – returns HTTP 503 if A2A is disabled.
5. Method dispatch – routes to handlers based on the method field:
   • "message/send" – synchronous skill execution
   • "message/stream" – SSE streaming execution
   • "tasks/get" – fetch task status
   • "tasks/cancel" – cancel a running task

The helper functions jsonRpcError and jsonRpcResult construct properly formatted responses (lines 79-88 in route.ts).

Message Normalization

The toMessageArray helper (lines 22-62 in route.ts) normalizes input to handle both the canonical shape:

{ "messages": [{ "role": "user", "content": "..." }] }

And legacy shapes (message.content, message.parts), always returning an array of {role, content} objects for skill handlers.

Task Lifecycle Management

The A2ATaskManager class in src/lib/a2a/taskManager.ts maintains an in-memory state machine for each task. It tracks transitions from submitted → working → completed, failed, or cancelled. Each task receives:

• UUID for identification
• Timestamps for creation and updates
• TTL (default 5 minutes) for automatic cleanup
• Storage for input, artifacts, events, and metadata (lines 37-48)

The manager also tracks activeStreams for monitoring SSE connections.

Skill Execution Registry

Skill handlers are registered in src/lib/a2a/taskExecution.ts within the A2A_SKILL_HANDLERS record:

export const A2A_SKILL_HANDLERS: Record<string, A2ASkillHandler> = {
  "smart-routing": executeSmartRouting,
  "quota-management": executeQuotaManagement,
  // ...additional built-in skills
};

The executeA2ATaskWithState function wraps these handlers, capturing artifacts and updating task state before returning a StreamTaskResult. Skill implementations reside in src/lib/a2a/skills/.

SSE Streaming Infrastructure

For streaming responses, src/lib/a2a/streaming.ts exports createA2AStream, which returns a ReadableStream that yields artifacts as SSE data: events:

export function createA2AStream(
  task: A2ATask,
  exec: (t: A2ATask) => Promise<StreamTaskResult>,
  abortSignal: AbortSignal,
  hooks: { onStart?: () => void; onEnd?: () => void }
): ReadableStream

The function accepts an abort signal and lifecycle hooks (onStart, onEnd) to track active stream counts in the task manager. The route returns a Response with SSE_HEADERS and the readable stream (lines 21-22).

JSON-RPC 2.0 Method Reference

message/send (Synchronous Execution)

The message/send method accepts parameters including skill, messages, and optional metadata. It executes the handler synchronously and returns a complete result:

{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "task": { "id": "uuid", "state": "completed" },
    "artifacts": [...],
    "metadata": {...}
  }
}

message/stream (SSE Streaming)

Uses identical parameters to message/send but returns an SSE stream of partial results. Each chunk contains JSON-encoded data with artifact updates, suitable for long-running agent operations. The stream uses TransformStream to wrap async skill execution.

tasks/get and tasks/cancel

• tasks/get – Retrieves current task state by ID via tm.getTask(taskId)
• tasks/cancel – Transitions a running task to cancelled state via tm.cancelTask(taskId)

Both methods return the task object wrapped in a JSON-RPC result envelope.

Practical Implementation Examples

Synchronous Skill Execution with cURL

curl -X POST http://localhost:20128/a2a \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer $OMNIROUTE_API_KEY" \
  -d '{
        "jsonrpc": "2.0",
        "id": 1,
        "method": "message/send",
        "params": {
          "skill": "smart-routing",
          "messages": [{"role": "user", "content": "Explain the A2A protocol"}]
        }
      }'

Response:

{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "task": {"id": "c71f...", "state": "completed"},
    "artifacts": [{"type": "text", "content": "..."}],
    "metadata": {}
  }
}

Streaming Responses with Node.js

import fetch from "node-fetch";

const resp = await fetch("http://localhost:20128/a2a", {
  method: "POST",
  headers: {
    "Content-Type": "application/json",
    "Authorization": `Bearer ${process.env.OMNIROUTE_API_KEY}`,
  },
  body: JSON.stringify({
    jsonrpc: "2.0",
    id: "stream-1",
    method: "message/stream",
    params: {
      skill: "smart-routing",
      messages: [{ role: "user", content: "Summarize this log file..." }],
    },
  }),
});

for await (const chunk of resp.body) {
  console.log("SSE chunk →", chunk.toString());
}

Each chunk is prefixed with data: as required by the SSE specification.

Task Management Operations

Query task status:

curl -X POST http://localhost:20128/a2a \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer $OMNIROUTE_API_KEY" \
  -d '{
        "jsonrpc": "2.0",
        "id": 2,
        "method": "tasks/get",
        "params": {"taskId": "c71f..."}
      }'

Cancel a running task:

curl -X POST http://localhost:20128/a2a \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer $OMNIROUTE_API_KEY" \
  -d '{
        "jsonrpc": "2.0",
        "id": 3,
        "method": "tasks/cancel",
        "params": {"taskId": "c71f..."}
      }'

Agent Discovery and Protocol Versioning

OmniRoute publishes an agent card at /.well-known/agent.json (implemented in src/app/.well-known/agent.json/route.ts). This document advertises:

• Protocol version (v0.3)
• Supported JSON-RPC methods
• Available skills and their capabilities

Clients use this endpoint for auto-discovery of the A2A interface, eliminating the need for hardcoded configuration.

Extending the Protocol with Custom Skills

To add new capabilities to your A2A v0.3 implementation:

1. Create a handler in src/lib/a2a/skills/<name>.ts exporting an async function
2. Register it in A2A_SKILL_HANDLERS in taskExecution.ts
3. Update the agent card in src/app/.well-known/agent.json/route.ts

The system automatically inherits JSON-RPC error handling (codes -32600, -32601, -32602, -32700, -32603), SSE streaming support, and task-state tracking.

For custom authentication, replace the authenticate() function in src/app/a2a/route.ts with your own token validation or OAuth flow. To persist tasks beyond the default 5-minute TTL, swap the in-memory Map for a SQLite table—the A2ATask schema aligns with standard database conventions.

Summary

• JSON-RPC 2.0 envelope parsing in src/app/a2a/route.ts provides standardized request handling with proper error codes.
• Bearer-token authentication secures the endpoint while remaining replaceable for custom auth schemes.
• A2ATaskManager maintains strict state machine semantics with TTL cleanup and active stream tracking.
• Skill registry in taskExecution.ts encapsulates agent logic while exposing a consistent handler interface.
• SSE streaming via createA2AStream enables real-time agent communication with proper abort handling.
• Agent discovery through the well-known agent.json endpoint automates client configuration.

Frequently Asked Questions

What is the A2A v0.3 protocol?

The A2A (Agent-to-Agent) v0.3 protocol is a standardized communication specification that enables AI agents to interoperate through JSON-RPC 2.0 requests and Server-Sent Events. It defines methods for sending messages, streaming responses, and managing task lifecycles, allowing agents to discover and invoke skills on remote systems.

How does OmniRoute handle authentication for A2A requests?

OmniRoute implements Bearer token authentication in src/app/a2a/route.ts, returning a JSON-RPC error with code -32600 for invalid credentials. The implementation is stateless and designed to be replaced with custom logic such as OAuth2, JWT validation, or API key databases by modifying the authenticate() function.

Can I persist tasks to a database instead of using in-memory storage?

Yes. The A2ATaskManager in src/lib/a2a/taskManager.ts uses a private Map for storage, but you can replace this with a SQLite or PostgreSQL backend. The A2ATask interface (lines 37-48) already includes fields compatible with database persistence, such as UUIDs, timestamps, and JSON-serializable metadata.

What is the difference between message/send and message/stream?

The message/send method executes skills synchronously and returns a complete JSON-RPC response with all artifacts. The message/stream method initiates an SSE connection that streams partial results as they become available, using createA2AStream in src/lib/a2a/streaming.ts to wrap the execution in a ReadableStream with proper data: event formatting.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}