a2a（agent2agent 协议）的介绍和使用

fouryears_23417

455人浏览 · 2026-06-17 00:26:34

fouryears_23417 · 2026-06-17 00:26:34 发布

官网地址: https://a2a-protocol.org/latest/
1.0 版本地址：https://a2a-protocol.org/v1.0.0/

Agent2Agent (A2A) 协议是由 Google 提出的开放标准，旨在让不同 AI Agent 之间能够互联互通、相互协作。

A2A 协议解决的问题

假设用户请求 AI 助手帮忙规划一次国际旅行。这个任务需要协调多个专业化的 Agent，例如：

航班预订 Agent
酒店预订 Agent
当地旅游推荐 Agent
汇率转换 Agent

在没有 A2A 协议的情况下，整合这些不同的 Agent 会面临以下挑战：

Agent 暴露问题：开发者通常会把 Agent 包装成工具的形式来暴露给其他 Agent，类似于通过 Model Context Protocol (MCP) 暴露工具的方式。但这种做法效率低下，因为 Agent 本身是设计为可以直接协商的。将 Agent 包装成工具会限制其能力。A2A 协议允许 Agent 以原始形态直接暴露，无需这种包装。

定制化集成： 每次交互都需要定制的、点对点的解决方案，造成巨大的工程开销。

创新缓慢： 为每个新集成进行定制开发会拖慢创新速度。

可扩展性问题： 随着 Agent 数量和交互复杂度的增长，系统变得难以扩展和维护。

互操作性受限： 这种做法限制了互操作性，阻碍了复杂 AI 生态系统的有机形成。

安全漏洞：临时拼凑的通信方式通常缺乏一致的安全措施。

A2A 协议通过为 AI Agent 之间建立互操作性标准，使它们能够可靠且安全地进行交互，从而解决了上述所有挑战。

如图：
在这里插入图片描述

版本说明

目前公开出来的有两个版本一个是 0.3版本一个是1.0版本，1.0版本已经是稳定版本，所以教程以此版本为准，语言以 ts 为主，ts 仓库：here 目前 ts 的版本 1.0 为还没合入主分支，主仓库：here

从简单示例入手

服务端 demo 链接：here

文件结构

src/samples/agents/sample-agent/
├── index.ts          # 服务启动入口，定义 AgentCard，挂载 Express 路由
├── agent_executor.ts # 核心业务逻辑，实现 AgentExecutor 接口
└── README.md

1. `index.ts` — 服务启动与 AgentCard

AgentCard（`sampleAgentCard`）

AgentCard 是 A2A 协议中的代理名片，用于向外界描述本 agent 的能力。重要字段：

字段	说明
`name` / `description`	代理的名称和描述
`supportedInterfaces`	关键字段：声明 agent 暴露的接口地址、协议绑定类型（`JSONRPC`）、协议版本
`provider`	提供方信息（组织名、URL）
`version`	agent 版本号
`capabilities`	关键字段：声明能力——`streaming: true` 支持 SSE 流式返回；`pushNotifications: false` 不支持服务端推送；`extendedAgentCard: false` 不提供扩展名片
`defaultInputModes` / `defaultOutputModes`	默认输入输出模态，如 `['text']`
`skills`	关键字段：声明 agent 支持的技能列表，每个 skill 有独立的 `id`、`name`、`description`、`tags`、`examples`、`inputModes`、`outputModes`
`securitySchemes` / `securityRequirements`	安全认证配置（本 sample 为空，无认证）

启动流程（`main()`）

InMemoryTaskStore  →  SampleAgentExecutor  →  DefaultRequestHandler  →  Express 路由

InMemoryTaskStore：任务存储，内存中保存 task 状态
SampleAgentExecutor：你实现的核心逻辑类
DefaultRequestHandler：SDK 提供的请求处理器，串联上面两者
Express 路由挂载：
- GET /.well-known/agent-card.json → agentCardHandler（返回 AgentCard）
- POST / → jsonRpcHandler（处理 JSON-RPC 请求）
- UserBuilder.noAuthentication：不做身份认证

2. `agent_executor.ts` — 核心业务逻辑

`AgentExecutor` 接口

你必须实现两个方法：

interface AgentExecutor {
  execute(requestContext: RequestContext, eventBus: ExecutionEventBus): Promise<void>;
  cancelTask(taskId: string, eventBus: ExecutionEventBus): Promise<void>;
}

`execute()` 方法 — 重要步骤

Step 1: 发布初始 Task（`TASK_STATE_SUBMITTED`）

const initialTask: Task = {
  id: taskId,
  contextId: contextId,
  status: { state: TaskState.TASK_STATE_SUBMITTED, ... },
  artifacts: [],
  history: [userMessage],   // 历史消息，从当前用户消息开始
  metadata: userMessage.metadata,
};
eventBus.publish(AgentEvent.task(initialTask));

Task：代表一个任务的完整状态，包含 id、contextId、status、artifacts、history
contextId：对话上下文 ID，同一对话的多个 task 共享同一个 contextId
history：消息历史，用于多轮对话

Step 2: 发布 `TASK_STATE_WORKING` 状态更新

const workingStatusUpdate: TaskStatusUpdateEvent = {
  taskId, contextId,
  status: { state: TaskState.TASK_STATE_WORKING, message: { role: 'agent', ... } },
};
eventBus.publish(AgentEvent.statusUpdate(workingStatusUpdate));

TaskStatusUpdateEvent：任务状态更新事件，通过 SSE 推送给客户端
Message：一条消息，包含 role（user / agent）、messageId、parts（内容分片）

Step 3: 模拟处理 + 取消检查

