# 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 streaming for synchronous execution, real-time data, and efficient artifact handling.

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

---

**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`](https://github.com/diegosouzapw/OmniRoute/blob/main/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`](https://github.com/diegosouzapw/OmniRoute/blob/main/taskExecution.ts) |
| `message/stream` | Asynchronous SSE streaming of partial results | `createA2AStream()` in [`src/lib/a2a/streaming.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/src/lib/a2a/streaming.ts) |
| `tasks/get` | Retrieves current task state and stored artifacts | `taskManager.getTask()` in [`src/lib/a2a/taskManager.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/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`](https://github.com/diegosouzapw/OmniRoute/blob/main/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`](https://github.com/diegosouzapw/OmniRoute/blob/main/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`](https://github.com/diegosouzapw/OmniRoute/blob/main/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`](https://github.com/diegosouzapw/OmniRoute/blob/main/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:

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

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

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