CopilotKit 现已支持 AngularAngular 支持 · 使用 Angular 构建智能体应用和生成式 UI· 现已推出

在本指南中，你将学习如何使用 A2A Protocol、AG-UI Protocol 和 CopilotKit，在来自不同 AI 智能体框架的 AI 智能体之间构建全栈 Agent-to-Agent（A2A）通信。

在正式开始之前，本文将涵盖以下内容：

什么是 A2A Protocol？

使用 CLI 设置 A2A 多智能体通信

通过 A2A Protocol 集成来自不同智能体框架的 AI 智能体

使用 CopilotKit 为 AG-UI 和 A2A 多智能体通信构建前端

以下是我们将要构建的内容预览：

什么是 A2A Protocol？

A2A（Agent-to-Agent）Protocol 是 Google 推出的一套标准化通信框架，它使分布式系统中使用不同框架的 AI 智能体能够相互发现、通信与协作。

A2A Protocol 旨在促进智能体之间的通信，让智能体能够将其他智能体作为工具或服务调用，从而构建由专业化 AI 能力组成的网络。

A2A Protocol 的主要特性包括：

A2A Client：它是负责统筹全局的智能体（我们称之为“客户端智能体”），由它启动整个流程。它会判断需要完成哪些工作，找到合适的辅助智能体，并将任务分派给它们。你可以把它理解为代码中的项目经理。

A2A Agent：一种按照 A2A 规则建立简单 Web 地址（HTTP 端点）的 AI 智能体。它监听传入的请求、处理任务，然后返回结果或进度更新。对于将智能体“公开”并使其能够参与协作来说，这非常实用。

Agent Card：可以把它想象成一张采用 JSON 格式的数字身份证，易于读取和共享。它包含 A2A 智能体的基本信息，例如名称、功能以及连接方式。

Agent Skills：它们就像智能体的职位描述。每项技能都说明该智能体特别擅长完成的一项具体工作，例如“总结文章”或“生成图片”。客户端通过读取这些信息，可以准确判断应该分配什么任务，无须猜测！

A2A Executor：幕后运行的核心逻辑。它是代码中的一个函数，负责执行实际工作：接收请求、运行解决任务的逻辑，然后输出响应或触发事件。

A2A Server：Web 服务器端的组成部分。它将智能体的技能转换为可通过互联网共享的服务。你需要使用 A2A 的请求处理程序进行配置，借助 Starlette（一个 Python Web 框架）构建轻量级 Web 应用，再使用 Uvicorn（一个高速服务器运行程序）将其启动。就这样，你的智能体上线了，可以开始工作！

如果想深入了解 A2A Protocol 的工作原理及其配置方式，请查看相关文档：A2A Protocol 文档。

现在你已经了解了什么是 A2A Protocol，接下来让我们看看如何结合 AG-UI 和 CopilotKit，构建全栈 A2A AI 智能体。

要充分理解本教程，你需要具备 React 或 Next.js 的基础知识。

我们还会使用以下工具：

Python——一种常用于通过 AI 智能体框架构建 AI 智能体的流行编程语言；请确保你的计算机上已经安装了它。

AG-UI Protocol——Agent User Interaction Protocol（AG-UI）由 CopilotKit 开发，是一种开源、轻量级、基于事件的协议，用于在前端与 AI 智能体后端之间实现丰富的实时交互。

Google ADK——由 Google 设计的开源框架，旨在简化复杂且生产就绪的 AI 智能体的构建流程。

LangGraph——一个用于创建和部署 AI 智能体的框架。它还可以帮助定义智能体需要执行的控制流和操作。

Gemini API Key——一个 API 密钥，使你能够使用 Gemini 模型为 ADK 智能体执行各种任务。

CopilotKit——一个开源 Copilot 框架，用于构建自定义 AI 聊天机器人、应用内 AI 智能体和文本区域。

使用 CLI 设置 A2A 多智能体通信

在本节中，你将学习如何使用一条 CLI 命令设置 A2A Client（编排智能体）和 A2A 智能体。该命令会使用 Google ADK 和 AG-UI Protocol 配置后端，并使用 CopilotKit 配置前端。

欢迎查看 AG-UI 的 GitHub ⭐️

第 1 步：运行 CLI 命令

如果你还没有预配置的 AG-UI 智能体，可以在终端中运行以下 CLI 命令快速创建一个：

npx copilotkit@latest create -f a2a

然后按下图所示为项目命名。

第 2 步：安装前端依赖

