手把手教你构建多代理AI系统：MCP+A2A+LangGraph实战！

引言

在 AI Agent 开发领域，MCP（Model Context Protocol，模型上下文协议）用于标准化工具和资源的访问，让 LLM 能无缝调用外部数据源；A2A（Agent2Agent，代理间协议）则实现代理间的标准化通信，支持多代理协作；LangGraph 是 LangChain 的扩展库，用于构建状态ful、多演员的代理工作流，支持复杂决策图谱。这些技术结合，能打造高效、可扩展的智能体系统。本指南基于实际案例（如天气代理、货币转换代理和采购管家），提供全流程开发实战，从环境搭建到部署集成。目标：构建一个多代理系统，其中一个 LangGraph 代理通过 MCP 调用工具，并经 A2A 与远程代理协作。

适用人群：有 Python 和 LLM 基础的开发者。预计耗时：2-4 小时。

前置条件

• 环境

  ：Python 3.10+，Node.js（若集成 Chrome DevTools MCP）。

• 库安装

  ：

pip install langgraph langchain-openai fastmcp a2a-sdk httpx pydantic requests asyncio

• API 密钥

  ：OpenAI 或 Google Gemini API key；可选：OpenWeatherMap API key（天气示例）、Frankfurter API（货币示例）。

• 部署

  ：Google Cloud 项目（免费额度内），gcloud CLI。

• GitHub 资源

  ：克隆示例仓库，如 MCP-A2A-step-by-step。

步骤 1: 构建 MCP Server（工具暴露层）

MCP Server 作为工具后端，提供标准化 API（如天气查询）。我们用 FastMCP 实现一个天气工具服务器。

代码实现（mcp_server.py）

import asyncioimport osimport requestsfrom fastmcp import FastMCPfrom pydantic import BaseModelmcp = FastMCP("Weather MCP Server")class CurrentWeather(BaseModel): location: str temperature: float description: str@mcp.tool()def get_current_weather(location: str) -> str: """获取指定地点的当前天气""" url = f"http://api.openweathermap.org/data/2.5/weather?q={location}&appid={os.getenv('OPENWEATHERMAP_API_KEY')}&units=metric" response = requests.get(url).json() temp = response['main']['temp'] desc = response['weather'][0]['description'] return f"{location} 当前天气：{temp}°C，{desc}"@mcp.prompt()def weather_analysis_prompt(location: str, data: str) -> str: """天气分析提示模板""" return f"基于以下数据分析 {location} 的穿衣建议：{data}"if __name__ == "__main__": asyncio.run(mcp.run_async(transport="streamable-http", host="0.0.0.0", port=8080))

运行与测试

• 运行：python mcp_server.py。
• 测试：用 curl 发送 JSON-RPC 请求：

curl -X POST http://localhost:8080/jsonrpc -H "Content-Type: application/json" -d '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{}}'

• 部署到 Cloud Run：构建 Docker 镜像并gcloud run deploy mcp-weather --source .。

此服务器暴露工具，供代理发现和调用。

步骤 2: 使用 LangGraph 构建核心代理（MCP 集成）

LangGraph 用于编排代理逻辑：LLM + MCP 工具 + 状态管理。我们构建一个“天气顾问”代理，查询 MCP 工具并生成建议。

代码实现（langgraph_agent.py）

from langgraph.prebuilt import create_react_agentfrom langchain_openai import ChatOpenAIfrom langchain_core.tools import toolfrom fastmcp import Clientimport asyncioimport jsonimport osllm = ChatOpenAI(model="gpt-4o-mini", api_key=os.getenv("OPENAI_API_KEY"))mcp_url = "http://localhost:8080" # 或 Cloud Run URLasync def discover_mcp_tools(): async with Client(mcp_url) as client: tools = await client.list_tools() return toolsdef create_langchain_tool(mcp_tool): def tool_func(input_str: str) -> str: args = json.loads(input_str) if input_str.startswith('{') else {"location": input_str} # 同步调用 MCP loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) result = loop.run_until_complete(call_mcp_tool(mcp_tool.name, args)) loop.close() return result return tool(name=mcp_tool.name, description=mcp_tool.description, func=tool_func)async def call_mcp_tool(tool_name: str, args: dict): async with Client(mcp_url) as client: result = await client.call_tool(tool_name, args) return result# 发现工具并创建代理tools_data = asyncio.run(discover_mcp_tools())tools = [create_langchain_tool(t) for t in tools_data]agent = create_react_agent(llm, tools, checkpointer=MemorySaver()) # 需 pip install langgraph-checkpointdef run_agent(query: str): return agent.invoke({"messages": [("user", query)]})

