AI Agent多智能体协作实战：基于MCP与A2A协议的餐厅预订系统开发

如果你还在为AI Agent开发感到困惑，觉得多智能体协作、MCP协议、A2A架构这些概念听起来高大上但实际操作无从下手，那么这篇文章就是为你准备的。2026年已经到来，AI Agent技术正在从单兵作战向多智能体协作演进，但很多开发者仍然停留在“对话式AI”的认知层面，无法构建真正具备专业分工的智能体系统。

本文将通过一个完整的餐厅预订系统实战案例，带你从零构建一个包含菜单查询代理和预订代理的多智能体系统。你将学会如何让不同专业能力的AI Agent通过MCP协议访问工具和数据，通过A2A协议进行智能体间通信，并最终部署到生产环境。最重要的是，本文提供的代码完全兼容DeepSeek模型，让你可以用更经济的成本实现相同的功能。

1. 为什么单智能体已经不够用了？

在传统的AI应用开发中，我们通常构建一个“全能型”智能体来处理所有任务。比如一个餐厅助手，既要能查询菜单，又要能处理预订，还要回答各种咨询问题。这种架构在简单场景下还能应付，但随着业务复杂度增加，问题就暴露出来了：

单智能体的局限性：

• 功能臃肿：一个智能体需要掌握所有领域的知识，模型容易产生混淆
• 更新困难：修改预订逻辑可能影响菜单查询功能，风险集中
• 资源浪费：简单的菜单查询也要动用复杂的预订处理能力
• 扩展性差：新增功能需要重新训练或大幅调整现有智能体

多智能体的优势：

• 专业分工：每个智能体专注特定领域，表现更专业
• 独立部署：不同智能体可以有不同的更新周期和部署策略
• 资源优化：根据任务复杂度分配合适的智能体
• 易于扩展：新增功能只需添加新的专业智能体

在实际项目中，我们采用的多智能体架构是这样的：一个“餐厅礼宾代理”作为总协调员，负责理解用户意图并将任务分发给专业的“菜单查询代理”（通过MCP协议）和“预订代理”（通过A2A协议）。

2. 核心概念解析：MCP vs A2A vs 传统技能

2.1 MCP（Model Context Protocol）：智能体与工具的桥梁

MCP的核心作用是让AI智能体能够安全、标准化地使用外部工具和数据源。可以把MCP理解为智能体的“手和眼睛”——它让智能体能够操作工具、获取信息。

# MCP工具示例：菜单查询工具 class MenuQueryTool: def search_by_category(self, category: str) -> List[MenuItem]: """根据分类查询菜单""" pass def search_by_description(self, description: str) -> List[MenuItem]: """语义搜索菜单项""" pass

2.2 A2A（Agent-to-Agent）：智能体间的协作协议

A2A协议解决了不同智能体之间如何发现彼此、如何通信、如何协作的问题。与MCP主要连接工具不同，A2A连接的是其他智能体。

关键区别：

• MCP工具：无状态，单一功能，执行完即结束
• A2A代理：有状态，具备推理能力，支持多轮对话

2.3 智能体卡片（Agent Card）：智能体的“名片”

每个通过A2A协议暴露的智能体都需要一个标准化的描述文件，这就是智能体卡片。它包含智能体的功能描述、技能列表、输入输出格式等信息。

{ "agent_name": "Reservation Agent", "description": "处理餐厅预订创建、查询和取消", "skills": [ { "id": "manage_reservations", "name": "餐厅预订管理", "examples": [ "预订4人周五晚上7点的座位", "查询电话号码555-0101的预订" ] } ] }

3. 环境准备与项目结构

3.1 开发环境要求

• Python版本：3.9或更高版本
• 包管理：uv（快速Python包管理器）
• AI框架：Google ADK（Agent Development Kit）
• 模型支持：Gemini系列或DeepSeek（本文兼容）

3.2 项目目录结构

adk-a2a-agent-runtime-starter/ ├── reservation_agent/ # 预订代理 │ ├── agent.py # 代理核心逻辑 │ ├── a2a_config.py # A2A配置和智能体卡片 │ └── executor.py # A2A执行器 ├── restaurant_agent/ # 餐厅总代理 │ └── agent.py # 集成了预订代理的餐厅代理 ├── tools.yaml # MCP工具箱配置 ├── scripts/ # 部署和测试脚本 └── .env # 环境配置

3.3 初始环境设置

