【实践】实现 A2A通信：让 AI Agents 互相协作

互联网运维杂谈

206人浏览 · 2025-12-23 07:28:32

互联网运维杂谈 · 2025-12-23 07:28:32 发布

我的初衷：这是一个AI原生能力，可以给不同部门去调用使用。也让我联想到RCA故障根因分析，我把故障上下文通过A2A协议传递给研发的Agent，研发侧进行直接代码定位，此时无需开放代码给运维，运维这等于把系统变成了一个白盒。

从零到一实现 A2A 协议，让 DevOps Architect Agent 成为可远程调用的智能服务

引言

想象这样一个场景：你的交付管理 Agent 需要设计一个 CI/CD 流水线，它不需要自己精通 DevOps，只需要"问一问"团队里的 DevOps Architect Agent 就能得到专业的架构建议。这不是科幻，而是 Agent-to-Agent (A2A) 协议让 AI Agents 之间互相协作的真实能力。

本文将带你从零开始，实现一个完整的 A2A 协议集成，让你的 AI Agent 变成可以被其他 Agent 远程调用的"微服务"。

什么是 A2A 协议？

A2A (Agent-to-Agent) 协议是一种让 AI Agents 之间进行标准化通信的协议规范。类似于 HTTP 协议让不同的 Web 服务互相调用，A2A 协议让不同的 AI Agent 能够：

发现彼此：通过标准化的 Agent Card 描述能力
互相调用：通过 JSON-RPC 或 REST 接口传递消息
协同工作：组合多个 Agent 的专业能力完成复杂任务

为什么需要 A2A？

传统方式的局限

在没有 A2A 之前，AI Agent 的能力是"孤岛式"的：

交付管理 Agent → 用户
DevOps Architect Agent → 用户  
前端专家 Agent → 用户

每个 Agent 只能单独服务用户，无法互相协作。

A2A 带来的变革

有了 A2A 协议，Agent 之间可以组网协作：

用户 → 交付管理 Agent → DevOps Architect Agent
                      → 前端专家 Agent
                      → 测试专家 Agent

交付管理 Agent 可以编排其他专家 Agent，形成"AI 团队"。

实战案例：DevOps Architect Agent

让我们以一个真实的 Agent 为例，展示如何实现完整的 A2A 集成。

Agent 基本信息

DevOps Architect Agent 是一个专注于基础设施和部署流程的专家 Agent：

名称: devops-architect
版本:1.1.0
描述:自动化基础设施和部署流程，专注可靠性和可观测性
技能:
-CI/CD流水线设计
-Kubernetes架构
-容器化方案
-监控和告警
-基础设施即代码

技术架构

A2A 协议的完整实现包括三个核心组件：

┌─────────────────────────────────────────────┐
│           A2A Protocol Stack                │
├─────────────────────────────────────────────┤
│  1. Agent Card (Agent 能力描述)            │
│  2. Discovery Endpoint (Agent 发现)        │
│  3. JSON-RPC Endpoint (Agent 调用)         │
└─────────────────────────────────────────────┘

实现步骤

步骤 1: 生成 Agent Card

Agent Card 是 Agent 的"名片"，描述了它的能力和调用方式。

// src/server/a2a/agent-card-generator.ts

exportclass AgentCardGenerator {
async generateAgentCard(
    agentId: string,
    workspaceId: string
  ): Promise<AgentCard> {
    // 1. 从数据库获取 Agent 信息
    const agent = await db.agent.findUnique({
      where: { id: agentId },
      include: {
        versions: {
          where: { isPublished: true },
          orderBy: { createdAt: "desc" },
          take: 1,
        },
        persona: true,
      },
    });

    // 2. 构造 Agent Card
    return {
      // 基本信息
      name: agent.name,
      description: agent.description,
      version: agent.versions[0].version,
      
      // 能力声明
      capabilities: {
        streaming: true,          // 支持流式响应
        pushNotifications: false, // 不支持推送通知
        stateHistory: true,       // 支持状态历史
      },
      
      // 技能列表
      skills: agent.tags.map(tag => ({
        name: tag,
        description: `${tag} 相关能力`,
      })),
      
      // 调用端点
      url: `${baseUrl}/api/a2a/agent/${agentId}/jsonrpc?workspaceId=${workspaceId}`,
      preferredTransport: "JSONRPC",
      
      // 支持的输入输出格式
      defaultInputModes: ["text/plain"],
      defaultOutputModes: ["text/plain"],
    };
  }
}

