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

OmniRoute's A2A server exposes a JSON-RPC 2.0 endpoint at POST /a2a that supports both synchronous skill execution and asynchronous streaming via Server-Sent Events (SSE), maintaining connections with 15-second heartbeats and standardized event schemas.

The diegosouzapw/OmniRoute repository provides a production-ready Agent-to-Agent (A2A) protocol server that bridges traditional JSON-RPC 2.0 request-response patterns with modern SSE streaming capabilities. This implementation enables real-time communication between AI agents while maintaining strict compliance with the JSON-RPC specification for method dispatch and error handling.

Core Architecture and Request Handling

Entry Point and Protocol Validation

All A2A traffic enters through src/app/a2a/route.ts, which acts as the main JSON-RPC router. The server expects a standard JSON-RPC 2.0 payload structure containing jsonrpc, id, method, and params fields. Upon receipt, the system immediately validates the request format before proceeding to authentication.

Authentication and Configuration Checks

The authenticate() function validates the optional Authorization: Bearer <token> header against the OMNIROUTE_API_KEY environment variable. If the A2A endpoint is disabled via the a2aEnabled setting, the server returns a JSON-RPC error response with code -32000, indicating a server-specific configuration error.

SSE Streaming Implementation for Real-Time Agent Communication

The message/stream Method

When handling the message/stream method, OmniRoute creates a persistent connection using a ReadableStream that emits Server-Sent Events. This approach allows skills to stream partial results, artifacts, and status updates back to the client without closing the HTTP connection.

Stream Lifecycle and Event Types

The streaming implementation in src/lib/a2a/streaming.ts manages four distinct event types through dedicated helper functions:

• Heartbeat events: Sent every 15 seconds via createHeartbeat() to prevent connection timeouts
• Chunk events: Generated by createChunkEvent() for each text fragment or artifact produced by the skill
• Completion events: Created with createCompletionEvent() to signal successful task termination along with final metadata
• Failure events: Produced by createFailureEvent() when a task is cancelled or encounters an execution error

All events are formatted using formatSSE(), which ensures proper SSE encoding as data: <JSON>\n\n lines. The response headers explicitly set Content-Type: text/event-stream with caching disabled to comply with the SSE specification.

Synchronous vs. Asynchronous Method Dispatch

OmniRoute supports four primary JSON-RPC methods, each mapped to specific handlers in the codebase:

message/send

Handled through direct skill invocation in src/lib/a2a/taskExecution.ts. This synchronous method executes the specified skill (such as smart-routing or quota-management) and returns the complete task ID, artifacts, and metadata in a single response.

message/stream

Managed by createA2AStream() in src/lib/a2a/streaming.ts. This asynchronous method initializes a long-lived connection that streams partial results as they are generated by the skill handler.

tasks/get

Retrieves current task state and stored artifacts via taskManager.getTask() in src/lib/a2a/taskManager.ts.

tasks/cancel

Updates task state to failed through taskManager.cancelTask(), effectively terminating running operations.

Skill Execution and Message Normalization

Before dispatching to skill handlers, OmniRoute normalizes incoming payloads using toMessageArray(), which converts various legacy payload shapes into a canonical [{role, content}] array structure. The system then routes requests through A2A_SKILL_HANDLERS, a registry defined in src/lib/a2a/taskExecution.ts that maps skill names to their implementation functions.

For the smart-routing skill, the system additionally logs routing decisions via logRoutingDecision(), providing observability into agent selection logic.

Practical Implementation Examples

The following examples demonstrate how to interact with OmniRoute's A2A server using standard fetch requests.

Synchronous skill execution:

// Send a prompt to the default smart-routing skill
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 response consumption:

// Receive partial results via SSE
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 queries:

// Retrieve a task's current state
await fetch("/api/a2a/tasks/req-001", { method: "GET" })
  .then(r => r.json())
  .then(console.log);

Key Source Files

The A2A implementation spans five critical files in the OmniRoute codebase:

• src/app/a2a/route.ts: Main JSON-RPC router handling authentication, method dispatch, and error formatting
• src/lib/a2a/streaming.ts: SSE utilities, event formatting, and the createA2AStream() implementation
• src/lib/a2a/taskManager.ts: In-memory task store with creation, state updates, cancellation, and TTL cleanup logic
• src/lib/a2a/taskExecution.ts: Skill registry (A2A_SKILL_HANDLERS) and the executeA2ATaskWithState helper
• src/app/api/a2a/tasks/route.ts: REST endpoints for task creation, listing, and cancellation used by the administrative interface

Summary

• OmniRoute's A2A server implements a JSON-RPC 2.0 protocol at POST /a2a with optional Bearer token authentication
• SSE streaming for the message/stream method uses a ReadableStream with 15-second heartbeats to maintain persistent connections
• The system supports both synchronous (message/send) and asynchronous execution patterns through the same endpoint
• All SSE events follow strict JSON-RPC formatting via formatSSE() and helpers like createChunkEvent() and createCompletionEvent()
• Task lifecycle management is centralized in src/lib/a2a/taskManager.ts, enabling state queries and cancellation via tasks/get and tasks/cancel

Frequently Asked Questions

What authentication mechanism does OmniRoute's A2A server use?

The server implements optional Bearer token authentication via the Authorization header, validating tokens against the OMNIROUTE_API_KEY environment variable. If authentication fails or the endpoint is disabled, it returns JSON-RPC error code -32000.

How does the SSE streaming format comply with JSON-RPC 2.0?

Each SSE event contains a JSON-RPC 2.0 object wrapped in the data: field, maintaining the standard jsonrpc, id, and result/error structure while using SSE's text/event-stream format for transport. This allows streaming of partial results while preserving protocol compliance.

Can clients cancel long-running streaming tasks?

Yes, clients can cancel running tasks by calling the tasks/cancel method or accessing the REST endpoint at /api/a2a/tasks/[id], which invokes taskManager.cancelTask() to update the task state to failed and terminate the stream.

What is the purpose of the 15-second heartbeat in SSE connections?

The heartbeat, implemented via createHeartbeat(), prevents proxy servers and load balancers from terminating idle connections during long-running skill executions, ensuring the stream remains active until the task completes or fails.

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