How to Implement A2A Protocol Skills for Agent Communication in OmniRoute

OmniRoute's A2A (Agent-to-Agent) server transforms the router into a discoverable AI agent that external systems can invoke via JSON-RPC 2.0, with built-in task management, skill execution, and real-time streaming.

The A2A protocol skills in OmniRoute (diegosouzapw/OmniRoute) enable the router to participate in multi-agent workflows as a first-class peer. This guide walks through the architecture, existing skills, and how to build your own.

---

Understanding the A2A Architecture

OmniRoute's A2A implementation consists of three core layers: discovery, task lifecycle management, and skill execution. Each layer is implemented as dedicated TypeScript modules in src/lib/a2a/.

The router exposes its capabilities through a well-known endpoint, manages asynchronous task states with automatic cleanup, and executes pure TypeScript skill handlers that integrate with the existing routing pipeline.

---

Discovery: The Agent Card

External agents discover OmniRoute's capabilities by fetching /.well-known/agent.json. This Agent Card—documented in src/lib/a2a/README.md—contains metadata, authentication requirements, and the complete skill registry.

The discovery payload lists all available A2A protocol skills including:

• smart-routing — Routes prompts through OmniRoute's normal chat pipeline with full diagnostics
• quota-management — Answers natural-language queries about provider quota availability

Clients cache this response to determine which skills to invoke and what parameters each expects.

---

Task Lifecycle and State Management

Every A2A request creates a tracked task. The A2ATaskManager class in src/lib/a2a/taskManager.ts handles the complete lifecycle:

// Task states: submitted → working → completed | failed | cancelled

Key responsibilities include:

• Assigning UUIDs and tracking timestamps
• Enforcing a default TTL of 5 minutes for automatic cleanup
• Logging all state transitions for observability
• Providing task lookup for streaming reconnections

When a client disconnects mid-stream, the task manager preserves state, allowing seamless reconnection via the same task ID.

---

Built-in A2A Protocol Skills

OmniRoute ships with two production-ready skills demonstrating different integration patterns.

Smart-Routing Skill

Located in src/lib/a2a/skills/smartRouting.ts, this skill forwards the user's prompt to OmniRoute's standard /v1/chat/completions endpoint and returns enriched metadata.

The handler:

1. Normalizes incoming messages into the chat completions format
2. Invokes the internal routing engine with the specified model and budget parameters
3. Captures routing decisions via routingLogger.ts
4. Returns the generated content plus metadata.routing_explanation and metadata.cost_envelope

This skill is the primary way external agents leverage OmniRoute's provider-agnostic routing without managing provider configurations themselves.

Quota-Management Skill

Found in src/lib/a2a/skills/quotaManagement.ts, this skill provides natural-language access to provider quota tables.

Rather than requiring clients to parse raw quota data, the skill interprets queries like "Which provider has the most quota?" and returns formatted answers with current utilization percentages and estimated remaining capacity.

---

Execution Flow: From Request to Response

The complete request path flows through src/app/a2a/route.ts:

1. Validate the JSON-RPC 2.0 request structure
2. Resolve the requested skill from the skill registry
3. Invoke taskExecution.ts to run the skill within a managed task context
4. Format the response—or initiate SSE streaming for message/stream requests

For synchronous calls (message/send), the route waits for completion and returns the full result. For streaming calls, it hands off to the streaming layer immediately.

---

Real-Time Streaming with Server-Sent Events

Long-running or incremental responses use the streaming.ts module in src/lib/a2a/streaming.ts. This layer constructs a TransformStream that emits properly formatted SSE events:

data: {"jsonrpc":"2.0","method":"tasks/push","params":{"taskId":"...","chunk":{"content":"..."}}}
data: {"jsonrpc":"2.0","method":"tasks/push","params":{"task":{"state":"working"}}}
data: {"jsonrpc":"2.0","method":"tasks/push","params":{"task":{"state":"completed"}}}

Each event includes:

• Content chunks with role attribution
• Heartbeat messages to keep connections alive
• Final state transitions with complete metadata

Clients subscribe by calling the message/stream method with identical parameters to the synchronous variant.

---

Implementing a Custom A2A Protocol Skill

Adding a new skill requires three steps. Here's a complete example implementing an echo service:

Step 1: Create the Skill Handler

Create src/lib/a2a/skills/echoSkill.ts:

import { A2ATaskManager, getTaskManager } from "../taskManager";

export interface EchoParams {
  skill: string;
  messages: { role: string; content: string }[];
  metadata?: Record<string, unknown>;
}

export async function handler(params: EchoParams) {
  const userMessage = params.messages.find((m) => m.role === "user")?.content ?? "";
  const prefix = (params.metadata?.prefix as string) || "Echo";
  
  return {
    artifact: {
      type: "text",
      content: `${prefix}: ${userMessage}`,
    },
    metadata: {
      originalLength: userMessage.length,
      timestamp: Date.now(),
    },
  };
}

Step 2: Register the Skill

Add the skill to src/lib/a2a/README.md so it appears in the Agent Card:

{
  "id": "echo-skill",
  "name": "Echo Skill",
  "description": "Returns user input with optional prefix customization",
  "parameters": {
    "messages": "Array of role/content objects (standard chat format)",
    "metadata": {
      "prefix": "String to prepend before the echo (default: 'Echo')"
    }
  }
}

Step 3: Extend Routing Diagnostics (Optional)

If your skill makes routing decisions, import and use routingLogger.ts:

import { logRoutingDecision } from "../routingLogger";

