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

> Learn how to implement A2A v0.3 for agent-to-agent communication using JSON-RPC 2.0 and SSE with OmniRoute's stateless HTTP server implementation.

- Repository: [Diego Rodrigues de Sa e Souza/OmniRoute](https://github.com/diegosouzapw/OmniRoute)
- Tags: how-to-guide
- Published: 2026-07-05

---

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

### Message Normalization

The `toMessageArray` helper (lines 22-62 in [`route.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/route.ts)) normalizes input to handle both the canonical shape:

```json
{ "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`](https://github.com/diegosouzapw/OmniRoute/blob/main/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`](https://github.com/diegosouzapw/OmniRoute/blob/main/src/lib/a2a/taskExecution.ts) within the `A2A_SKILL_HANDLERS` record:

```typescript
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`](https://github.com/diegosouzapw/OmniRoute/blob/main/src/lib/a2a/streaming.ts) exports `createA2AStream`, which returns a `ReadableStream` that yields artifacts as SSE `data:` events:

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

```json
{
  "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

```bash
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:**

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

```

### Streaming Responses with Node.js

```javascript
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:**

```bash
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:**

```bash
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`](https://github.com/diegosouzapw/OmniRoute/blob/main//.well-known/agent.json) (implemented in [`src/app/.well-known/agent.json/route.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/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`](https://github.com/diegosouzapw/OmniRoute/blob/main/taskExecution.ts)
3. Update the agent card in [`src/app/.well-known/agent.json/route.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/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`](https://github.com/diegosouzapw/OmniRoute/blob/main/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`](https://github.com/diegosouzapw/OmniRoute/blob/main/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`](https://github.com/diegosouzapw/OmniRoute/blob/main/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`](https://github.com/diegosouzapw/OmniRoute/blob/main/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`](https://github.com/diegosouzapw/OmniRoute/blob/main/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`](https://github.com/diegosouzapw/OmniRoute/blob/main/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`](https://github.com/diegosouzapw/OmniRoute/blob/main/src/lib/a2a/streaming.ts) to wrap the execution in a `ReadableStream` with proper `data:` event formatting.