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

> Discover how OmniRoute's A2A protocol server implements JSON-RPC 2.0 with SSE for real-time data streaming. Explore synchronous and asynchronous skill execution with 15-second heartbeats.

- Repository: [Diego Rodrigues de Sa e Souza/OmniRoute](https://github.com/diegosouzapw/OmniRoute)
- Tags: deep-dive
- Published: 2026-07-21

---

**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`](https://github.com/diegosouzapw/OmniRoute/blob/main/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`](https://github.com/diegosouzapw/OmniRoute/blob/main/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`](https://github.com/diegosouzapw/OmniRoute/blob/main/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`](https://github.com/diegosouzapw/OmniRoute/blob/main/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`](https://github.com/diegosouzapw/OmniRoute/blob/main/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`](https://github.com/diegosouzapw/OmniRoute/blob/main/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:

```typescript
// 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:

```typescript
// 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:

```typescript
// 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`](https://github.com/diegosouzapw/OmniRoute/blob/main/src/app/a2a/route.ts)**: Main JSON-RPC router handling authentication, method dispatch, and error formatting
- **[`src/lib/a2a/streaming.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/src/lib/a2a/streaming.ts)**: SSE utilities, event formatting, and the `createA2AStream()` implementation
- **[`src/lib/a2a/taskManager.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/src/lib/a2a/taskManager.ts)**: In-memory task store with creation, state updates, cancellation, and TTL cleanup logic
- **[`src/lib/a2a/taskExecution.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/src/lib/a2a/taskExecution.ts)**: Skill registry (`A2A_SKILL_HANDLERS`) and the `executeA2ATaskWithState` helper
- **[`src/app/api/a2a/tasks/route.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/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`](https://github.com/diegosouzapw/OmniRoute/blob/main/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.