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

官网地址: 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 的能力。重要字段：

启动流程（main()）

InMemoryTaskStore → SampleAgentExecutor → DefaultRequestHandler → Express 路由

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. 关键类型总结

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';

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 是客户端的核心工厂，负责：

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: [],
};

关键字段说明

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 事件类型与处理

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. 关键类型对照

高级特性

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

如何取消请求？

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);
}

关键字段说明

客户端实现

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) ———|
| |

关键字段说明

推送通知

当 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,
},
},
};

关键字段说明

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

功能概览

核心代码片段

接入 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."

总结