AI Agents系列：如何利用 ADK + MCP + Gemini AI 构建这个多署理系统 ...

架构

我们将利用与 [旅行规划器]

相同的架构，并在此底子上扩展 A2A + MCP 协议。
   下面的演示只是为了说明 A2A 协议在多个署理之间的通信，仅用于说明目的。
  

上面的架构利用了模块化的多署理 AI 系统，其中每个署理都是独立可摆设的，并且通过谷歌的A2A（Agent-to-Agent）协议进行通信。
核心组件

• 用户界面层 —— 向前端服务器发送 HTTP 哀求
  
• 署理层 —— 和谐宿主署理、署理 1 和署理 2 之间的交互
  
• 协议层 —— 署理之间通过 A2A 协议进行通信
  
• 外部数据层 —— 利用 MCP 访问外部 API
  

署理脚色：

• 行程规划署理 —— 作为中心和谐者 —— 宿主署理，和谐用户与专业署理之间的交互。
  
• 航班搜索署理 —— 一个专门负责根据用户输入获取航班选项的署理
  
• 旅店搜索署理 —— 一个专门负责根据用户偏好获取旅店过夜的署理
  

MCP 在本项目中的实现：

航班搜索 MCP 服务器

• 连接：航班搜索（署理 1）连接到 MCP 航班服务器
  
• 功能：连接到航班预订 API 和数据库
  

旅店搜索 MCP 服务器

• 连接：旅店搜索（署理 2）连接到 MCP 旅店服务器
  
• 功能：连接到旅店预订系统和聚合器
  

署理通信流程

以下是利用 Mermaid 语法绘制的流程图，形貌了用户通过 Streamlit UI 提交旅行查询并天生完备行程的过程：
     流程图说明：

• 用户通过 Streamlit UI 提交旅行查询：用户在前端界面输入旅行需求。
  
• 旅行规划器解析查询以提取关键信息：旅行规划器解析用户输入，提取出发地、目的地、日期等关键信息。
  
• 旅行规划器向航班搜索署理哀求航班信息：旅行规划器将关键信息发送给航班搜索署理。
  
• 航班搜索署理通过调用 MCP 服务器返回可用航班：航班搜索署理查询航班信息并返回结果。
  
• 旅行规划器提取目的地详细信息：旅行规划器从航班信息中提取目的地相干细节。
  
• 旅行规划器向旅店搜索署理哀求旅店信息：旅行规划器将目的地信息发送给旅店搜索署理。
  
• 旅店搜索署理返回过夜选项：旅店搜索署理查询旅店信息并返回结果。
  
• 旅行规划器将所有数据综合成一个完备的行程：旅行规划器整合航班和旅店信息，天生完备的行程。
  

这个流程图清晰地展示了从用户输入到天生完备行程的整个过程。
实现

让我们深入相识一下如何利用 ADK + MCP + Gemini AI 构建这个多署理系统，我们将把它分解为关键的实现步调。
准备工作

• 安装 Python 3.11+
  

2. 获取谷歌 Gemini 天生式 AI 的 API 密钥
3. 获取有效的 SerpAPI 密钥
4. 获取有效的 OpenAI GPT 密钥
项目文件布局

1. ├── common
2. │   ├── __init__.py
3. │   ├── client
4. │   │   ├── __init__.py
5. │   │   ├── card_resolver.py
6. │   │   └── client.py
7. │   ├── server
8. │   │   ├── __init__.py
9. │   │   ├── server.py
10. │   │   ├── task_manager.py
11. │   │   └── utils.py
12. │   ├── types.py
13. │   └── utils
14. │       ├── in_memory_cache.py
15. │       └── push_notification_auth.py
16. ├── flight_search_app
17. │   ├── a2a_agent_card.json
18. │   ├── agent.py
19. │   ├── main.py
20. │   ├── static
21. │   │   └── .well-known
22. │   │       └── agent.json
23. │   └── streamlit_ui.py
24. ├── hotel_search_app
25. │   ├── README.md
26. │   ├── a2a_agent_card.json
27. │   ├── langchain_agent.py
28. │   ├── langchain_server.py
29. │   ├── langchain_streamlit.py
30. │   ├── static
31. │   │   └── .well-known
32. │   │       └── agent.json
33. │   └── streamlit_ui.py
34. └── itinerary_planner
35.     ├── __init__.py
36.     ├── a2a
37.     │   ├── __init__.py
38.     │   └── a2a_client.py
39.     ├── a2a_agent_card.json
40.     ├── event_log.py
41.     ├── itinerary_agent.py
42.     ├── itinerary_server.py
43.     ├── run_all.py
44.     ├── static
45.     │   └── .well-known
46.     │       └── agent.json
47.     └── streamlit_ui.py