生成的 Agent Card 示例：

{
  "name": "devops-architect",
"description": "Automate infrastructure and deployment processes with focus on reliability and observability",
"version": "1.1.0",
"capabilities": {
    "streaming": true,
    "pushNotifications": false,
    "stateHistory": true
  },
"skills": [
    { "name": "engineering", "description": "工程实践" },
    { "name": "superclaude", "description": "SuperClaude 框架" },
    { "name": "devops", "description": "DevOps 实践" }
  ],
"url": "http://localhost:3000/api/a2a/agent/cmj3xjkx7000dbqqqcym47w6c/jsonrpc?workspaceId=ws-tars-default",
"preferredTransport": "JSONRPC",
"defaultInputModes": ["text/plain"],
"defaultOutputModes": ["text/plain"]
}

步骤 2: 实现 Discovery 端点

按照 A2A 协议规范，Agent Card 应该通过 .well-known/agent-card.json 端点公开。

挑战：Next.js 不支持以 . 开头的路由！

解决方案：使用 well-known/agent-card.json（无点），并在 SDK 调用时指定自定义路径。

// src/app/api/a2a/agent/[agentId]/well-known/agent-card.json/route.ts

exportasyncfunction GET(
  request: NextRequest,
  { params }: { params: Promise<{ agentId: string }> }
) {
const { agentId } = await params;
const session = await auth();

// 1. 获取 workspaceId（支持可选的查询参数）
let workspaceId = request.nextUrl.searchParams.get("workspaceId");

if (!workspaceId) {
    // 从 Agent 记录中获取
    const agent = await db.agent.findUnique({
      where: { id: agentId },
      select: { workspaceId: true },
    });
    workspaceId = agent?.workspaceId;
  }

// 2. 验证访问权限（可选认证）
if (session?.user?.id) {
    const membership = await db.membership.findUnique({
      where: {
        userId_workspaceId: {
          userId: session.user.id,
          workspaceId: workspaceId!,
        },
      },
    });

    if (!membership) {
      return NextResponse.json(
        { error: "Access denied" },
        { status: 403 }
      );
    }
  }

// 3. 生成并返回 Agent Card
const generator = new AgentCardGenerator();
const card = await generator.generateAgentCard(agentId, workspaceId!);

return NextResponse.json(card, {
    headers: {
      "Content-Type": "application/json",
      "Access-Control-Allow-Origin": "*",
      "Cache-Control": "public, max-age=300",
    },
  });
}

访问示例：

# 公开访问（无需认证）
curl http://localhost:3000/api/a2a/agent/cmj3xjkx7000dbqqqcym47w6c/well-known/agent-card.json

# 带 workspaceId 参数
curl "http://localhost:3000/api/a2a/agent/cmj3xjkx7000dbqqqcym47w6c/well-known/agent-card.json?workspaceId=ws-tars-default"

步骤 3: 实现 JSON-RPC 端点

JSON-RPC 端点负责处理 Agent 之间的实际消息传递。

// src/app/api/a2a/agent/[agentId]/jsonrpc/route.ts

exportasyncfunction POST(
  request: NextRequest,
  { params }: { params: Promise<{ agentId: string }> }
) {
const { agentId } = await params;
const session = await auth();

// 1. 获取 workspaceId 和 userId
let workspaceId = request.nextUrl.searchParams.get("workspaceId");
let userId = session?.user?.id;

if (!workspaceId) {
    const agent = await db.agent.findUnique({
      where: { id: agentId },
      select: { workspaceId: true },
    });
    workspaceId = agent?.workspaceId;
  }

// 2. 如果未认证，使用系统用户（支持 A2A 协议）
if (!userId) {
    const defaultMembership = await db.membership.findFirst({
      where: {
        workspaceId: workspaceId!,
        role: "ADMIN",
      },
      select: { userId: true },
    });
    userId = defaultMembership?.userId;
  }

// 3. 创建 A2A Handler
const handlerFactory = getA2AHandlerFactory();
const handler = await handlerFactory.getHandler(
    agentId,
    workspaceId!,
    userId!
  );

// 4. 处理 JSON-RPC 请求
const body = await request.json();
const transportHandler = new JsonRpcTransportHandler(handler);
const context = new ServerCallContext();

const response = await transportHandler.handle(body, context);

return NextResponse.json(response);
}