await new Promise((resolve) => setTimeout(resolve, 1000)); // 模拟耗时
if (this.cancelledTasks.has(taskId)) {
  eventBus.publish(AgentEvent.statusUpdate({ state: TaskState.TASK_STATE_CANCELED }));
  return;
}

取消机制：cancelTask() 被调用时把 taskId 记入 cancelledTasks 集合，execute() 在每个 yield 点检查并中止

Step 4: 发布 Artifact（结果产物）

const resultArtifact: Artifact = {
  artifactId: uuidv4(),
  name: 'Result',
  parts: [{ content: { $case: 'text', value: agentReplyText }, ... }],
};
const artifactUpdate: TaskArtifactUpdateEvent = {
  taskId, contextId,
  artifact: resultArtifact,
  lastChunk: true,   // 是否为最后一个 chunk（流式场景下可分批发送）
  append: false,      // 是否追加到已有 artifact（而非覆盖）
};
eventBus.publish(AgentEvent.artifactUpdate(artifactUpdate));

Artifact：agent 产生的输出产物（如文本、文件等），由 parts 组成
parts：内容分片，每种内容类型用 $case 区分（如 text、file、data）

Step 5: 发布最终状态（`TASK_STATE_COMPLETED`）

eventBus.publish(AgentEvent.statusUpdate({
  status: { state: TaskState.TASK_STATE_COMPLETED },
}));

3. 关键类型总结

类型	作用
`Task`	任务实体，贯穿整个生命周期
`TaskState`	任务状态枚举：`TASK_STATE_SUBMITTED` → `TASK_STATE_WORKING` → `TASK_STATE_COMPLETED` / `TASK_STATE_FAILED` / `TASK_STATE_CANCELLED`
`Message`	一条消息（`user` 或 `agent` 角色），由 `parts` 组成
`Artifact`	agent 输出的产物，由 `parts` 组成
`TaskStatusUpdateEvent`	状态变更事件，推送给客户端
`TaskArtifactUpdateEvent`	产物更新事件，推送给客户端（支持流式 chunk）
`ExecutionEventBus`	事件总线，`publish()` 方法把事件推给请求处理器，再由处理器通过 SSE 返回给客户端
`RequestContext`	请求上下文，包含 `userMessage`、`taskId`、`contextId`、`task`（已有任务，若为续聊）

4. 消息流示意

Client 发起请求
  → DefaultRequestHandler 接收
    → SampleAgentExecutor.execute() 被调用
      → 发布 Task(TASK_STATE_SUBMITTED)       ← SSE 推给客户端
      → 发布 StatusUpdate(TASK_STATE_WORKING)  ← SSE 推给客户端
      → 执行业务逻辑
      → 发布 ArtifactUpdate          ← SSE 推给客户端
      → 发布 StatusUpdate(TASK_STATE_COMPLETED) ← SSE 推给客户端

客户端如何发起请求

文件链接：here

1. 客户端初始化

1.1 解析命令行参数

const preferredTransport = process.argv
  .find((arg) => arg.startsWith('--transport='))
  ?.split('=')[1];
const serverUrlArg = process.argv.slice(2).find((arg) => !arg.startsWith('--'));
const serverUrl = serverUrlArg || 'http://localhost:41241';

参数	说明
`--transport=xxx`	指定优先使用的传输协议，如 `JSONRPC`、`REST`、`GRPC`
`--agent-engine`	使用 Google ADC 认证（用于 Agent Engine 部署的 agent）
位置参数	agent 的 base URL，默认 `http://localhost:41241`

1.2 创建 `ClientFactory`

const factory = new ClientFactory(
  ClientFactoryOptions.createFrom(ClientFactoryOptions.default, {
    cardResolver: new DefaultAgentCardResolver({ fetchImpl }),
    transports: [
      new JsonRpcTransportFactory({ fetchImpl }),
      new RestTransportFactory({ fetchImpl }),
      new GrpcTransportFactory(),
    ],
    preferredTransports: preferredTransport ? [preferredTransport] : undefined,
  })
);

ClientFactory 是客户端的核心工厂，负责：

cardResolver：用于从 agent 的 base URL 获取 AgentCard（默认请求 /.well-known/agent-card.json）
transports：注册支持的传输协议工厂（JSON-RPC、REST、gRPC）
preferredTransports：优先选择的传输协议，会覆盖 AgentCard 中的声明

1.3 创建 `Client`

const client = await factory.createFromUrl(serverUrl, agentCardPath);

createFromUrl 的内部流程：

createFromUrl(serverUrl)
  → cardResolver.resolve(serverUrl)       // 1. 获取 AgentCard
  → 从 AgentCard.supportedInterfaces      // 2. 选择匹配的传输协议
     中选择优先级最高的 interface
  → factory.create(transportUrl, agentCard) // 3. 创建 Transport 实例
  → new Client(transport, agentCard)        // 4. 创建 Client

Client 创建后会自动与 agent 建立连接，后续所有请求都通过选定的 Transport 发送。

2. 构造并发送消息

2.1 用户输入与 `Message` 构造

用户在终端输入文字后，CLI 构造一个 Message 对象：

const messageId = generateId(); // crypto.randomUUID()

const messagePayload: Message = {
  messageId: messageId,
  role: Role.ROLE_USER,           // 消息发送者角色：USER
  parts: [
    {
      content: {
        $case: 'text',            // 内容类型：纯文本
        value: input,              // 用户输入的文本
      },
      metadata: undefined,
      filename: '',
      mediaType: 'text/plain',
    },
  ],
  taskId: '',                     // 新任务为空；续聊时填入已有 taskId
  contextId: '',                  // 新会话为空；续聊时填入已有 contextId
  extensions: [],
  metadata: {},
  referenceTaskIds: [],
};

关键字段说明