复制代码

第一步：设置虚拟环境

安装依赖项

1. # 设置虚拟环境
2. python -m venv .venv # 激活虚拟环境
3. source .venv/bin/activate# 安装依赖项
4. pip install fastapi uvicorn streamlit httpx python-dotenv pydantic
5. pip install google-generativeai google-adk langchain langchain-openai

复制代码

第二步：安装 MCP 服务器包

mcp 旅店服务器 — https://pypi.org/project/mcp-hotel-search/
mcp 航班服务器 — https://pypi.org/project/mcp-flight-search/

1. # 安装 mcp 酒店搜索
2. pip install mcp-hotel-search# 安装 mcp 航班搜索
3. pip install mcp-flight-search

复制代码

第三步：设置 Gemini、OpenAI、SerpAI 的环境变量

设置上面准备工作中提到的环境变量

1. GOOGLE_API_KEY=your_google_api_key
2. OPENAI_API_KEY=your_openai_api_key
3. SERP_API_KEY=your_serp_api_key

复制代码

第四步：利用 ADK 设置航班搜索（署理）作为 MCP 客户端，利用 Gemini 2.0 Flash

利用https://github.com/google/A2A/tree/main/samples/python/common 中的可复用模块。

1. ├── common/                          # 共享 A2A 协议组件
2. │   ├── __init__.py
3. │   ├── client/                      # 客户端实现
4. │   │   ├── __init__.py
5. │   │   └── client.py                # 基础 A2A 客户端
6. │   ├── server/                      # 服务器实现
7. │   │   ├── __init__.py
8. │   │   ├── server.py                # A2A 服务器实现
9. │   │   └── task_manager.py          # 任务管理工具
10. │   └── types.py                     # A2A 共享类型定义

复制代码

1. ├── flight_search_app/               # 航班搜索代理（代理 1）
2. │   ├── __init__.py
3. │   ├── a2a_agent_card.json          # 代理能力声明
4. │   ├── agent.py                     # ADK 代理实现
5. │   ├── main.py                      # ADK 服务器入口点，使用 Gemini LLM
6. │   └── static/                      # 静态文件
7. │       └── .well-known/             # 代理发现目录
8. │           └── agent.json           # 标准化代理发现文件

复制代码

4.1 利用 ADK 署理实现作为 MCP 客户端从 MCP 服务器获取工具

1. from google.adk.agents.llm_agent import LlmAgent
2. from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset, StdioServerParameters..
3. ..
4. # 从 MCP 服务器获取工具
5. server_params = StdioServerParameters(
6.             command="mcp-flight-search",
7.             args=["--connection_type", "stdio"],
8.             env={"SERP_API_KEY": serp_api_key},)        tools, exit_stack = await MCPToolset.from_server(
9.             connection_params=server_params)
10. ..
11. ..

复制代码

4.2 利用通用 A2A 服务器组件和类型以及谷歌 ADK 运行器、会话和署理定义 ADK 服务器入口点

