LangGraph与MCP实战：构建能调用工具的AI智能体工作流

在 AI 应用开发领域如何让大模型不仅能够回答问题还能主动调用工具、执行多步骤任务是提升智能体Agent能力的关键。LangGraph 作为 LangChain 生态中专注于工作流编排的框架结合模型上下文协议MCP为构建具备复杂推理和行动能力的 AI 智能体提供了强大的基础设施。本文将以打造一个“专属智能小秘书”为目标带你从零开始深度掌握 LangGraph 与 MCP 的实战集成理解其设计哲学并规避开发中的常见陷阱。这个智能小秘书将能够理解你的自然语言指令例如“帮我查一下北京明天天气然后提醒我下午三点开会”并自动分解任务先调用天气查询工具获取信息再创建日历提醒。我们将重点剖析 LangGraph 的状态管理、节点编排与循环控制机制以及 MCP 如何以标准化方式为模型提供丰富的工具调用能力。1. 理解 LangGraph 与 MCP 的核心价值1.1 为什么需要 LangGraph超越 LangChain 的简单链式调用LangChain 提供了Chain的概念能够将大模型调用、工具使用、数据检索等环节链接起来但其流程通常是线性的。当任务需要根据中间结果进行条件判断、循环执行或并行处理时简单的链式结构就显得力不从心。LangGraph 应运而生它引入了**有状态、可循环的工作流Stateful Graph**概念。你可以将整个应用建模为一个图Graph其中节点Nodes代表执行单元如调用模型、运行工具边Edges代表控制流决定下一个执行哪个节点。关键优势在于状态持久化整个工作流维护一个状态对象State所有节点都可以读取和修改这个状态使得信息能够在多个步骤间传递。循环与条件分支支持基于当前状态动态决定下一步走向实现if-else、while等逻辑这是构建能“思考”的 Agent 的核心。清晰的责任分离将复杂的任务分解为多个专注的节点每个节点只负责一件事代码更易维护。对于智能小秘书场景查询天气和创建提醒是两个独立的步骤但需要共享用户指令的解析结果如时间、地点。LangGraph 的状态管理正好用于传递这些共享信息。1.2 MCPModel Context Protocol是什么为什么它比传统工具调用更优传统的大模型工具调用如 OpenAI 的 Function Calling需要开发者在代码中硬编码工具的定义和实现。当工具数量多、来源杂如来自不同团队或第三方服务时管理和集成会变得复杂。MCP 是一个开放协议旨在标准化模型与工具或数据源之间的交互方式。它的核心思想是解耦MCP Server负责实际提供工具如天气查询、数据库操作、日历管理。它向客户端暴露一组标准化的工具列表。MCP Client通常是你的应用或 LangGraph 节点发现 Server 提供的工具并按照协议调用它们。这样做的好处是工具可发现性Client 可以动态地发现 Server 上有哪些工具可用无需提前硬编码。标准化接口不同的工具提供商只要遵循 MCP 协议就能轻松接入你的 AI 应用。更好的安全性可以对工具访问进行统一的权限控制。在我们的项目中智能小秘书所需的“天气查询”和“日历创建”就可以作为两个 MCP 工具由独立的 MCP Server 提供。1.3 LangGraph MCP 的协同工作模式结合两者典型的工作流如下LangGraph 工作流作为总控中心管理任务执行的整个生命周期和状态。工作流中的某个节点如agent_node负责与大模型交互。大模型根据当前状态和任务决定是否需要调用工具并选择要调用的 MCP 工具名称和参数。LangGraph 节点将模型的请求转发给对应的MCP Client。MCP Client通过 MCP 协议调用远端的MCP Server。MCP Server执行具体工具逻辑如调用天气 API并返回结果。结果被写回 LangGraph 的状态驱动工作流进入下一步。这种架构使得智能体能力扩展变得非常清晰要增加新功能只需开发并部署一个新的 MCP Server然后在 Client 端配置连接即可。2. 环境准备与项目初始化2.1 环境与依赖配置我们将使用 Python 作为开发语言。确保你的环境已安装 Python 3.10 或更高版本。首先创建项目目录并初始化虚拟环境。mkdir ai-personal-assistant cd ai-personal-assistant python -m venv venv # Windows 激活: venv\Scripts\activate # macOS/Linux 激活: source venv/bin/activate接下来安装核心依赖库。我们将使用langgraph来构建工作流langchain-openai来接入 OpenAI 模型或其他兼容 API 的模型并安装 MCP 相关的客户端库。pip install langgraph langchain-openai # 安装一个基础的 MCP 客户端库例如来自 LangChain 社区的 mcp-client # 注意MCP 生态正在快速发展具体包名可能变化请以最新文档为准。 # pip install mcp-client由于 MCP 的 Python 客户端库尚在快速迭代中一个更稳定且易于理解的方式是使用subprocess模块或requests库与 MCP Server可能是用其他语言如 TypeScript 编写进行 HTTP 通信。本文将以一个模拟的 HTTP MCP Server 为例进行说明。2.2 项目结构设计一个清晰的项目结构有助于管理复杂度。建议如下ai-personal-assistant/ ├── requirements.txt # 项目依赖 ├── .env # 环境变量如 API Keys ├── src/ │ ├── __init__.py │ ├── mcp_client.py # MCP 客户端封装 │ ├── graph/ │ │ ├── __init__.py │ │ ├── state.py # 定义工作流状态类 │ │ └── assistant_graph.py # 核心 LangGraph 定义 │ └── tools/ │ ├── __init__.py │ └── weather.py # 模拟天气工具 MCP Server (简易版) └── main.py # 应用入口在.env文件中配置你的 OpenAI API KeyOPENAI_API_KEYyour_openai_api_key_here2.3 创建基础状态类LangGraph 的工作流围绕一个状态对象运转。我们首先在src/graph/state.py中定义状态类。from typing import Annotated, List, Dict, Any, Optional from typing_extensions import TypedDict from langgraph.graph.message import add_messages class AssistantState(TypedDict): # 消息历史LangGraph 内置支持 messages: Annotated[List[Dict[str, Any]]], add_messages] # 当前最新的用户输入 current_input: str # 模型决定要调用的工具名称如果有 next_tool_to_call: Optional[str] # 模型为工具调用准备的参数 tool_arguments: Optional[Dict[str, Any]] # 最近一次工具调用的结果 latest_tool_result: Optional[str] # 工作流是否应该继续用于控制循环 should_continue: bool这个AssistantState类型定义了一个字典的结构它包含了工作流运行过程中需要跟踪的所有信息。Annotated和add_messages用于方便地处理消息列表的追加操作。3. 构建 MCP 客户端与工具模拟3.1 实现一个简易的 MCP 客户端如前所述我们将模拟一个通过 HTTP 与 MCP Server 交互的客户端。在src/mcp_client.py中实现import requests import json from typing import List, Dict, Any class SimpleMCPClient: def __init__(self, server_base_url: str): self.server_base_url server_base_url def list_tools(self) - List[Dict[str, Any]]: 向 MCP Server 请求可用的工具列表 try: response requests.get(f{self.server_base_url}/tools) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: print(fError fetching tools from MCP Server: {e}) return [] def call_tool(self, tool_name: str, arguments: Dict[str, Any]) - str: 调用指定的 MCP 工具 try: payload {arguments: arguments} response requests.post( f{self.server_base_url}/tools/{tool_name}/execute, jsonpayload, headers{Content-Type: application/json} ) response.raise_for_status() result response.json() return result.get(content, Tool executed but returned no content.) except requests.exceptions.RequestException as e: error_msg fError calling tool {tool_name}: {e} print(error_msg) return error_msg # 假设我们的 MCP Server 运行在本地 3000 端口 mcp_client SimpleMCPClient(http://localhost:3000)3.2 创建模拟的 MCP 工具 Server为了演示我们需要一个简单的 MCP Server。这里用 Python 的Flask快速创建一个。在src/tools/weather.py中from flask import Flask, jsonify, request app Flask(__name__) # 模拟工具列表 app.route(/tools, methods[GET]) def list_tools(): tools [ { name: get_weather, description: Get the current weather for a city., parameters: { type: object, properties: { city: {type: string, description: The city name.} }, required: [city] } }, { name: create_reminder, description: Create a calendar reminder., parameters: { type: object, properties: { title: {type: string, description: The reminder title.}, time: {type: string, description: The time of the reminder.} }, required: [title, time] } } ] return jsonify(tools) # 模拟天气查询工具 app.route(/tools/get_weather/execute, methods[POST]) def execute_get_weather(): data request.json city data.get(arguments, {}).get(city, Unknown City) # 模拟返回数据 weather_info fThe weather in {city} is sunny, 25°C. return jsonify({content: weather_info}) # 模拟创建提醒工具 app.route(/tools/create_reminder/execute, methods[POST]) def execute_create_reminder(): data request.json title data.get(arguments, {}).get(title, Untitled) time data.get(arguments, {}).get(time, Unknown Time) # 模拟创建成功 reminder_info fReminder {title} set for {time}. return jsonify({content: reminder_info}) if __name__ __main__: app.run(port3000, debugTrue)运行python src/tools/weather.py启动这个模拟 MCP Server。在生产环境中MCP Server 可能会用更高效的语言如 Node.js实现并部署为独立的服务。4. 组装 LangGraph 智能体工作流4.1 定义工作流节点核心逻辑在src/graph/assistant_graph.py。我们需要定义几个关键节点。首先引入依赖并初始化模型和 MCP 客户端。import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI from src.mcp_client import mcp_client from src.graph.state import AssistantState load_dotenv() # 初始化大语言模型 llm ChatOpenAI(modelgpt-3.5-turbo, api_keyos.getenv(OPENAI_API_KEY)) # 让模型具备工具调用的能力这里我们手动构建工具列表实际可从 MCP Client 动态获取 # 动态获取示例: tools_from_mcp mcp_client.list_tools() tools_from_mcp [ { name: get_weather, description: Get the current weather for a city., parameters: { type: object, properties: { city: {type: string, description: The city name.} }, required: [city] } }, { name: create_reminder, description: Create a calendar reminder., parameters: { type: object, properties: { title: {type: string, description: The reminder title.}, time: {type: string, description: The time of the reminder.} }, required: [title, time] } } ] # 将工具描述绑定到模型简化版实际 LangChain 有更优雅的绑定方式 llm_with_tools llm.bind_tools(tools_from_mcp)接下来定义agent_node这是与大模型交互的核心节点。from langgraph.prebuilt import ToolNode from langchain_core.messages import HumanMessage, AIMessage, ToolMessage def agent_node(state: AssistantState): 节点与大模型交互决定下一步行动回复或调用工具 print(f[Agent Node] Current state: {state}) # 准备输入给模型的消息历史 model_messages state[messages] # 如果上一步有工具执行结果需要作为 ToolMessage 加入对话历史 if state.get(latest_tool_result): # 为工具结果创建一个消息。通常需要关联一个 tool_call_id这里简化处理。 tool_message ToolMessage(contentstate[latest_tool_result], tool_call_idcall_1) model_messages.append(tool_message) # 清空结果避免下次重复添加 state[latest_tool_result] None # 调用模型 response llm_with_tools.invoke(model_messages) # 更新消息历史 new_messages model_messages [response] state[messages] new_messages # 解析模型的响应判断是直接回复还是要求调用工具 if hasattr(response, tool_calls) and response.tool_calls: # 模型要求调用工具本例假设一次只调用一个工具 tool_call response.tool_calls[0] state[next_tool_to_call] tool_call[name] state[tool_arguments] tool_call[args] state[should_continue] True # 需要继续执行以调用工具 else: # 模型直接给出文本回复 state[next_tool_to_call] None state[tool_arguments] None state[should_continue] False # 工作流可以结束 return state然后定义tool_node负责执行模型指定的工具。def tool_node(state: AssistantState): 节点执行模型指定的工具调用 tool_name state[next_tool_to_call] arguments state[tool_arguments] if not tool_name: print([Tool Node] No tool to call.) return state print(f[Tool Node] Calling tool: {tool_name} with args: {arguments}) # 通过 MCP Client 调用工具 tool_result mcp_client.call_tool(tool_name, arguments) # 将工具执行结果存入状态 state[latest_tool_result] tool_result # 清空工具调用指令避免重复执行 state[next_tool_to_call] None state[tool_arguments] None # 工具执行后需要继续让 Agent 节点处理结果 state[should_continue] True return state4.2 编排节点与定义条件边现在我们将节点组装成图并定义控制流逻辑。from langgraph.graph import StateGraph, END def should_continue(state: AssistantState): 条件判断函数决定工作流下一步是调用工具还是结束 if state.get(should_continue, False): # 如果需要继续且下一个动作是调用工具则走向 Tool 节点 if state.get(next_tool_to_call): return call_tool else: # 否则继续让 Agent 思考例如工具执行后需要模型总结 return agent else: # 不需要继续则结束工作流 return END # 创建图构建器 graph_builder StateGraph(AssistantState) # 添加节点 graph_builder.add_node(agent, agent_node) graph_builder.add_node(call_tool, tool_node) # 设置入口点 graph_builder.set_entry_point(agent) # 定义边控制流 graph_builder.add_conditional_edges( agent, # 从 agent 节点出发 should_continue, # 根据条件判断下一个节点 { call_tool: call_tool, # 条件返回 call_tool则去 call_tool 节点 agent: agent, # 条件返回 agent则循环回 agent 节点 END: END # 条件返回 END则结束 } ) # 从工具节点执行完后总是回到 Agent 节点去处理结果 graph_builder.add_edge(call_tool, agent) # 编译图得到可执行的工作流 assistant_graph graph_builder.compile()5. 运行与验证智能小秘书5.1 创建应用入口在main.py中我们创建一个简单的交互循环。from src.graph.assistant_graph import assistant_graph from src.graph.state import AssistantState from langchain_core.messages import HumanMessage import asyncio async def main(): print(智能小秘书已启动输入您的要求如北京明天天气怎么样 或 提醒我下午三点开会输入 quit 退出。) # 确保模拟 MCP Server 正在运行 (http://localhost:3000) while True: user_input input(\n您: ).strip() if user_input.lower() in [quit, exit, q]: print(再见) break if not user_input: continue # 初始化工作流状态 initial_state: AssistantState { messages: [HumanMessage(contentuser_input)], current_input: user_input, next_tool_to_call: None, tool_arguments: None, latest_tool_result: None, should_continue: True } print(小秘书正在思考...) # 执行工作流 final_state None # LangGraph 的 compile().stream 可以流式输出这里用 invoke 获取最终结果 for step in assistant_graph.stream(initial_state): # stream 会返回每个节点执行后的状态 for node_name, node_state in step.items(): print(f[Graph Stream] Node {node_name} completed.) final_state node_state # 可以在这里实时打印模型或工具的输出 # 从最终状态中提取模型的最后一条消息作为回复 if final_state and messages in final_state: # 最后一条消息通常是 AI 的回复 last_message final_state[messages][-1] if hasattr(last_message, content): print(f小秘书: {last_message.content}) else: print(小秘书: 执行了操作但无直接回复) else: print(小秘书: 处理过程出现意外。) if __name__ __main__: asyncio.run(main())5.2 启动与测试在一个终端启动模拟 MCP Serverpython src/tools/weather.py你应该看到输出表明 Server 运行在http://127.0.0.1:3000。在另一个终端运行主程序python main.py进行测试单步任务输入“北京明天天气怎么样”。模型应识别出需要调用get_weather工具工作流执行工具后模型将结果组织成自然语言回复给你。多步任务输入“帮我查一下上海天气然后提醒我晚上八点健身”。模型会先调用天气工具将结果作为上下文再调用创建提醒工具。直接对话输入“你好”。模型判断无需工具直接回复。观察控制台输出你可以清晰地看到工作流在agent和call_tool节点之间的跳转以及状态的演变。6. 常见问题排查与优化6.1 典型问题与解决方案问题现象可能原因检查与解决启动报错ModuleNotFoundError依赖未安装或虚拟环境未激活确认激活 venv 并执行pip install -r requirements.txtMCP Server 连接失败Server 未启动、端口被占用或 URL 错误检查python src/tools/weather.py是否成功运行确认mcp_client.py中的server_base_url正确模型不调用工具直接回复1. 工具描述不够清晰。2. 模型指令System Prompt未强调使用工具。3. 用户输入意图不明显。1. 优化工具的description和parameters。2. 在发给模型的首条消息如 System Message中明确其助手身份和可用工具。3. 在agent_node中确保消息历史包含工具定义。工作流陷入死循环should_continue逻辑有误状态should_continue始终为 True。检查agent_node和tool_node中对state[should_continue]的赋值逻辑确保在最终回复后将其设为 False。工具调用参数错误模型生成的参数格式与 MCP Server 期望不符。在tool_node中打印arguments对比 MCP Server 的日志调整工具定义或模型指令。6.2 性能与生产环境优化建议状态序列化当前状态存储在内存中。生产环境需要将其序列化如到数据库或 Redis以支持长时间运行的任务和容错。异步调用将agent_node和tool_node改为异步函数使用ainvoke和异步 HTTP 客户端提升并发性能。工具路由如果工具很多可以设计更复杂的路由机制而不是在单个tool_node中处理所有调用。错误处理与重试在 MCP 客户端和工具节点中加入更健壮的错误处理、超时控制和重试逻辑。可观测性集成日志记录如structlog和指标监控如 Prometheus跟踪工作流执行时长、工具调用成功率等。安全性对 MCP Server 进行认证和授权确保只有合法的 Client 可以调用工具。对用户输入和工具参数进行验证和过滤。7. 扩展方向与深入学习掌握了本项目的核心模式后你可以从以下几个方向深化集成真实工具将模拟的天气和日历工具替换为真实的 API如和风天气、Google Calendar API。探索复杂工作流实现需要多次“思考-行动”循环的任务如网上购物、旅行规划等。动态工具加载实现 MCP Client 在启动时或运行时动态从多个 MCP Server 发现工具真正体现 MCP 的优势。UI 界面使用Gradio或Streamlit为你的智能小秘书构建一个 Web 界面。深入研究 LangGraph学习其更高级的特性如检查点Checkpointing用于持久化状态、并行执行、子图等。通过这个实战项目你不仅学会了 LangGraph 和 MCP 的基本用法更重要的是理解了构建具备工具使用能力的 AI 智能体的核心架构思想。这种“规划-执行-观察”的循环是迈向更高级 AI 应用的基础。