# 克隆项目代码 git clone https://github.com/alphinside/adk-a2a-agent-runtime-starter.git cd adk-a2a-agent-runtime-starter # 使用uv安装依赖（uv比pip快10倍以上） uv sync # 环境配置 cp .env.example .env # 编辑.env文件，配置项目ID和区域

4. 构建专业预订代理

4.1 预订代理的核心功能设计

我们的预订代理需要处理三种核心操作：

• 创建预订：收集用户信息并创建新的预订记录
• 查询预订：根据电话号码查询现有预订
• 取消预订：取消已存在的预订

# reservation_agent/agent.py import os from google.adk.agents import LlmAgent from google.adk.tools import ToolContext # 使用应用级状态前缀，确保预订数据在所有会话间持久化 STATE_PREFIX = "app:reservation:" def create_reservation( phone_number: str, name: str, party_size: int, date: str, time: str, tool_context: ToolContext, ) -> dict: """创建新的餐厅预订""" reservation = { "name": name, "party_size": party_size, "date": date, "time": time, "status": "confirmed", } # 使用电话号码作为唯一标识符存储预订 tool_context.state[f"{STATE_PREFIX}{phone_number}"] = reservation return { "status": "confirmed", "message": f"Reservation created for {name}, party of {party_size} on {date} at {time}", } def check_reservation(phone_number: str, tool_context: ToolContext) -> dict: """根据电话号码查询预订""" reservation = tool_context.state.get(f"{STATE_PREFIX}{phone_number}") if reservation: return {"found": True, "reservation": reservation} return {"found": False, "message": f"No reservation found for {phone_number}"} def cancel_reservation(phone_number: str, tool_context: ToolContext) -> dict: """取消现有预订""" key = f"{STATE_PREFIX}{phone_number}" reservation = tool_context.state.get(key) if not reservation: return {"success": False, "message": f"No reservation found for {phone_number}"} reservation["status"] = "cancelled" tool_context.state[key] = reservation return {"success": True, "message": "Reservation cancelled"} # 创建主要的LLM代理 root_agent = LlmAgent( name="reservation_agent", model="gemini-3.5-flash", # 可替换为DeepSeek模型 instruction="""你是Foodie Finds餐厅的友好预订助手。 帮助食客创建、查询和取消餐桌预订。 创建预订时需要收集以下信息： - 预订姓名 - 电话号码（作为预订ID） - 人数 - 日期 - 时间 在创建前始终确认详细信息。 查询或取消时，如果未提供电话号码请询问。 保持简洁专业。""", tools=[create_reservation, check_reservation, cancel_reservation], )

4.2 状态管理的设计考量

在AI Agent开发中，状态管理是一个关键问题。我们采用了以下几种策略：

1. 会话状态：在同一对话中保持上下文
2. 应用状态：跨会话持久化数据（如预订记录）
3. 临时状态：单次请求的临时数据

对于预订数据，我们使用应用级状态确保即使对话结束，预订信息仍然保留。这种设计比使用数据库更轻量，适合中小型应用。

5. 配置A2A协议支持

5.1 创建智能体卡片

智能体卡片是其他智能体发现和理解我们预订代理功能的途径。

# reservation_agent/a2a_config.py from a2a.types import AgentSkill from vertexai.preview.reasoning_engines.templates.a2a import create_agent_card reservation_skill = AgentSkill( id="manage_reservations", name="餐厅预订管理", description="为Foodie Finds餐厅创建、查询和取消餐桌预订", tags=["reservations", "restaurant", "booking"], examples=[ "预订4人周五晚上7点的座位", "查询电话号码555-0101的预订", "取消我的预订，电话号码555-0101" ], input_modes=["text/plain"], output_modes=["text/plain"], ) agent_card = create_agent_card( agent_name="预订代理", description="处理餐厅餐桌预订——为Foodie Finds餐厅创建、查询和取消预订", skills=[reservation_skill], )

5.2 实现A2A执行器

执行器是A2A协议与ADK代理之间的桥梁，负责处理传入的A2A请求并通过我们的代理执行。