字段	说明
`messageId`	每条消息的唯一 ID，客户端生成
`role`	`ROLE_USER`（用户消息）或 `ROLE_AGENT`（agent 消息）
`parts`	消息内容分片，支持多种 `$case`：`text`、`url`、`raw`、`data`
`taskId`	关联的 task ID。首次发送为空，agent 会在响应中返回新 task 的 ID；续聊时填入此 ID
`contextId`	对话上下文 ID。首次发送为空，同一对话的多个 task 共享同一个 `contextId`
`metadata`	附加元数据，可携带自定义信息

2.2 续聊：`taskId` / `contextId` 的传递

CLI 维护两个状态变量：

let currentTaskId: string | undefined = undefined;
let currentContextId: string | undefined = undefined;

首次发送：currentTaskId 和 currentContextId 均为 undefined，构造的 Message.taskId 和 contextId 为空字符串
agent 响应后：从返回的事件中提取 taskId 和 contextId，更新这两个变量
续聊发送：把 currentTaskId 和 currentContextId 写入 Message，agent 据此关联到已有任务
/new 命令：清空两个变量，开启全新会话

2.3 构造 `SendMessageRequest`

const params: SendMessageRequest = {
  tenant: '',              // 租户标识，多租户场景下使用
  message: messagePayload, // 上面构造的 Message
  configuration: undefined, // 可选：配置 streaming、blocking 等行为
  metadata: {},            // 附加元数据
};

SendMessageRequest 是 sendMessage（非流式）和 sendMessageStream（流式）的请求参数类型。

3. 发送请求与处理流式响应

3.1 发送流式请求

const stream = client.sendMessageStream(params);

for await (const event of stream) {
  // 处理每一个事件
}

sendMessageStream 返回一个异步可迭代对象（AsyncIterable），每个 event 是一个包含 payload 的对象，payload.$case 区分事件类型。

3.2 事件类型与处理

`payload.$case`	类型	说明
`statusUpdate`	`TaskStatusUpdateEvent`	任务状态变更（TASK_STATE_WORKING、TASK_STATE_COMPLETED 等）
`artifactUpdate`	`TaskArtifactUpdateEvent`	agent 产生了新的输出产物
`message`	`Message`	agent 直接返回的消息（非标准任务流）
`task`	`Task`	完整的任务对象（含历史、产物等）

`statusUpdate` 处理

case 'statusUpdate': {
  const typedEvent = payload.value as TaskStatusUpdateEvent;
  printAgentEvent(AgentEvent.statusUpdate(typedEvent));

  // 任务进入终态时，清空 currentTaskId
  if (
    typedEvent.status?.state === TaskState.TASK_STATE_COMPLETED ||
    typedEvent.status?.state === TaskState.TASK_STATE_FAILED ||
    typedEvent.status?.state === TaskState.TASK_STATE_CANCELED ||
    typedEvent.status?.state === TaskState.TASK_STATE_REJECTED
  ) {
    currentTaskId = undefined;
  }
  break;
}

任务终态：TASK_STATE_COMPLETED、TASK_STATE_FAILED、TASK_STATE_CANCELLED、TASK_STATE_REJECTED 均视为终态，CLI 在收到这些状态后清空 currentTaskId，下次发送将开启新任务。

`task` 事件处理（提取 `taskId` / `contextId`）

case 'task': {
  const task = payload.value as Task;
  // 更新 currentTaskId 和 currentContextId
  currentTaskId = task.id;
  currentContextId = task.contextId;
  break;
}

agent 在首次处理消息时会创建新 Task，并通过 task 事件或 statusUpdate 事件返回新 taskId。CLI 提取后用于后续续聊。

4. 完整请求流程图

用户启动 cli.ts
  → ClientFactory 创建
  → factory.createFromUrl(serverUrl)
    → GET /.well-known/agent-card.json  ← 获取 AgentCard
    → 选择传输协议（JSONRPC / REST / GRPC）
    → 创建 Client
  → 打印 AgentCard 信息（name、skills、streaming 支持等）
  → 进入 readline 主循环

用户输入文字（或 /new、/exit）
  → 构造 Message { role: USER, parts: [{ text: input }] }
  → 若为续聊，填入 taskId / contextId
  → client.sendMessageStream(params)
    → 通过 Transport 发送 JSON-RPC / REST / gRPC 请求
    → 返回 AsyncIterable 事件流

  收到 event:
    → statusUpdate: 打印状态（TASK_STATE_WORKING → TASK_STATE_COMPLETED）
    → artifactUpdate: 打印产物内容
    → task: 更新 currentTaskId / currentContextId
    → message: 打印 agent 消息

  → 任务终态（TASK_STATE_COMPLETED 等）: 清空 currentTaskId
  → rl.prompt() 等待下一次输入

5. 关键类型对照

类型	位置	作用
`Message`	`src/index.ts`	一条消息，包含 `role`、`parts`、`taskId`、`contextId`
`Part`	`src/index.ts`	消息/产物的最小内容单元，`$case` 区分类型
`SendMessageRequest`	`src/types/pb/a2a.ts`	`sendMessage` / `sendMessageStream` 的请求参数
`TaskStatusUpdateEvent`	`src/index.ts`	任务状态变更事件
`TaskArtifactUpdateEvent`	`src/index.ts`	产物更新事件
`AgentExecutionEvent`	`src/server/index.ts`	客户端收到的事件联合类型（`statusUpdate` \| `artifactUpdate`）
`Client`	`src/client/multitransport-client.ts`	SDK 客户端，提供 `sendMessage`、`sendMessageStream` 等方法
`ClientFactory`	`src/client/factory.ts`	客户端工厂，根据 AgentCard 创建适配的 `Client`

高级特性

上面我们解析了如何简单的建立请求，那么接下来讲一下高级特性

