「让运维 Agent 自己去问数据库 Agent」
四月份我们遇到一个需求：客服 Agent 在处理客诉时，需要判断「这个订单的物流是不是真的延误了」。这个判断需要的权限（查物流商内部系统、看历史延误率）我们不打算给客服 Agent。
常规解法是加一个工具。但那意味着把物流系统的权限开放出去，而且物流团队不想让客服团队碰他们的接口。
最后我们用了 A2A：让客服 Agent 把这件事「委派」给物流团队的 Agent。这篇记录 A2A 协议的设计逻辑、我们的 Java 实现，以及用了两个月后的判断。
A2A 解决的是 MCP 不管的那部分
先把分工说清楚，这是最容易搞混的地方。
MCP 解决的是「Agent 怎么用工具」。工具是无状态的、确定性的、不参与决策的——你调 queryOrder 它返回一个订单，仅此而已。
但真实场景里有另一类需求：你需要的是一个能自己做判断的主体，不是一个函数。比如「判断这个物流是不是延误了」，这不是查一次数据库就能回答的，需要看多个数据源、需要行业经验、可能需要反问、可能给出「不确定」的结论。
这就是 A2A 的场景：Agent 之间怎么协作。
| 维度 | MCP | A2A |
|---|---|---|
| 交互对象 | Agent → 工具（无状态服务） | Agent → Agent（有自主性的主体） |
| 交互模式 | 请求-响应，同步 | 任务（Task），可异步、可长周期 |
| 对方能力 | 固定的工具签名 | 用自然语言描述的「技能」 |
| 协商 | 无，参数对了就执行 | 有，可以要求澄清、可以拒绝 |
| 产出 | 结构化数据 | 工件（Artifact）+ 状态 |
| 典型耗时 | 毫秒~秒 | 秒~分钟（可能涉及人工） |
最简单的判断标准：如果你需要对方「想一想」，用 A2A；如果只是「做一下」，用 MCP。
它们不是竞争关系。我们的物流 Agent 自己就用 MCP 连着物流系统的各种工具，对客服 Agent 来说它是一个 A2A 对端，对它自己来说它是 MCP 客户端。分层很清楚。
协议核心：Agent Card、Task、Artifact
A2A 的三个核心概念。
Agent Card：能力的自描述
每个 A2A 服务端要提供一个 Agent Card，相当于它的「名片」。这是个约定路径的 JSON：
GET /.well-known/agent-card.json{
"name": "logistics-agent",
"description": "物流异常诊断智能体。可以判断订单是否真的发生延误、延误原因、以及预计到达时间。",
"url": "https://logistics-agent.internal/a2a",
"version": "1.4.0",
"protocolVersion": "0.3.0",
"capabilities": {
"streaming": true,
"pushNotifications": false
},
"defaultInputModes": ["text"],
"defaultOutputModes": ["text", "data"],
"skills": [
{
"id": "delay-diagnosis",
"name": "延误诊断",
"description": "判断订单物流是否真实延误。需要提供订单号。可以给出延误原因分类和预计到达时间。",
"tags": ["物流", "延误", "诊断"],
"examples": [
"订单 SO20260512001 的物流是不是延误了",
"帮我看看这个件为什么还没到"
]
},
{
"id": "eta-prediction",
"name": "到货时间预测",
"description": "基于历史数据和当前状态预测到货时间。",
"tags": ["物流", "时效"]
}
]
}这张卡片的设计我挺喜欢的：它描述的是「能帮你解决什么问题」，不是「有哪些接口」。调用方 Agent 可以根据 description 和 examples 判断该不该找它，这正是模型擅长的事。
注意 description 里我写了「可以给出延误原因分类」这种模糊表述。这是刻意的——它描述能力边界，不承诺精确输出格式。
Task：异步的、有生命周期的工作单元
A2A 的基本交互单元是 Task，不是 Request。这个区别很重要。
POST /a2a HTTP/1.1
Content-Type: application/json
{
"jsonrpc": "2.0",
"id": "req-001",
"method": "tasks/send",
"params": {
"id": "task-7f3a2c91",
"sessionId": "conv-abc123",
"message": {
"role": "user",
"parts": [
{"type": "text", "text": "订单 SO20260512001 的物流是不是延误了？"}
]
},
"metadata": {
"callerTenant": "T001",
"priority": "normal"
}
}
}响应里 Task 有状态：
{
"jsonrpc": "2.0",
"id": "req-001",
"result": {
"id": "task-7f3a2c91",
"status": {
"state": "completed",
"timestamp": "2026-05-27T10:23:41Z"
},
"artifacts": [
{
"name": "diagnosis",
"parts": [
{
"type": "data",
"data": {
"delayed": true,
"delayHours": 38,
"reasonCategory": "WEATHER",
"confidence": 0.82,
"evidence": [
"5月25日 途中遭遇暴雨预警（来源：气象局）",
"同线路其他 12 个包裹均延误 24-48 小时"
]
}
}
]
},
{
"name": "summary",
"parts": [
{"type": "text", "text": "该订单确实延误，原因是天气，预计明天下午到达。"}
]
}
]
}
}Task 的状态机：submitted → working → (input-required) → completed / failed / canceled。
input-required 这个状态是 A2A 最有价值的设计。 它意味着被调用方可以反问：「你给我的信息不够，能不能再提供一下承运商单号？」这在 MCP 里是没有的——工具要么成功要么失败，不能说「我需要更多信息」。
Artifact：结构化 + 非结构化的混合产出
上面那个响应里有两个 artifact：一个是结构化数据（给调用方 Agent 的程序逻辑用），一个是文本（给模型读）。这个设计很实用。
我们的约定是：每个 Task 至少返回一个 data artifact 和一个 text artifact。前者让调用方可以做确定性判断（比如 delayed == true 才触发补偿流程），后者让模型能生成自然语言回复。
Java 实现
Java 侧的 A2A 库还不算成熟，我们评估了几个，最后选了自己实现核心部分（约 800 行）。理由后面说。
服务端
@RestController
@RequestMapping("/a2a")
public class A2aEndpoint {
private final TaskRegistry tasks;
private final LogisticsAgent agent;
@GetMapping("/.well-known/agent-card.json")
public AgentCard card() {
return AgentCard.loadFromClasspath("agent-card.json");
}
@PostMapping
public Mono<JsonRpcResponse> handle(@RequestBody JsonRpcRequest req) {
return switch (req.method()) {
case "tasks/send" -> send(req);
case "tasks/get" -> get(req);
case "tasks/cancel" -> cancel(req);
case "tasks/sendSubscribe" -> subscribe(req); // SSE 流式
default -> Mono.just(JsonRpcResponse.methodNotFound(req.id()));
};
}
private Mono<JsonRpcResponse> send(JsonRpcRequest req) {
SendParams p = req.paramsAs(SendParams.class);
// 幂等：同一个 task id 重复提交直接返回已有结果
if (tasks.exists(p.id())) {
return Mono.just(JsonRpcResponse.result(req.id(), tasks.get(p.id())));
}
Task task = tasks.create(p.id(), p.sessionId());
// 异步执行，长任务不阻塞 HTTP 连接
return Mono.fromFuture(
CompletableFuture.supplyAsync(() -> execute(task, p), taskExecutor))
.map(t -> JsonRpcResponse.result(req.id(), t));
}
private Task execute(Task task, SendParams p) {
task.markWorking();
try {
AgentResult r = agent.handle(p.message(), p.metadata());
task.complete(r.artifacts());
} catch (NeedMoreInfoException e) {
// 关键：可以要求补充信息
task.requireInput(e.question());
} catch (Exception e) {
task.fail(e.getMessage());
}
return task;
}
}三个必须做的工程点：
- 幂等。网络重试会导致同一个 task 被提交两次，不幂等就会重复执行。我们用 task id 做幂等键，重复提交直接返回已存在的结果；
- 异步执行。A2A 的 task 可能跑几十秒甚至几分钟（我们的物流诊断平均 8 秒，但涉及人工的能到几小时）。绝不能占着 HTTP 连接；
- 状态持久化。Task 状态要落库，否则实例重启就丢了。
CREATE TABLE a2a_task (
task_id varchar(64) PRIMARY KEY,
session_id varchar(64),
caller varchar(32) NOT NULL, -- 调用方 Agent 标识
tenant_id varchar(16) NOT NULL,
state varchar(20) NOT NULL,
input jsonb NOT NULL,
artifacts jsonb,
error text,
created_at timestamptz DEFAULT now(),
updated_at timestamptz DEFAULT now()
);
CREATE INDEX ON a2a_task (session_id, created_at DESC);客户端
客户端这边我们做成一个 Spring AI 的 Tool，这样调用方 Agent 不需要感知 A2A 的存在：
@Component
public class A2aDelegationTool {
private final A2aClientRegistry registry;
@Tool(description = """
向物流智能体提问物流相关问题。
适用于：判断物流是否延误、延误原因分析、到货时间预测。
不适用于：查询订单基本信息（用 queryOrder）、发起物流投诉（转人工）。
这是一个异步任务，可能需要 10~30 秒，请耐心等待。
""")
public DelegationResult askLogistics(
@ToolParam(description = "要向物流智能体提出的问题，必须包含订单号") String question) {
AgentCard card = registry.get("logistics-agent");
if (!card.hasSkillMatching(question)) {
return DelegationResult.unavailable("物流智能体不支持该类问题");
}
Task task = a2aClient.send(card.url(), TaskRequest.builder()
.sessionId(SessionContext.currentId())
.text(question)
.metadata(Map.of("callerTenant", TenantContext.currentId()))
.build());
// 轮询或订阅，最长等 60 秒
Task completed = awaitCompletion(task, Duration.ofSeconds(60));
return switch (completed.state()) {
case COMPLETED -> DelegationResult.of(completed.artifacts());
case INPUT_REQUIRED -> DelegationResult.needInfo(completed.question());
case FAILED -> DelegationResult.unavailable(completed.error());
default -> DelegationResult.timeout();
};
}
}关键点是 INPUT_REQUIRED 的处理：调用方 Agent 收到后会转头问用户「物流那边需要承运商单号，能提供一下吗」。这个链路走通的时候我们挺兴奋的——Agent 之间和 Agent 与用户之间用同一套交互范式，不需要为「被反问」写特殊逻辑。
为什么没用现成库
Java 侧现在有几个 A2A 的实现，我们都试过，最后还是自己写了。原因：
- 协议版本迭代快。我们开始做的时候是 0.2.x，两个月内升到 0.3.0，字段有变化。现成库跟不上；
- 我们需要的东西不多。完整协议里有推送通知、流式更新、多模态 parts，我们只用到了同步 send、get、cancel 和 SSE 订阅。自己实现 800 行就够了；
- 和我们现有体系的整合。鉴权、多租户、链路追踪、成本计量都要接进来，用现成库反而要包一层。
这个决定是场景相关的。如果你需要完整的协议支持（尤其是多模态和推送通知），还是用库更合适。我们的场景简单，自己写更快也更可控。
踩的坑
一、超时策略
第一个版本我们设了 30 秒超时，然后发现大量超时——物流诊断平均 8 秒，但 P99 是 47 秒。
正确的做法是要区分两种「慢」：任务还在跑和任务卡死了。我们的解法是让被调用方定期更新 task 的 updated_at，客户端看这个时间戳，只要还在更新就继续等。
Task awaitCompletion(Task task, Duration timeout) {
Instant deadline = Instant.now().plus(timeout);
Instant lastProgress = Instant.now();
while (Instant.now().isBefore(deadline)) {
Task current = client.get(task.id());
if (current.state().isTerminal()) return current;
// 有进展就重置「无进展计时」，但不超过总 deadline
if (current.updatedAt().isAfter(lastProgress)) {
lastProgress = current.updatedAt();
} else if (Duration.between(lastProgress, Instant.now())
.compareTo(Duration.ofSeconds(45)) > 0) {
return current.markStalled(); // 45 秒无进展判定卡死
}
Thread.sleep(1000);
}
return task.markTimeout();
}改完之后超时率从 18% 降到 0.7%。
二、上下文传递的边界
客服 Agent 派任务给物流 Agent 时，要传多少上下文？传多了浪费 token 且有数据泄露风险，传少了对方答不上来。
我们第一版是把整个对话历史传过去，被安全团队拦下了——客服对话里有用户的手机号、地址，物流 Agent 不需要这些，也不应该有权限看。
现在的规矩是：只传任务必需的字段，且经过脱敏。
public class ContextSanitizer {
public Map<String, Object> forDelegation(String targetAgent,
ConversationContext ctx) {
Set<String> allowed = allowedFields.get(targetAgent); // 白名单
Map<String, Object> out = new HashMap<>();
out.put("orderNo", ctx.fact("order_no"));
out.put("userId", mask(ctx.fact("user_id"))); // 脱敏
out.put("question", ctx.lastUserMessage());
// 手机号、地址、支付信息一律不传
return out;
}
}白名单是按「目标 Agent」配置的，每个 Agent 能收到什么字段是显式声明的。这个配置在安全评审时过了一遍。
三、循环委派
Agent A 派给 B，B 又派回给 A，无限循环。我们在测试环境就遇到了，一个任务跑了 40 分钟才被超时终止。
解法是在请求头里带上调用链：
public void checkNoCycle(String callerChain, String selfName) {
List<String> chain = Arrays.asList(callerChain.split(">"));
if (chain.contains(selfName)) {
throw new DelegationCycleException(
"检测到循环委派：" + callerChain + ">" + selfName);
}
if (chain.size() >= 3) {
throw new DelegationDepthException("委派层级超过 3 层");
}
}
// 请求头
X-A2A-Delegation-Chain: customer-service-agent>order-agent深度限制设 3 层。实际用下来，超过 2 层的委派我们就没见成功过——越深的链路越容易在某个环节断掉。
四、责任归属
这是个非技术问题，但很重要：委派出去的任务出错了，算谁的？
我们的约定是：Agent Card 的 skill description 里必须写明能力边界和免责范围，调用方对最终用户负责。物流 Agent 说「没延误」但实际延误了，导致客服给了错误答复——对外是客服 Agent 的责任，内部再看是物流 Agent 的判断错误。
这个责任模型定下来之后，调用方会做二次校验（我们的客服 Agent 对物流 Agent 的 confidence < 0.7 的结论会转人工），而不是无脑采信。
两个月的数据
| 指标 | 数值 | 说明 |
|---|---|---|
| 日均委派任务 | 3,420 | 占客服对话的 13% |
| 平均耗时 | 6.8s | P99 是 41s |
| 超时率 | 0.7% | 优化前 18% |
| input-required 比例 | 4.1% | 需要反问用户 |
| 委派后任务完成率 | 91.3% | 相比自己判断的 78.2% |
| 循环委派 | 0 | 加了链路检查后 |
最关键的一行是「委派后任务完成率 91.3% vs 自己判断 78.2%」。这个对比来自我们 A/B 测试的两组数据——同样的客诉，一组让客服 Agent 自己判断物流是否延误（用它能拿到的有限数据），一组委派给物流 Agent。
差异主要来自物流 Agent 能拿到客服 Agent 拿不到的数据（物流商内部状态、同线路其他包裹的情况）。这是 A2A 最实在的价值：让专业的事交给有权限、有数据、有领域知识的 Agent 去做。
我的判断
用了两个月，说说我对 A2A 的看法。
一、协议设计里最有价值的是 input-required 状态。它承认了一个事实：复杂任务的信息往往是不完备的，需要在执行过程中反补。这是 Agent 协作和函数调用的本质区别。光这一条就值得用 A2A 而不是简单地互相调 HTTP 接口。
二、现在（2026 年中）还处在很早期的阶段。 协议版本 0.3.0，还在快速迭代。Java 侧的实现都不成熟。我们的建议是：如果你只有一个明确的点对点协作需求，A2A 有点重，直接定义内部接口更快。真正需要它是当你有多个团队各自维护 Agent，且协作关系会变化时。
三、最大的风险是它让系统变得难以理解。 一次用户请求可能穿过 3 个 Agent，每个 Agent 内部又有多次 LLM 调用和工具调用。出问题时排查链路很长。我们现在强制要求全链路 traceId 贯通，并且每个委派点都记日志。这个成本不小，但必须有。
四、跨组织的 A2A 我暂时不看好。 协议本身支持（Agent Card 是公开可发现的），但实际操作中，跨公司的 Agent 协作涉及责任、计费、数据合规，这些都不是协议能解决的。我们目前只在公司内部跨团队用。
留个问题
关于《A2A 协议：智能体之间的协作标准》里这个坑，你当时是怎么处理的？欢迎在评论区聊聊你踩过的类似情况。