这次我们不聊概念直接走一遍从 MCP Server 到 LangChain Agent、再到 LangGraph 工作流、最后通过 Deepseek 和 Claude Code 完成落地的完整链路。2026 年再做 Agent 工程化工具接入方式已经绕不开 MCP 这个标准协议了。很多人卡在“原理听懂了代码不知道从哪写”。这篇文章会先用一个最简单的 MCP Server 把协议跑通然后分别演示 LangChain 如何加载 MCP 工具、LangGraph 如何做条件路由、Deepseek 如何作为模型后端接入以及 Claude Code 如何连接同一个服务。整套代码会保持最小可运行方便直接复制到本地验证。1. 核心能力速览能力项说明核心协议MCPModel Context Protocol标准化 LLM 与外部工具/数据源之间的通信Agent 编排框架LangChain负责工具加载、Prompt 组装、Agent 执行流程控制框架LangGraph基于状态图实现条件路由、循环、子图、并行分支模型接入Deepseek通过 OpenAI 兼容 API 接入 LangChain编程 Agent 示例Claude Code可加载 MCP Server 作为工具来源启动方式命令行启动 MCP ServerPython 脚本运行 AgentFastAPI 暴露 HTTP 接口API 支持支持可封装成 HTTP 服务批量任务支持可通过 asyncio 并发调用但需要关注限流与资源占用环境依赖Python 3.10Node.js 18仅 Claude Code 场景需要模型 API Key 或本地模型服务适合读者想了解 MCP 与 Agent 落地、准备做工具接入和流程编排的开发者表格里没有写死显存参数因为 Deepseek 既可以走官方 API也可以本地部署。如果本地跑模型显存需要按实际模型大小判断后面性能章节会给出观察方法。2. MCP、LangChain、LangGraph 到底分别解决什么问题MCP 解决的是工具接入的碎片化问题。最早想让大模型调用工具需要在提示词里写 JSON Schema再针对每个模型厂商做一版 function calling 适配。换一个模型工具定义可能要重写。MCP 把工具做成了独立服务模型端通过协议发现工具、调用工具、拿到结果工具本身不需要关心模型是谁。LangChain 解决的是 Agent 的组装问题。它提供统一的工具输入输出格式负责把用户问题、系统提示词、工具描述交给模型再把模型决定调用的参数解析出来传给具体函数。你可以用 LangChain 自带的BaseTool封装普通函数也可以用适配器加载 MCP 暴露的工具。LangGraph 解决的是流程控制问题。普通 Agent 是单轮“模型决定动作、执行工具、再回到模型”的循环但真实业务往往是多步骤流程先判断问题类型再走不同分支可能需要并行查询多个数据源最后汇总生成报告。LangGraph 把 Agent 流程建模成一张有向图节点是动作边是状态转移条件边负责路由非常适合做这种可控流程。LangChain 和 LangGraph 的区别可以这样理解LangChain 提供组件库LangGraph 提供执行引擎。实际项目中经常两者混用LangChain 负责工具与模型封装LangGraph 负责流程编排这也是本文采用的组合方式。MCP 则在整个结构中下沉为工具层让这些工具可以被 LangChain、Claude Code 等不同平台共同消费。MCP 和 Agent Skill 不是同一个层面的东西。Agent Skill 更像是一组提示词和能力模板比如“这个 Agent 擅长写单元测试附带相关的代码风格指南”MCP 则是工具调用协议解决的是“如何安全地让模型执行一个外部函数”。两者可以同时使用不要混为一谈。3. 环境准备与前置条件安装前先确认下面几项操作系统Windows / macOS / Linux 均可本文命令以 Linux/macOS 为准Windows 注意 Python 路径和命令行差异。Python 版本建议 3.10 或更高。MCP SDK 和 LangChain 生态对低版本 Python 支持有限。Node.js只有在准备使用 Claude Code 或部分基于 Node 的 MCP Server 时才需要建议 18 以上。模型后端准备一个 Deepseek API Key或在本机启动一个兼容 OpenAI 接口的本地模型服务。本地部署可以选 Ollama 或 vLLM接口地址通常是http://localhost:11434/v1。网络Deepseek API 和部分依赖包需要公网访问请确保网络可达。建议使用独立的 Python 虚拟环境避免依赖冲突python -m venv .venv source .venv/bin/activateWindows 下激活命令换成.venv\Scripts\activate接下来安装核心依赖pip install -U langchain langchain-openai langgraph mcp langchain-mcp-adapters fastapi uvicorn这里的langchain-mcp-adapters负责把 MCP 工具转换成 LangChain 的BaseTool。如果安装时提示找不到该包说明版本暂不兼容可以去对应仓库按官方文档调整安装方式或者先改用 LangChain 原生BaseTool封装 MCP 返回结果。4. 从零写一个最小 MCP Server先写一个最简单的 MCP Server暴露两个工具一个做加法一个做本地文件读取。文件读取仅用于演示生产环境必须限制路径范围防止任意文件读取。# mcp_server.py from mcp.server.fastmcp import FastMCP mcp FastMCP(demo-server) mcp.tool() def add(a: int, b: int) - int: 计算 a 和 b 的和。 return a b mcp.tool() def read_file(path: str) - str: 读取指定文本文件仅限本地测试环境使用。 return open(path, r, encodingutf-8).read() if __name__ __main__: mcp.run(transportstdio)这个文件可以单独启动测试。MCP Server 默认以 STDIO 方式运行也就是通过标准输入输出和客户端通信。先直接运行一下确认没有语法错误python mcp_server.py正常情况进程会一直挂起等待输入因为它在等待客户端通过标准输入发送请求。使用 CtrlC 可以终止。后面所有客户端接入都会以子进程方式启动这个服务所以这里能跑通后面的集成才有基础。真实项目中 MCP Server 也可以换成 Streamable HTTP 传输方式便于远程部署但本地调试阶段 STDIO 最简单日志和错误信息都直接在终端可见。5. 在 LangChain 中加载 MCP 工具接下来写客户端启动刚才的mcp_server.py通过langchain-mcp-adapters加载工具再交给 LangChain Agent 执行。# langchain_mcp_client.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.agents import create_react_agent, AgentExecutor from langchain_openai import ChatOpenAI server_params StdioServerParameters( commandpython, args[mcp_server.py], ) async def main(): llm ChatOpenAI( modeldeepseek-chat, api_keysk-your-api-key, base_urlhttps://api.deepseek.com, temperature0, ) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: tools await load_mcp_tools(session) prompt ( 你是一个工具调用助手。请根据用户问题选择合适的工具 并严格根据工具返回结果组织回答。 ) agent create_react_agent(llm, tools, prompt) executor AgentExecutor(agentagent, toolstools, verboseTrue) result await executor.ainvoke({input: 计算 123 456 的结果}) print(最终回答:, result[output]) if __name__ __main__: asyncio.run(main())运行前先设置DEEPSEEK_API_KEY环境变量或者直接替换代码里的sk-your-api-key。更稳妥的做法是使用环境变量export DEEPSEEK_API_KEYsk-你的key python langchain_mcp_client.py这段代码的关键流程是通过stdio_client启动mcp_server.py子进程。ClientSession建立 MCP 会话。load_mcp_tools自动发现并加载 MCP Server 中注册的工具。create_react_agent构建一个 ReAct Agent让模型决定是否调用工具。AgentExecutor负责实际循环模型输出动作 - 执行工具 - 结果返回模型 - 模型生成最终回答。如果 Deepseek API 返回正常执行后应该能看到 Agent 决定调用add工具并给出计算结果。不同 LangChain 版本的create_react_agent参数略有差异如果传入 prompt 报错可以查一下当前版本的方法签名。核心思路不变把模型和工具组装成一个 Agent 执行器。6. 用 LangGraph 编排带条件路由的 Agent 工作流普通 Agent 适合“问题 - 工具 - 回答”这种单层循环。但生产环境更多是流程图先分析问题判断是否需要查库再决定调用哪些 MCP 工具最后汇总结果。LangGraph 适合做这件事。下面是一个状态图示例包含一个条件路由节点。节点之间传递WorkflowStatesteps字段使用 reducer 累加保证多节点写入时不会覆盖历史记录。# langgraph_workflow.py import operator from typing import Annotated, Literal, TypedDict from langgraph.graph import END, StateGraph from langgraph.checkpoint.memory import MemorySaver class WorkflowState(TypedDict): question: str steps: Annotated[list[str], operator.add] answer: str def analyze(state: WorkflowState) - dict: print(节点 analyze 执行) return {steps: [analyze]} def query_database(state: WorkflowState) - dict: print(节点 query_database 执行) return { steps: [query_database], answer: 模拟数据库查询结果订单数量 128, } def generate_report(state: WorkflowState) - dict: print(节点 generate_report 执行) return { steps: [generate_report], answer: f报告{state[answer]}, } def router(state: WorkflowState) - Literal[query_database, generate_report]: if 订单 in state[question] or 查询 in state[question]: return query_database return generate_report builder StateGraph(WorkflowState) builder.add_node(analyze, analyze) builder.add_node(query_database, query_database) builder.add_node(generate_report, generate_report) builder.set_entry_point(analyze) builder.add_conditional_edges(analyze, router) builder.add_edge(query_database, generate_report) builder.add_edge(generate_report, END) graph builder.compile(checkpointerMemorySaver()) config {configurable: {thread_id: demo-1}} def run(question: str): result graph.invoke( {question: question, steps: [], answer: }, configconfig, ) print(最终状态:, result) if __name__ __main__: run(帮我查询订单数据) run(写一份日常总结)这个例子展示了 LangGraph 最核心的几个能力状态化执行所有节点共享一份WorkflowState节点返回值会更新状态。条件路由router根据问题内容决定下一步走向。checkpointer 持久化MemorySaver保存每次运行状态支持多轮对话记忆和断点续跑。实际项目中query_database节点内部完全可以直接调用上一节加载的 MCP 工具把 MCP 工具从“单体 Agent 循环”下沉为“流程节点中的一个动作”。这也是 LangGraph 比平铺式 Agent 更容易维护的原因。7. Deepseek 作为模型后端的接入姿势Deepseek 官方提供 OpenAI 兼容接口所以在 LangChain 里接入非常直接核心就是用ChatOpenAI指定base_url和model。import os from langchain_openai import ChatOpenAI api_key os.getenv(DEEPSEEK_API_KEY) llm ChatOpenAI( modeldeepseek-chat, api_keyapi_key, base_urlhttps://api.deepseek.com, temperature0, )deepseek-reasoner这类推理模型也可以按相同方式切换具体模型名以官方文档为准。如果不想走 API而是在本地部署模型思路也类似。Ollama 启动后默认暴露一个 OpenAI 兼容接口地址通常是http://localhost:11434/v1llm ChatOpenAI( modelqwen2.5:7b, api_keyollama, base_urlhttp://localhost:11434/v1, temperature0, )本地部署时要关注显存。7B 模型量化版通常 6G 显存可以跑14B 或 32B 则需要更大显存。实际占用受上下文长度和并发数影响很大不能只看模型参数量。先小批量测试再看显存曲线是比较稳妥的做法。如果你的业务需要多轮对话记忆单纯给 LangChain 传ChatOpenAI是不够的要用 LangGraph 的 checkpointer 保存会话状态。上面的MemorySaver只适合单机开发测试生产环境可以换成 Redis 或数据库实现的持久化方案。8. Claude Code 如何消费同一个 MCP ServerClaude Code 是面向终端场景的编程 Agent本身支持通过 MCP 协议连接外部工具。我们可以把上面的mcp_server.py注册成 Claude Code 的 MCP Server。在项目根目录创建.mcp.json内容如下{ mcpServers: { demo-server: { command: python, args: [mcp_server.py] } } }然后启动 Claude Code输入/mcp查看连接状态。如果配置正确应该能看到demo-server处于已连接状态并且能发现add和read_file两个工具。这里有一个重点Claude Code 默认使用的是 Claude 模型配置里不需要也不能通过base_url直接指定 Deepseek。更好的做法是保持架构分层在 LangChain 或 LangGraph 里接入 Deepseek构建你自己的 Agent 服务。把这个 Agent 服务暴露成 MCP Server。Claude Code 作为客户端再消费这个 MCP Server。这样模型选择、业务流程、工具调用都在你自己的代码里控制Claude Code 只充当智能终端入口。比起硬改 Claude Code 内部模型配置这种方案更干净也不容易受到版本更新影响。9. 接口 API 化与批量任务落地Agent 写好之后最终要变成一个可用的服务。这里最直接的做法是用 FastAPI 把 Agent 执行器包成一个 HTTP 接口。# agent_service.py import os from fastapi import FastAPI, HTTPException from pydantic import BaseModel from langchain.agents import AgentExecutor, create_react_agent from langchain_openai import ChatOpenAI app FastAPI(titleAgent Service) class TaskRequest(BaseModel): question: str class TaskResponse(BaseModel): output: str def build_agent() - AgentExecutor: llm ChatOpenAI( modeldeepseek-chat, api_keyos.getenv(DEEPSEEK_API_KEY), base_urlhttps://api.deepseek.com, temperature0, ) prompt 你是一个工具调用助手。请选择合适的工具回答用户问题。 agent create_react_agent(llm, [], prompt) return AgentExecutor(agentagent, tools[], verboseTrue) executor build_agent() app.post(/agent/run, response_modelTaskResponse) async def run_agent(req: TaskRequest): try: result await executor.ainvoke({input: req.question}) return TaskResponse(outputresult[output]) except Exception as exc: raise HTTPException(status_code500, detailstr(exc)) from exc注意上面为了展示服务骨架把tools留空了。实际使用时要通过load_mcp_tools或其他方式把工具列表传进来否则 Agent 无工具可调。批量任务方面最简单的并发方式是用asyncio.gatherimport asyncio async def run_batch(questions: list[str]): async def run_one(q: str): return await executor.ainvoke({input: q}) return await asyncio.gather(*(run_one(q) for q in questions))批量处理要注意几个问题限流Deepseek API 有速率限制并发太高会返回 429。超时单个任务可能因为模型推理慢或工具执行异常而卡住要给每次调用设置超时。失败重试设置最大重试次数并对瞬时错误做指数退避。幂等如果是写操作类的工具重复执行可能产生重复数据批量任务前必须确认工具是否安全可重放。更正式的批量任务架构会引入消息队列把任务先入队再由 worker 消费但原理都是同一个 Agent 执行器区别只在于调度和异常处理。10. 资源占用与性能观察方法MCP Server 本身资源占用很低因为它只是协议层真正消耗资源的是模型推理或外部工具调用。如果你用的是 Deepseek API本地不需要显存主要观察的是 API 延迟和请求并发。ChatOpenAI调用是阻塞型 IO异步场景下要关注连接池和超时时间。如果你本地部署模型建议用下面命令实时观察显存nvidia-smi -l 1重点看这几个指标单次请求的显存峰值。请求并发升高后的显存增长曲线。上下文长度对显存的影响。是否触发显存不足报错信息通常是CUDA out of memory。降低显存占用有几种常见手段降低最大上下文长度控制输入 token。使用量化模型比如 GGUF、GPTQ、AWQ。减少并发或在推理服务端设置排队。使用 vLLM 这类带 PagedAttention 的服务显存利用率比普通推理框架高。如果不确定合适的参数先跑一个小 batch逐步增大记录不同并发下的显存和响应延迟再定上线参数。不要一上来就并发 32 路很容易直接把服务打满。11. 常见问题与排查方法问题现象可能原因排查方式解决方案MCP Server 启动后立即退出Python 路径不对导入库失败命令行直接运行python mcp_server.py看报错日志手动修复依赖或语法错误确认虚拟环境已激活LangChain 加载不到工具MCP Server 没有注册任何 tool或协议版本不匹配打印tools列表确认mcp.tool()装饰器已注册确认 SDK 版本LangChain Agent 一直不调用工具提示词描述不清晰模型选择跳过工具打开 verbose 日志查看中间推理过程在系统提示词中明确说明“必须调用工具”调低 temperatureDeepseek API 返回 401API Key 错误或未设置检查环境变量是否生效重新设置DEEPSEEK_API_KEYDeepseek API 超时网络连接不稳定或请求上下文过长在调用处打印请求时间和 token 数加大超时时间减少上下文长度FastAPI 接口返回 500代码异常未捕获或 API Key 为空查看 uvicorn 控制台完整 traceback补异常日志确认模型服务和 API Key 正常批量任务部分失败并发触达限流或某个工具执行报错统计失败任务类型和错误码增加重试与退避控制并发数Claude Code 看不到 MCP Server.mcp.json位置不对或命令启动失败在终端手动运行python mcp_server.py调整配置文件路径重启 Claude Code排查问题时最有效的手段是保留一份完整日志。无论是 MCP Server 的运行日志还是 LangChain 的 verbose 输出都能快速定位问题出在协议层、模型层还是工具层。12. 最佳实践与安全边界Agent 项目落地时建议从一开始就保持清晰的分层和权限控制。MCP Server 暴露的工具要遵循最小权限原则。文件读取、数据库操作、网络请求这类高敏感工具必须做白名单校验。比如文件读取工具要限制只能访问指定目录不能允许任意路径。不要把 API Key 写死在代码仓库里。环境变量或者密钥管理服务是底线。特别是团队协作时密钥泄漏到公开仓库是非常常见的事故。批量任务要加日志和失败重试。每个任务记录开始时间、结束时间、耗时、输入摘要、输出状态失败任务要有独立的错误日志方便事后分析。模型输出不等于可信输出。Agent 执行工具返回的数据、生成的代码、生成的文档都要经过复核再用于生产。涉及代码生成时尤其要注意 SQL 注入、命令注入和越权访问问题。涉及用户数据、版权素材、人脸、声音等场景时必须确认授权和合规边界。MCP 工具可能会访问内部系统务必在工具层做权限校验和操作审计。13. 总结与下一步这条技术链路其实可以拆成三个最小闭环先用 LangChain 加载 MCP 工具跑通“工具调用”再用 LangGraph 把工具调用放进可控的流程节点最后用 FastAPI 把整个 Agent 服务化。Claude Code 在这条链路中的角色只是一个 MCP 客户端真正的业务逻辑和模型策略都留在你自己的代码里。建议第一次跑通时只保留一个add工具验证整条链路没有问题再逐步加文件读取、数据库查询、HTTP 请求等复杂工具。最容易踩的坑集中在协议版本不匹配、MCP Server 子进程启动失败、API Key 未设置这三类遇到问题先看日志基本都能解决。如果你已经在生产环境使用 Agent下一步可以重点验证 LangGraph 的 checkpointer 换成 Redis 或数据库后多轮记忆和故障恢复是否符合预期。工具多了以后还可以把 MCP Server 迁移到 Streamable HTTP 传输方式让多个应用共用同一套工具服务这会让 Agent 架构真正走向平台化。