1. from google.adk.runners import Runner
2. from google.adk.sessions import InMemorySessionService
3. from google.adk.agents import Agent
4. from .agent import get_agent_async# 导入通用 A2A 服务器组件和类型
5. from common.server.server import A2AServer
6. from common.server.task_manager import InMemoryTaskManager
7. from common.types import (
8.     AgentCard,
9.     SendTaskRequest,
10.     SendTaskResponse,
11.     Task,
12.     TaskStatus,
13.     Message,
14.     TextPart,
15.     TaskState,
16. )# --- 自定义航班搜索任务管理器 ---
17. class FlightAgentTaskManager(InMemoryTaskManager):
18.     """特定于 ADK 航班搜索代理的任务管理器。"""
19.     def __init__(self, agent: Agent, runner: Runner, session_service: InMemorySessionService):
20.         super().__init__()
21.         self.agent = agent
22.         self.runner = runner
23.         self.session_service = session_service
24.         logger.info("FlightAgentTaskManager 初始化完成。")...
25. ...

复制代码

4.3 利用署理卡片创建 A2A 服务器实例

1. # --- 主执行块 ---
3. async def run_server():
4. """初始化服务并启动航班搜索 A2A 服务器。"""
5. logger.info("开始航班搜索 A2A 服务器初始化...") session_service = None
6. exit_stack = None
7. try:
8. session_service = InMemorySessionService()
9. agent, exit_stack = await get_agent_async()
10. runner = Runner(
11. app_name='flight_search_a2a_app',
12. agent=agent,
13. session_service=session_service,
14. ) # 创建特定的任务管理器
15. task_manager = FlightAgentTaskManager(
16. agent=agent,
17. runner=runner,
18. session_service=session_service
19. ) # 定义代理卡片
20. port = int(os.getenv("PORT", "8000"))
21. host = os.getenv("HOST", "localhost")
22. listen_host = "0.0.0.0" agent_card = AgentCard(
23. name="Flight Search Agent (A2A)",
24. description="根据用户查询提供航班信息。",
25. url=f"http://{host}:{port}/",
26. version="1.0.0",
27. defaultInputModes=["text"],
28. defaultOutputModes=["text"],
29. capabilities={"streaming": False},
30. skills=[
31. {
32. "id": "search_flights",
33. "name": "Search Flights",
34. "description": "根据出发地、目的地和日期搜索航班。",
35. "tags": ["flights", "travel"],
36. "examples": ["Find flights from JFK to LAX tomorrow"]
37. }
38. ]
39. ) # 创建 A2AServer 实例
40. a2a_server = A2AServer(
41. agent_card=agent_card,
42. task_manager=task_manager,
43. host=listen_host,
44. port=port
45. ) # 配置 Uvicorn
46. config = uvicorn.Config(
47. app=a2a_server.app, # 传递 A2AServer 的 Starlette 应用
48. host=listen_host,
49. port=port,
50. log_level="info"
51. )
52. server = uvicorn.Server(config)
53. ...
54. ...

复制代码

4.4 让我们启动航班搜索应用

第五步：利用 LangChain 配置旅店搜索署理作为 MCP 客户端，并利用 OpenAI（GPT-4o）作为 LLM

1. ├── hotel_search_app/                # 酒店搜索代理（代理 2）
2. │   ├── __init__.py
3. │   ├── a2a_agent_card.json          # 代理能力声明
4. │   ├── langchain_agent.py           # LangChain 代理实现
5. │   ├── langchain_server.py          # 服务器入口点
6. │   └── static/                      # 静态文件
7. │       └── .well-known/             # 代理发现目录
8. │           └── agent.json           # 标准化代理发现文件

复制代码

图片由作者提供
5.1. LangChain 署理 实现作为 MCP 客户端，利用 OpenAI LLM 作为语言模子