项目成功创建后，使用你偏好的包管理器安装依赖：

npm install

第 3 步：安装后端依赖

安装前端依赖后，继续安装后端依赖：

cd agents
python3 -m venv .venv
source .venv/bin/activate  # On Windows: .venv\Scripts\activate
pip install -r requirements.txt
cd ..

第 4 步：设置环境变量

安装完后端依赖后，设置环境变量：

cp .env.example .env
# Edit .env and add your API keys:
# GOOGLE_API_KEY=your_google_api_key
# OPENAI_API_KEY=your_openai_api_key

第 5 步：启动所有服务

设置好环境变量后，启动包括后端和前端在内的所有服务。

npm run dev

开发服务器运行后，访问 http://localhost:3000/，你应该会看到 A2A 多智能体前端已经成功运行。

恭喜！你已经成功设置了 A2A 多智能体通信。可以尝试让智能体研究某个主题，例如“Please research quantum computing”。你会看到它向研究智能体和分析智能体发送消息，随后向用户展示完整的研究与分析结果。

在后端中将编排智能体与 Google ADK 和 AG-UI Protocol 集成

在本节中，你将学习如何将编排智能体与 Google ADK 和 AG-UI Protocol 集成，并将其作为 ASGI 应用暴露给前端。

第 1 步：设置后端

首先，克隆 A2A-Travel 仓库，其中包含基于 Python 的后端（agents）和 Next.js 前端。

接下来，进入后端目录：

cd agents

然后创建一个新的 Python 虚拟环境：

python -m venv .venv

之后，激活虚拟环境：

source .venv/bin/activate  # On Windows: .venv\Scripts\activate

最后，安装 requirements.txt 文件中列出的全部 Python 依赖。

pip install -r requirements.txt

第 2 步：配置 Orchestrator ADK 智能体

后端设置完成后，配置 Orchestrator ADK 智能体：定义智能体名称，指定 Gemini 2.5 Pro 作为大语言模型（LLM），并定义智能体指令，如下面的 agents/orchestrator.py 文件所示。

# Import Google ADK components for LLM agent creation
from google.adk.agents import LlmAgent

# === ORCHESTRATOR AGENT CONFIGURATION ===
# Create the main orchestrator agent using Google ADK's LlmAgent
# This agent coordinates all travel planning activities and manages the workflow

orchestrator_agent = LlmAgent(
    name="OrchestratorAgent",
    model="gemini-2.5-pro",  # Use the more powerful Pro model for complex orchestration
    instruction="""
    You are a travel planning orchestrator agent. Your role is to coordinate specialized agents
    to create personalized travel plans.

AVAILABLE SPECIALIZED AGENTS:

1. **Itinerary Agent** (LangGraph) - Creates day-by-day travel itineraries with activities
    2. **Restaurant Agent** (LangGraph) - Recommends restaurants for breakfast, lunch, and dinner by day
    3. **Weather Agent** (ADK) - Provides weather forecasts and packing advice
    4. **Budget Agent** (ADK) - Estimates travel costs and creates budget breakdowns

CRITICAL CONSTRAINTS:
    - You MUST call agents ONE AT A TIME, never make multiple tool calls simultaneously
    - After making a tool call, WAIT for the result before making another tool call
    - Do NOT make parallel/concurrent tool calls - this is not supported

RECOMMENDED WORKFLOW FOR TRAVEL PLANNING:

// ...

""",
)

第 3 步：创建 ADK 中间件智能体实例

配置好 Orchestrator ADK 智能体后，创建一个 ADK 中间件智能体实例。该实例会包装 Orchestrator ADK 智能体，使其与 AG-UI Protocol 集成，如下面的 agents/orchestrator.py 文件所示。

# Import AG-UI ADK components for frontend integration
from ag_ui_adk import ADKAgent

// ...

# === AG-UI PROTOCOL INTEGRATION ===
# Wrap the orchestrator agent with AG-UI Protocol capabilities
# This enables frontend communication and provides the interface for user interactions

