How OmniRoute's A2A Protocol Server Implements JSON-RPC 2.0 with SSE Streaming

OmniRoute exposes a JSON-RPC 2.0 compliant Agent-to-Agent (A2A) endpoint at POST /a2a that supports both synchronous skill execution and real-time streaming via Server-Sent Events (SSE), implementing 15-second heartbeats, artifact chunking, and comprehensive lifecycle management through a dedicated Task Manager.

The OmniRoute repository provides a production-ready implementation of an A2A protocol server that combines strict JSON-RPC 2.0 request handling with SSE streaming capabilities. This architecture enables AI agents to invoke skills either synchronously or through real-time streams, with all requests validated, authenticated, and dispatched through a centralized routing layer.

Core Endpoint Architecture

The A2A protocol server is anchored in src/app/a2a/route.ts, which implements the main JSON-RPC router. When a request arrives at POST /a2a, the server executes a strict validation pipeline:

• Request Parsing: The body is parsed into a canonical JSON-RPC object containing jsonrpc, id, method, and params fields.
• Authentication: The authenticate() function validates the optional Authorization: Bearer <token> header against the OMNIROUTE_API_KEY environment variable.
• Feature Flag Check: If the A2A endpoint is disabled via settings (a2aEnabled), the server returns a JSON-RPC error with code -32000.
• Method Dispatch: Valid requests are routed to their respective handlers based on the method name.

Method Dispatch and Skill Registry

OmniRoute supports four primary A2A methods, each mapped to specific handlers in the execution layer:

Method	Description	Handler Location
message/send	Synchronous execution returning task ID, artifacts, and metadata	Direct skill invocation via taskExecution.ts
message/stream	Asynchronous SSE streaming of partial results	createA2AStream() in src/lib/a2a/streaming.ts
tasks/get	Retrieves current task state and stored artifacts	taskManager.getTask() in src/lib/a2a/taskManager.ts
tasks/cancel	Cancels a running task and updates state to failed	taskManager.cancelTask()

The skill registry (A2A_SKILL_HANDLERS in src/lib/a2a/taskExecution.ts) maps skill names to their implementation functions. Before execution, toMessageArray() normalizes various legacy payload shapes into a canonical [{role, content}] array format. When the smart-routing skill is invoked, the system logs routing decisions via logRoutingDecision().

SSE Streaming Implementation

For the message/stream method, OmniRoute establishes a persistent SSE connection that streams JSON-RPC 2.0 compliant events. The implementation in src/lib/a2a/streaming.ts creates a ReadableStream that manages the entire transmission lifecycle:

1. Heartbeat Mechanism: The stream emits a heartbeat event every 15 seconds to prevent connection timeouts.
2. Chunk Transmission: Each artifact fragment produced by the skill generates a chunk event containing partial data.
3. Completion Signaling: Upon successful skill execution, the stream emits a completion event with final metadata.
4. Failure Handling: If the task is cancelled or encounters an error, the stream emits a failure event and terminates.

All SSE events are formatted using formatSSE(), which constructs properly encoded data: <JSON>\n\n lines. The response headers are set to Content-Type: text/event-stream with caching explicitly disabled. Helper functions createChunkEvent(), createCompletionEvent(), createHeartbeat(), and createFailureEvent() ensure all events conform to the JSON-RPC 2.0 event schema.

Task Lifecycle Management

The Task Manager (src/lib/a2a/taskManager.ts) coordinates the full lifecycle of every A2A request:

• Creation: Initializes task records with unique IDs and stores the JSON-RPC request parameters.
• State Tracking: Maintains in-memory task states (pending, running, completed, failed) and artifact collections.
• Cancellation: Provides cancelTask() to update task states and signal streaming handlers to terminate.
• TTL Cleanup: Automatically removes stale tasks to prevent memory leaks.

For REST-based task management, src/app/api/a2a/tasks/route.ts exposes additional endpoints for creating, listing, and cancelling tasks through standard HTTP methods.

Practical Implementation Examples

Synchronous Skill Invocation

Use the message/send method for blocking execution that returns complete results:

await fetch("/a2a", {
  method: "POST",
  headers: { 
    "Content-Type": "application/json", 
    Authorization: "Bearer $YOUR_KEY" 
  },
  body: JSON.stringify({
    jsonrpc: "2.0",
    id: "req-001",
    method: "message/send",
    params: {
      skill: "smart-routing",
      messages: [{ role: "user", content: "Explain quantum entanglement" }],
    },
  }),
}).then(r => r.json()).then(console.log);

Streaming Execution via SSE

Use the message/stream method for real-time token streaming:

const resp = await fetch("/a2a", {
  method: "POST",
  headers: { "Content-Type": "application/json" },
  body: JSON.stringify({
    jsonrpc: "2.0",
    id: "req-002",
    method: "message/stream",
    params: {
      skill: "smart-routing",
      messages: [{ role: "user", content: "Write a poem about sunrise" }],
    },
  }),
});

const reader = resp.body?.getReader();
const decoder = new TextDecoder();
while (true) {
  const { done, value } = await reader!.read();
  if (done) break;
  console.log(decoder.decode(value)); // prints SSE lines like `data: {"jsonrpc":"2.0",...}`
}

Task Status Retrieval

Query the state of any task using the REST endpoint:

await fetch("/api/a2a/tasks/req-001", { method: "GET" })
  .then(r => r.json())
  .then(console.log);

Summary

• OmniRoute implements a JSON-RPC 2.0 A2A endpoint at POST /a2a with comprehensive request validation and authentication.
• The server supports four primary methods: message/send, message/stream, tasks/get, and tasks/cancel.
• SSE streaming in src/lib/a2a/streaming.ts provides real-time updates with 15-second heartbeats, chunk events, and completion signaling.
• The Task Manager (src/lib/a2a/taskManager.ts) handles in-memory state, lifecycle transitions, and cancellation signals.
• Skill dispatch is managed through A2A_SKILL_HANDLERS in src/lib/a2a/taskExecution.ts, with automatic message normalization via toMessageArray().

Frequently Asked Questions

What is the endpoint URL for the A2A protocol server?

The A2A protocol server exposes a single JSON-RPC 2.0 endpoint at POST /a2a. This route, defined in src/app/a2a/route.ts, accepts all method calls including both synchronous message/send requests and streaming message/stream requests.

How does authentication work for A2A requests?

The server validates the optional Authorization: Bearer <token> header against the OMNIROUTE_API_KEY environment variable through the authenticate() function. If the header is missing or invalid, the request is rejected before reaching the skill dispatch layer.

What is the format of SSE events in the streaming implementation?

All SSE events follow the Server-Sent Events specification with data: <JSON>\n\n formatting. Each event is a JSON-RPC 2.0 compliant object created by helpers like createChunkEvent() and createCompletionEvent(), transmitted with Content-Type: text/event-stream and caching disabled.

How do I cancel a running A2A task?

Send a JSON-RPC request with method tasks/cancel to the /a2a endpoint, or call the REST endpoint src/app/api/a2a/tasks/route.ts directly. The Task Manager updates the task state to failed and emits a failure event through the active SSE stream, causing createA2AStream() to terminate the connection.

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