1. from langchain_openai import ChatOpenAI
2. from langchain.agents import AgentExecutor, create_openai_functions_agent
3. from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
4. from langchain_mcp_adapters.client import MultiServerMCPClient# MCP 客户端配置
5. MCP_CONFIG = {
6.     "hotel_search": {
7.         "command": "mcp-hotel-search",
8.         "args": ["--connection_type", "stdio"],
9.         "transport": "stdio",
10.         "env": {"SERP_API_KEY": os.getenv("SERP_API_KEY")},
11.     }
12. }class HotelSearchAgent:
13.     """使用 LangChain MCP 适配器的酒店搜索代理。"""        def __init__(self):
14.         self.llm = ChatOpenAI(model="gpt-4o", temperature=0)            def _create_prompt(self):
15.         """创建一个带有自定义系统消息的提示模板。"""
16.         system_message = """你是一个有用的酒店搜索助手。
17.         """                return ChatPromptTemplate.from_messages([
18.             ("system", system_message),
19.             ("human", "{input}"),
20.             MessagesPlaceholder(variable_name="agent_scratchpad"),
21.         ])
22. ..
23. ..
24. async def process_query(self, query):
25. ...            # 为这个查询创建一个 MCP 客户端实例
26.             async with MultiServerMCPClient(MCP_CONFIG) as client:
27.                 # 从这个客户端实例获取工具
28.                 tools = client.get_tools()                                # 创建一个提示
29.                 prompt = self._create_prompt()                                # 使用这些工具创建一个代理
30.                 agent = create_openai_functions_agent(
31.                     llm=self.llm,
32.                     tools=tools,
33.                     prompt=prompt
34.                 )                                # 使用这些工具创建一个执行器
35.                 executor = AgentExecutor(
36.                     agent=agent,
37.                     tools=tools,
38.                     verbose=True,
39.                     handle_parsing_errors=True,
40.                 )

复制代码

5.2 利用通用 A2A 服务器组件和类型创建 A2AServer 实例

2. # 直接使用底层代理
3. from hotel_search_app.langchain_agent import get_agent, HotelSearchAgent # 导入通用 A2A 服务器组件和类型
4. from common.server.server import A2AServer
5. from common.server.task_manager import InMemoryTaskManager
6. from common.types import (
7.     AgentCard,
8.     SendTaskRequest,
9.     SendTaskResponse,
10.     Task,
11.     TaskStatus,
12.     Message,
13.     TextPart,
14.     TaskState
15. )
16. ..
17. ..class HotelAgentTaskManager(InMemoryTaskManager):
18.     """特定于酒店搜索代理的任务管理器。"""
19.     def __init__(self, agent: HotelSearchAgent):
20.         super().__init__()
21.         self.agent = agent # HotelSearchAgent 实例
22.         logger.info("HotelAgentTaskManager 初始化完成。")    async def on_send_task(self, request: SendTaskRequest) -> SendTaskResponse:
23.         """通过调用代理的 process_query 处理 tasks/send 请求。"""
24.         task_params = request.params
25.         task_id = task_params.id
26.         user_message_text = None        logger.info(f"HotelAgentTaskManager 正在处理任务 {task_id}")# --- 主执行块 ---
27. async def run_server():
28.     """初始化服务并启动酒店搜索 A2A 服务器。"""
29.     logger.info("开始酒店搜索 A2A 服务器初始化...")        agent_instance: Optional[HotelSearchAgent] = None
30.     try:
31.         agent_instance = await get_agent()
32.         if not agent_instance:
33.              raise RuntimeError("初始化 HotelSearchAgent 失败")        # 创建特定的任务管理器
34.         task_manager = HotelAgentTaskManager(agent=agent_instance)                # 定义代理卡片
35.         port = int(os.getenv("PORT", "8003")) # 默认端口 8003
36.         host = os.getenv("HOST", "localhost")
37.         listen_host = "0.0.0.0"        agent_card = AgentCard(
38.             name="Hotel Search Agent (A2A)",
39.             description="根据位置、入住/退房日期和客人数量提供酒店信息。",
40.             url=f"http://{host}:{port}/",
41.             version="1.0.0",
42.             defaultInputModes=["text"],
43.             defaultOutputModes=["text"],
44.             capabilities={"streaming": False},
45.             skills=[
46.                 {
47.                     "id": "search_hotels",
48.                     "name": "Search Hotels",
49.                     "description": "根据位置、入住/退房日期和客人数量搜索酒店。",
50.                     "tags": ["hotels", "travel", "accommodation"],
51.                     "examples": ["Find hotels in London from July 1st to July 5th for 2 adults"]
52.                 }
53.             ]
54.         )        # 创建 A2AServer 实例，不使用 endpoint 参数
55.         a2a_server = A2AServer(
56.             agent_card=agent_card,
57.             task_manager=task_manager,
58.             host=listen_host,
59.             port=port
60.         )        config = uvicorn.Config(
61.             app=a2a_server.app, # 传递 A2AServer 的 Starlette 应用
62.             host=listen_host,
63.             port=port,
64.             log_level="info"
65.         )