如何取消请求？

A2A 协议支持客户端主动取消正在执行中的任务（§3.1.5 Cancel Task）。取消请求涉及服务端和客户端两方面的配合。

服务端实现

Demo 位置：here

核心思路

cancelTask(taskId) 被调用
  → 把 taskId 记录到 cancelledTasks 集合（Set）
  → execute() 在每个步骤前检查集合
  → 若发现 taskId 在集合中，发布 CANCELED 状态并 return
  → finally 块中清理集合（防止内存泄漏）

关键代码

1. 实现 cancelTask 方法（记录取消标记）：

private readonly cancelledTasks = new Set<string>();

public cancelTask = async (taskId: string, _eventBus: ExecutionEventBus): Promise<void> => {
  console.log(`[CancellableAgentExecutor] Cancellation requested for task ${taskId}`);
  this.cancelledTasks.add(taskId);
};

2. 在 execute() 中轮询检查（在每一步耗时操作前检查）：

const totalSteps = 5;
for (let step = 1; step <= totalSteps; step++) {
  // 检查取消标记
  if (this.cancelledTasks.has(taskId)) {
    console.log(`[CancellableAgentExecutor] Aborting task ${taskId} at step ${step}.`);
    // 发布 CANCELED 状态更新（附上带原因的消息）
    const cancelledUpdate: TaskStatusUpdateEvent = {
      taskId,
      contextId,
      status: {
        state: TaskState.TASK_STATE_CANCELED,
        timestamp: new Date().toISOString(),
        message: {
          role: Role.ROLE_AGENT,
          messageId: uuidv4(),
          parts: [
            {
              content: { $case: 'text', value: `Task cancelled by user at step ${step}/${totalSteps}.` },
              metadata: undefined,
              filename: '',
              mediaType: 'text/plain',
            },
          ],
          taskId,
          contextId,
          extensions: [],
          metadata: {},
          referenceTaskIds: [],
        },
      },
      metadata: {},
    };
    eventBus.publish(AgentEvent.statusUpdate(cancelledUpdate));
    return; // 终止执行
  }
  console.log(`[CancellableAgentExecutor] Task ${taskId}: step ${step}/${totalSteps}`);
  await new Promise((resolve) => setTimeout(resolve, 1000)); // 每步模拟耗时
}

3. finally 块清理（防止内存泄漏，覆盖正常完成、取消、异常三种情况）：

try {
  // ... 执行逻辑
} finally {
  this.cancelledTasks.delete(taskId);
}

关键字段说明

字段 / 方法	说明
`cancelledTasks`（Set）	内存中的取消标记集合，`cancelTask` 写入，`execute` 读取
`cancelTask(taskId, eventBus)`	SDK 在客户端调用 `cancelTask` 时触发，由 `DefaultRequestHandler` 回调
`TASK_STATE_CANCELED`	任务取消后的终态，发布后客户端收到该事件
`finally { delete }`	必须清理，否则 Set 会无限增长；三种退出路径（完成/取消/异常）都能覆盖

客户端实现

Demo 位置：here

核心流程

1. 发起流式请求 client.sendMessageStream(params)
2. 从返回事件中捕获 taskId（首个 task 事件）
3. 用 setTimeout 延迟调用 client.cancelTask({ id: taskId })
4. 继续消费 stream，直到收到 TASK_STATE_CANCELED 事件

关键代码

1. 发起流式请求并捕获 taskId + 安排取消定时器（完整代码）：

const stream = client.sendMessageStream(params);

let taskId: string | undefined;
let cancelTimer: NodeJS.Timeout | undefined;

for await (const event of stream) {
  const payload = event.payload;
  if (!payload) {
    continue;
  }

  switch (payload.$case) {
    case 'task': {
      const task = payload.value;
      taskId = task.id;
      console.log(
        `[Client] Task created id=${task.id} state=${taskStateToJSON(task.status!.state)}`
      );

      // 获取到 taskId 后，延迟 CANCEL_AFTER_MS 毫秒发送取消请求
      cancelTimer = setTimeout(async () => {
        console.log(`[Client] Sending cancelTask for ${taskId} after ${CANCEL_AFTER_MS}ms`);
        try {
          const cancelled = await client.cancelTask({
            tenant: '',
            id: taskId!,
            metadata: {},
          });
          console.log(
            `[Client] cancelTask returned state=${taskStateToJSON(cancelled.status!.state)}`
          );
        } catch (err) {
          console.error(`[Client] cancelTask failed:`, err);
        }
      }, CANCEL_AFTER_MS);
      break;
    }
    case 'statusUpdate': {
      const update = payload.value;
      console.log(
        `[Client] statusUpdate task=${update.taskId} state=${taskStateToJSON(update.status!.state)}`
      );
      if (update.status?.state === TaskState.TASK_STATE_CANCELED) {
        console.log(`[Client] Confirmed task ${update.taskId} was cancelled.`);
      }
      break;
    }
    case 'artifactUpdate': {
      const update = payload.value;
      console.log(
        `[Client] artifactUpdate task=${update.taskId} artifact=${update.artifact?.name}`
      );
      break;
    }
    case 'message':
      console.log(`[Client] Received direct message event.`);
      break;
  }
}

if (cancelTimer) {
  clearTimeout(cancelTimer);
}
console.log('[Client] Stream complete.');

关键点：

从 case 'task' 中获取 taskId 后，用 setTimeout 延迟调用 client.cancelTask()
CANCEL_AFTER_MS 默认为 2500ms，可通过环境变量覆盖
循环结束后用 clearTimeout(cancelTimer) 清理定时器

调用时序图