测试

• 查询：run_agent("纽约明天穿什么？")→ 代理调用 MCP 获取天气，生成建议。
• 扩展：集成 Chrome DevTools MCP（npx chrome-devtools-mcp@latest），添加浏览器自动化工具，如@mcp.tool() def navigate_page(url: str)。

此步实现单代理的工具调用与推理。

步骤 3: 暴露代理为 A2A Server（代理间通信）

A2A 让 LangGraph 代理成为远程服务，支持发现（Agent Card）和任务执行。我们构建货币转换代理作为示例。

Agent Card（agent.json）

{ "name": "currency_agent", "description": "货币转换代理", "protocolVersion": "0.2.6", "capabilities": {"streaming": true}, "defaultInputModes": ["text"], "defaultOutputModes": ["text"], "skills": [{"id": "convert_currency", "name": "Convert Currency", "description": "转换货币汇率"}], "url": "https://your-cloud-run-url.com"}

放置在/static/.well-known/agent.json。

代码实现（a2a_server.py）

from langgraph.prebuilt import create_react_agentfrom langchain_google_genai import ChatGoogleGenerativeAI # 或 OpenAIfrom a2a import A2AStarletteApplication, InMemoryTaskStorefrom a2a.events import TaskStatusUpdateEvent, TaskArtifactUpdateEventimport requestsfrom pydantic import BaseModelllm = ChatGoogleGenerativeAI(model="gemini-1.0-pro")store = InMemoryTaskStore()@tooldef get_exchange_rate(from_currency: str, to_currency: str) -> str: url = f"https://api.frankfurter.app/latest?from={from_currency}&to={to_currency}" data = requests.get(url).json() return f"1 {from_currency} = {data['rates'][to_currency]} {to_currency}"tools = [get_exchange_rate]graph = create_react_agent(llm, tools)class CurrencyAgent: def __init__(self): self.graph = graph def invoke(self, query: str, context_id: str): result = self.graph.invoke({"messages": [("user", query)]}, {"configurable": {"thread_id": context_id}}) return result["messages"][-1].content async def stream(self, query: str, context_id: str): for chunk in self.graph.stream({"messages": [("user", query)]}, {"configurable": {"thread_id": context_id}}): yield chunkclass CurrencyExecutor: def __init__(self): self.agent = CurrencyAgent() self.store = store async def on_message_send(self, request): # A2A 入口 task = self.store.create_task(request.context_id, request.message_id) task.status = "working" self.store.enqueue_event(task.id, TaskStatusUpdateEvent(status="working")) result = self.agent.invoke(request.params["message"]["parts"][0]["text"], request.context_id) artifact = TaskArtifactUpdateEvent(artifact={"text": result}) self.store.enqueue_event(task.id, artifact) task.status = "completed" return taskapp = A2AStarletteApplication(executor=CurrencyExecutor(), task_store=store)if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8080)

部署

• Docker 化并gcloud run deploy currency-a2a --source . --allow-unauthenticated。
• 测试：客户端 GET/ .well-known/agent.json发现卡片。

步骤 4: 构建 A2A Client 代理（多代理协作）

客户端代理调用远程 A2A Server，实现协作。如采购管家调用汉堡/披萨卖家代理。

代码实现（a2a_client.py）

import httpxfrom a2a import A2AClient, SendMessageRequestfrom langgraph.prebuilt import create_react_agentfrom langchain_openai import ChatOpenAIimport uuidllm = ChatOpenAI(model="gpt-4o-mini")remote_url = "https://currency-a2a-xxxx.run.app" # A2A Server URL@tooldef call_remote_agent(agent_name: str, task: str) -> str: async def _call(): async with httpx.AsyncClient() as client: resolver = A2ACardResolver(httpx_client=client, base_url=remote_url) card = await resolver.get_agent_card() a2a_client = A2AClient(httpx_client=client, agent_card=card) req = SendMessageRequest( id=str(uuid.uuid4()), params={"message": {"role": "user", "parts": [{"kind": "text", "text": task}]}} ) resp = await a2a_client.send_message(req) return resp.root.result.artifacts[0].parts[0].root.text return asyncio.run(_call())tools = [call_remote_agent]client_agent = create_react_agent(llm, tools)def run_client(query: str): return client_agent.invoke({"messages": [("user", query)]})# 示例：run_client("将100美元转换为欧元")

