A2A v0.3 Protocol Server Implementation in OmniRoute: Complete Technical Guide

The A2A v0.3 protocol server implementation in OmniRoute is a lightweight JSON-RPC 2.0 server that exposes internal capabilities—such as smart routing and quota management—via standardized HTTP endpoints with support for real-time streaming and task lifecycle management.

The OmniRoute project implements the Agent-to-Agent (A2A) v0.3 specification to create a bridge between internal services and external consumers. Located primarily under src/lib/a2a and exposed through Next.js API routes in src/app/api/a2a, this implementation allows clients to invoke remote procedures, monitor long-running operations via Server-Sent Events (SSE), and manage task state through a clean JSON-RPC interface.

Core Architecture and Protocol Design

The A2A v0.3 server implementation follows a layered architecture that separates transport concerns from business logic. It operates on standard HTTP with JSON-RPC 2.0 as the messaging envelope, enabling compatibility with existing tools while supporting OmniRoute-specific extensions for streaming and task management.

JSON-RPC 2.0 Foundation

All communication uses the JSON-RPC 2.0 specification. Incoming requests must include jsonrpc: "2.0", a unique request id, a method string identifying the skill to invoke, and optional params. The server validates these envelopes using Zod schemas before routing to the appropriate handler. Responses follow the standard JSON-RPC shape: either a result field on success or an error object with code and message on failure.

Task Lifecycle Management

The server maintains an in-memory task registry that tracks execution state through a defined lifecycle: pending, running, completed, and canceled. When a client POSTs to the task creation endpoint, the system generates a UUID, assigns the task to a worker, and updates state transitions atomically. This registry is consulted by both the streaming layer to emit progress events and the cancellation endpoint to abort ongoing work.

Key Components and Source Files

The implementation is modularized across several critical files in the src/lib/a2a directory, each handling distinct responsibilities from execution to transport.

Task Execution Core (taskExecution.ts)

The src/lib/a2a/taskExecution.ts file exports the runTask() function, the primary entry point for task dispatch. This function:

