OmniRoute A2A Server 全解析：把 AI 网关变成 A2A 协议智能路由 Agent 的完整实践

OmniRoute A2A Server 全解析：把 AI 网关变成 A2A 协议智能路由 Agent 的完整实践

【免费下载链接】OmniRouteNever stop coding. Free MIT AI gateway: one endpoint, 352 providers (150+ free), 1200+ models Kimi, Claude, GPT, Gemini, GLM, DeepSeek, MiniMax. Works with Claude Code, Codex, Cursor, OpenCode, Cline & Copilot. Quota-aware auto-fallback, RTK+Caveman compression saves 15-95% tokens, MCP/A2A, Desktop/PWA. Built by 550+ contributors项目地址: https://gitcode.com/GitHub_Trending/om/OmniRoute

OmniRoute 除了作为 OpenAI 兼容网关，还内置了一个 A2A（Agent-to-Agent Protocol v0.3）服务端，让外部智能体可以通过标准的 JSON-RPC 2.0 接口把"请求路由、配额查询、成本分析、健康报告"等能力当作技能（Skill）调用。本文以仓库中的 A2A 文档为骨架，结合 JSON-RPC 路由、任务管理器、技能执行器 等源码，讲清 A2A 端点的发现、认证、四个核心方法、六类技能、任务生命周期，以及如何扩展新技能——读完后你既能直接接入 OmniRoute 的 A2A 服务，也能读懂其底层的任务状态机与流式实现。

整体架构：一个协议，两张"脸"

A2A 表面在 OmniRoute 中分为两个入口，这一点在源码中体现得非常清晰：