// Inside your handler after selecting a provider
logRoutingDecision({
  provider: "selected-provider",
  combo: "model-variant",
  estimatedCost: 0.0012,
  latencyMs: 245,
  reasoning: "Selected based on budget constraint and quota availability",
});

This ensures metadata.routing_explanation populates correctly for clients.

---

Python Client Example: Discovery and Invocation

import requests
import json

BASE_URL = "http://localhost:20128"
API_KEY = "YOUR_API_KEY"
HEADERS = {
    "Content-Type": "application/json",
    "Authorization": f"Bearer {API_KEY}",
}

# 1️⃣ Discover capabilities

agent = requests.get(f"{BASE_URL}/.well-known/agent.json").json()
print("Agent:", agent["name"], agent["version"])
print("Skills:", [s["id"] for s in agent["skills"]])

# 2️⃣ Smart-routing task (synchronous)

payload = {
    "jsonrpc": "2.0",
    "id": "task-1",
    "method": "message/send",
    "params": {
        "skill": "smart-routing",
        "messages": [{"role": "user", "content": "Write a Python quicksort"}],
        "metadata": {"model": "auto", "budget": 0.10},
    },
}
resp = requests.post(f"{BASE_URL}/a2a", headers=HEADERS, json=payload).json()
result = resp["result"]
print("📝 Code:", result["artifacts"][0]["content"][:200])
print("🔀 Route:", result["metadata"]["routing_explanation"])
print("💰 Cost:", result["metadata"]["cost_envelope"]["actual"])

# 3️⃣ Quota-management query

quota_payload = {
    "jsonrpc": "2.0",
    "id": "task-2",
    "method": "message/send",
    "params": {
        "skill": "quota-management",
        "messages": [{"role": "user", "content": "Which provider has the most quota?"}],
    },
}
quota_resp = requests.post(f"{BASE_URL}/a2a", headers=HEADERS, json=quota_payload).json()
print("📊 Quota:", quota_resp["result"]["artifacts"][0]["content"])

---

TypeScript Client Example: Streaming Smart-Routing

const BASE_URL = "http://localhost:20128";
const API_KEY = "YOUR_API_KEY";

async function streamSmartRouting(prompt: string) {
  const resp = await fetch(`${BASE_URL}/a2a`, {
    method: "POST",
    headers: {
      "Content-Type": "application/json",
      Authorization: `Bearer ${API_KEY}`,
    },
    body: JSON.stringify({
      jsonrpc: "2.0",
      id: "stream-1",
      method: "message/stream",
      params: { 
        skill: "smart-routing", 
        messages: [{ role: "user", content: prompt }] 
      },
    }),
  });

  const reader = resp.body!.getReader();
  const decoder = new TextDecoder();

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    const chunk = decoder.decode(value);
    for (const line of chunk.split("\n")) {
      if (line.startsWith("data: ")) {
        const ev = JSON.parse(line.slice(6));
        if (ev.params.chunk) process.stdout.write(ev.params.chunk.content);
        if (ev.params.task.state === "completed") console.log("\n✅ Done");
      }
    }
  }
}

streamSmartRouting("Explain microservices architecture");

---

Key Files Reference

Component	Path	Purpose
Agent Card	src/lib/a2a/README.md	Discovery endpoint definition and skill registry
Task Manager	src/lib/a2a/taskManager.ts	UUID assignment, state machine, TTL enforcement
Task Execution	src/lib/a2a/taskExecution.ts	Skill invocation orchestration
Streaming	src/lib/a2a/streaming.ts	SSE formatting and transport
Routing Logger	src/lib/a2a/routingLogger.ts	Decision capture for diagnostics
Smart-Routing	src/lib/a2a/skills/smartRouting.ts	Core routing integration skill
Quota-Management	src/lib/a2a/skills/quotaManagement.ts	Natural-language quota queries
API Route	src/app/a2a/route.ts	Entry point for all A2A requests
Skill Directory	src/lib/a2a/skills/	Auto-discovered skill modules

---

Summary

• Discovery happens via /.well-known/agent.json, served from documentation in src/lib/a2a/README.md
• Tasks are managed by A2ATaskManager with 5-minute TTL, UUID tracking, and state transitions
• Skills are pure TypeScript modules in src/lib/a2a/skills/ that export a handler(params) function
• Streaming uses Server-Sent Events through message/stream for real-time response delivery
• Extending requires only creating a skill file and registering it in the README

---

Frequently Asked Questions

How do I authenticate with the OmniRoute A2A server?

Request an API key from your OmniRoute administrator and include it in the Authorization: Bearer header. The src/app/a2a/route.ts handler validates this token against configured credentials before processing any JSON-RPC request.

Can I use OmniRoute A2A skills from LangChain or CrewAI?

Yes. Any framework that speaks HTTP/JSON-RPC 2.0 can discover and invoke OmniRoute skills. Use the Python client pattern above wrapped in a LangChain Tool or CrewAI Agent delegate. The streaming endpoint supports LangChain's async callback patterns for real-time token delivery.

What happens if a streaming connection drops mid-task?

The A2ATaskManager preserves task state for the full TTL duration. Reconnect with the same task ID and the message/stream method to resume receiving events. Completed tasks return the full result immediately upon reconnection.

How do I add routing explanations to my custom skill?

Import logRoutingDecision from src/lib/a2a/routingLogger.ts and call it whenever your skill selects providers or makes cost/latency tradeoffs. The next task state push automatically includes this data in metadata.routing_explanation.

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