这是一条完整的学习与实战路线,不是单个开源项目:LangChain + MCP。只要在做 LLM 应用开发,2026 年大概率绕不开这两个词。LangChain 解决的是 Agent、检索、编排这些上层问题,MCP 解决的是 Agent 如何标准化连接外部工具和数据源。把两者打通,等于把“AI 能做什么”从聊天扩展到了真正可执行的工具系统。
网上讲 LangChain 的教程很多,讲 MCP 的也有,但能把两者串成一条可落地的学习路径、并且包含可直接运行代码的内容确实少。这篇文章会按照“概念 -> 环境 -> MCP Server 开发 -> LangChain 集成 -> Agent 实战 -> 批量与 API -> 排查清单”的顺序展开,每一步都有可复制的代码。看完你能回答这几个问题:MCP 在 LangChain 里到底怎么接入;Agent 为什么有时候调不到工具;LangGraph 和 LangChain 是什么关系;怎么把一个 MCP Server 封成批量任务和 HTTP 接口。
先说结论:这套组合适合做 Agent 工具链、内部知识库问答、办公自动化、业务系统接入 AI 能力等场景;不需要本地大模型,CPU 机器也能完成代码开发和联调;真正消耗资源的是模型推理。下面直接进入内容。
1. 核心能力速览
| 能力项 | 说明 |
|---|---|
| 技术栈 | LangChain、LangGraph、MCP(Model Context Protocol) |
| 核心目标 | 让 Agent 通过标准化协议调用外部工具和数据源 |
| 开发语言 | Python 3.10+ |
| 硬件要求 | CPU 可开发,推理阶段依赖模型服务(云端 API 或本地模型) |
| 模型支持 | OpenAI 兼容接口、Ollama、vLLM、各类国产模型 API |
| 启动方式 | 命令行启动 MCP Server、Python 脚本运行 Agent、FastAPI 暴露 HTTP 接口 |
| 接口能力 | 可通过 FastAPI 封装为 HTTP API |
| 批量任务 | 支持,通过遍历任务目录调用 Agent 实现 |
| 核心依赖 | langchain、langchain-openai、langgraph、mcp、langchain-mcp-adapters、fastmcp |
| 适合场景 | RAG 问答、Agent 工具调用、办公自动化、业务系统接入、MCP Server 开发测试 |
这里说明一下:LangChain 是应用框架,MCP 是工具连接协议,LangGraph 是有状态的任务编排引擎,三者可以组合使用,也可以单独使用。下面会逐个讲清楚。
2. LangChain、MCP、LangGraph 到底分别解决什么问题
2.1 LangChain:LLM 应用的组件库
LangChain 不是一个大模型,也不是一个单独的工具,而是把 LLM 应用开发需要的通用能力抽象成了可组合的组件,包括模型封装、提示词管理、检索器、记忆、输出解析、Agent 框架等。
一个典型 LangChain 应用的流程是:用户输入 -> 检索相关文档(RAG)-> 拼装提示词 -> 调用模型 -> 解析输出 -> 返回结果。如果只做简单问答,LangChain 的优势不明显;但一旦涉及工具调用、多轮对话、知识库检索,组件化设计的价值就体现出来了。
热词里经常有人问“LangChain 过时了吗”。准确地说,LangChain 的核心抽象仍然是当前很多生产应用的基础,但更复杂的流程编排确实在往 LangGraph 迁移。这不是过时,是抽象层级在变。
2.2 MCP:Agent 连接外部工具的标准协议
MCP(Model Context Protocol)是一个开放协议,目标是统一 AI 应用与外部工具、数据源之间的连接方式。在没有 MCP 之前,每个 Agent 项目都要自己写工具调用逻辑,调用方式各不相同。MCP 定义了 Client、Server、Tool、Resource 这些标准概念,让工具提供方实现一次,就能被多种支持 MCP 的客户端复用。
MCP Server 可以运行在本地,通过 stdio 和父进程通信;也可以部署为远程服务,通过 HTTP/SSE 方式调用。你可以在 MCP Server 里暴露一个查询库存的函数、一个读写文件的工具、一个调用内部 API 的操作,然后让 Agent 在需要时自动调用。
需要特别注意的是:MCP 本身不限制权限。一个 MCP Server 暴露了什么工具,Agent 就能调什么工具。权限边界靠开发者在实现 Server 时控制,这也是后文会反复强调的安全点。
2.3 LangGraph:有状态、可控制的任务编排引擎
LangGraph 提供了图执行能力,支持循环、分支、条件跳转、checkpoint 记忆,适合编排多步骤、有状态、需要人工参与确认的复杂任务。它不是替代 LangChain,而是在 LangChain 基础上补充了更细粒度的流程控制。
热词里有大量“LangChain 和 LangGraph 的区别”。简单区分:
- LangChain 偏组件库,适合快速搭一条处理链路。
- LangGraph 偏编排引擎,适合 Agent 循环、人工审批、多分支判断。
- LangChain 底层已经大量使用 LangGraph,两者不是竞争,而是不同抽象层。
3. 适用场景与使用边界
适合用这套组合的场景非常明确:
- Agent 工具接入:让 LLM 调用业务 API、数据库查询、文档读取、计算器等外部能力。
- RAG 知识库问答:接入向量库,结合 LangChain 的检索组件做带引用的问答。
- 办公自动化:用 MCP Server 暴露文件处理和表格操作工具,Agent 按指令执行批量操作。
- 内部系统智能助手:通过 MCP 接入工单、权限、数据平台,实现自然语言操作入口。
- MCP Server 开发与验证:独立开发一个标准 MCP Server,再被 LangChain Agent 或其他客户端调用。
不适合的场景也要说清楚:如果只是单轮普通问答,直接调模型 API 即可,不需要 LangChain + MCP;如果工具数量少且固定,手写工具函数可能比引入 MCP 更轻;如果流程极度简单,用 LangGraph 属于过度设计。
合规边界非常重要。LangChain 接入 MCP 后,Agent 具备了调用外部系统的能力,这意味着权限管理、数据隐私、操作审计都必须提前设计:
- 调用第三方平台、内部系统接口前,必须确认已经获得授权。
- 涉及用户个人信息、企业敏感数据的工具,要设置访问白名单和操作审计。
- 使用网络搜索、文件读写、数据库操作类 MCP Server 时,要严格限定访问范围。
- 涉及人脸、声音、版权素材等内容生成或处理时,必须确认素材来源合法、用途合规。
- 不要随意连接来路不明的远程 MCP 服务,避免数据被回传到不可控的第三方。
4. 环境准备与前置条件
以下环境信息基于常见开发配置编写,具体版本以你本机环境为准,建议不要直接照抄版本号,而是先确认兼容性。
4.1 操作系统与 Python
开发环境推荐:
- 操作系统:Windows 10/11、macOS、主流 Linux 发行版均可。
- Python 版本:3.10 或更高,推荐 3.11/3.12。
- 包管理工具:建议使用
uv或venv + pip。
这里给出使用venv初始化环境的通用命令:
python -m venv .venv # Windows .venv\Scripts\activate # macOS / Linux source .venv/bin/activate4.2 安装核心依赖
pip install "langchain>=0.3" langchain-openai langgraph mcp langchain-mcp-adapters fastmcp fastapi uvicorn如果部分包安装较慢,可以按项目实际需要分步安装。langchain-mcp-adapters是 LangChain 官方提供的 MCP 适配层,版本需要和mcp包匹配,后面会专门讲兼容性问题。
4.3 模型服务
LangChain 默认支持 OpenAI 接口,也兼容所有实现了 OpenAI 接口的服务。如果你有本地模型环境,也可以通过 Ollama、vLLM 等以 OpenAI 兼容协议接入。
配置方式推荐使用环境变量:
export OPENAI_API_KEY="你的密钥" export OPENAI_BASE_URL="https://api.openai.com/v1"如果使用其他兼容服务,把OPENAI_BASE_URL改成对应服务地址即可。注意:不要把真实密钥硬编码在代码里,后续排错和权限管理都会更麻烦。
4.4 验证安装是否成功
python -c "import langchain, langgraph, mcp; print('deps ok')"能正常打印deps ok,说明基础依赖装好了。如果报 ModuleNotFoundError,按提示补齐对应包。
5. 第一个 MCP Server:从零开发可调用工具
下面写一个最小可用的 MCP Server。它不依赖 LangChain,可以独立运行,也可以被任何 MCP Client 连接。
5.1 代码实现
# mcp_demo_server.py from mcp.server.fastmcp import FastMCP mcp = FastMCP("demo-tools") @mcp.tool() def add(a: int, b: int) -> int: """两数相加,适合做最简单验证""" return a + b @mcp.tool() def get_city_weather(city: str) -> str: """查询指定城市天气。演示接口,返回模拟数据。""" # 实际项目应改为调用真实天气服务,并确认数据使用授权 return f"{city} 天气:晴,26 摄氏度(模拟数据)" if __name__ == "__main__": mcp.run()这里用了 FastMCP,它是 MCP 官方提供的快速开发封装。@mcp.tool()注册的是 Agent 可以直接调用的工具,@mcp.resource()可以注册资源,不在这里展开。
5.2 启动与本地验证
python mcp_demo_server.py运行后终端没有明显输出,这是正常的,因为 stdio 传输模式下客户端和 Server 通过标准输入输出通信。可以用官方 MCP Inspector 做可视化验证:
npx -y @modelcontextprotocol/inspector python mcp_demo_server.py这个命令需要本机有 Node.js。如果不想安装,也可以写一个最小 Client 连接测试。
5.3 写一个最小 Client 验证工具列表
# mcp_client_test.py import asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client async def main(): server_params = StdioServerParameters( command="python", args=["mcp_demo_server.py"] ) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() tools = await session.list_tools() for tool in tools.tools: print(tool.name, "->", tool.description) asyncio.run(main())执行后如果能看到add和get_city_weather,说明 MCP Server 工作正常。这一步是整个学习路径的第一道关卡,先确保工具可以在标准协议下被发现和调用,再进入 LangChain 集成。
6. LangChain 集成 MCP:让 Agent 调用 MCP 工具
MCP Server 本身不依赖 LangChain,但 LangChain 的 Agent 需要工具。langchain-mcp-adapters负责把 MCP Server 暴露的工具转换成 LangChain 工具对象,交给 Agent 使用。
6.1 集成代码
# langchain_mcp_demo.py import asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client from langchain_mcp_adapters.tools import load_mcp_tools from langchain_openai import ChatOpenAI from langgraph.prebuilt import create_react_agent async def main(): server_params = StdioServerParameters( command="python", args=["mcp_demo_server.py"] ) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() tools = await load_mcp_tools(session) model = ChatOpenAI( model="你的模型名,例如 gpt-4o-mini", api_key="你的密钥或从环境变量读取", base_url="OpenAI 兼容接口地址,否则留空" ) agent = create_react_agent(model, tools) result = await agent.ainvoke({ "messages": [{"role": "user", "content": "计算 12345 + 67890"}] }) print(result) asyncio.run(main())load_mcp_tools把 MCP 的 Tool 转换成 LangChain Tool,create_react_agent是 LangGraph 内置的 ReAct Agent,可以自动完成“推理 -> 调用工具 -> 观察结果 -> 继续推理”的循环。
6.2 效果验证
运行上述脚本时,模型如果收到“计算 12345 + 67890”这种请求,会触发add工具,返回结果后生成最终回答。在终端打印结果中通常可以看到ToolMessage,里面包含工具返回值。
判断是否成功的标准:
- Agent 能识别需要调用哪个工具。
- 工具调用后能拿到正确结果。
- Agent 能基于工具结果生成最终回复。
如果 Agent 没有调用工具,而是直接给了一个错误答案,最常见的原因是模型本身不支持工具调用,或者工具描述不清晰。
6.3 使用 HTTP/SSE 方式连接 MCP Server
stdio 适合本地开发,远程部署时更推荐 HTTP/SSE 方式。MCP 客户端连接方式会不同,但 LangChain 侧的工具加载逻辑基本不变。示例:
from mcp.client.streamable_http import streamablehttp_client # 伪代码示意,需要按你部署的 MCP Server 地址调整 # async with streamablehttp_client(url="http://127.0.0.1:8000/mcp") as (read, write): # async with ClientSession(read, write) as session: # await session.initialize() # tools = await load_mcp_tools(session)实际参数以你使用的mcp包版本和 MCP Server 部署方式为准。开发阶段先用 stdio 跑通,再迁移到 HTTP 模式更稳妥。
7. Agent 实战:从 Prompt 模板到多工具调用
7.1 工具描述决定 Agent 是否会用
很多初学者遇到的问题是:工具已经加载,但 Agent 就是不调用。观察下来,最常见的原因是工具描述写得太模糊。
不推荐:
查询信息
推荐:
根据用户提供的城市名查询该城市当前天气,城市名为中文或拼音时先转成标准城市代码再查询
工具描述越具体,模型就越容易在合适的场景下调用它。如果 Agent 有多个工具,描述里要说明工具之间的边界,避免模型选错。
7.2 多工具协同
把第 5 节的 MCP Server 扩展一下,让 Agent 可以同时调用多个工具完成复杂任务。例如一个“会议纪要助手”的 MCP Server:
# mcp_meeting_server.py from mcp.server.fastmcp import FastMCP mcp = FastMCP("meeting-tools") @mcp.tool() def extract_actions(markdown_text: str) -> list: """从会议纪要文本中提取待办事项,返回列表,每一项包含负责人和截止时间(演示用简单规则提取)""" lines = [line.strip() for line in markdown_text.splitlines() if line.strip()] actions = [line for line in lines if line.startswith("- [ ]")] return actions @mcp.tool() def summarize_doc(content: str, max_words: int = 100) -> str: """对传入文本做摘要,max_words 控制最大字数。真实项目应该调用模型或摘要服务实现""" return content[:max_words] + "..." if __name__ == "__main__": mcp.run()这类工具适合演示,但要注意:真实项目不要用简单截断代替摘要,应该调用专门的摘要服务或模型。
7.3 控制 Agent 的迭代次数和超时
Agent 循环如果设计不好,可能陷入反复调用工具的循环,消耗大量 token。LangGraph 的create_react_agent可以配置递归限制。更多控制项需要查看当前版本文档,但记住一个原则:第一次跑通时把最大迭代次数控制在 5 到 10 次,避免异常场景下长时间挂起。
8. LangChain 与 LangGraph:什么时候切到图编排
对初学者来说,直接用 LangChain 的简化接口最省事,比如create_react_agent算是对 LangGraph 的封装。但当需求变成下面这些情况,就需要显式使用 LangGraph:
- 有状态多轮流程:Agent 需要记住前面的步骤,并根据中间结果决定分支。
- 人工确认环节:执行写操作前需要停下来等待用户确认,LangGraph 可以设计专门的确认节点。
- 工具调用失败重试策略:需要细粒度控制“失败了就换一个工具”还是“直接结束”。
- 多 Agent 协作:一个 Agent 负责规划,一个 Agent 负责执行,一个 Agent 负责质检,用 LangGraph 更容易组织。
简单对比:
| 对比项 | LangChain 简化接口 | LangGraph |
|---|---|---|
| 上手成本 | 低 | 中 |
| 流程控制 | 少 | 强 |
| 状态管理 | 基础 | 支持 checkpoint |
| 适合场景 | 快速验证、简单链路 | 生产级复杂编排 |
热词里还有一些问题,比如“LangChain 中的任务规划能力是怎么实现的”,这通常是通过 ReAct 循环或计划-执行模式实现的,LangGraph 里可以显式拆分规划节点和执行节点,控制力更强。
9. 批量任务与 HTTP API 封装
工具链跑通之后,下一步就是工程化。这里给出两个方向:批量任务处理和 HTTP API 封装。
9.1 批量任务处理
把待处理问题放到tasks/目录,每个问题一个 JSON 文件,循环调用 Agent,输出结果写入results/。示例:
# batch_runner.py import asyncio import json from pathlib import Path from langchain_openai import ChatOpenAI from langgraph.prebuilt import create_react_agent # 这里假设 tools 已经通过前面的 MCP 方式加载 # 实际使用时把 tools 替换成你的工具列表 tools = [] model = ChatOpenAI( model="你的模型名", api_key="你的密钥", base_url="OpenAI 兼容接口地址,否则留空" ) agent = create_react_agent(model, tools) async def process_one(file_path: Path, output_dir: Path): task = json.loads(file_path.read_text(encoding="utf-8")) try: result = await agent.ainvoke({ "messages": [{"role": "user", "content": task["question"]}] }) answer = result.get("messages", [])[-1].content if result.get("messages") else str(result) output_path = output_dir / f"{file_path.stem}.json" output_path.write_text( json.dumps({"question": task["question"], "answer": answer}, ensure_ascii=False, indent=2), encoding="utf-8" ) print(f"[OK] {file_path.name}") except Exception as exc: print(f"[FAIL] {file_path.name}: {exc}") async def main(): input_dir = Path("./tasks") output_dir = Path("./results") output_dir.mkdir(exist_ok=True) for file_path in sorted(input_dir.glob("*.json")): await process_one(file_path, output_dir) asyncio.run(main())批量任务的关键点不在代码本身,而在于:
- 每个任务独立记录日志,失败不能中断整个批次。
- 控制并发数量,避免模型服务限流。
- 输出结果要包含输入、输出、模型名、时间,方便回溯。
- 涉及写操作的任务,先在小批量样本上验证。
9.2 封装 FastAPI 接口
# api_server.py from fastapi import FastAPI from pydantic import BaseModel app = FastAPI() class AskRequest(BaseModel): question: str # 假设已经创建好了 agent,这里省略 MCP 连接代码 agent = None @app.post("/ask") async def ask(req: AskRequest): result = await agent.ainvoke({ "messages": [{"role": "user", "content": req.question}] }) answer = result.get("messages", [])[-1].content if result.get("messages") else str(result) return {"question": req.question, "answer": answer} if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)启动:
python api_server.py然后用 curl 测试:
curl -X POST http://127.0.0.1:8000/ask \ -H "Content-Type: application/json" \ -d '{"question": "计算 12345 + 67890"}'注意:接口服务不要直接暴露到公网,至少加一层身份验证。密钥、模型服务地址、MCP Server 地址都应该通过环境变量或配置中心管理。
10. 资源占用与性能观察
10.1 显存与内存
LangChain + MCP 本身不直接消耗显存,显存占用取决于你使用哪个模型服务。如果调用云端 API,本机内存占用很低;如果本地跑 Ollama 或 vLLM,显存占用由模型大小和推理参数决定。以常见 7B 模型为例,FP16 精度通常需要 14GB 左右显存,量化后可能降到 6GB 左右,但具体要以你本地模型实测为准,这里不给死数字。
10.2 性能瓶颈
这套链路里最耗时的通常是模型推理,其次是多次工具调用的往返。假设一个 Agent 任务需要调用 3 次工具,每轮推理 3 秒,总耗时可能接近 10 秒甚至更高。优化方向:
- 减少工具调用次数:把多个简单查询合并成一个工具。
- 减少上下文长度:工具描述和中间结果不要全量塞进提示词。
- 限制最大迭代次数:避免死循环。
- 使用更快的模型:工具调用场景对模型要求其实不高,响应速度比“智商”更重要。
10.3 观察显存占用的方法
- Linux 下用
nvidia-smi -l 1动态查看显存。 - Windows 下用任务管理器或
nvidia-smi命令。 - 本地跑模型时,重点观察峰值显存和持续时间,判断是否出现显存溢出。
11. 常见问题与排查方法
| 问题现象 | 可能原因 | 排查方式 | 解决方案 |
|---|---|---|---|
| 执行 LangChain 脚本时提示 ModuleNotFoundError: langchain_mcp_adapters | langchain-mcp-adapters 未安装或版本不兼容 | `pip list | findstr mcp` 查看版本 |
| MCP Server 启动后 Client 端连不上 | stdio 参数 command 或 args 路径错误 | 先手动运行python mcp_demo_server.py确认能启动 | 检查绝对路径,Windows 下注意 python 命令对应关系 |
| 工具已加载但 Agent 不调用 | 工具描述不清、模型不支持 tool calling、上下文太长 | 先单独调用session.list_tools()确认工具存在 | 改写工具描述;换支持 function call 的模型;缩短上下文 |
| 批量任务执行到一半卡住 | 单次推理超时、模型服务限流、工具调用死循环 | 加日志,观察卡在哪个文件 | 给单任务加超时和重试机制;减少并发;限制 Agent 最大迭代次数 |
| 本地模型启动后显存溢出 | 模型过大、量化未开启、并发数过高 | 用 nvidia-smi 观察显存 | 换更小的模型或开启量化;降低 batch_size;分批处理 |
| Windows 下 MCP 输出乱码 | 控制台编码问题 | 查看启动日志编码 | 设置 PYTHONIOENCODING=utf-8;脚本开头加编码声明 |
| 使用 HTTP 方式连接 MCP Server 失败 | URL 地址错误;Server 未开启对应 transport | 检查 MCP Server 启动参数和网络连通性 | 先用 stdio 本地验证,再迁移到 HTTP |
| API 服务启动后端口冲突 | 8000 端口被占用 | `netstat -ano | findstr 8000` |
| Agent 输出不稳定 | 模型随机性、提示词顺序变化 | 固定 temperature 参数,对比多次输出 | 对输出加校验和二次处理,必要时增加质量校验环节 |
12. 最佳实践与使用建议
第一,第一次跑通时用最简单的方式。先只加载一个add工具,确认 Agent 能调用,再逐步增加工具。跳过 MCP 工具加载的单独验证,直接调试 Agent,会很难定位问题。
第二,工具权限最小化。MCP Server 暴露的工具越少越好,不要为了方便把所有系统操作都暴露出去。文件读写工具限制在指定目录,数据库工具只开放只读查询,网络请求工具做域名白名单。涉及删除、覆盖、转账、发布一类的高危操作,代码里要加二次确认。
第三,密钥和配置分离。API Key、数据库连接、MCP Server 地址全部走环境变量或配置中心,不写进代码仓库。日志里不要打印完整密钥和敏感字段。
第四,建立可观测性。每个任务记录输入、输出、调用链、耗时、token 消耗、失败原因。批量任务尤其重要,失败要能重跑,不能丢数据。
第五,模型选型不要贪大。工具调用任务用中等规模模型往往就够了,关键是模型支持 function call 并且指令遵循能力合格。优先在真实任务上做 50 条测试样例验证准确率,再决定最终模型。
第六,数据合规先于功能。涉及个人信息、企业机密、版权素材的数据,在使用前确认授权范围;对外发布或商用前,人工复核 Agent 输出,避免出现错误信息和合规风险。
13. 总结与后续扩展
从这条路线最值得先验证的点是:让 LangChain Agent 成功调用一个自己实现的 MCP 工具。能完成这一步,后面的批量任务、HTTP API、LangGraph 编排都是加分项。最容易踩的坑是依赖版本不匹配和工具描述不清晰,这两个问题占掉初学者大量调试时间。
接下来可以继续扩展的方向:
- 把 MCP Server 部署成远程 HTTP 服务,接入 Dify、Cursor 等支持 MCP 的客户端。
- 增加向量检索,把 RAG 和 MCP 工具组合,让 Agent 既能查文档又能调系统。
- 用 LangGraph 设计人工确认节点,让写操作先经过审核再执行。
- 研究 Skill 与 MCP 的配合:Skill 定义任务 SOP,MCP 提供执行工具。
- 针对具体业务封装专用 MCP Server,沉淀成团队内部可复用的工具市场。
以上内容足矣,建议按顺序从第 5 节开始本地实践,先跑通,再扩展。