# reservation_agent/executor.py import os import vertexai from a2a.server.agent_execution import AgentExecutor, RequestContext from google.adk.runners import Runner from google.adk.sessions import InMemorySessionService, VertexAiSessionService class ReservationAgentExecutor(AgentExecutor): """连接A2A协议和ADK预订代理的桥梁""" def __init__(self) -> None: self.agent = None self.runner = None def _init_agent(self): if self.agent is not None: return from reservation_agent.agent import root_agent as reservation_agent self.agent = reservation_agent # 根据环境选择会话服务 engine_id = os.environ.get("GOOGLE_CLOUD_AGENT_ENGINE_ID") if engine_id: # 生产环境：使用持久化会话服务 project = os.environ.get("GOOGLE_CLOUD_PROJECT") location = os.environ.get("GOOGLE_CLOUD_LOCATION", "us-central1") vertexai.init(project=project, location=location) session_service = VertexAiSessionService( project=project, location=location, agent_engine_id=engine_id ) app_name = engine_id else: # 本地开发：使用内存会话服务 session_service = InMemorySessionService() app_name = self.agent.name self.runner = Runner( app_name=app_name, agent=self.agent, session_service=session_service, # 其他服务配置... )

这种设计的关键优势在于环境自适应——在本地开发时使用轻量级的内存存储，部署到生产环境时自动切换为持久化存储。

6. 本地测试与验证

6.1 完整的本地测试流程

在部署之前，我们需要确保A2A代理在本地正常工作。以下是完整的测试脚本：

# scripts/test_a2a_agent_local.py import asyncio import os from vertexai.preview.reasoning_engines import A2aAgent from reservation_agent.a2a_config import agent_card from reservation_agent.executor import ReservationAgentExecutor async def main(): # 创建本地A2A代理 a2a_agent = A2aAgent( agent_card=agent_card, agent_executor_builder=ReservationAgentExecutor ) a2a_agent.set_up() # 测试1：获取智能体卡片 print("=== 测试智能体卡片 ===") card_response = await a2a_agent.handle_authenticated_agent_card() print(f"代理名称: {card_response.get('name')}") print(f"技能列表: {[s.get('name') for s in card_response.get('skills', [])]}") # 测试2：创建预订 print("\n=== 测试创建预订 ===") message_data = { "message": { "messageId": "test-001", "content": [{"text": "预订2人周六晚上6点的座位，姓名：Bob，电话：555-0202"}], "role": "ROLE_USER", }, } response = await a2a_agent.on_message_send(**message_data) task_id = response["task"]["id"] print(f"任务ID: {task_id}") # 等待任务完成并获取结果 result = await wait_for_task_completion(a2a_agent, task_id) print(f"结果: {result}") if __name__ == "__main__": asyncio.run(main())

6.2 测试结果验证

运行测试脚本后，你应该看到类似以下的输出：

=== 测试智能体卡片 === 代理名称: 预订代理 技能列表: ['餐厅预订管理'] === 测试创建预订 === 任务ID: f7f7004d-cfea-49c2-b57d-5bca9959e193 结果: 您的预订已确认：Bob，2人，周六晚上6点，电话555-0202

这个测试验证了：

1. A2A代理能够正确暴露其功能描述
2. 代理能够处理预订请求并返回正确结果
3. 整个A2A协议流程正常工作

7. 部署到Agent运行时环境

7.1 部署脚本实现

将代理部署到生产环境需要处理依赖管理、环境配置和部署流程。

# scripts/deploy_a2a_agent_runtime.py import os from pathlib import Path import vertexai from vertexai.preview.reasoning_engines import A2aAgent from reservation_agent.a2a_config import agent_card from reservation_agent.executor import ReservationAgentExecutor def deploy_reservation_agent(): """部署预订代理到Agent运行时""" # 环境配置 project_id = os.environ["GOOGLE_CLOUD_PROJECT"] region = os.environ["REGION"] staging_bucket = os.environ.get("STAGING_BUCKET") vertexai.init(project=project_id, location=region, staging_bucket=staging_bucket) # 创建A2A代理实例 a2a_agent = A2aAgent( agent_card=agent_card, agent_executor_builder=ReservationAgentExecutor, ) print("开始部署预订代理到Agent运行时...") print("预计需要3-5分钟...") # 执行部署 client = vertexai.Client(project=project_id, location=region) remote_agent = client.agent_engines.create( agent=a2a_agent, config={ "display_name": agent_card.name, "description": agent_card.description, "requirements": [ "google-cloud-aiplatform[agent_engines,adk]==1.149.0", "a2a-sdk==0.3.26", "google-adk==1.29.0", ], "extra_packages": ["./reservation_agent"], }, ) print(f"部署完成！资源名称: {remote_agent.api_resource.name}") return remote_agent.api_resource.name