复制代码

5.3 让我们启动旅店搜索应用（Langchain）作为入口点以调用 MCP 服务器

第六步：实现宿主署理作为署理之间的和谐者，利用 A2A 协议

行程规划器是上述旅行规划的核心组件，利用 A2A 协议与航班和旅店服务进行通信。

1. ├── itinerary_planner/               # 行程规划宿主代理（代理 3）
2. │   ├── __init__.py
3. │   ├── a2a/                         # A2A 客户端实现
4. │   │   ├── __init__.py
5. │   │   └── a2a_client.py            # 航班和酒店代理的客户端
6. │   ├── a2a_agent_card.json          # 代理能力声明
7. │   ├── event_log.py                 # 事件日志工具
8. │   ├── itinerary_agent.py           # 主规划器实现
9. │   ├── itinerary_server.py          # FastAPI 服务器
10. │   ├── run_all.py                   # 运行所有组件的脚本
11. │   ├── static/                      # 静态文件
12. │   │   └── .well-known/             # 代理发现目录
13. │   │       └── agent.json           # 标准化代理发现文件
14. │   └── streamlit_ui.py              # 主用户界面

复制代码

6.1 利用航班和旅店 API URL 实现 A2A 协议

• 包罗与服务通信的客户端代码
  
• 实现 Agent-to-Agent 协议
  
• 包罗调用航班和旅店搜索服务的模块
  

1. # A2A 兼容代理 API 的根端点
2. FLIGHT_SEARCH_API_URL = os.getenv("FLIGHT_SEARCH_API_URL", "http://localhost:8000")
3. HOTEL_SEARCH_API_URL = os.getenv("HOTEL_SEARCH_API_URL", "http://localhost:8003")class A2AClientBase:
4.     """通过根端点与 A2A 兼容代理通信的基础客户端。"""    async def send_a2a_task(self, user_message: str, task_id: Optional[str] = None, agent_type: str = "generic") -> Dict[str, Any]:
5.     ...
6.     ....
7.         # 构造带有 A2A 方法和修正后的参数结构的 JSON-RPC 负载
8.         payload = {
9.             "jsonrpc": "2.0",
10.             "method": "tasks/send",
11.             "params": {
12.                 "id": task_id,
13.                 "taskId": task_id,
14.                 "message": {
15.                     "role": "user",
16.                     "parts": [
17.                         {"type": "text", "text": user_message}
18.                     ]
19.                 }
20.             },
21.             "id": task_id
22.         }

复制代码

6.2 行程规划署理卡片

JSON 元数据文件，形貌署理的本领、端点、认证要求和技能。在 A2A 协议中用于服务发现。