步骤 4: 实现 Agent Executor

Agent Executor 是连接 A2A 协议和实际 Agent 执行逻辑的桥梁。

// src/server/a2a/tars-agent-executor.ts

exportclass TarsA2AAgentExecutor implements AgentExecutor {
constructor(
    private agentId: string,
    private agentVersionId: string,
    private workspaceId: string,
    private userId: string
  ) {}

async execute(
    requestContext: RequestContext,
    eventBus: ExecutionEventBus
  ): Promise<void> {
    const { taskId, contextId, userMessage } = requestContext;

    // 1. 提取消息内容
    const messageText =
      userMessage?.parts
        ?.filter((part) => part.kind === "text")
        .map((part) => part.text)
        .join("\n") || "";

    // 2. 创建 Agent 执行请求
    const tarsRequest: RunRequest = {
      agentId: this.agentId,
      workspaceId: this.workspaceId,
      userId: this.userId,
      targetType: "agent",
      targetVersionId: this.agentVersionId,
      threadId: contextId,
      input: messageText,
      options: { stream: true },
    };

    // 3. 执行 Agent 并流式返回结果
    let fullResponse = "";
    
    forawait (const event of this.tarsExecutor.execute(tarsRequest)) {
      if (event.type === "TEXT_DELTA") {
        fullResponse += event.delta;
        
        // 发布增量更新
        const responseMessage: Message = {
          kind: "message",
          messageId: crypto.randomUUID(),
          role: "agent",
          parts: [{ kind: "text", text: event.delta }],
          contextId: contextId,
        };

        eventBus.publish({
          kind: "message",
          message: responseMessage,
          final: false,
        });
      } elseif (event.type === "STREAM_END") {
        // 发布最终结果
        eventBus.publish({
          kind: "result",
          result: {
            kind: "message",
            message: {
              kind: "message",
              messageId: crypto.randomUUID(),
              role: "agent",
              parts: [{ kind: "text", text: fullResponse }],
              contextId: contextId,
            },
          },
        });
      }
    }
  }
}

实际调用演示

方法 1: 通过 Web UI 调用

访问 A2A 集成页面
```
http://localhost:3000/dashboard/a2a
```
选择"远程调用"标签

输入 Agent URL

http://localhost:3000/api/a2a/agent/cmj3xjkx7000dbqqqcym47w6c/jsonrpc?workspaceId=ws-tars-default

输入任务描述

帮我设计一个 Node.js 微服务的 CI/CD 流水线，要求：
- GitLab CI 自动触发
- Docker 镜像构建
- 自动部署到 Kubernetes 测试环境
- 生产环境需要手动审批

查看 Agent 响应

DevOps Architect Agent 会返回详细的 CI/CD 流水线设计方案。

方法 2: 通过代码调用

import { ClientFactory } from"@a2a-js/sdk/client";

asyncfunction callDevOpsArchitect() {
// 1. 创建 A2A Client
const factory = new ClientFactory();
const client = await factory.createFromUrl(
    "http://localhost:3000/api/a2a/agent/cmj3xjkx7000dbqqqcym47w6c/jsonrpc?workspaceId=ws-tars-default",
    "well-known/agent-card.json"// 自定义 Agent Card 路径
  );

// 2. 发送消息
const response = await client.sendMessage({
    message: {
      messageId: crypto.randomUUID(),
      role: "user",
      parts: [{
        kind: "text",
        text: "帮我设计一个高可用的 Kubernetes 集群架构"
      }],
      kind: "message",
    },
  });

// 3. 处理响应
const result = response.message?.parts
    ?.filter(part => part.kind === "text")
    .map(part => part.text)
    .join("\n");

console.log("DevOps Architect 建议：", result);
}

方法 3: Agent-to-Agent 编排

更强大的是，你可以让一个 Agent 调用另一个 Agent：

// 在交付管理 Agent 的代码中