7.2 部署后的验证

部署完成后，我们需要验证代理在生产环境的正常运行：

# scripts/test_deployed_agent.py import asyncio import vertexai from dotenv import load_dotenv load_dotenv() async def test_deployed_agent(): """测试已部署的代理""" project_id = os.environ["GOOGLE_CLOUD_PROJECT"] region = os.environ["REGION"] resource_name = os.environ["RESERVATION_AGENT_RESOURCE_NAME"] vertexai.init(project=project_id, location=region) client = vertexai.Client(project=project_id, location=region) # 获取已部署的代理 agent = client.agent_engines.get(name=resource_name) # 测试智能体卡片 card = await agent.handle_authenticated_agent_card() print(f"已部署代理: {card.name}") print(f"访问URL: {card.url}") # 测试预订功能 response = await agent.on_message_send( messageId="prod-test-001", role="user", parts=[{"kind": "text", "text": "预订3人周日下午12点的座位"}] ) print("生产环境测试通过！") if __name__ == "__main__": asyncio.run(test_deployed_agent())

8. 集成多智能体系统

8.1 升级餐厅主代理

现在我们需要将预订代理集成到餐厅主代理中，实现真正的多智能体协作。

# restaurant_agent/agent.py import os import httpx from google.adk.agents import LlmAgent from google.adk.agents.remote_a2a_agent import RemoteA2aAgent from google.auth import default from toolbox_adk import ToolboxToolset class GoogleCloudAuth(httpx.Auth): """Google Cloud身份验证处理器""" def __init__(self): self.credentials, _ = default( scopes=["https://www.googleapis.com/auth/cloud-platform"] ) def auth_flow(self, request): if not self.credentials.valid: self.credentials.refresh(httpx.Request("POST", "https://oauth2.googleapis.com/token")) request.headers["Authorization"] = f"Bearer {self.credentials.token}" yield request # 配置工具和代理 TOOLBOX_URL = os.environ.get("TOOLBOX_URL", "http://127.0.0.1:5000") RESERVATION_AGENT_CARD_URL = os.environ.get("RESERVATION_AGENT_CARD_URL", "") toolbox = ToolboxToolset(TOOLBOX_URL) # 创建远程A2A代理实例 reservation_remote_agent = RemoteA2aAgent( name="reservation_agent", description="处理餐厅餐桌预订——创建、查询和取消预订", agent_card=RESERVATION_AGENT_CARD_URL, httpx_client=httpx.AsyncClient(auth=GoogleCloudAuth(), timeout=60), ) # 主代理：协调菜单查询和预订功能 root_agent = LlmAgent( name="restaurant_agent", model="gemini-3.5-flash", instruction="""你是Foodie Finds餐厅的知识渊博的礼宾员。 你的工作： - 帮助食客按分类或菜系浏览菜单 - 提供菜品的完整详细信息，包括成分、价格和饮食信息 - 根据食客渴望的描述推荐菜品 - 当用户想要预订、查询或取消餐桌时，委托给reservation_agent 当食客按名称或菜系询问特定菜品时，使用get-item-details工具。 当食客要求特定分类或菜系类型时，使用search-menu工具。 当食客描述他们想要的食物类型时，使用search-menu-by-description工具进行语义搜索。 对于预订请求（预订、查询或取消餐桌），委托给reservation_agent。 保持对话性、知识渊博且简洁。""", tools=[toolbox], sub_agents=[reservation_remote_agent], )

8.2 多智能体协作流程

这个集成实现了以下协作流程：

1. 用户请求分析：主代理首先分析用户意图
2. 任务路由：如果是菜单相关请求，路由到MCP工具箱；如果是预订相关请求，路由到A2A预订代理
3. 结果整合：主代理将子代理的结果整合后返回给用户

9. 完整系统测试与验证

9.1 端到端测试场景

我们需要测试整个多智能体系统的各种使用场景：