测试多代理

• 查询：”帮我订购一个汉堡和披萨” → 客户端发现远程代理，发送任务，聚合响应。

步骤 5: 全流程集成与优化

• Chrome DevTools MCP 集成

  ：在 LangGraph 中添加浏览器工具：

# 运行 npx chrome-devtools-mcp@latest # 在工具中调用：@tool def debug_page(url: str): ... 使用 MCP Client

用于前端调试，如代理生成代码后自动测试。

• 状态管理

  ：用 LangGraph 的 Checkpointer 持久化会话。

• 部署全栈

  ：MCP → LangGraph A2A Server → Client on Agent Engine。

• 性能优化

  ：添加流式响应（A2A streaming），错误处理（重试 MCP 调用）。

项目案例

1. 天气顾问

   ：MCP 天气工具 + LangGraph 分析 → A2A 分享给旅行代理。

2. 货币转换

   ：如上，处理多轮澄清（如缺失目标货币）。

3. 采购管家

   ：LangGraph 协调远程汉堡/披萨 A2A 代理，处理订单。

注意事项

• 安全

  ：A2A 用认证（如 OAuth），MCP 限本地访问。

• 调试

  ：用 LangSmith 追踪图执行。

• 扩展

  ：结合 RAG 或多模态（图像工具）。

• 资源

  ：参考 Google Codelab A2A 采购示例，完整代码见 GitHub。

通过此流程，你能从零构建生产级多代理系统。实践时从天气示例起步，逐步添加 A2A。

如何学习AI大模型？

如果你对AI大模型入门感兴趣，那么你需要的话可以点击这里大模型重磅福利：入门进阶全套104G学习资源包免费分享！

这份完整版的大模型 AI 学习和面试资料已经上传CSDN，朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】

这是一份大模型从零基础到进阶的学习路线大纲全览，小伙伴们记得点个收藏！

第一阶段：从大模型系统设计入手，讲解大模型的主要方法；

第二阶段：在通过大模型提示词工程从Prompts角度入手更好发挥模型的作用；

第三阶段：大模型平台应用开发借助阿里云PAI平台构建电商领域虚拟试衣系统；

第四阶段：大模型知识库应用开发以LangChain框架为例，构建物流行业咨询智能问答系统；

第五阶段：大模型微调开发借助以大健康、新零售、新媒体领域构建适合当前领域大模型；

第六阶段：以SD多模态大模型为主，搭建了文生图小程序案例；

第七阶段：以大模型平台应用与开发为主，通过星火大模型，文心大模型等成熟大模型构建大模型行业应用。

100套AI大模型商业化落地方案

大模型全套视频教程

200本大模型PDF书籍

👉学会后的收获：👈

• 基于大模型全栈工程实现（前端、后端、产品经理、设计、数据分析等），通过这门课可获得不同能力；

• 能够利用大模型解决相关实际项目需求： 大数据时代，越来越多的企业和机构需要处理海量数据，利用大模型技术可以更好地处理这些数据，提高数据分析和决策的准确性。因此，掌握大模型应用开发技能，可以让程序员更好地应对实际项目需求；

• 基于大模型和企业数据AI应用开发，实现大模型理论、掌握GPU算力、硬件、LangChain开发框架和项目实战技能， 学会Fine-tuning垂直训练大模型（数据准备、数据蒸馏、大模型部署）一站式掌握；

• 能够完成时下热门大模型垂直领域模型训练能力，提高程序员的编码能力： 大模型应用开发需要掌握机器学习算法、深度学习框架等技术，这些技术的掌握可以提高程序员的编码能力和分析能力，让程序员更加熟练地编写高质量的代码。

LLM面试题合集

大模型产品经理资源合集

大模型项目实战合集

👉获取方式：
😝有需要的小伙伴，可以保存图片到wx扫描二v码免费领取【保证100%免费】🆓