asyncfunction designCICDPipeline(projectInfo: string) {
// 调用 DevOps Architect Agent 获取专业建议
const devopsAdvice = await callA2AAgent(
    "devops-architect",
    `请为以下项目设计 CI/CD 流水线：${projectInfo}`
  );

// 调用前端专家 Agent 获取前端构建方案
const frontendBuild = await callA2AAgent(
    "frontend-expert",
    "Vue3 + Vite 项目的最佳构建配置是什么？"
  );

// 综合两个专家的建议，生成最终方案
return synthesizeAdvice(devopsAdvice, frontendBuild);
}

技术难点与解决方案

在实现过程中，我们遇到并解决了几个关键技术难点：

难点 1: Next.js 路由限制

问题：Next.js App Router 不支持以 . 开头的目录名，但 A2A 协议规定使用 .well-known/agent-card.json。

解决方案：

使用 well-known/agent-card.json（无点）
SDK 调用时指定自定义路径："well-known/agent-card.json"
符合 A2A 协议精神，同时兼容 Next.js 限制

难点 2: 查询参数丢失

问题：A2A SDK 在构造 Agent Card URL 时，会丢失原始 URL 中的查询参数。

解决方案：

Discovery 端点自动从 Agent 记录获取 workspaceId
同时支持可选的 workspaceId 查询参数
向后兼容标准 A2A 协议调用

难点 3: 认证与授权

问题：Agent-to-Agent 调用时，A2A SDK 不会携带认证信息。

解决方案：

实现可选认证机制
未认证请求使用工作空间的 ADMIN 用户
既支持已认证调用，也支持标准 A2A 协议调用

性能优化

1. Agent Card 缓存

// 设置 5 分钟缓存
headers: {
  "Cache-Control": "public, max-age=300"
}

2. 连接池复用

export class A2AClientManager {
private clients = new Map<string, Client>();

async getClient(agentUrl: string): Promise<Client> {
    if (!this.clients.has(agentUrl)) {
      const factory = new ClientFactory();
      const client = await factory.createFromUrl(agentUrl);
      this.clients.set(agentUrl, client);
    }
    returnthis.clients.get(agentUrl)!;
  }
}

3. 流式响应

支持流式响应，减少首字节时间（TTFB）：

configuration: {
  blocking: false,  // 非阻塞模式
  acceptedOutputModes: ["text/plain"]
}

实际应用场景

场景 1: 多专家协作

交付项目时，项目经理 Agent 可以：

调用 DevOps Architect Agent 设计 CI/CD 流水线
调用前端专家 Agent 确定构建方案
调用测试专家 Agent 设计测试策略
综合所有建议，生成完整的交付方案

场景 2: 知识共享

不同组织的 Agent 可以互相调用，分享专业知识：

公司 A 的数据库优化 Agent
公司 B 的安全审计 Agent
公司 C 的性能测试 Agent

场景 3: Agent 市场

就像 API 市场一样，可以构建 Agent 市场：

开发者发布专业 Agent
其他开发者通过 A2A 协议调用
按调用次数计费

监控与调试

日志记录

const logger = new Logger("a2a-jsonrpc-api", "access");

logger.info("JSON-RPC request received", {
  requestId,
  agentId,
  userId,
  workspaceId,
});

错误追踪

logger.error("A2A execution error", {
  taskId,
  error: error.message,
  stack: error.stack,
});

性能监控

const startTime = Date.now();
await client.sendMessage(params);
const duration = Date.now() - startTime;

logger.info("A2A call completed", {
  duration,
  agentUrl,
});

未来展望

1. 多模态支持

未来可以扩展支持：

图片输入/输出
文件传输
结构化数据

parts: [
  { kind: "text", text: "分析这张架构图" },
  { kind: "image", url: "https://..." }
]

2. Agent 编排引擎

实现声明式的 Agent 编排：

workflow:
  name:"完整交付流程"
steps:
    -agent:devops-architect
      input:"设计 CI/CD 流水线"
      output:cicd_design
    
    -agent:frontend-expert
      input:"基于 {{cicd_design}} 优化前端构建"
      output:frontend_build
    
    -agent:project-manager
      input:"综合 {{cicd_design}} 和 {{frontend_build}} 生成交付方案"