# scripts/test_full_system.py import asyncio import requests async def test_full_system(): """测试完整的多智能体系统""" agent_url = os.environ.get("AGENT_URL", "http://localhost:8080") test_scenarios = [ { "name": "菜单查询测试", "message": "你们有哪些意大利菜？", "expected_keywords": ["pasta", "pizza", "意大利"] }, { "name": "预订创建测试", "message": "我想预订周五晚上7点4个人的座位，姓名Eve，电话555-0505", "expected_keywords": ["确认", "预订", "Eve"] }, { "name": "预订查询测试", "message": "查询电话555-0505的预订", "expected_keywords": ["预订", "Eve", "周五"] } ] for scenario in test_scenarios: print(f"\n=== {scenario['name']} ===") # 发送请求到代理 response = requests.post( f"{agent_url}/api/chat", json={"message": scenario["message"]} ) if response.status_code == 200: result = response.json() print(f"响应: {result['response']}") # 验证响应包含预期关键词 for keyword in scenario["expected_keywords"]: if keyword in result['response']: print(f"✓ 包含预期关键词: {keyword}") else: print(f"✗ 缺少预期关键词: {keyword}") else: print(f"请求失败: {response.status_code}")

9.2 性能与稳定性考量

在生产环境中，我们还需要考虑：

1. 错误处理：网络超时、代理不可用等情况
2. 重试机制：临时故障的自动恢复
3. 限流控制：防止单个用户过度使用资源
4. 监控告警：实时监控系统健康状态

10. 生产环境最佳实践

10.1 安全配置

# 生产环境安全配置示例 security: # API密钥管理 api_keys: rotation_period: 90 days emergency_rotation: enabled # 访问控制 access_control: ip_whitelist: ["10.0.0.0/8"] rate_limiting: 100 requests/minute # 数据保护 data_protection: encryption: enabled retention_period: 30 days

10.2 监控与日志

建立完整的可观测性体系：

# 监控配置示例 from opentelemetry import trace from opentelemetry.sdk.trace import TracerProvider # 设置分布式追踪 tracer_provider = TracerProvider() trace.set_tracer_provider(tracer_provider) # 关键指标监控 monitoring_metrics = [ "agent_response_time", "a2a_request_success_rate", "concurrent_users", "error_rate_by_agent" ]

10.3 成本优化策略

特别是使用DeepSeek等成本较低的模型时：

1. 缓存策略：对常见查询结果进行缓存
2. 请求批处理：合并多个小请求
3. 模型选择：根据任务复杂度选择合适的模型规格
4. 自动扩缩：根据负载动态调整资源

11. 常见问题与解决方案

11.1 部署问题排查

问题现象	可能原因	解决方案
部署超时	网络问题或依赖下载慢	检查网络连接，使用国内镜像源
依赖冲突	版本不兼容	使用uv管理依赖，确保版本一致
权限错误	服务账号权限不足	配置正确的IAM角色

11.2 运行时问题

问题现象	可能原因	解决方案
A2A调用失败	身份验证令牌过期	实现自动令牌刷新机制
响应缓慢	模型推理时间过长	优化提示词，启用流式响应
状态丢失	会话配置错误	检查会话服务配置

11.3 模型兼容性问题

由于我们设计时考虑了多模型兼容性，从Gemini切换到DeepSeek只需要修改配置：

# 修改模型配置 # 从Gemini切换到DeepSeek model_config = { "gemini": "gemini-3.5-flash", "deepseek": "deepseek-chat" # 根据实际模型名称调整 } # 在agent初始化时选择模型 selected_model = model_config[os.environ.get("MODEL_PROVIDER", "deepseek")]

12. 扩展与进阶方向

12.1 添加新的专业智能体

基于这个架构，你可以轻松添加新的智能体：

1. 支付处理代理：处理订单支付和退款
2. 推荐代理：基于用户历史提供个性化推荐
3. 客服代理：处理投诉和复杂咨询

12.2 性能优化建议

1. 异步处理：对耗时操作使用异步模式
2. 连接池：重用数据库和API连接
3. 预加载：提前加载常用数据到内存
4. 压缩传输：对大量数据使用压缩算法

12.3 企业级特性

对于大型企业应用，可以考虑：

1. 多租户支持：隔离不同客户的数据和配置
2. 审计日志：记录所有操作用于合规审查
3. 备份恢复：定期备份智能体状态和配置
4. 蓝绿部署：实现零停机更新

通过本文的实战指南，你应该已经掌握了构建生产级多智能体系统的核心技能。从单智能体到多智能体协作的转变，不仅仅是技术架构的升级，更是开发思维的转变。这种架构为构建复杂、可扩展的AI应用提供了坚实的基础。

记住，最好的学习方式是实践。建议你按照本文的步骤亲手实现这个系统，然后在基础上进行扩展和优化。随着经验的积累，你将能够设计出更加复杂和强大的多智能体解决方案。