adk_orchestrator_agent = ADKAgent( adk_agent=orchestrator_agent, # The core LLM agent we created above app_name="orchestrator_app", # Unique application identifier user_id="demo_user", # Default user ID for demo purposes session_timeout_seconds=3600, # Session timeout (1 hour) use_in_memory_services=True # Use in-memory storage for simplicity )

## 第 4 步：配置 FastAPI 端点

创建 ADK 中间件智能体实例后，配置一个 FastAPI 端点，将由 AG-UI 封装的编排器 ADK 智能体暴露给前端，如 `agents/orchestrator.py` 文件所示。

Import necessary libraries for web server and environment variables

import os import uvicorn

Import FastAPI for creating HTTP endpoints

from fastapi import FastAPI

// ...

=== FASTAPI WEB APPLICATION SETUP ===

Create the FastAPI application that will serve the orchestrator agent

This provides HTTP endpoints for the AG-UI Protocol communication

app = FastAPI(title="Travel Planning Orchestrator (ADK)")

Add the ADK agent endpoint to the FastAPI application

This creates the necessary routes for AG-UI Protocol communication

add_adk_fastapi_endpoint(app, adk_orchestrator_agent, path="/")

=== MAIN APPLICATION ENTRY POINT ===

if name == "main": """ Main entry point when the script is run directly.

This function: 1. Checks for required environment variables (API keys) 2. Configures the server port 3. Starts the uvicorn server with the FastAPI application """

Check for required Google API key

if not os.getenv("GOOGLE_API_KEY"):
    print("⚠️  Warning: GOOGLE_API_KEY environment variable not set!")
    print("   Set it with: export GOOGLE_API_KEY='your-key-here'")
    print("   Get a key from: https://aistudio.google.com/app/apikey")
    print()

Get server port from environment variable, default to 9000

port = int(os.getenv("ORCHESTRATOR_PORT", 9000))

Start the server with detailed information

print(f"🚀 Starting Orchestrator Agent (ADK + AG-UI) on http://localhost:{port}")

Run the FastAPI application using uvicorn

# host="0.0.0.0" allows external connections
# port is configurable via the environment variable
uvicorn.run(app, host="0.0.0.0", port=port)

恭喜！你已经成功将编排器 ADK Agent 与 AG-UI 协议集成，现在可以通过 `http://localhost:9000`（或指定的端口）端点访问它。

## 使用 A2A 协议集成来自不同智能体框架的 AI 智能体

在本节中，你将学习如何使用 A2A 集成来自不同智能体框架的 AI 智能体。

### 第 1 步：配置 A2A 远程智能体

首先，配置你的 A2A 远程智能体，例如使用 LangGraph 框架的行程智能体，如 `agents/itinerary_agent.py` 文件所示。

import json from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI

=== MAIN AGENT CLASS ===

class ItineraryAgent: """ Main agent class that handles itinerary generation using LangGraph workflow. """

def init(self): self.llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)

Build and compile the LangGraph workflow

    self.graph = self._build_graph()

def _build_graph(self): workflow = StateGraph(ItineraryState) workflow.add_node("parse_request", self._parse_request) workflow.add_node("create_itinerary", self._create_itinerary) workflow.set_entry_point("parse_request") workflow.add_edge("parse_request", "create_itinerary") workflow.add_edge("create_itinerary", END)

Compile the workflow into an executable graph

    return workflow.compile()

def _parse_request(self, state: "ItineraryState") -> "ItineraryState": message = state["message"]

Create a focused prompt for the extraction task

    prompt = f"""
    Extract the destination and number of days from this travel request.
    Return ONLY a JSON string with 'destination' and 'days' fields.

Request: {message}

Example output: {{"destination": "Tokyo", "days": 3}} """

Get LLM response for parsing

    response = self.llm.invoke(prompt)
    print(response.content)  # Debug: print LLM response

try: parsed = json.loads(response.content) state["destination"] = parsed.get("destination", "Unknown") state["days"] = int(parsed.get("days", 3)) except Exception as e: print(f"⚠️ Failed to parse request: {e}") state["destination"] = "Unknown" state["days"] = 3

return state

def _create_itinerary(self, state: "ItineraryState") -> "ItineraryState": destination = state["destination"] days = state["days"]

prompt = f""" Create a detailed {days}-day travel itinerary for {destination}. Make it realistic, interesting, and include specific place names. Return ONLY valid JSON, no markdown, no other text. """

Generate itinerary using LLM

    response = self.llm.invoke(prompt)
    content = response.content.strip()

Clean up response (remove code fences if present)

    if "```json" in content:
        content = content.split("```json")[1].split("```")[0].strip()
    elif "```" in content:
        content = content.split("```")[1].split("```")[0].strip()

try: structured_data = json.loads(content) validated_itinerary = StructuredItinerary(**structured_data)

state["structured_itinerary"] = validated_itinerary.model_dump() state["itinerary"] = json.dumps( validated_itinerary.model_dump(), indent=2 )