• Validates the incoming RPC method against the skill registry
• Instantiates the task context with a generated UUID
• Invokes the corresponding skill handler from src/lib/a2a/skills/*
• Wraps return values in JSON-RPC success responses or converts thrown errors into sanitized RPC error objects

If a skill returns a plain object, runTask() serializes it immediately. If the skill returns a ReadableStream, execution delegates to the streaming adapter.

Task Manager (taskManager.ts)

Located at src/lib/a2a/taskManager.ts, this module maintains the canonical state of active tasks. It exports cancelTask(), which is invoked by the cancellation API endpoint to signal_abort controllers and update task status to canceled. The manager uses a Map structure keyed by task UUID to store metadata, start timestamps, and AbortController instances for ongoing operations.

Streaming Adapter (streaming.ts)

The src/lib/a2a/streaming.ts module handles the conversion between internal Node.js streams and the SSE transport expected by A2A clients. When runTask() detects a stream return, the adapter:

• Pipes chunks through a transformation layer
• Formats each chunk as a JSON-encoded SSE data event
• Manages backpressure and client disconnections gracefully

Skill Registry (src/lib/a2a/skills/*)

Skills are modular business logic units implementing specific RPC methods. Common implementations include:

• smartRouting: Generates optimized LLM routing plans based on model and prompt parameters
• quotaManagement: Queries and updates rate-limiting metadata
• providerDiscovery: Lists available upstream providers and health status

Each skill exports a handler function matching the (params: unknown) => Promise<unknown> | ReadableStream signature.

API Endpoints and HTTP Routing

The server exposes three primary HTTP interfaces through Next.js App Router conventions, all residing under src/app/api/a2a.

Task Creation Endpoint (/api/a2a/tasks)

The route file src/app/api/a2a/tasks/route.ts accepts POST requests containing JSON-RPC envelopes. It delegates immediately to taskExecution.runTask() after Zod validation. This endpoint supports both synchronous responses and stream initiation depending on the skill's return type.

Task Cancellation (/api/a2a/tasks/[id]/cancel)

The dynamic route at src/app/api/a2a/tasks/[id]/cancel/route.ts handles POST requests to abort running tasks. It extracts the task ID from the URL path parameter and invokes taskManager.cancelTask() to trigger the AbortController and update state.

Status Monitoring (/api/a2a/status)

The src/app/api/a2a/status/route.ts endpoint provides health checks and task inspection capabilities. It returns JSON-RPC compatible status objects containing the server's current load, active task counts, and individual task metadata when queried with a specific task ID.

Practical Implementation Examples

Invoking the Smart Routing Skill

To execute a task against the A2A v0.3 server, POST a JSON-RPC envelope to the tasks endpoint:

const payload = {
  jsonrpc: "2.0",
  id: "req-123",
  method: "smartRouting",
  params: { 
    model: "gpt-4o", 
    prompt: "Explain quantum entanglement." 
  }
};

const response = await fetch("/api/a2a/tasks", {
  method: "POST",
  headers: { "Content-Type": "application/json" },
  body: JSON.stringify(payload)
});

const result = await response.json();
if (result.error) {
  console.error("RPC Error:", result.error.code, result.error.message);
} else {
  console.log("Routing plan:", result.result);
}

Consuming Real-Time Streams

For skills returning streams, connect via SSE after obtaining the task ID:

const eventSource = new EventSource(`/api/a2a/tasks/${taskId}/stream`);

eventSource.onmessage = (event) => {
  const chunk = JSON.parse(event.data);
  console.log("Received chunk:", chunk);
};

eventSource.onerror = () => {
  console.error("Stream connection failed");
  eventSource.close();
};

Canceling Long-Running Tasks

To abort an operation in progress:

await fetch(`/api/a2a/tasks/${taskId}/cancel`, { 
  method: "POST" 
});

Security and Error Sanitization

All A2A routes share a unified error handling pipeline located in open-sse/utils/error.ts. This middleware sanitizes stack traces and internal error details before serialization, ensuring that the JSON-RPC responses contain only safe error codes and messages compliant with the A2A v0.3 security guidelines. The implementation prevents information leakage while preserving diagnostic utility through structured error codes.

Summary

• OmniRoute's A2A v0.3 server implements a JSON-RPC 2.0 layer over HTTP for agent-to-agent communication, located in src/lib/a2a and src/app/api/a2a.
• Task execution is handled by taskExecution.runTask() in src/lib/a2a/taskExecution.ts, which routes methods to skills and manages response formatting.
• State management uses an in-memory registry in taskManager.ts with cancelTask() support for graceful aborts.
• Streaming capabilities via src/lib/a2a/streaming.ts convert internal streams to SSE for real-time client updates.
• API surface includes endpoints for task creation (/api/a2a/tasks), cancellation (/api/a2a/tasks/[id]/cancel), and status checks (/api/a2a/status).
• Security relies on centralized error sanitization in open-sse/utils/error.ts to prevent data leakage in RPC error responses.

Frequently Asked Questions

What is the A2A v0.3 protocol and how does OmniRoute implement it?

The A2A (Agent-to-Agent) v0.3 protocol is a specification for inter-agent communication using JSON-RPC 2.0 over HTTP. OmniRoute implements this as a lightweight server in the src/lib/a2a directory, exposing internal capabilities like routing and quota management through standardized RPC methods. The implementation adds OmniRoute-specific extensions for Server-Sent Events (SSE) streaming and task lifecycle management while maintaining strict protocol compliance.

How does task streaming work in the OmniRoute A2A server?

When a skill handler returns a ReadableStream instead of a plain value, taskExecution.runTask() delegates to the streaming adapter in src/lib/a2a/streaming.ts. This module pipes stream chunks to the client using Server-Sent Events, encoding each chunk as a JSON data event. The connection remains open until the stream completes, the client disconnects, or taskManager.cancelTask() triggers an abort, enabling real-time updates for long-running operations like progressive LLM responses.

Where are skills registered and how are they invoked?

Skills are implemented as modules in src/lib/a2a/skills/*, such as smartRouting and quotaManagement. The taskExecution.ts core maintains a registry mapping method names to these handler functions. When a JSON-RPC request arrives, the server validates the method exists in this registry, then invokes the corresponding skill with the provided parameters. This modular architecture allows adding new capabilities without modifying the core RPC infrastructure.

How does the server handle task cancellation?

Cancellation requests route to src/app/api/a2a/tasks/[id]/cancel/route.ts, which extracts the task ID from the URL and calls taskManager.cancelTask() from src/lib/a2a/taskManager.ts. The task manager looks up the active task in its in-memory Map, triggers the associated AbortController, and updates the task state to canceled. This propagates the abort signal to any ongoing stream processing or async operations, ensuring resources are released promptly.

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