1. {
2.   "name": "Travel Itinerary Planner",
3.   "displayName": "Travel Itinerary Planner",
4.   "description": "一个协调航班和酒店信息以创建综合旅行行程的代理",
5.   "version": "1.0.0",
6.   "contact": "code.aicloudlab@gmail.com",
7.   "endpointUrl": "http://localhost:8005",
8.   "authentication": {
9.     "type": "none"
10.   },
11.   "capabilities": ["streaming"],
12.   "skills": [
13.     {
14.       "name": "createItinerary",
15.       "description": "创建包含航班和住宿的综合旅行行程",
16.       "inputs": [
17.         {
18.           "name": "origin",
19.           "type": "string",
20.           "description": "出发城市或机场代码"
21.         },
22.         {
23.           "name": "destination",
24.           "type": "string",
25.           "description": "目的地城市或地区"
26.         },
27.         {
28.           "name": "departureDate",
29.           "type": "string",
30.           "description": "出发日期，格式为 YYYY-MM-DD"
31.         },
32.         {
33.           "name": "returnDate",
34.           "type": "string",
35.           "description": "返回日期，格式为 YYYY-MM-DD（可选）"
36.         },
37.         {
38.           "name": "travelers",
39.           "type": "integer",
40.           "description": "旅行者人数"
41.         },
42.         {
43.           "name": "preferences",
44.           "type": "object",
45.           "description": "其他偏好，如预算、酒店设施等"
46.         }
47.       ],
48.       "outputs": [
49.         {
50.           "name": "itinerary",
51.           "type": "object",
52.           "description": "包含航班、酒店和行程的完整旅行行程"
53.         }
54.       ]
55.     }
56.   ]
57. }

复制代码

6.3 利用谷歌天生式 AI SDK 的行程署理

   为了简化演示，这里利用了 GenAI SDK（也可以利用 ADK、CrewAI 或其他框架）
  行程署理 是系统的中心宿主署理，它和谐与航班和旅店搜索服务的通信，并利用语言模子解析自然语言哀求。

1. import google.generativeai as genai # 直接使用 SDK
2. ..
3. ..
4. from itinerary_planner.a2a.a2a_client import FlightSearchClient, HotelSearchClient# 配置谷歌生成式 AI SDK
5. genai.configure(api_key=api_key)class ItineraryPlanner:
6.     """一个使用谷歌生成式 AI SDK 协调航班和酒店搜索代理以创建行程的规划器。"""        def __init__(self):
7.         """初始化行程规划器。"""
8.         logger.info("使用谷歌生成式 AI SDK 初始化行程规划器")
9.         self.flight_client = FlightSearchClient()
10.         self.hotel_client = HotelSearchClient()                # 使用 SDK 创建 Gemini 模型实例
11.         self.model = genai.GenerativeModel(
12.             model_name="gemini-2.0-flash",
13.         )
14. ..
15. ..

复制代码

6.4 行程服务器 —— FastAPI 服务器，暴露行程规划器的端点，处理传入的 HTTP 哀求并将哀求路由到行程署理

1. from fastapi import FastAPI, HTTPException, Requestfrom itinerary_planner.itinerary_agent import ItineraryPlanner@app.post("/v1/tasks/send")
2. async def send_task(request: TaskRequest):
3.     """处理 A2A tasks/send 请求。"""
4.     global planner        if not planner:
5.         raise HTTPException(status_code=503, detail="规划器未初始化")        try:
6.         task_id = request.taskId                # 提取用户的消息
7.         user_message = None
8.         for part in request.message.get("parts", []):
9.             if "text" in part:
10.                 user_message = part["text"]
11.                 break                if not user_message:
12.             raise HTTPException(status_code=400, detail="请求中未找到文本消息")                # 根据查询生成行程
13.         itinerary = await planner.create_itinerary(user_message)                # 创建 A2A 响应
14.         response = {
15.             "task": {
16.                 "taskId": task_id,
17.                 "state": "completed",
18.                 "messages": [
19.                     {
20.                         "role": "user",
21.                         "parts": [{"text": user_message}]
22.                     },
23.                     {
24.                         "role": "agent",
25.                         "parts": [{"text": itinerary}]
26.                     }
27.                 ],
28.                 "artifacts": []
29.             }
30.         }                return response

复制代码

6.5 Streamlit_ui —— 利用 Streamlit 构建的用户界面，为旅行规划提供表单，并以用户友好的格式显示结果