print("✅ Successfully created structured itinerary")

except Exception as e: print(f"⚠️ Failed to parse itinerary JSON: {e}") state["itinerary"] = json.dumps( {"error": "Failed to generate valid itinerary"}, indent=2 )

return state

async def invoke(self, message: "Message") -> str: message_text = message.parts[0].root.text print("Invoking itinerary agent with message:", message_text)

result = self.graph.invoke({ "message": message_text, "destination": "", "days": 3, "itinerary": "" })

return result["itinerary"]

### 第 2 步：设置 A2A 远程智能体技能和智能体卡片

配置好 A2A 远程智能体后，需要对智能体进行设置，使其他智能体能够发现并调用它。

为此，请定义每个智能体提供的具体技能，以及其他智能体能够发现的公开智能体卡片，如 `agents/itinerary_agent.py` 文件所示。

from a2a.types import ( AgentCapabilities, AgentCard, AgentSkill)

// ...

Define the specific skill this agent provides

skill = AgentSkill( id='itinerary_agent', name='Itinerary Planning Agent', description='Creates detailed day-by-day travel itineraries using LangGraph', tags=['travel', 'itinerary', 'langgraph'], examples=[ 'Create a 3-day itinerary for Tokyo', 'Plan a week-long trip to Paris', 'What should I do in New York for 5 days?' ], )

Define the public agent card that other agents can discover