Client                              Agent（服务端）
  |                                      |
  |-- sendMessageStream(req) ----------->|
  |                                      |-- execute() 开始执行
  |<-- Task(TASK_STATE_SUBMITTED) ---------------|
  |<-- StatusUpdate(TASK_STATE_WORKING) ----------|
  |                                      |
  |   （2.5 秒后）                      |
  |-- cancelTask({ id }) -------------->|
  |                                      |-- cancelTask() 被调用
  |                                      |-- cancelledTasks.add(taskId)
  |                                      |-- execute() 检测到标记
  |<-- StatusUpdate(CANCELED) ---------|
  |                                      |

关键字段说明

字段 / 方法	说明
`client.cancelTask(params)`	客户端取消请求，对应 A2A 协议的 `CancelTask` 操作（JSON-RPC: `CancelTask` / REST: `POST /tasks/{id}:cancel`）
`params.id`	要取消的 task ID，从流式响应中的 `task` 事件获取
`params.tenant`	多租户场景下的租户标识，单租户传空字符串
取消后的行为	服务端发布 `TASK_STATE_CANCELED` 状态事件，流式请求正常结束（不是异常中断）

推送通知

当 agent 执行长时间任务时，客户端不必一直保持连接等待，可以通过推送通知让 agent 把任务进度主动 POST 到指定的 webhook URL。

Demo 位置：here

协议背景

A2A 规范 §3.5.3（Push Notification Delivery）：agent 在每次任务状态变更时，向客户端预先配置的 URL 发送 POST 请求。

服务端配置

1. AgentCard 声明支持推送

const pushNotificationAgentCard: AgentCard = {
  capabilities: {
    pushNotifications: true, // 必须设为 true
    streaming: true,
    // ...
  },
  // ...
};

2. 创建 PushNotification 组件

// 存储客户端发来的 webhook 配置
const pushNotificationStore = new InMemoryPushNotificationStore();

// 发送器：把事件 POST 到配置的 URL
const pushNotificationSender = new DefaultPushNotificationSender(
  pushNotificationStore,
  {
    timeout: 5000,                                 // 每次 POST 的超时
    tokenHeaderName: 'X-A2A-Notification-Token',  // 认证 token 的 header 名
  }
);

3. 注入到 DefaultRequestHandler

const requestHandler = new DefaultRequestHandler(
  pushNotificationAgentCard,
  taskStore,
  agentExecutor,
  undefined,                      // eventBusManager（默认）
  pushNotificationStore,           // ← 关键：注入 store
  pushNotificationSender           // ← 关键：注入 sender
);

原理：agent executor 每次调用 eventBus.publish() 发布事件时，DefaultRequestHandler 会自动通过 pushNotificationSender 把事件 POST 到客户端配置的 webhook URL。

客户端配置

1. 在 `SendMessageRequest` 中配置 webhook

const params: SendMessageRequest = {
  tenant: '',
  message: messagePayload,
  configuration: {
    acceptedOutputModes: ['text/plain'],
    returnImmediately: true, // 立即返回，不等待任务完成
    taskPushNotificationConfig: {
      id: '',          // 由服务端填充
      taskId: '',      // 由服务端填充
      tenant: '',
      url: WEBHOOK_URL,    // ← webhook 地址
      token: WEBHOOK_TOKEN, // ← 认证 token（可选）
      authentication: undefined,
    },
  },
};

关键字段说明

字段	说明
`returnImmediately: true`	非流式调用时立即返回 Task 对象，不等待执行完成（配合推送通知使用）
`taskPushNotificationConfig.url`	webhook URL，agent 会把所有事件 POST 到这里
`taskPushNotificationConfig.token`	认证令牌，agent 发送 POST 时会在 `X-A2A-Notification-Token` header 中带上
`WEBHOOK_URL`	客户端需要提前启动一个 HTTP 服务来接收推送

2. 启动 webhook 接收服务

Demo 位置：here

app.post('/webhook/task-updates', (req, res) => {
  // 1. 验证 token
  const token = req.header('X-A2A-Notification-Token');
  if (token !== EXPECTED_TOKEN) {
    res.status(401).json({ error: 'Unauthorized' });
    return;
  }

  // 2. 解析事件（每次 POST 只包含一种事件）
  const body = req.body;
  if (body.statusUpdate) {
    console.log(`[Webhook] statusUpdate: state=${body.statusUpdate.status.state}`);
  } else if (body.artifactUpdate) {
    console.log(`[Webhook] artifactUpdate: artifact=${body.artifactUpdate.artifact?.name}`);
  } else if (body.task) {
    console.log(`[Webhook] task: id=${body.task.id}`);
  }

  res.status(200).json({ received: true });
});

完整调用流程

Client                            Agent                          Webhook（客户端）
  |                                 |                                 |
  |-- sendMessage(req + config) --->|                                 |
  |<-- Task(TASK_STATE_SUBMITTED) ------------|                                 |
  |                                 |-- eventBus.publish(TASK_STATE_WORKING) -->|--- POST /webhook (TASK_STATE_WORKING)
  |                                 |                                 |
  |                                 |-- eventBus.publish(TASK_STATE_WORKING) -->|--- POST /webhook (TASK_STATE_WORKING)
  |                                 |                                 |
  |                                 |-- eventBus.publish(artifact) -->|--- POST /webhook (artifact)
  |                                 |                                 |
  |                                 |-- eventBus.publish(TASK_STATE_COMPLETED) →|--- POST /webhook (TASK_STATE_COMPLETED)
  |                                 |                                 |

多传输协议

同一个 agent 可以同时暴露多种传输协议（JSON-RPC、REST、gRPC），客户端根据自己的需求选择合适的协议。

Demo 位置：here

AgentCard 声明多个接口