1. ...
2. ...
4. # API 端点
6. API_URL = "http://localhost:8005/v1/tasks/send"def generate_itinerary(query: str):
7. """向行程规划器 API 发送查询。"""
8. try:
9. task_id = "task-" + datetime.now().strftime("%Y%m%d%H%M%S") payload = {
10. "taskId": task_id,
11. "message": {
12. "role": "user",
13. "parts": [
14. {
15. "text": query
16. }
17. ]
18. }
19. } # 将用户查询和请求记录到事件日志
20. log_user_query(query)
21. log_itinerary_request(payload) response = requests.post(
22. API_URL,
23. json=payload,
24. headers={"Content-Type": "application/json"}
25. )
26. response.raise_for_status() result = response.json() # 提取代理的响应消息
27. agent_message = None
28. for message in result.get("task", {}).get("messages", []):
29. if message.get("role") == "agent":
30. for part in message.get("parts", []):
31. if "text" in part:
32. agent_message = part["text"]
33. break
34. if agent_message:
35. break
36. ..
37. ..
38. ...

复制代码

第七步：最终演示

在每个终端中，按照以下方式启动服务器署理，正如下面的演示所看到的那样：

1. # 启动航班搜索代理 - 1，端口 8000
3. python -m flight_search_app.main# 启动酒店搜索代理 - 2，端口 8003
4. python -m hotel_search_app.langchain_server# 启动行程宿主代理 - 端口 8005
5. python -m itinerary_planner.itinerary_server# 启动前端 UI - 端口 8501
6. streamlit run itinerary_planner/streamlit_ui.py

复制代码

航班搜索日志，显示从宿主代剃头起的任务 ID

旅店搜索日志，显示从宿主代剃头起的任务

行程规划器 —— 宿主署理，显示所有哀求/响应

署理事件日志

这个演示实现了 谷歌 A2A 协议 的核心原则，使署理能够以布局化、可互操作的方式进行通信。在上面的演示中实现的组件如下：

• 署理卡片 —— 所有署理都暴露了 .well-known/agent.json 文件以供发现。
  
• A2A 服务器 —— 每个署理都作为一个 A2A 服务器运行：flight_search_app、hotel_search_app 和 itinerary_planner
  
• A2A 客户端 —— 行程规划器包罗了针对航班和旅店署理的专用 A2A 客户端。
  
• 任务管理 —— 每个哀求/响应都被建模为一个 A2A 任务，状态包括已提交、正在处理和已完成。
  
• 消息布局 —— 利用标准的 JSON-RPC 格式，包罗脚色（用户/署理）和部分（重要是 TextPart）。
  

以下组件在我们的演示中尚未实现，但可以扩展为适用于企业级署理：

• 流式传输（SSE） —— A2A 支持服务器发送事件用于长时间运行的任务，但我们的演示利用的是简单的哀求/响应，耗时不到 3-5 秒。
  
• 推送关照 —— 尚未利用 Webhook 更新机制。
  
• 复杂部分 —— 只利用了 TextPart。可以添加 DataPart、FilePart 等以支持更丰富的负载。
  
• 高级发现 —— 实现了根本的 .well-known/agent.json，但尚未实现高级认证、JWKS 或授权流程。
  

A2A 协议 —— 官方文档

总结

在本文中，我们探索了如何利用可复用的 A2A 组件、ADK、LangChain 和 MCP 构建一个功能齐全的多署理系统，用于旅行规划场景。通过联合这些开源工具和框架，我们的署理能够做到以下几点：

• 利用 A2A 动态发现并调用彼此
  
• 通过 MCP 以模子友好的方式连接到外部 API
  
• 利用当代框架，如 ADK 和 LangChain
  
• 以异步方式通信，具有清晰的任务生命周期和布局化结果
  

同样的原理可以扩展到更多领域，如零售、客户服务主动化、操作工作流以及 AI 辅助的企业工具。
感谢Arjun Prabhulal的分享https://medium.com/ai-cloud-lab/building-multi-agent-ai-app-with-googles-a2a-agent2agent-protocol-adk-and-mcp-a-deep-a94de2237200

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。