public_agent_card = AgentCard( name='Itinerary Agent', description='LangGraph-powered agent that creates detailed day-by-day travel itineraries in plain text format with activities and meal recommendations.', url=f'http://localhost:{port}/', version='1.0.0', defaultInputModes=['text'], # Accepts text input defaultOutputModes=['text'], # Returns text output capabilities=AgentCapabilities(streaming=True), # Supports streaming responses skills=[skill], # List of skills this agent provides supportsAuthenticatedExtendedCard=False, # No authentication required )

### 第 3 步：配置 A2A 智能体执行器

设置好每个智能体的技能和智能体卡片后，为每个智能体配置一个用于处理 A2A 请求与响应的 A2A 智能体执行器，如 `agents/itinerary_agent.py` 文件所示。

from a2a.server.agent_execution import AgentExecutor, RequestContext

// ...

=== A2A PROTOCOL EXECUTOR ===

class ItineraryAgentExecutor(AgentExecutor): """ An executor class that bridges the A2A Protocol with our ItineraryAgent.

This class handles the A2A Protocol lifecycle: - Receives execution requests from other agents - Delegates to our ItineraryAgent for processing - Sends results back through the event queue """

def init(self): """Initialize the executor with an instance of our agent""" self.agent = ItineraryAgent()

async def execute( self, context: RequestContext, event_queue: EventQueue, ) -> None: """ Execute an itinerary generation request.

此方法：
        1. 使用传入的消息调用我们的智能体
        2. 将结果格式化为 A2A 文本消息
        3. 通过事件队列发送响应

参数：
            context：包含消息和元数据的请求上下文
            event_queue：用于将响应事件发回调用方的队列
        """
        # 使用我们的智能体生成行程
        result = await self.agent.invoke(context.message)

# 通过 A2A Protocol 事件队列发回结果
        await event_queue.enqueue_event(new_agent_text_message(result))

async def cancel(
        self, context: RequestContext, event_queue: EventQueue
    ) -> None:
        """
        处理取消请求（尚未实现）。

对于此智能体，我们不支持取消，因为行程生成通常很快，
        且无法中断。
        """
        raise Exception('cancel not supported')

第 4 步：设置 A2A 智能体服务器

为每个远程智能体配置好 A2A 智能体执行器后，需要设置各智能体的 A2A 智能体服务器。该服务器负责配置 A2A Protocol 请求处理器、创建 Starlette Web 应用程序并启动 uvicorn 服务器，如 agents/itineray_agent.py 文件所示。

# Import A2A Protocol components for inter-agent communication
from a2a.server.apps import A2AStarletteApplication
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.tasks import InMemoryTaskStore

// ...

# Get port from environment variable, default to 9001
port = int(os.getenv("ITINERARY_PORT", 9001))

# === MAIN APPLICATION SETUP ===
def main():
    """
    Main function that sets up and starts the A2A Protocol server.

This function:
    1. Checks for required environment variables
    2. Sets up the A2A Protocol request handler
    3. Creates the Starlette web application
    4. Starts the uvicorn server
    """

# Check for required OpenAI API key
    if not os.getenv("OPENAI_API_KEY"):
        print("⚠️  Warning: OPENAI_API_KEY environment variable not set!")
        print("   Set it with: export OPENAI_API_KEY='your-key-here'")
        print()

# Create the A2A Protocol request handler
    # This handles incoming requests and manages the task lifecycle
    request_handler = DefaultRequestHandler(
        agent_executor=ItineraryAgentExecutor(),  # Our custom executor
        task_store=InMemoryTaskStore(),           # Simple in-memory task storage
    )

# Create the A2A Starlette web application
    # This provides the HTTP endpoints for A2A Protocol communication
    server = A2AStarletteApplication(
        agent_card=public_agent_card,           # Public agent information
        http_handler=request_handler,           # Request processing logic
        extended_agent_card=public_agent_card,  # Extended agent info (same as public)
    )

# Start the server
    print(f"🗺️  Starting Itinerary Agent (LangGraph + A2A) on http://localhost:{port}")
    uvicorn.run(server.build(), host='0.0.0.0', port=port)

# === ENTRY POINT ===
if __name__ == '__main__':
    """
    Entry point when the script is run directly.

This allows the agent to be started as a standalone service:
    python itinerary_agent.py
    """
    main()

恭喜！你已经成功将远程智能体与 A2A Protocol 集成，现在编排器智能体可以将任务委派给这些智能体了。

使用 CopilotKit 为 AG-UI 和 A2A 多智能体通信构建前端

在本节中，你将学习如何使用 CopilotKit 为 AG-UI 和 A2A 多智能体通信添加前端。CopilotKit 可以在任何支持 React 的环境中运行。

第 1 步：设置前端

首先，在之前克隆的 A2A-Travel 仓库中安装前端依赖项。

npm install

然后配置环境变量：编辑 .env 文件，并添加你的 GOOGLE_API_KEY 和 OPENAI_API_KEY。

cp .env.example .env

最后，启动所有后端和前端服务器：

npm run dev

该命令会启动以下服务：UI 位于 http://localhost:3000，编排器位于 http://localhost:9000，行程智能体位于 http://localhost:9001，预算智能体位于 http://localhost:9002，餐厅智能体位于 http://localhost:9003，天气智能体位于 http://localhost:9005。

如果访问 http://localhost:3000/，你应该会看到旅行规划 A2A 多智能体前端已经启动并正在运行。

第 2 步：使用 A2A Middleware 配置 CopilotKit API 路由

设置好前端后，使用 A2A Middleware 配置 CopilotKit API 路由，以建立前端、AG-UI + ADK 编排器智能体和 A2A 智能体之间的连接，如 app/api/copilotkit/route.ts 文件所示。

import {
  CopilotRuntime,
  ExperimentalEmptyAdapter,
  copilotRuntimeNextJSAppRouterEndpoint,
} from "@copilotkit/runtime";
import { HttpAgent } from "@ag-ui/client";
import { A2AMiddlewareAgent } from "@ag-ui/a2a-middleware";
import { NextRequest } from "next/server";

export async function POST(request: NextRequest) {
  // STEP 1: Define A2A agent URLs
  const itineraryAgentUrl =
    process.env.ITINERARY_AGENT_URL || "http://localhost:9001";
  const budgetAgentUrl =
    process.env.BUDGET_AGENT_URL || "http://localhost:9002";
  const restaurantAgentUrl =
    process.env.RESTAURANT_AGENT_URL || "http://localhost:9003";
  const weatherAgentUrl =
    process.env.WEATHER_AGENT_URL || "http://localhost:9005";

// STEP 2: Define orchestrator URL (speaks AG-UI Protocol)
  const orchestratorUrl =
    process.env.ORCHESTRATOR_URL || "http://localhost:9000";

// STEP 3: Wrap orchestrator with HttpAgent (AG-UI client)
    // the orchestrator agent we pass to the middleware needs to be an instance of a derivative of an ag-ui `AbstractAgent`
    // In this case, we have access to the agent via url, so we can gain an instance using the `HttpAgent` class
  const orchestrationAgent = new HttpAgent({
    url: orchestratorUrl,
  });

// STEP 4: Create A2A Middleware Agent
  // This bridges AG-UI and A2A protocols by:
  // 1. Wrapping the orchestrator
  // 2. Registering all A2A agents
  // 3. Injecting send_message_to_a2a_agent tool
  // 4. Routing mes