• JSON-RPC 2.0 规范入口：POST /a2a，定义在 src/app/a2a/route.ts，是 A2A 协议客户端（如 a2a-sdk、Hermes 等）的标准对接面；
• REST 辅助入口：/api/a2a/*（状态、任务列表、取消），面向仪表盘与外部工具，代码位于 src/app/api/a2a/。

任务统一由A2ATaskManager（src/lib/a2a/taskManager.ts，默认 5 分钟 TTL）跟踪，技能则通过 src/lib/a2a/taskExecution.ts 中的A2A_SKILL_HANDLERS注册表分发。route.ts的主处理流程可以概括为：认证 → 解析 JSON-RPC 信封 → 检查 A2A 开关 → 解析调用方身份（owner）→ 按method分发到四个 case 之一。

Agent 发现（Agent Discovery）

其他 Agent 首先通过 well-known 地址发现 OmniRoute 的能力：

curl http://localhost:20128/.well-known/agent.json

该端点返回 Agent Card，描述 OmniRoute 的能力、技能清单和认证要求。实现位于 src/app/.well-known/agent.json/route.ts，有三个值得注意的实现细节：

1. 版本自动同步：Card 的version字段取自process.env.npm_package_version（见 route.ts），因此每次发版都与package.json保持一致，无需手工维护；
2. 动态技能合并：Card 的技能数组在内置 6 个技能之后，还会追加getFleetSkills()返回的 OmniConductor 车队技能（缓存约 60 秒，Hub 离线时为空数组，Card 依然有效）；
3. 缓存策略：响应头携带Cache-Control: public, max-age=3600，即 Agent Card 被 CDN/客户端缓存 1 小时。

Agent Card 同时声明capabilities.streaming: true、pushNotifications: false，以及认证方式api-key+Authorization头，调用方据此即可完成对接。文档也提醒：Agent Card 应始终与实时的 352 提供商目录保持对齐，提供商数量与 free/no-auth 元数据均来自运行时注册表。

认证与启用开关

认证

所有/a2a请求都要求通过Authorization头携带 API Key：

Authorization: Bearer YOUR_OMNIROUTE_API_KEY

"如果服务器未配置 API Key 则跳过认证"——这句话在源码中对应一个更完整的三级策略，见 src/lib/a2a/authenticate.ts 的authenticateA2ARequest：

• REQUIRE_API_KEY开启时：必须携带有效的 OmniRoute API Key；没有 Key 的请求还会回退到仪表盘会话认证（isDashboardSessionAuthenticated），这样仪表盘内置的 A2A playground 仅凭会话 Cookie 也能工作；
• 配置了OMNIROUTE_API_KEY时：用timingSafeEqual做常量时间比较校验 Bearer Key，防止时序侧信道；
• 两者都未配置：保持 keyless local-first 默认姿态，直接放行。

JSON-RPC 路由与 REST 任务路由共用这一份实现，注释中明确提到这是安全修复 GHSA-jcm5-6wpp-wjj8 的成果——此前 REST 任务路由完全没有认证调用，两个入口从此不可能再各自漂移。

启用开关

A2A 由Endpoints → A2A开关控制，默认关闭。从 route.ts 的rejectIfA2ADisabled可以看到具体行为：

• 关闭时GET /api/a2a/status报告status: "disabled"、online: false；
• 对POST /a2a的 JSON-RPC 调用返回 HTTP 503，并附带 JSON-RPC 错误码-32000，错误消息为 "A2A endpoint is disabled. Enable it from the Endpoints page."。

判断依据是getSettings()返回的settings.a2aEnabled字段。

JSON-RPC 2.0 方法详解

message/send— 同步执行

向某个技能发送消息并等待完整响应。skill参数缺省时回退为smart-routing；messages既支持数组形式，也兼容单条message.content/message.parts旧格式（见 route.ts 的toMessageArray归一化逻辑）。

curl -X POST http://localhost:20128/a2a \ -H "Content-Type: application/json" \ -H "Authorization: Bearer YOUR_KEY" \ -d '{ "jsonrpc": "2.0", "id": "1", "method": "message/send", "params": { "skill": "smart-routing", "messages": [{"role": "user", "content": "Write a hello world in Python"}], "metadata": {"model": "auto", "combo": "fast-coding"} } }'

响应：

{ "jsonrpc": "2.0", "id": "1", "result": { "task": { "id": "uuid", "state": "completed" }, "artifacts": [{ "type": "text", "content": "..." }], "metadata": { "routing_explanation": "Selected claude-sonnet via provider \"anthropic\" (latency: 1200ms, cost: $0.003)", "cost_envelope": { "estimated": 0.005, "actual": 0.003, "currency": "USD" }, "resilience_trace": [ { "event": "primary_selected", "provider": "anthropic", "timestamp": "..." } ], "policy_verdict": { "allowed": true, "reason": "within budget and quota limits" } } } }

这些 metadata 字段不是摆设——对照 src/lib/a2a/skills/smartRouting.ts 的实现：技能内部以 30 秒超时调用本机的/v1/chat/completions（combo映射到x-combo请求头），然后自己拼出routing_explanation（含模型、provider、实测延迟与成本）、cost_envelope（按 prompt token 数粗估 vs 上游返回的实际值）、resilience_trace（上游触发回退时会追加fallback_needed事件），以及policy_verdict（当metadata.budget给出时，用实际成本与预算比较得到 allowed/denied）。执行成功后，路由层还会把这次决策写入 routingLogger 的路由决策日志。

message/stream— SSE 流式执行

用法与message/send相同，但返回 Server-Sent Events 实时流：

curl -N -X POST http://localhost:20128/a2a \ -H "Content-Type: application/json" \ -H "Authorization: Bearer YOUR_KEY" \ -d '{ "jsonrpc": "2.0", "id": "1", "method": "message/stream", "params": { "skill": "smart-routing", "messages": [{"role": "user", "content": "Explain quantum computing"}] } }'

SSE 事件：

data: {"jsonrpc":"2.0","method":"message/stream","params":{"task":{"id":"...","state":"working"},"chunk":{"type":"text","content":"..."}}} : heartbeat 2026-03-03T17:00:00Z data: {"jsonrpc":"2.0","method":"message/stream","params":{"task":{"id":"...","state":"completed"},"metadata":{...}}}

streaming.ts 给出了这条流的完整机制：

• 心跳：每 15 秒发一条: heartbeat <ISO时间>注释行保活，防止中间代理掐断空闲连接；
• 响应头：Content-Type: text/event-stream、Cache-Control: no-cache, no-transform、X-Accel-Buffering: no（关闭 Nginx 缓冲），确保事件即时送达；
• 事件序列：技能执行完成后逐条 artifact 发chunk事件（state 为working），最后发一条携带完整 metadata 的completed事件；失败则发failed事件并附metadata.error；
• 取消传播：监听请求的AbortSignal，客户端断开时向流内写入 "Cancelled" 失败事件并关闭。

tasks/get— 查询任务状态

curl -X POST http://localhost:20128/a2a \ -H "Content-Type: application/json" \ -H "Authorization: Bearer YOUR_KEY" \ -d '{"jsonrpc":"2.0","id":"2","method":"tasks/get","params":{"taskId":"TASK_UUID"}}'

taskId也可写成id（源码同时接受两种写法）。任务不存在时返回-32601。

tasks/cancel— 取消任务

curl -X POST http://localhost:20128/a2a \ -H "Content-Type: application/json" \ -H "Authorization: Bearer YOUR_KEY" \ -d '{"jsonrpc":"2.0","id":"3","method":"tasks/cancel","params":{"taskId":"TASK_UUID"}}'

取消走updateTask(id, "cancelled", ...)状态迁移；注意终态任务无法再迁移（见下文状态机），对已终态任务调用取消会返回-32603。

A2A 1.0 方法名兼容层

除了 v0.3 的四个方法，route.ts 还内置了一个 v1.0 ↔ v0.3 兼容层：A2A 1.0 把方法重命名为SendMessage/SendStreamingMessage，且同步响应中回复文本位于task.status.message.parts[].text。OmniRoute 通过V1_METHOD_ALIASES把 1.0 方法名映射回 v0.3，再用buildV1Task把结果重塑成 1.0 客户端期望的形状（state 字符串加TASK_STATE_前缀），使 a2a-sdk 1.x 等客户端可以不改代码直接调用；v0.3 客户端不受影响。该行为由 tests/unit/a2a-v1-compat-10839.test.ts 覆盖。

可用技能（Skills）

OmniRoute 暴露 6 个 A2A 技能，统一注册在 taskExecution.ts 的A2A_SKILL_HANDLERS中，每个技能模块位于 src/lib/a2a/skills/：

对应实现文件依次为 smartRouting.ts、quotaManagement.ts、providerDiscovery.ts、costAnalysis.ts、healthReport.ts、listCapabilities.ts。

list-capabilities技能详解

list-capabilities对需要"先发现能力、再发调用"的外部 Agent 尤其有用。它返回结构化的 Markdown 表格 artifact：

| ID | Name | Category | Area | Endpoints/Commands | Raw URL | | --- | --- | --- | --- | --- | --- | | omni-auth | Auth & Sessions | api | auth | POST /api/auth/login, ... | https://raw.githubusercontent.com/... | ...

每一行都带rawUrl列，Agent 可以立即拉取完整的 SKILL.md 注入上下文；metadata.totalSkills字段镜像目录规模。技能清单与仓库 skills/ 目录下的 SKILL.md 文件一一对应（如 skills/omni-auth/ 等）。

技能执行前的记忆命中采集

还有一个源码层面才可见的细节：每个技能真正执行前，taskExecution.ts 的collectMemoryHits会先按"最后一条 user 消息"做一次记忆召回。它有严格的防御边界：默认 1500ms 超时（MEMORY_RECALL_TIMEOUT_MS），任何失败/超时都退化为空命中、绝不让任务失败；结果只写入task.metadata.memoryHits并追加memory_hits历史事件，纯粹用于可观测性，不会注入技能提示词。环境变量OMNIROUTE_A2A_MEMORY_HITS=0是总开关，直接短路返回空数组。

REST 辅助 API

JSON-RPC 端点/a2a是规范入口，以下 REST 端点为仪表盘和外部工具提供辅助访问（实现位于 src/app/api/a2a/）：

入站 Conductor 委派（POST /api/a2a/tasks）：外部 A2A Agent 可以通过 OmniRoute 把编码工作委派给 OmniConductor 车队。请求体形如{ skill: "conductor" | "conductor-cli-<profile>", messages: [{role, content}], metadata: { conductor: { repo: { url, base_ref? }, mode?, cli?, model? } } }。只有 Agent Card 上宣布过的 Conductor 车队技能可被委派；metadata.conductor.repo.url必填（车队在 git 仓库上工作）。路由层用服务端的CONDUCTOR_ORCHESTRATOR_TOKEN（回退CONDUCTOR_HUB_TOKEN）将其翻译为 Hub 的POST /v1/tasks，并返回201 { conductor_task_id, state: "submitted" }；任务状态经 SSE→A2A 镜像回流（RF1），可通过GET /api/a2a/tasks?skill=conductor查询。该路径有专门的测试 tests/unit/conductor-a2a-post.test.ts。

任务生命周期与 TTL

submitted → working → completed → failed → cancelled

• TTL：任务默认 5 分钟后过期。TTL 在A2ATaskManager构造函数中配置（taskManager.ts，ttlMinutes参数默认 5）；如需定制，可自行实例化，例如new A2ATaskManager(15)得到 15 分钟 TTL。注意生产路由通过getTaskManager()获取的全局单例使用默认值；
• 过期处理：一个 60 秒周期的后台定时器扫描过期任务——未到期前访问getTask若发现非终态任务已过期，会即时将其迁移为failed（消息 "Task expired"）；
• 回收：终态任务在超过 2 倍 TTL 后从内存中删除；
• 终态：completed、failed、cancelled，一旦进入不可再迁移；
• 事件日志：每次状态迁移都推入task.events数组（时间戳 + 状态 + 可选消息）。

状态机的合法性由 taskManager.ts 的VALID_TRANSITIONS表严格约束：

const VALID_TRANSITIONS: Record<TaskState, TaskState[]> = { submitted: ["working", "failed", "cancelled"], working: ["completed", "failed", "cancelled"], completed: [], failed: [], cancelled: [], };

所有者隔离（Owner Scoping）：每次调用都通过resolveA2AOwner解析出 owner——有 API Key 时为 Key 的 SHA-256 前 32 位，仪表盘会话为"dashboard"，keyless 姿态下为undefined。带 owner 的任务只对同一 owner 可见/可取消；取消前会先做 owner 检查，并用与"任务不存在"相同的错误掩盖"存在但不是你的"，使 IDOR 探测无法区分两者。keyless 创建的任务保持对所有人可见（本地优先姿态）。行为由 tests/unit/a2a-task-owner-idor.test.ts 锁定。

持久化与保留：内存Map是活动任务的唯一事实来源，同时每次写入都会 best-effort 落到 SQLite 历史表（upsertA2ATask/appendA2ATaskEvent，见 src/lib/db/a2aTasks.ts 的引入），持久化失败只记日志、不阻断写入路径。历史行保留天数由OMNIROUTE_A2A_HISTORY_RETENTION_DAYS控制（缺省 30 天），清理动作被节流为每天至多一次。

错误码

HTTP 状态码与 JSON-RPC 码存在映射：-32600→ 400，-32601→ 404，-32603→ 500，其余为 200（见 route.ts 的jsonRpcError），而禁用端点-32000单独返回 503。

如何新增一个技能

文档给出五步扩展流程，与源码结构完全对应：

1. 创建技能文件：src/lib/a2a/skills/<your-skill>.ts，导出异步函数(task: A2ATask) => Promise<{ artifacts, metadata }>，可参照 smartRouting.ts 的形状；

2. 注册处理器：在 taskExecution.ts 的A2A_SKILL_HANDLERS中追加条目：

   export const A2A_SKILL_HANDLERS = { // ...existing skills "your-skill": async (task) => { const skillModule = await import("./skills/yourSkill"); return skillModule.executeYourSkill(task); }, };

   采用动态import()意味着新技能模块只在首次被调用时才加载，不影响启动时延；

3. 在 Agent Card 中暴露：在 src/app/.well-known/agent.json/route.ts 的skills数组追加：

   { "id": "your-skill", "name": "Your Skill", "description": "Brief, intent-focused description", "tags": ["routing", "quota"], "examples": ["Sample natural-language invocation"] }

   Agent Card 是外部 Agent 的发现依据，不登记的技能对协议客户端不可见；

4. 编写测试：tests/unit/a2a-<your-skill>.test.ts，覆盖正常路径与错误路径。可参考现有测试组织，如 tests/unit/t09-a2a-lifecycle.test.ts（生命周期）、tests/unit/a2a-enabled-route.test.ts（开关行为）、tests/unit/a2a-cost-analysis-numeric-fallback.test.ts（数值回退）；

5. 文档：在本文件的Available Skills表格中补上新技能条目，保持文档与A2A_SKILL_HANDLERS一致。

集成示例

Python（requests）

import requests resp = requests.post("http://localhost:20128/a2a", json={ "jsonrpc": "2.0", "id": "1", "method": "message/send", "params": { "skill": "smart-routing", "messages": [{"role": "user", "content": "Hello"}] } }, headers={"Authorization": "Bearer YOUR_KEY"}) result = resp.json()["result"] print(result["artifacts"][0]["content"]) print(result["metadata"]["routing_explanation"])

TypeScript（fetch）

const resp = await fetch("http://localhost:20128/a2a", { method: "POST", headers: { "Content-Type": "application/json", Authorization: "Bearer YOUR_KEY", }, body: JSON.stringify({ jsonrpc: "2.0", id: "1", method: "message/send", params: { skill: "smart-routing", messages: [{ role: "user", content: "Hello" }], }, }), }); const { result } = await resp.json(); console.log(result.metadata.routing_explanation);

两个示例都省略了metadata：如果需要约束路由，可以追加{"metadata": {"model": "auto", "combo": "fast-coding", "budget": 0.5}}——model缺省为auto，combo会映射为上游请求的x-combo头，budget触发policy_verdict的预算裁决。

小结与适用前提

• 适用前提：A2A 端点默认关闭，需先在 Endpoints 页打开 A2A 开关；默认端口为 20128，示例中的localhost:20128需按实际部署地址替换；
• 本地 keyless 部署下/a2a直接放行（与/v1的本地优先姿态一致），生产环境建议启用OMNIROUTE_API_KEY或REQUIRE_API_KEY以启用 Key 校验与按 owner 的任务隔离；
• 协议版本为 A2A v0.3，并附带 1.0 方法名兼容层；同步响应中路由解释、成本包络、韧性轨迹都在metadata内，流式场景则在最后的 completed 事件中给出；
• 任务默认 5 分钟 TTL、终态 2 倍 TTL 后从内存回收、历史保留 30 天（可经OMNIROUTE_A2A_HISTORY_RETENTION_DAYS调整），长任务或审计场景需要留意这些边界。

由此，OmniRoute 把"智能路由网关"的每一项核心能力——combo 路由、配额、成本、健康度、技能目录——都收敛成一个对 Agent 世界开放的 A2A 面，外部智能体只需一次 Agent Card 发现加标准的 JSON-RPC 调用，就能把整个网关当作自己的路由后端使用。

【免费下载链接】OmniRouteNever stop coding. Free MIT AI gateway: one endpoint, 352 providers (150+ free), 1200+ models Kimi, Claude, GPT, Gemini, GLM, DeepSeek, MiniMax. Works with Claude Code, Codex, Cursor, OpenCode, Cline & Copilot. Quota-aware auto-fallback, RTK+Caveman compression saves 15-95% tokens, MCP/A2A, Desktop/PWA. Built by 550+ contributors项目地址: https://gitcode.com/GitHub_Trending/om/OmniRoute