const multiTransportAgentCard: AgentCard = {
  supportedInterfaces: [
    {
      url: `http://localhost:${HTTP_PORT}/a2a/jsonrpc`,
      protocolBinding: 'JSONRPC',
      tenant: '',
      protocolVersion: A2A_PROTOCOL_VERSION,
    },
    {
      url: `http://localhost:${HTTP_PORT}/a2a/rest`,
      protocolBinding: 'HTTP+JSON',
      tenant: '',
      protocolVersion: A2A_PROTOCOL_VERSION,
    },
    {
      url: `localhost:${GRPC_PORT}`,
      protocolBinding: 'GRPC',
      tenant: '',
      protocolVersion: A2A_PROTOCOL_VERSION,
    },
  ],
  // ...
};

ClientFactory 在创建 Client 时，会根据 supportedInterfaces 和本地注册的传输工厂自动选择匹配的协议。

服务端同时启动多个传输处理器

// 共享同一个 requestHandler（业务逻辑只需写一次）
const requestHandler = new DefaultRequestHandler(/* ... */);

// HTTP 服务：同时挂载 JSON-RPC 和 REST 路由
const app = express();
app.use('/a2a/jsonrpc', jsonRpcHandler({ requestHandler, userBuilder }));
app.use('/a2a/rest',    restHandler({ requestHandler, userBuilder }));
app.listen(HTTP_PORT);

// gRPC 服务：独立端口
const grpcServer = new Server();
grpcServer.addService(A2AService, grpcService({ requestHandler, userBuilder }));
grpcServer.bindAsync(`localhost:${GRPC_PORT}`, ServerCredentials.createInsecure(), /* ... */);

客户端指定传输协议

# 使用 JSON-RPC 传输
npx tsx src/samples/cli.ts --transport=JSONRPC http://localhost:41241

# 使用 REST 传输
npx tsx src/samples/cli.ts --transport=REST http://localhost:41241

# 使用 gRPC 传输
npx tsx src/samples/cli.ts --transport=GRPC http://localhost:41241

身份认证

生产环境中 agent 需要认证才能访问。A2A 通过 AgentCard 的 securitySchemes 和 securityRequirements 字段声明认证方式，服务端通过 Express 中间件实施认证。

Demo 位置：here

AgentCard 中声明认证方式

const authenticationAgentCard: AgentCard = {
  securitySchemes: {
    Bearer: {
      scheme: {
        $case: 'httpAuthSecurityScheme',
        value: {
          description: 'Bearer Token',
          scheme: 'bearer',
          bearerFormat: 'JWT',
        },
      },
    },
  },
  securityRequirements: [
    { schemes: { Bearer: { list: [] } } }, // 要求客户端提供 Bearer token
  ],
  // ...
};

服务端：认证中间件 + UserBuilder

// authentication_middleware.ts：Express 中间件，校验 token
function authenticationHandler(req: Request, res: Response, next: NextFunction) {
  const authHeader = req.headers.authorization;
  if (!authHeader?.startsWith('Bearer ')) {
    res.status(401).json({ error: 'Missing or invalid Authorization header' });
    return;
  }
  const token = authHeader.slice(7);
  // 验证 token（示例：简单比对）
  if (token !== 'expected-token') {
    res.status(401).json({ error: 'Invalid token' });
    return;
  }
  next();
}

// user_builder.ts：从请求中提取用户信息，注入到 RequestContext
function userBuilder(req: Request): User {
  return { id: 'authenticated-user', /* ... */ };
}

服务端挂载

const app = express();
app.use(authenticationHandler); // 先认证
app.use(jsonRpcHandler({
  requestHandler,
  userBuilder,                 // 传入 userBuilder
}));

客户端：配置认证 Header

// 使用 ADCHandler（Google Application Default Credentials）
class ADCHandler implements AuthenticationHandler {
  async headers(): Promise<Record<string, string>> {
    const token = await getAccessToken();
    return { Authorization: `Bearer ${token}` };
  }
}

const fetchImpl = createAuthenticatingFetchWithRetry(fetch, new ADCHandler());
const factory = new ClientFactory(/* ... transports with fetchImpl ... */);

客户端拦截器（Interceptor）

CallInterceptor 允许在客户端每次调用前后注入自定义逻辑，例如注入请求 ID、记录耗时、添加认证头等。

Demo 位置：here

实现 CallInterceptor

class RequestIdInterceptor implements CallInterceptor {
  async before(args: BeforeArgs): Promise<void> {
    const requestId = crypto.randomUUID();
    // 通过 serviceParameters 传递（传输无关，HTTP 下会变成 header，gRPC 下会变成 metadata）
    args.options = {
      ...args.options,
      serviceParameters: {
        ...(args.options?.serviceParameters ?? {}),
        'X-Request-ID': requestId,
      },
    };
    console.log(`[RequestIdInterceptor] ${args.input?.method} -> X-Request-ID=${requestId}`);
  }

  async after(_args: AfterArgs): Promise<void> {
    // No-op
  }
}

class LoggingInterceptor implements CallInterceptor {
  async before(args: BeforeArgs): Promise<void> {
    const start = performance.now();
    // 把开始时间存到 serviceParameters 中（before 和 after 都能访问）
    args.options = {
      ...args.options,
      serviceParameters: {
        ...(args.options?.serviceParameters ?? {}),
        'X-Demo-Start-Ms': String(start),
      },
    };
  }

  async after(args: AfterArgs): Promise<void> {
    const start = Number(args.options?.serviceParameters?.['X-Demo-Start-Ms']);
    const elapsed = `${(performance.now() - start).toFixed(1)}ms`;
    console.log(`[LoggingInterceptor] <- ${args.result?.method} (${elapsed})`);
  }
}

注册拦截器

