手撕AI Agent框架多智能体MCPA2A全链路实战DeepSeek可跑如果你还在为AI Agent开发感到困惑觉得多智能体协作、MCP协议、A2A通信这些概念听起来高大上但实际操作无从下手那么这篇文章就是为你准备的。今天我们不谈空洞的理论直接手把手带你从零构建一个完整的餐厅多智能体系统让你真正理解AI Agent开发的完整链路。很多人以为AI Agent开发就是调用API那么简单但实际上真正的挑战在于如何让多个智能体协同工作、如何管理状态、如何实现跨框架通信。本文将以餐厅预订场景为例带你实战MCP工具集成、A2A多智能体通信、状态管理等核心概念所有代码都兼容DeepSeek模型确保你可以实际运行和修改。1. 这篇文章真正要解决的问题为什么你需要关注多智能体开发传统的单智能体系统在处理复杂任务时存在明显瓶颈一个智能体要同时处理菜单查询、预订管理、用户对话等多个功能导致系统臃肿、维护困难、更新周期长。而多智能体架构通过职责分离让专业的人做专业的事菜单查询智能体专注于菜品搜索和推荐预订管理智能体专门处理餐厅预订逻辑对话协调智能体负责任务分发和用户交互这种架构的优势在于单个智能体功能单一易于开发和测试不同智能体可以由不同团队独立开发部署和扩缩容更加灵活系统容错性更强本文要解决的核心痛点如何让多个AI智能体发现彼此并协同工作如何实现智能体间的标准化通信如何管理智能体的状态和会话如何将整个系统部署到生产环境2. 基础概念与核心原理2.1 MCPModel Context Protocol是什么MCP是连接AI智能体与工具数据的协议标准。简单来说它让智能体能够安全、标准化地访问外部工具和数据源。在我们的餐厅场景中MCP Toolbox就是通过MCP协议为智能体提供菜单查询能力的。MCP vs 传统API调用MCP提供了统一的工具发现和调用机制支持工具描述和参数验证具备更好的安全控制和权限管理2.2 A2AAgent-to-Agent协议的核心价值A2A协议解决了智能体间的通信问题就像人类工作中的团队协作一样# A2A通信的基本流程 用户请求 → 协调智能体 → A2A协议 → 专业智能体 → 返回结果A2A三大核心组件智能体卡片Agent Card相当于智能体的名片描述其功能和调用方式消息Message智能体间的通信内容任务Task具有生命周期的执行单元2.3 多智能体架构的优势对比架构类型开发复杂度维护成本扩展性适用场景单智能体低高差简单任务多智能体中低优秀复杂业务流程3. 环境准备与前置条件3.1 基础环境要求确保你的开发环境满足以下要求# 检查Python版本 python --version # 需要Python 3.9 uv --version # 需要uv包管理器 # 检查DeepSeek API访问 # 确保你有可用的DeepSeek API密钥3.2 项目结构准备创建项目目录结构# 创建项目目录 mkdir multi-agent-restaurant cd multi-agent-restaurant # 创建核心目录结构 mkdir -p reservation_agent restaurant_agent scripts logs touch .env.example .env README.md # 初始化uv环境 uv venv source .venv/bin/activate # Linux/Mac # 或 .venv\Scripts\activate # Windows3.3 依赖包安装创建pyproject.toml文件[project] name multi-agent-restaurant version 0.1.0 description Multi-agent restaurant system with MCP and A2A requires-python 3.9 dependencies [ google-cloud-aiplatform[agent_engines,adk]1.149.0, a2a-sdk0.3.26, google-adk1.29.0, httpx0.27.0, python-dotenv1.0.0, pydantic2.0.0, ] [build-system] requires [hatchling] build-backend hatchling.build安装依赖uv sync4. 构建预订管理智能体4.1 创建基础智能体框架首先构建专门处理餐厅预订的智能体# reservation_agent/agent.py import os from typing import Dict, Any 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[str, Any]: 创建新的餐厅预订 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预订成功{name}{party_size}人{date} {time}电话{phone_number}, } def check_reservation(phone_number: str, tool_context: ToolContext) - Dict[str, Any]: 通过电话号码查询预订 reservation tool_context.state.get(f{STATE_PREFIX}{phone_number}) if reservation: return {found: True, reservation: reservation} return {found: False, message: f未找到电话{phone_number}的预订} def cancel_reservation(phone_number: str, tool_context: ToolContext) - Dict[str, Any]: 取消现有预订 key f{STATE_PREFIX}{phone_number} reservation tool_context.state.get(key) if not reservation: return {success: False, message: f未找到电话{phone_number}的预订} if reservation.get(status) cancelled: return {success: False, message: f电话{phone_number}的预订已取消} reservation[status] cancelled tool_context.state[key] reservation return { success: True, message: f{reservation[name]}{phone_number}的预订已取消 } # 创建预订智能体 reservation_agent LlmAgent( namereservation_agent, modelgemini-3.5-flash, # 可替换为DeepSeek模型 instruction你是Foodie Finds餐厅的友好预订助手。 帮助顾客创建、查询和取消餐桌预订。 当顾客要预订时请收集以下信息 - 预订姓名 - 电话号码作为预订ID - 用餐人数 - 日期 - 时间 创建预订前请确认详细信息。 查询或取消时如果未提供电话号码请询问。 保持简洁专业。, tools[create_reservation, check_reservation, cancel_reservation], )4.2 配置A2A智能体卡片为了让其他智能体能够发现和使用预订智能体我们需要创建智能体卡片# reservation_agent/a2a_config.py from a2a.types import AgentSkill from vertexai.preview.reasoning_engines.templates.a2a import create_agent_card reservation_skill AgentSkill( idmanage_reservations, name餐厅预订管理, description为Foodie Finds餐厅创建、查询和取消餐桌预订, tags[reservations, restaurant, booking], examples[ 预订周五晚上7点4人桌, 查询电话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], )4.3 实现A2A执行器执行器是A2A协议与ADK智能体之间的桥梁# reservation_agent/executor.py import os from typing import NoReturn import vertexai from a2a.server.agent_execution import AgentExecutor, RequestContext from a2a.server.events import EventQueue from a2a.server.tasks import TaskUpdater from a2a.types import TaskState, TextPart, UnsupportedOperationError from a2a.utils import new_agent_text_message from a2a.utils.errors import ServerError from google.adk.artifacts import InMemoryArtifactService from google.adk.memory.in_memory_memory_service import InMemoryMemoryService from google.adk.runners import Runner from google.adk.sessions import InMemorySessionService, VertexAiSessionService from google.genai import types from reservation_agent.agent import reservation_agent class ReservationAgentExecutor(AgentExecutor): A2A协议与ADK预订智能体之间的桥梁 def __init__(self) - None: self.agent None self.runner None def _init_agent(self) - None: if self.agent is not None: return self.agent reservation_agent engine_id os.environ.get(GOOGLE_CLOUD_AGENT_ENGINE_ID) if engine_id: # 生产环境使用VertexAiSessionService实现持久会话 project os.environ.get(GOOGLE_CLOUD_PROJECT) location os.environ.get(GOOGLE_CLOUD_LOCATION, us-central1) vertexai.init(projectproject, locationlocation) session_service VertexAiSessionService( projectproject, locationlocation, agent_engine_idengine_id, ) app_name engine_id else: # 本地开发使用内存会话服务 session_service InMemorySessionService() app_name self.agent.name self.runner Runner( app_nameapp_name, agentself.agent, artifact_serviceInMemoryArtifactService(), session_servicesession_service, memory_serviceInMemoryMemoryService(), ) async def execute(self, context: RequestContext, event_queue: EventQueue) - None: if self.agent is None: self._init_agent() query context.get_user_input() updater TaskUpdater(event_queue, context.task_id, context.context_id) user_id context.message.metadata.get(user_id, a2a-user) if context.message.metadata else a2a-user if not context.current_task: await updater.submit() await updater.start_work() try: session await self._get_or_create_session(context.context_id, user_id) content types.Content(roleuser, parts[types.Part(textquery)]) async for event in self.runner.run_async( session_idsession.id, user_iduser_id, new_messagecontent, ): if event.is_final_response(): parts event.content.parts answer .join(p.text for p in parts if p.text) or 无响应 await updater.add_artifact([TextPart(textanswer)], nameanswer) await updater.complete() break except Exception as e: await updater.update_status( TaskState.failed, messagenew_agent_text_message(f错误{e!s}), ) raise async def _get_or_create_session(self, context_id: str, user_id: str): app_name self.runner.app_name if context_id: session await self.runner.session_service.get_session( app_nameapp_name, session_idcontext_id, user_iduser_id, ) if session: return session session await self.runner.session_service.create_session( app_nameapp_name, user_iduser_id, session_idcontext_id, ) return session async def cancel(self, context: RequestContext, event_queue: EventQueue) - NoReturn: raise ServerError(errorUnsupportedOperationError())5. 本地测试A2A智能体在部署之前我们需要在本地测试A2A智能体的功能# scripts/test_a2a_agent_local.py import asyncio import json import os from dotenv import load_dotenv from starlette.requests import Request from vertexai.preview.reasoning_engines import A2aAgent from reservation_agent.a2a_config import agent_card from reservation_agent.executor import ReservationAgentExecutor load_dotenv() def build_post_request(data: dict None, path_params: dict None) - Request: 构建模拟POST请求 scope { type: http, http_version: 1.1, headers: [(bcontent-type, bapplication/json)], app: None } if path_params: scope[path_params] path_params async def receive(): byte_data json.dumps(data).encode(utf-8) if data else b return {type: http.request, body: byte_data, more_body: False} return Request(scope, receive) async def wait_for_task(a2a_agent, task_id, max_retries30): 等待任务完成 for _ in range(max_retries): request build_post_request() result await a2a_agent.on_get_task(requestrequest, contextNone) state result.get(status, {}).get(state, ) if state in [completed, failed]: return result await asyncio.sleep(1) return result async def main(): # 创建本地A2A智能体 a2a_agent A2aAgent( agent_cardagent_card, agent_executor_builderReservationAgentExecutor ) a2a_agent.set_up() print( * 50) print(1. 测试智能体卡片检索...) print( * 50) # 测试智能体卡片 request build_post_request() card_response await a2a_agent.handle_authenticated_agent_card( requestrequest, contextNone ) print(f智能体: {card_response.get(name)}) print(f技能: {[s.get(name) for s in card_response.get(skills, [])]}) # 测试预订创建 print(\n * 50) print(2. 测试创建预订...) print( * 50) message_data { message: { messageId: fmsg-{os.urandom(4).hex()}, content: [{text: 预订周六晚上6点2人桌姓名Bob电话555-0202}], role: ROLE_USER, }, } request build_post_request(message_data) response await a2a_agent.on_message_send(requestrequest, contextNone) task_id response[task][id] context_id response[task].get(contextId) print(f任务ID: {task_id}) # 等待结果 result await wait_for_task(a2a_agent, task_id) print(f状态: {result.get(status, {}).get(state)}) for artifact in result.get(artifacts, []): if artifact.get(parts) and text in artifact[parts][0]: print(f响应: {artifact[parts][0][text]}) print(\n * 50) print(本地测试完成) print( * 50) if __name__ __main__: asyncio.run(main())运行测试uv run python scripts/test_a2a_agent_local.py6. 构建餐厅协调智能体现在创建主协调智能体负责将用户请求路由到相应的专业智能体# 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 google.auth.transport.requests import Request as AuthRequest # 配置环境变量 TOOLBOX_URL os.environ.get(TOOLBOX_URL, http://127.0.0.1:5000) RESERVATION_AGENT_CARD_URL os.environ.get(RESERVATION_AGENT_CARD_URL, ) 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(AuthRequest()) request.headers[Authorization] fBearer {self.credentials.token} yield request # 创建远程A2A预订智能体 reservation_remote_agent RemoteA2aAgent( namereservation_agent, description处理餐厅餐桌预订——创建、查询和取消预订。当用户要预订、查询或取消时委托给此智能体, agent_cardRESERVATION_AGENT_CARD_URL, httpx_clienthttpx.AsyncClient(authGoogleCloudAuth(), timeout60), ) # 创建主餐厅智能体 restaurant_agent LlmAgent( namerestaurant_agent, modelgemini-3.5-flash, # 可替换为DeepSeek模型 instruction你是Foodie Finds餐厅的友好知识渊博的门房。你的工作 - 帮助顾客按类别或菜系浏览菜单 - 提供菜品的完整详细信息包括成分、价格和饮食信息 - 根据顾客对想吃什么的自然语言描述推荐菜品 - 当被要求时添加新菜单项 - 对于预订请求预订、查询或取消餐桌委托给reservation_agent 当顾客按名称或菜系询问特定菜品时使用get-item-details工具。 当顾客要求特定类别或菜系类型时使用search-menu工具。 当顾客描述他们想要的食物类型——通过风味、质地、饮食需求或渴望——使用search-menu-by-description工具进行语义搜索。 在search-menu和search-menu-by-description之间不确定时优先使用search-menu-by-description——它搜索菜品描述并找到更相关的匹配项。 如果菜品不可用available为false告知顾客并从搜索结果中建议类似的替代品。 保持对话式、知识渊博且简洁。, tools[], # 这里可以添加MCP工具箱工具 sub_agents[reservation_remote_agent], )7. 系统集成与部署7.1 配置MCP工具箱集成创建MCP工具箱配置文件# tools.yaml toolsets: - name: restaurant-tools description: Tools for restaurant menu management tools: - name: search-menu description: Search menu items by category or cuisine inputSchema: type: object properties: category: type: string description: Category or cuisine to search for required: [category] - name: get-item-details description: Get detailed information about a specific menu item inputSchema: type: object properties: item_id: type: string description: ID of the menu item required: [item_id] - name: search-menu-by-description description: Semantic search for menu items based on natural language description inputSchema: type: object properties: description: type: string description: Natural language description of desired food required: [description]7.2 部署脚本创建完整的部署脚本# scripts/deploy_system.py import os import asyncio from pathlib import Path import vertexai from dotenv import load_dotenv from google.genai import types load_dotenv() async def deploy_reservation_agent(): 部署预订智能体到Agent Runtime PROJECT_ID os.environ[GOOGLE_CLOUD_PROJECT] REGION os.environ[REGION] STAGING_BUCKET os.environ.get(STAGING_BUCKET, f{PROJECT_ID}-adk-a2a-agent-runtime) vertexai.init(projectPROJECT_ID, locationREGION, staging_bucketfgs://{STAGING_BUCKET}) client vertexai.Client( projectPROJECT_ID, locationREGION, http_optionstypes.HttpOptions(api_versionv1beta1), ) print(部署预订智能体到Agent Runtime...) print(这可能需要3-5分钟...) # 这里需要导入实际的A2A智能体配置 from reservation_agent.a2a_config import agent_card from reservation_agent.executor import ReservationAgentExecutor from vertexai.preview.reasoning_engines import A2aAgent a2a_agent A2aAgent( agent_cardagent_card, agent_executor_builderReservationAgentExecutor, ) remote_agent client.agent_engines.create( agenta2a_agent, config{ display_name: agent_card.name, description: agent_card.description, requirements: [ google-cloud-aiplatform[agent_engines,adk]1.149.0, a2a-sdk0.3.26, google-adk1.29.0, ], extra_packages: [./reservation_agent], }, ) resource_name remote_agent.api_resource.name print(f部署完成资源名称: {resource_name}) # 保存资源名称到环境文件 env_path Path(.env) lines env_path.read_text().splitlines() if env_path.exists() else [] lines [l for l in lines if not l.startswith(RESERVATION_AGENT_RESOURCE_NAME)] lines.append(fRESERVATION_AGENT_RESOURCE_NAME{resource_name}) env_path.write_text(\n.join(lines) \n) return resource_name async def resolve_agent_card_url(resource_name: str): 解析智能体卡片URL PROJECT_ID os.environ[GOOGLE_CLOUD_PROJECT] REGION os.environ[REGION] vertexai.init(projectPROJECT_ID, locationREGION) client vertexai.Client( projectPROJECT_ID, locationREGION, http_optionstypes.HttpOptions(api_versionv1beta1), ) agent client.agent_engines.get(nameresource_name) card await agent.handle_authenticated_agent_card() card_url f{card.url}/v1/card print(f智能体: {card.name}) print(f卡片URL: {card_url}) # 保存到环境文件 for env_path in [Path(restaurant_agent/.env), Path(.env)]: lines env_path.read_text().splitlines() if env_path.exists() else [] lines [l for l in lines if not l.startswith(RESERVATION_AGENT_CARD_URL)] lines.append(fRESERVATION_AGENT_CARD_URL{card_url}) env_path.write_text(\n.join(lines) \n) return card_url async def main(): # 1. 部署预订智能体 resource_name await deploy_reservation_agent() # 2. 解析卡片URL card_url await resolve_agent_card_url(resource_name) print(\n *50) print(部署完成) print(*50) print(f预订智能体资源: {resource_name}) print(fA2A卡片URL: {card_url}) if __name__ __main__: asyncio.run(main())8. 完整系统测试创建端到端测试脚本验证整个多智能体系统# scripts/test_full_system.py import asyncio import os from dotenv import load_dotenv load_dotenv() async def test_menu_query(): 测试菜单查询功能 print(测试菜单查询...) # 这里实现菜单查询测试逻辑 print(菜单查询测试完成) async def test_reservation_flow(): 测试完整预订流程 print(测试预订流程...) # 1. 创建预订 print(1. 创建预订...) # 实现预订创建逻辑 # 2. 查询预订 print(2. 查询预订...) # 实现预订查询逻辑 # 3. 取消预订 print(3. 取消预订...) # 实现预订取消逻辑 print(预订流程测试完成) async def test_multi_agent_coordination(): 测试多智能体协调 print(测试多智能体协调...) # 测试协调智能体如何路由请求 test_cases [ 你们有什么意大利菜, # 应该路由到菜单查询 我想预订明天晚上7点的4人桌, # 应该路由到预订智能体 查询我电话123456的预订, # 应该路由到预订智能体 推荐一些辣的菜品 # 应该路由到菜单查询 ] for query in test_cases: print(f测试查询: {query}) # 实现路由测试逻辑 print(多智能体协调测试完成) async def main(): print(开始完整系统测试...) print( * 50) await test_menu_query() print(- * 30) await test_reservation_flow() print(- * 30) await test_multi_agent_coordination() print(- * 30) print(所有测试完成) print( * 50) if __name__ __main__: asyncio.run(main())9. 常见问题与排查思路9.1 部署问题排查问题现象可能原因排查方式解决方案部署失败提示权限不足服务账号缺少必要权限检查IAM权限设置授予aiplatform.user角色智能体卡片无法访问网络配置问题检查VPC和防火墙规则配置正确的网络规则会话状态不持久SessionService配置错误检查环境变量和配置确保使用VertexAiSessionService9.2 通信问题排查# 通信诊断脚本 import asyncio import aiohttp import os from dotenv import load_dotenv load_dotenv() async def diagnose_communication(): 诊断A2A通信问题 card_url os.environ.get(RESERVATION_AGENT_CARD_URL) if not card_url: print(错误未找到RESERVATION_AGENT_CARD_URL环境变量) return try: async with aiohttp.ClientSession() as session: # 测试卡片访问 async with session.get(card_url) as response: if response.status 200: print(✓ 智能体卡片可访问) card_data await response.json() print(f智能体名称: {card_data.get(name)}) else: print(f✗ 卡片访问失败: {response.status}) except Exception as e: print(f通信诊断失败: {e}) if __name__ __main__: asyncio.run(diagnose_communication())9.3 状态管理问题常见状态问题状态丢失检查SessionService配置和权限状态冲突确保使用正确的状态前缀和范围状态同步延迟验证网络连接和超时设置10. 最佳实践与工程建议10.1 智能体设计原则单一职责原则每个智能体只负责一个明确的业务领域接口标准化使用A2A协议确保智能体间的标准化通信状态隔离不同的智能体应该管理独立的状态错误处理实现完善的错误处理和重试机制10.2 部署和运维建议# 生产环境配置示例 监控配置: 指标监控: - 请求延迟 - 错误率 - 并发连接数 日志记录: - 访问日志 - 错误日志 - 性能日志 告警规则: - 错误率 1% - 平均延迟 2秒 - 服务不可用10.3 安全最佳实践身份验证使用Google Cloud IAM进行访问控制网络隔离在VPC内部署敏感服务数据加密启用CMEK进行数据加密审计日志启用完整的操作审计11. 扩展和优化方向11.1 性能优化# 智能体性能优化示例 from google.adk.agents import LlmAgent from google.adk.caching import DiskCache # 启用响应缓存 optimized_agent LlmAgent( nameoptimized_agent, modelgemini-3.5-flash, cacheDiskCache(), # 磁盘缓存 instruction..., tools[...], )11.2 扩展更多智能体基于现有架构可以轻松扩展更多专业智能体支付处理智能体处理订单支付配送跟踪智能体跟踪外卖配送状态客户服务智能体处理客户投诉和咨询11.3 与DeepSeek模型集成要将系统适配DeepSeek模型只需修改模型配置# 配置DeepSeek模型 deepseek_agent LlmAgent( namedeepseek_agent, modelCustomLLM( # 自定义DeepSeek模型适配器 modeldeepseek-chat, api_keyos.environ.get(DEEPSEEK_API_KEY) ), instruction..., tools[...], )通过本文的实战教程你应该已经掌握了多智能体系统开发的核心技能。从MCP工具集成到A2A智能体通信从本地测试到生产部署这套架构可以应用于各种复杂的业务场景。记住良好的智能体设计就像组建一个高效的团队让每个成员专注自己最擅长的工作通过标准化协议协同合作才能发挥最大的效能。建议将本文代码作为起点根据你的具体业务需求进行定制和扩展。多智能体架构的未来充满可能性现在正是深入实践的最佳时机。
多智能体系统实战：MCP协议与A2A通信构建餐厅AI Agent框架
发布时间：2026/7/15 1:44:15
手撕AI Agent框架多智能体MCPA2A全链路实战DeepSeek可跑如果你还在为AI Agent开发感到困惑觉得多智能体协作、MCP协议、A2A通信这些概念听起来高大上但实际操作无从下手那么这篇文章就是为你准备的。今天我们不谈空洞的理论直接手把手带你从零构建一个完整的餐厅多智能体系统让你真正理解AI Agent开发的完整链路。很多人以为AI Agent开发就是调用API那么简单但实际上真正的挑战在于如何让多个智能体协同工作、如何管理状态、如何实现跨框架通信。本文将以餐厅预订场景为例带你实战MCP工具集成、A2A多智能体通信、状态管理等核心概念所有代码都兼容DeepSeek模型确保你可以实际运行和修改。1. 这篇文章真正要解决的问题为什么你需要关注多智能体开发传统的单智能体系统在处理复杂任务时存在明显瓶颈一个智能体要同时处理菜单查询、预订管理、用户对话等多个功能导致系统臃肿、维护困难、更新周期长。而多智能体架构通过职责分离让专业的人做专业的事菜单查询智能体专注于菜品搜索和推荐预订管理智能体专门处理餐厅预订逻辑对话协调智能体负责任务分发和用户交互这种架构的优势在于单个智能体功能单一易于开发和测试不同智能体可以由不同团队独立开发部署和扩缩容更加灵活系统容错性更强本文要解决的核心痛点如何让多个AI智能体发现彼此并协同工作如何实现智能体间的标准化通信如何管理智能体的状态和会话如何将整个系统部署到生产环境2. 基础概念与核心原理2.1 MCPModel Context Protocol是什么MCP是连接AI智能体与工具数据的协议标准。简单来说它让智能体能够安全、标准化地访问外部工具和数据源。在我们的餐厅场景中MCP Toolbox就是通过MCP协议为智能体提供菜单查询能力的。MCP vs 传统API调用MCP提供了统一的工具发现和调用机制支持工具描述和参数验证具备更好的安全控制和权限管理2.2 A2AAgent-to-Agent协议的核心价值A2A协议解决了智能体间的通信问题就像人类工作中的团队协作一样# A2A通信的基本流程 用户请求 → 协调智能体 → A2A协议 → 专业智能体 → 返回结果A2A三大核心组件智能体卡片Agent Card相当于智能体的名片描述其功能和调用方式消息Message智能体间的通信内容任务Task具有生命周期的执行单元2.3 多智能体架构的优势对比架构类型开发复杂度维护成本扩展性适用场景单智能体低高差简单任务多智能体中低优秀复杂业务流程3. 环境准备与前置条件3.1 基础环境要求确保你的开发环境满足以下要求# 检查Python版本 python --version # 需要Python 3.9 uv --version # 需要uv包管理器 # 检查DeepSeek API访问 # 确保你有可用的DeepSeek API密钥3.2 项目结构准备创建项目目录结构# 创建项目目录 mkdir multi-agent-restaurant cd multi-agent-restaurant # 创建核心目录结构 mkdir -p reservation_agent restaurant_agent scripts logs touch .env.example .env README.md # 初始化uv环境 uv venv source .venv/bin/activate # Linux/Mac # 或 .venv\Scripts\activate # Windows3.3 依赖包安装创建pyproject.toml文件[project] name multi-agent-restaurant version 0.1.0 description Multi-agent restaurant system with MCP and A2A requires-python 3.9 dependencies [ google-cloud-aiplatform[agent_engines,adk]1.149.0, a2a-sdk0.3.26, google-adk1.29.0, httpx0.27.0, python-dotenv1.0.0, pydantic2.0.0, ] [build-system] requires [hatchling] build-backend hatchling.build安装依赖uv sync4. 构建预订管理智能体4.1 创建基础智能体框架首先构建专门处理餐厅预订的智能体# reservation_agent/agent.py import os from typing import Dict, Any 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[str, Any]: 创建新的餐厅预订 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预订成功{name}{party_size}人{date} {time}电话{phone_number}, } def check_reservation(phone_number: str, tool_context: ToolContext) - Dict[str, Any]: 通过电话号码查询预订 reservation tool_context.state.get(f{STATE_PREFIX}{phone_number}) if reservation: return {found: True, reservation: reservation} return {found: False, message: f未找到电话{phone_number}的预订} def cancel_reservation(phone_number: str, tool_context: ToolContext) - Dict[str, Any]: 取消现有预订 key f{STATE_PREFIX}{phone_number} reservation tool_context.state.get(key) if not reservation: return {success: False, message: f未找到电话{phone_number}的预订} if reservation.get(status) cancelled: return {success: False, message: f电话{phone_number}的预订已取消} reservation[status] cancelled tool_context.state[key] reservation return { success: True, message: f{reservation[name]}{phone_number}的预订已取消 } # 创建预订智能体 reservation_agent LlmAgent( namereservation_agent, modelgemini-3.5-flash, # 可替换为DeepSeek模型 instruction你是Foodie Finds餐厅的友好预订助手。 帮助顾客创建、查询和取消餐桌预订。 当顾客要预订时请收集以下信息 - 预订姓名 - 电话号码作为预订ID - 用餐人数 - 日期 - 时间 创建预订前请确认详细信息。 查询或取消时如果未提供电话号码请询问。 保持简洁专业。, tools[create_reservation, check_reservation, cancel_reservation], )4.2 配置A2A智能体卡片为了让其他智能体能够发现和使用预订智能体我们需要创建智能体卡片# reservation_agent/a2a_config.py from a2a.types import AgentSkill from vertexai.preview.reasoning_engines.templates.a2a import create_agent_card reservation_skill AgentSkill( idmanage_reservations, name餐厅预订管理, description为Foodie Finds餐厅创建、查询和取消餐桌预订, tags[reservations, restaurant, booking], examples[ 预订周五晚上7点4人桌, 查询电话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], )4.3 实现A2A执行器执行器是A2A协议与ADK智能体之间的桥梁# reservation_agent/executor.py import os from typing import NoReturn import vertexai from a2a.server.agent_execution import AgentExecutor, RequestContext from a2a.server.events import EventQueue from a2a.server.tasks import TaskUpdater from a2a.types import TaskState, TextPart, UnsupportedOperationError from a2a.utils import new_agent_text_message from a2a.utils.errors import ServerError from google.adk.artifacts import InMemoryArtifactService from google.adk.memory.in_memory_memory_service import InMemoryMemoryService from google.adk.runners import Runner from google.adk.sessions import InMemorySessionService, VertexAiSessionService from google.genai import types from reservation_agent.agent import reservation_agent class ReservationAgentExecutor(AgentExecutor): A2A协议与ADK预订智能体之间的桥梁 def __init__(self) - None: self.agent None self.runner None def _init_agent(self) - None: if self.agent is not None: return self.agent reservation_agent engine_id os.environ.get(GOOGLE_CLOUD_AGENT_ENGINE_ID) if engine_id: # 生产环境使用VertexAiSessionService实现持久会话 project os.environ.get(GOOGLE_CLOUD_PROJECT) location os.environ.get(GOOGLE_CLOUD_LOCATION, us-central1) vertexai.init(projectproject, locationlocation) session_service VertexAiSessionService( projectproject, locationlocation, agent_engine_idengine_id, ) app_name engine_id else: # 本地开发使用内存会话服务 session_service InMemorySessionService() app_name self.agent.name self.runner Runner( app_nameapp_name, agentself.agent, artifact_serviceInMemoryArtifactService(), session_servicesession_service, memory_serviceInMemoryMemoryService(), ) async def execute(self, context: RequestContext, event_queue: EventQueue) - None: if self.agent is None: self._init_agent() query context.get_user_input() updater TaskUpdater(event_queue, context.task_id, context.context_id) user_id context.message.metadata.get(user_id, a2a-user) if context.message.metadata else a2a-user if not context.current_task: await updater.submit() await updater.start_work() try: session await self._get_or_create_session(context.context_id, user_id) content types.Content(roleuser, parts[types.Part(textquery)]) async for event in self.runner.run_async( session_idsession.id, user_iduser_id, new_messagecontent, ): if event.is_final_response(): parts event.content.parts answer .join(p.text for p in parts if p.text) or 无响应 await updater.add_artifact([TextPart(textanswer)], nameanswer) await updater.complete() break except Exception as e: await updater.update_status( TaskState.failed, messagenew_agent_text_message(f错误{e!s}), ) raise async def _get_or_create_session(self, context_id: str, user_id: str): app_name self.runner.app_name if context_id: session await self.runner.session_service.get_session( app_nameapp_name, session_idcontext_id, user_iduser_id, ) if session: return session session await self.runner.session_service.create_session( app_nameapp_name, user_iduser_id, session_idcontext_id, ) return session async def cancel(self, context: RequestContext, event_queue: EventQueue) - NoReturn: raise ServerError(errorUnsupportedOperationError())5. 本地测试A2A智能体在部署之前我们需要在本地测试A2A智能体的功能# scripts/test_a2a_agent_local.py import asyncio import json import os from dotenv import load_dotenv from starlette.requests import Request from vertexai.preview.reasoning_engines import A2aAgent from reservation_agent.a2a_config import agent_card from reservation_agent.executor import ReservationAgentExecutor load_dotenv() def build_post_request(data: dict None, path_params: dict None) - Request: 构建模拟POST请求 scope { type: http, http_version: 1.1, headers: [(bcontent-type, bapplication/json)], app: None } if path_params: scope[path_params] path_params async def receive(): byte_data json.dumps(data).encode(utf-8) if data else b return {type: http.request, body: byte_data, more_body: False} return Request(scope, receive) async def wait_for_task(a2a_agent, task_id, max_retries30): 等待任务完成 for _ in range(max_retries): request build_post_request() result await a2a_agent.on_get_task(requestrequest, contextNone) state result.get(status, {}).get(state, ) if state in [completed, failed]: return result await asyncio.sleep(1) return result async def main(): # 创建本地A2A智能体 a2a_agent A2aAgent( agent_cardagent_card, agent_executor_builderReservationAgentExecutor ) a2a_agent.set_up() print( * 50) print(1. 测试智能体卡片检索...) print( * 50) # 测试智能体卡片 request build_post_request() card_response await a2a_agent.handle_authenticated_agent_card( requestrequest, contextNone ) print(f智能体: {card_response.get(name)}) print(f技能: {[s.get(name) for s in card_response.get(skills, [])]}) # 测试预订创建 print(\n * 50) print(2. 测试创建预订...) print( * 50) message_data { message: { messageId: fmsg-{os.urandom(4).hex()}, content: [{text: 预订周六晚上6点2人桌姓名Bob电话555-0202}], role: ROLE_USER, }, } request build_post_request(message_data) response await a2a_agent.on_message_send(requestrequest, contextNone) task_id response[task][id] context_id response[task].get(contextId) print(f任务ID: {task_id}) # 等待结果 result await wait_for_task(a2a_agent, task_id) print(f状态: {result.get(status, {}).get(state)}) for artifact in result.get(artifacts, []): if artifact.get(parts) and text in artifact[parts][0]: print(f响应: {artifact[parts][0][text]}) print(\n * 50) print(本地测试完成) print( * 50) if __name__ __main__: asyncio.run(main())运行测试uv run python scripts/test_a2a_agent_local.py6. 构建餐厅协调智能体现在创建主协调智能体负责将用户请求路由到相应的专业智能体# 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 google.auth.transport.requests import Request as AuthRequest # 配置环境变量 TOOLBOX_URL os.environ.get(TOOLBOX_URL, http://127.0.0.1:5000) RESERVATION_AGENT_CARD_URL os.environ.get(RESERVATION_AGENT_CARD_URL, ) 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(AuthRequest()) request.headers[Authorization] fBearer {self.credentials.token} yield request # 创建远程A2A预订智能体 reservation_remote_agent RemoteA2aAgent( namereservation_agent, description处理餐厅餐桌预订——创建、查询和取消预订。当用户要预订、查询或取消时委托给此智能体, agent_cardRESERVATION_AGENT_CARD_URL, httpx_clienthttpx.AsyncClient(authGoogleCloudAuth(), timeout60), ) # 创建主餐厅智能体 restaurant_agent LlmAgent( namerestaurant_agent, modelgemini-3.5-flash, # 可替换为DeepSeek模型 instruction你是Foodie Finds餐厅的友好知识渊博的门房。你的工作 - 帮助顾客按类别或菜系浏览菜单 - 提供菜品的完整详细信息包括成分、价格和饮食信息 - 根据顾客对想吃什么的自然语言描述推荐菜品 - 当被要求时添加新菜单项 - 对于预订请求预订、查询或取消餐桌委托给reservation_agent 当顾客按名称或菜系询问特定菜品时使用get-item-details工具。 当顾客要求特定类别或菜系类型时使用search-menu工具。 当顾客描述他们想要的食物类型——通过风味、质地、饮食需求或渴望——使用search-menu-by-description工具进行语义搜索。 在search-menu和search-menu-by-description之间不确定时优先使用search-menu-by-description——它搜索菜品描述并找到更相关的匹配项。 如果菜品不可用available为false告知顾客并从搜索结果中建议类似的替代品。 保持对话式、知识渊博且简洁。, tools[], # 这里可以添加MCP工具箱工具 sub_agents[reservation_remote_agent], )7. 系统集成与部署7.1 配置MCP工具箱集成创建MCP工具箱配置文件# tools.yaml toolsets: - name: restaurant-tools description: Tools for restaurant menu management tools: - name: search-menu description: Search menu items by category or cuisine inputSchema: type: object properties: category: type: string description: Category or cuisine to search for required: [category] - name: get-item-details description: Get detailed information about a specific menu item inputSchema: type: object properties: item_id: type: string description: ID of the menu item required: [item_id] - name: search-menu-by-description description: Semantic search for menu items based on natural language description inputSchema: type: object properties: description: type: string description: Natural language description of desired food required: [description]7.2 部署脚本创建完整的部署脚本# scripts/deploy_system.py import os import asyncio from pathlib import Path import vertexai from dotenv import load_dotenv from google.genai import types load_dotenv() async def deploy_reservation_agent(): 部署预订智能体到Agent Runtime PROJECT_ID os.environ[GOOGLE_CLOUD_PROJECT] REGION os.environ[REGION] STAGING_BUCKET os.environ.get(STAGING_BUCKET, f{PROJECT_ID}-adk-a2a-agent-runtime) vertexai.init(projectPROJECT_ID, locationREGION, staging_bucketfgs://{STAGING_BUCKET}) client vertexai.Client( projectPROJECT_ID, locationREGION, http_optionstypes.HttpOptions(api_versionv1beta1), ) print(部署预订智能体到Agent Runtime...) print(这可能需要3-5分钟...) # 这里需要导入实际的A2A智能体配置 from reservation_agent.a2a_config import agent_card from reservation_agent.executor import ReservationAgentExecutor from vertexai.preview.reasoning_engines import A2aAgent a2a_agent A2aAgent( agent_cardagent_card, agent_executor_builderReservationAgentExecutor, ) remote_agent client.agent_engines.create( agenta2a_agent, config{ display_name: agent_card.name, description: agent_card.description, requirements: [ google-cloud-aiplatform[agent_engines,adk]1.149.0, a2a-sdk0.3.26, google-adk1.29.0, ], extra_packages: [./reservation_agent], }, ) resource_name remote_agent.api_resource.name print(f部署完成资源名称: {resource_name}) # 保存资源名称到环境文件 env_path Path(.env) lines env_path.read_text().splitlines() if env_path.exists() else [] lines [l for l in lines if not l.startswith(RESERVATION_AGENT_RESOURCE_NAME)] lines.append(fRESERVATION_AGENT_RESOURCE_NAME{resource_name}) env_path.write_text(\n.join(lines) \n) return resource_name async def resolve_agent_card_url(resource_name: str): 解析智能体卡片URL PROJECT_ID os.environ[GOOGLE_CLOUD_PROJECT] REGION os.environ[REGION] vertexai.init(projectPROJECT_ID, locationREGION) client vertexai.Client( projectPROJECT_ID, locationREGION, http_optionstypes.HttpOptions(api_versionv1beta1), ) agent client.agent_engines.get(nameresource_name) card await agent.handle_authenticated_agent_card() card_url f{card.url}/v1/card print(f智能体: {card.name}) print(f卡片URL: {card_url}) # 保存到环境文件 for env_path in [Path(restaurant_agent/.env), Path(.env)]: lines env_path.read_text().splitlines() if env_path.exists() else [] lines [l for l in lines if not l.startswith(RESERVATION_AGENT_CARD_URL)] lines.append(fRESERVATION_AGENT_CARD_URL{card_url}) env_path.write_text(\n.join(lines) \n) return card_url async def main(): # 1. 部署预订智能体 resource_name await deploy_reservation_agent() # 2. 解析卡片URL card_url await resolve_agent_card_url(resource_name) print(\n *50) print(部署完成) print(*50) print(f预订智能体资源: {resource_name}) print(fA2A卡片URL: {card_url}) if __name__ __main__: asyncio.run(main())8. 完整系统测试创建端到端测试脚本验证整个多智能体系统# scripts/test_full_system.py import asyncio import os from dotenv import load_dotenv load_dotenv() async def test_menu_query(): 测试菜单查询功能 print(测试菜单查询...) # 这里实现菜单查询测试逻辑 print(菜单查询测试完成) async def test_reservation_flow(): 测试完整预订流程 print(测试预订流程...) # 1. 创建预订 print(1. 创建预订...) # 实现预订创建逻辑 # 2. 查询预订 print(2. 查询预订...) # 实现预订查询逻辑 # 3. 取消预订 print(3. 取消预订...) # 实现预订取消逻辑 print(预订流程测试完成) async def test_multi_agent_coordination(): 测试多智能体协调 print(测试多智能体协调...) # 测试协调智能体如何路由请求 test_cases [ 你们有什么意大利菜, # 应该路由到菜单查询 我想预订明天晚上7点的4人桌, # 应该路由到预订智能体 查询我电话123456的预订, # 应该路由到预订智能体 推荐一些辣的菜品 # 应该路由到菜单查询 ] for query in test_cases: print(f测试查询: {query}) # 实现路由测试逻辑 print(多智能体协调测试完成) async def main(): print(开始完整系统测试...) print( * 50) await test_menu_query() print(- * 30) await test_reservation_flow() print(- * 30) await test_multi_agent_coordination() print(- * 30) print(所有测试完成) print( * 50) if __name__ __main__: asyncio.run(main())9. 常见问题与排查思路9.1 部署问题排查问题现象可能原因排查方式解决方案部署失败提示权限不足服务账号缺少必要权限检查IAM权限设置授予aiplatform.user角色智能体卡片无法访问网络配置问题检查VPC和防火墙规则配置正确的网络规则会话状态不持久SessionService配置错误检查环境变量和配置确保使用VertexAiSessionService9.2 通信问题排查# 通信诊断脚本 import asyncio import aiohttp import os from dotenv import load_dotenv load_dotenv() async def diagnose_communication(): 诊断A2A通信问题 card_url os.environ.get(RESERVATION_AGENT_CARD_URL) if not card_url: print(错误未找到RESERVATION_AGENT_CARD_URL环境变量) return try: async with aiohttp.ClientSession() as session: # 测试卡片访问 async with session.get(card_url) as response: if response.status 200: print(✓ 智能体卡片可访问) card_data await response.json() print(f智能体名称: {card_data.get(name)}) else: print(f✗ 卡片访问失败: {response.status}) except Exception as e: print(f通信诊断失败: {e}) if __name__ __main__: asyncio.run(diagnose_communication())9.3 状态管理问题常见状态问题状态丢失检查SessionService配置和权限状态冲突确保使用正确的状态前缀和范围状态同步延迟验证网络连接和超时设置10. 最佳实践与工程建议10.1 智能体设计原则单一职责原则每个智能体只负责一个明确的业务领域接口标准化使用A2A协议确保智能体间的标准化通信状态隔离不同的智能体应该管理独立的状态错误处理实现完善的错误处理和重试机制10.2 部署和运维建议# 生产环境配置示例 监控配置: 指标监控: - 请求延迟 - 错误率 - 并发连接数 日志记录: - 访问日志 - 错误日志 - 性能日志 告警规则: - 错误率 1% - 平均延迟 2秒 - 服务不可用10.3 安全最佳实践身份验证使用Google Cloud IAM进行访问控制网络隔离在VPC内部署敏感服务数据加密启用CMEK进行数据加密审计日志启用完整的操作审计11. 扩展和优化方向11.1 性能优化# 智能体性能优化示例 from google.adk.agents import LlmAgent from google.adk.caching import DiskCache # 启用响应缓存 optimized_agent LlmAgent( nameoptimized_agent, modelgemini-3.5-flash, cacheDiskCache(), # 磁盘缓存 instruction..., tools[...], )11.2 扩展更多智能体基于现有架构可以轻松扩展更多专业智能体支付处理智能体处理订单支付配送跟踪智能体跟踪外卖配送状态客户服务智能体处理客户投诉和咨询11.3 与DeepSeek模型集成要将系统适配DeepSeek模型只需修改模型配置# 配置DeepSeek模型 deepseek_agent LlmAgent( namedeepseek_agent, modelCustomLLM( # 自定义DeepSeek模型适配器 modeldeepseek-chat, api_keyos.environ.get(DEEPSEEK_API_KEY) ), instruction..., tools[...], )通过本文的实战教程你应该已经掌握了多智能体系统开发的核心技能。从MCP工具集成到A2A智能体通信从本地测试到生产部署这套架构可以应用于各种复杂的业务场景。记住良好的智能体设计就像组建一个高效的团队让每个成员专注自己最擅长的工作通过标准化协议协同合作才能发挥最大的效能。建议将本文代码作为起点根据你的具体业务需求进行定制和扩展。多智能体架构的未来充满可能性现在正是深入实践的最佳时机。