const factory = new ClientFactory(
  ClientFactoryOptions.createFrom(ClientFactoryOptions.default, {
    transports: [new JsonRpcTransportFactory()],
    clientConfig: {
      interceptors: [new LoggingInterceptor(), new RequestIdInterceptor()],
      // 执行顺序：before 按声明顺序，after 按声明逆序
    },
  })
);

单次调用超时控制

// 通过 AbortSignal 实现单次调用超时
try {
  await client.sendMessage(buildMessage('Hello'), {
    signal: AbortSignal.timeout(5000), // 5 秒超时
  });
} catch (err) {
  console.error('Call timed out or aborted:', err);
}

扩展

A2A 扩展机制允许 agent 在协议标准字段之外，通过 extensions 字段声明和传递自定义数据。

Demo 位置：here

AgentCard 声明支持的扩展

const extensionAgentCard: AgentCard = {
  capabilities: {
    extensions: [
      {
        uri: 'https://example.com/extensions/timestamp/v1',
        description: 'Timestamp extension: every event will include a timestamp',
        required: false,   // 客户端可选是否使用该扩展
        params: {},
      },
    ],
    // ...
  },
  // ...
};

服务端：用 Executor 包装模式注入扩展数据

// extensions.ts：TimestampingAgentExecutor 包装了真实的 executor
export class TimestampingAgentExecutor implements AgentExecutor {
  constructor(private readonly delegate: AgentExecutor) {}

  async execute(requestContext: RequestContext, eventBus: ExecutionEventBus): Promise<void> {
    // 包装 eventBus，在每次 publish 之前注入扩展数据
    const wrappedBus: ExecutionEventBus = {
      ...eventBus,
      publish: (event) => {
        // 在 event 的 extensions 字段中注入时间戳
        event.extensions = [
          ...(event.extensions ?? []),
          { uri: EXTENSION_URI, params: { timestamp: new Date().toISOString() } },
        ];
        return eventBus.publish(event);
      },
    };
    // 把包装后的 eventBus 传给真实的 executor
    return this.delegate.execute(requestContext, wrappedBus);
  }

  cancelTask(taskId: string, eventBus: ExecutionEventBus): Promise<void> {
    return this.delegate.cancelTask(taskId, eventBus);
  }
}

客户端：读取扩展数据

扩展数据随事件一起到达，客户端从 event.extensions 中读取：

for await (const event of stream) {
  if (event.payload.$case === 'statusUpdate') {
    const exts = event.extensions ?? [];
    for (const ext of exts) {
      if (ext.uri === 'https://example.com/extensions/timestamp/v1') {
        console.log(`[Client] Event timestamp: ${ext.params.timestamp}`);
      }
    }
  }
}

Agent Card 签名验证

为了防止 Agent Card 在传输过程中被篡改，A2A 支持对 Agent Card 进行 JWS（JSON Web Signature）签名，客户端在获取 Agent Card 时验证签名。

Demo 位置：here

Agent Card 中的 `signatures` 字段

{
  "name": "My Signed Agent",
  "signatures": [
    {
      "protected": "eyJhbGc...",  // base64url 编码的 JOSE Header（含 alg、kid、jku 等）
      "signature": "AbCdEf..."    // base64url 编码的签名值
    }
  ]
}

客户端验证签名

import { verifyAgentCardSignature, canonicalizeAgentCard } from 'a2a-js';

// 1. 定义公钥获取函数（从 jku 端点获取）
async function retrievePublicKey(kid: string, jku?: string): Promise<CryptoKey> {
  const response = await fetch(jku!);
  const keys = await response.json();       // { kid: "PEM-encoded public key" }
  const pem = keys[kid];
  return crypto.subtle.importKey(/* 从 PEM 导入 CryptoKey */);
}

// 2. 创建验证器
const verifier = verifyAgentCardSignature(retrievePublicKey);

// 3. 验证 Agent Card
const agentCard = await fetchAgentCard();
try {
  await verifier(agentCard);
  console.log('PASS: Agent card signature is valid!');
} catch (err) {
  console.error('FAIL: Signature verification failed:', err);
}

// 4. 篡改检测（改为错误 name 后应验证失败）
const tamperedCard = { ...agentCard, name: 'TAMPERED Agent Name' };
try {
  await verifier(tamperedCard);
  console.error('FAIL: tampered card was accepted!');
} catch {
  console.log('PASS: tampered card was correctly rejected.');
}

规范化

签名验证前，Agent Card 会被规范化为确定性的 JSON 字符串（字段按固定顺序排序），确保不同实现产生的签名输入一致：

const canonical = canonicalizeAgentCard(cardWithoutSignatures);
// 多次对同一个 card 调用，结果必须一致（确定性）

通过 SDK 客户端自动验证

const client = await factory.createFromUrl(serverUrl);
// getAgentCard 时传入 verifier，自动验证签名
const verifiedCard = await client.getAgentCard(undefined, verifier);
console.log(`Verified card: ${verifiedCard.name}`);

综合示例：Movie Agent

前面的 demo 都是最小化示例，Movie Agent 是一个更接近真实场景的综合示例：接入 LLM（Google Genkit）、调用外部 API（TMDB）、支持多轮对话。

Demo 位置：here

功能概览

功能	实现方式
理解自然语言	Google Genkit + Gemini Pro
查询电影数据	TMDB API（`tools.ts` 中定义 `searchMovies`、`searchPeople`）
多轮对话	通过 `contextId` 关联同一对话的历史消息
任务状态管理	标准 `AgentExecutor` 模式（`TASK_STATE_SUBMITTED` → `TASK_STATE_WORKING` → `TASK_STATE_COMPLETED`）
取消支持	`cancelledTasks` 集合 + 轮询检查

核心代码片段

接入 Genkit prompt + 工具调用：

const response = await movieAgentPrompt(
  { goal: goal, now: new Date().toISOString() },
  {
    messages,              // 对话历史（格式化为 Genkit MessageData）
    tools: [searchMovies, searchPeople],  // ← 注册外部工具
  }
);

多轮对话：维护 context 历史：

// 每个 contextId 对应一个对话上下文
const historyForGenkit = contexts.get(contextId) || [];
historyForGenkit.push(userMessage);
contexts.set(contextId, historyForGenkit);

// agent 回复后也追加到历史
historyForGenkit.push(agentMessage);

运行方式

# 设置环境变量
export GEMINI_API_KEY=your_gemini_key
export TMDB_API_KEY=your_tmdb_key

# 启动 Movie Agent
npm run agents:movie-agent

# 使用 CLI 客户端交互
npx tsx src/samples/cli.ts http://localhost:41241
# 然后输入： "Tell me about the plot of Inception."

总结

特性	关键类型 / 方法	Demo
基础请求-响应	`AgentExecutor`、`Client`、`Message`	`sample-agent`、`cli.ts`
取消请求	`cancelTask()`、`client.cancelTask()`	`cancellable-agent`
推送通知	`taskPushNotificationConfig`、`DefaultPushNotificationSender`	`push-notification-agent`
多传输协议	`supportedInterfaces`、`ClientFactory`	`multi-transport-agent`
身份认证	`securitySchemes`、`AuthenticationHandler`	`authentication`
客户端拦截器	`CallInterceptor`、`serviceParameters`	`client/interceptors`
扩展	`capabilities.extensions`、Executor 包装模式	`extensions`
Agent Card 签名	`signatures`、`verifyAgentCardSignature()`	`verify-signing`
综合示例	Genkit + 外部工具 + 多轮对话	`movie-agent`

MCP技术社区

欢迎加入 MCP 技术社区！与志同道合者携手前行，一同解锁 MCP 技术的无限可能！

更多推荐

【Bug已解决】Workspace path with spaces / Special characters in path — Claude Code 路径含空格或特殊字符解决方案

MCP技术社区

【Bug已解决】Memory leak / High memory usage / Process killed — Claude Code 内存泄漏解决方案

MCP技术社区

MCP协议、熔断器与OAuth——Agent接入外部世界的工程方案

MCP技术社区

所有评论(0)

查看更多评论

fouryears_23417

@fouryears_23417

已为社区贡献2条内容

a2a（agent2agent 协议）的介绍和使用

fouryears_23417

A2A 协议解决的问题

版本说明

从简单示例入手

文件结构

1. index.ts — 服务启动与 AgentCard

AgentCard（sampleAgentCard）

启动流程（main()）

2. agent_executor.ts — 核心业务逻辑

AgentExecutor 接口

execute() 方法 — 重要步骤

Step 1: 发布初始 Task（TASK_STATE_SUBMITTED）

Step 2: 发布 TASK_STATE_WORKING 状态更新

Step 3: 模拟处理 + 取消检查

Step 4: 发布 Artifact（结果产物）

Step 5: 发布最终状态（TASK_STATE_COMPLETED）

3. 关键类型总结

4. 消息流示意

客户端如何发起请求

1. 客户端初始化

1.1 解析命令行参数

1.2 创建 ClientFactory

1.3 创建 Client

2. 构造并发送消息

2.1 用户输入与 Message 构造

关键字段说明

2.2 续聊：taskId / contextId 的传递

2.3 构造 SendMessageRequest

3. 发送请求与处理流式响应

3.1 发送流式请求

3.2 事件类型与处理

statusUpdate 处理

task 事件处理（提取 taskId / contextId）

4. 完整请求流程图

5. 关键类型对照

高级特性

如何取消请求？

服务端实现

核心思路

关键代码

关键字段说明

客户端实现

核心流程

关键代码

调用时序图

关键字段说明

推送通知

协议背景

服务端配置

1. AgentCard 声明支持推送

2. 创建 PushNotification 组件

3. 注入到 DefaultRequestHandler

客户端配置

1. 在 SendMessageRequest 中配置 webhook

关键字段说明

2. 启动 webhook 接收服务

完整调用流程

多传输协议

AgentCard 声明多个接口

服务端同时启动多个传输处理器

客户端指定传输协议

身份认证

AgentCard 中声明认证方式

服务端：认证中间件 + UserBuilder

服务端挂载

客户端：配置认证 Header

客户端拦截器（Interceptor）

实现 CallInterceptor

注册拦截器

单次调用超时控制

扩展

AgentCard 声明支持的扩展

服务端：用 Executor 包装模式注入扩展数据

客户端：读取扩展数据

Agent Card 签名验证

Agent Card 中的 signatures 字段

客户端验证签名

规范化

通过 SDK 客户端自动验证

1. `index.ts` — 服务启动与 AgentCard

AgentCard（`sampleAgentCard`）

启动流程（`main()`）

2. `agent_executor.ts` — 核心业务逻辑

`AgentExecutor` 接口

`execute()` 方法 — 重要步骤

Step 1: 发布初始 Task（`TASK_STATE_SUBMITTED`）

Step 2: 发布 `TASK_STATE_WORKING` 状态更新

Step 5: 发布最终状态（`TASK_STATE_COMPLETED`）

1.2 创建 `ClientFactory`

1.3 创建 `Client`

2.1 用户输入与 `Message` 构造

2.2 续聊：`taskId` / `contextId` 的传递

2.3 构造 `SendMessageRequest`

`statusUpdate` 处理

`task` 事件处理（提取 `taskId` / `contextId`）

1. 在 `SendMessageRequest` 中配置 webhook

Agent Card 中的 `signatures` 字段