MCP Agent实战：从零编写Server到LangChain/LangGraph工具调用集成

做 Agent 这个事最难的不是把大模型的对话接起来而是让模型在推理过程中真正能操作外部系统。MCPModel Context Protocol模型上下文协议就是把“工具服务”标准化文件、数据库、HTTP API、本地脚本统一暴露成 Agent 可以动态调用的接口。这次我们直接走完整条落地链路从零写一个 MCP Server用 LangChain 的 MCP 适配器把它变成 Agent 工具再升级到 LangGraph 做多工具编排LLM 侧用 DeepSeek 的 OpenAI 兼容接口最后在 Claude Code 里注册验证整套配置。文章会覆盖MCP 三种角色怎么跑通、LangChain 工具封装的代码写法、LangGraph 条件路由怎么设计、DeepSeek API 参数怎么填、以及本地调试时最容易踩的几个坑。适合刚接触 Agent 想搞懂工具调用原理的开发者也适合已经在用 LangChain 但还没接 MCP 的团队。先给结论这套方案的核心能力速览如下。1. 核心能力速览维度说明核心链路MCP Server - LangChain Tool - AgentExecutor / LangGraph - LLMDeepSeek APIMCP 角色HostAgent/Claude Code、ClientLangChain 适配层、Server工具服务工具类型本地计算、时间查询、文件读写、HTTP 请求、数据库等统一封装为标准工具LLM 接入方式DeepSeek APIOpenAI 兼容协议可直接使用 ChatOpenAIAgent 编排LangChain AgentExecutor 适合快速验证LangGraph 适合生产级多路由编排IDE AgentClaude Code 可以通过 MCP 配置注册本机工具服务批量任务可在 Agent 外层封装 FastAPI 服务再加任务队列本机资源占用MCP Server 和 Agent 进程占用较小大头在模型 API 调用和工具返回数据量接下来先把协议原理讲清楚否则后面写代码会卡在“为什么工具能被模型调用”这个点上。2. MCP 协议与 Agent 工具调用原理2.1 MCP 的三个角色MCP 协议把一次工具调用拆成了三个角色。Host用户直接面对的 Agent 程序例如 Claude Code、LangChain 应用负责接收用户问题、维护对话上下文、决定是否调用工具。Client嵌入在 Host 内部的协议客户端负责和 MCP Server 建连、发现工具、发起调用。Server独立的工具服务进程暴露工具列表和调用逻辑一个 Server 可以注册多个工具。从 LangChain 角度看Host 是 AgentExecutor 或 LangGraph 图Client 是langchain-mcp-adapters提供的适配器Server 是我们自己写的mcp_server.py。三者通过 JSON-RPC 消息通信传输层可以用 stdio也可以走 SSE / HTTP。2.2 工具调用为什么能让模型“自由发挥”大模型本身不会真的执行函数。模型做的只是“决定”根据用户问题和系统提示词输出一个结构化工具调用请求例如{name: get_current_time, arguments: {}}。LangChain 拿到这个结构后去本地工具列表里找到对应工具执行并把执行结果返回给模型让模型基于结果继续组织回答。这个循环叫作 ReAct 风格的 Agent 循环具体流程是用户输入问题。模型推理判断是否需要工具。如果需要输出工具名和参数。Agent 框架执行工具拿到结果。结果作为新的消息回传给模型。模型继续推理直到不再调用工具输出最终答案。MCP 的价值在于第 3 步和第 4 步之间工具不再是写死在代码里的 Python 函数而是由独立进程通过协议暴露出来的服务。这样工具可以跨项目复用也可以由不同团队分别维护。2.3 为什么 LangChain 要接 MCPLangChain 自己有tool装饰器写一个工具并不难。但真实项目里工具会被多个 Agent 复用甚至需要被 Claude Code、Cline、自研 Agent 同时使用。如果每个 Agent 都重新封装一遍工具维护成本很高。MCP 把工具定义成标准协议任何支持 MCP 的客户端都能发现并调用。LangChain 接 MCP 后团队只需要维护一份 MCP Server所有 Agent 都能复用同一套工具。这也是 LangChain 官方做langchain-mcp-adapters的原因把 MCP 工具直接转换成 LangChain 的 Tool 对象复用现有 Agent 执行链。3. 环境准备与前置条件在写代码之前先确认环境。操作系统Windows / macOS / Linux 均可但 MCP Server 用 stdio 启动时命令参数里要注意 Python 路径差异。Python建议 3.10 或更高版本。MCP Python SDK 和 LangChain 生态对 Python 3.10 支持更稳。包管理使用 venv 或 conda 创建独立虚拟环境避免和系统 Python 互相污染。LLM API注册 DeepSeek 开放平台获取 API Key。DeepSeek 提供 OpenAI 兼容接口base_url可以直接指向https://api.deepseek.com。网络模型调用走 HTTPS需要本机能正常访问 API 服务地址。创建虚拟环境并安装依赖python -m venv .venv # Windows .venv\Scripts\activate # macOS / Linux source .venv/bin/activate安装核心依赖pip install -U langchain langchain-openai langchain-community langgraph mcp langchain-mcp-adapters fastapi uvicorn这里说明一下mcp是官方 Python SDKlangchain-mcp-adapters负责把 MCP 工具转成 LangChain Tool。如果你用的版本较新接口导出名可能有变化以官方文档为准。4. 第一个 MCP Server从零写一个可被调用的工具服务4.1 创建mcp_server.py先用 FastMCP 写一个简单的服务包含两个工具一个是时间查询一个是数值计算。这是验证整条链路的最小可用案例。# mcp_server.py from datetime import datetime from mcp.server.fastmcp import FastMCP mcp FastMCP(agent-tool-server) mcp.tool() def get_current_time() - str: 返回服务器当前时间用于验证 Agent 工具调用链路。 return datetime.now().strftime(%Y-%m-%d %H:%M:%S) mcp.tool() def add_numbers(a: float, b: float) - float: 对两个数值执行加法运算适合测试数值类工具。 return a b if __name__ __main__: mcp.run(transportstdio)这段代码有两个要点。第一mcp.tool()装饰器会自动扫描函数签名把函数名和 docstring 作为工具描述传给客户端。docstring 不是可选项模型会依据描述判断什么时候调用这个工具描述越清晰调用准确率越高。第二transportstdio表示通过标准输入输出通信。这种方式适合本地 Agent 进程直接拉起 MCP Server不需要额外开端口也方便 Claude Code 这类命令行工具集成。启动验证python mcp_server.py如果程序没有报错并保持运行说明 MCP Server 已经正常启动。此时它不会打印内容因为真正消息交互走的是 stdin/stdout而不是终端日志。4.2 为什么用 stdio 而不是 HTTP本地工具用 stdio 更合适不用考虑端口冲突子进程生命周期由 Agent 进程管理。天然隔离不会暴露到局域网。启动快资源占用低。如果是跨机器、跨服务共享工具则建议改用 HTTP 或 SSE 传输方式把 MCP Server 部署成独立的服务。5. LangChain 接入 MCP把 MCP 工具变成 Agent 手里的工具5.1 用load_mcp_tools加载工具langchain-mcp-adapters提供了一个核心函数load_mcp_tools它接收一个 MCP ClientSession返回 LangChain Tool 列表。import asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client from langchain_mcp_adapters.tools import load_mcp_tools async def main(): server_params StdioServerParameters( commandpython, args[mcp_server.py], envNone, ) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: tools await load_mcp_tools(session) for tool in tools: print(tool.name) print(tool.description) print(tool.args_schema) print(---) asyncio.run(main())运行后可以看到MCP Server 里的get_current_time和add_numbers变成了 LangChain 的 StructuredToolargs_schema也自动生成。这一层是整条链路的核心MCP Server 负责工具执行LangChain 框架负责模型调用和工具调度。5.2 配置 DeepSeek 作为 Agent 的 LLMDeepSeek 提供 OpenAI 兼容的 API所以不需要写自定义封装直接用langchain-openai的ChatOpenAI即可。from langchain_openai import ChatOpenAI model ChatOpenAI( modeldeepseek-chat, api_keysk-xxxxxxxxxxxx, base_urlhttps://api.deepseek.com, temperature0, )有三个参数要特别注意。base_url必须指向 DeepSeek 的 OpenAI 兼容地址不要默认填 OpenAI 官方地址。model当前常用的是deepseek-chat具体模型名以 DeepSeek 开放平台文档为准。temperature工具调度类任务建议设置为 0降低模型随机性让工具调用更稳定。5.3 用 AgentExecutor 跑通工具调用加载工具之后使用create_tool_calling_agent构建 Agent再用AgentExecutor执行。from langchain.agents import create_tool_calling_agent, AgentExecutor from langchain_core.prompts import ChatPromptTemplate prompt ChatPromptTemplate.from_messages([ (system, 你是工具调度助手。请根据用户问题判断是否需要调用工具需要时直接调用。), (human, {input}), (placeholder, {agent_scratchpad}), ]) agent create_tool_calling_agent(model, tools, prompt) executor AgentExecutor(agentagent, toolstools, verboseTrue) result executor.invoke({input: 现在几点顺便算一下 12.5 加 7.3 等于多少}) print(result[output])执行时主要看两个信息。模型是否输出了正确的工具调用结构。工具结果是否被正确返回给模型并生成最终回答。如果执行成功verboseTrue会打印出每一步工具调用信息。这是验证 Agent 链路最直观的方式。6. 用 LangGraph 编排多工具 Agent6.1 为什么从 AgentExecutor 升级到 LangGraphAgentExecutor 适合快速验证但业务逻辑复杂后就会出现几个问题分支条件不好控制工具调用失败重试逻辑不透明状态管理不灵活。LangGraph 把 Agent 流程建模成一张有向图节点是处理步骤边是状态流转开发人员可以精确控制“什么时候调用工具、调用哪个工具、失败后怎么办”。6.2 用create_react_agent快速起步LangGraph 提供了预置的 React Agent直接用模型和工具列表构造。from langgraph.prebuilt import create_react_agent graph_agent create_react_agent(modelmodel, toolstools) response graph_agent.invoke( {messages: [{role: user, content: 现在几点顺便算一下 12.5 加 7.3}]} ) for msg in response[messages]: print(msg.type, msg.content)这里response[messages]保存了完整的调用链用户消息、模型工具调用请求、工具执行结果、最终回答。对排查问题很有帮助。6.3 手写状态图做条件路由如果业务需要自定义路由例如判断用户问题是“查询类”还是“分析类”可以手写一个StateGraph。from typing import TypedDict, Literal from langgraph.graph import StateGraph, START, END class TaskState(TypedDict): user_input: str route: Literal[tool, analysis, direct] def route_node(state: TaskState) - dict: return state def decide_route(state: TaskState) - str: text state[user_input] if 时间 in text or 计算 in text or 查询 in text: return tool if 分析 in text: return analysis return direct def tool_execute_node(state: TaskState) - dict: # 实际场景中在这里调用 MCP 工具并写入结果 return {route: tool} builder StateGraph(TaskState) builder.add_node(route, route_node) builder.add_node(tool_execute, tool_execute_node) builder.add_edge(START, route) builder.add_conditional_edges( route, decide_route, { tool: tool_execute, analysis: END, direct: END, } )这一段代码展示了 LangGraph 最核心的条件路由能力add_conditional_edges根据路由函数返回值决定下一跳走到哪个节点。这样就把“模型自由发挥”和“业务规则控制”结合起来了工具调度不再是黑盒。6.4 MCP 工具如何接入 LangGraph 节点LangGraph 的节点本质是一个接收状态、返回更新的函数。MCP 工具转成 LangChain Tool 后可以直接在节点内部调用。from langchain_core.messages import AIMessage def call_tool_node(state: TaskState) - dict: # 这里简化为把固定问题交给 Agent 执行 result graph_agent.invoke( {messages: [{role: user, content: state[user_input]}]} ) return {route: tool}生产环境建议把工具调用单独放在一个节点里并加上超时和重试逻辑避免单个工具卡死整条链路。7. Claude Code 接入 DeepSeek 与 MCP 注册7.1 Claude Code 在 Agent 工程里的位置Claude Code 是 Anthropic 推出的命令行 Agent 工具它能读取项目文件、执行命令、调用 MCP 工具。在很多团队里它被用来做代码重构、批量文件处理、自动化脚本编写。它的核心优势不是模型本身而是把“终端操作能力”直接交给了 Agent 编排层。7.2 通过环境变量配置 DeepSeek社区里最常见的做法是通过环境变量把 Claude Code 的请求端点指到 DeepSeek 的 Anthropic 兼容接口。具体是否可用以 DeepSeek 官方文档和版本支持为准。export ANTHROPIC_BASE_URLhttps://api.deepseek.com/anthropic export ANTHROPIC_AUTH_TOKENsk-xxxxxxxxxxxx配置完成后启动 Claude Code 时它会使用 DeepSeek 模型处理对话和工具调度。这种做法适合想用本地命令行 Agent 但不想额外搭建服务层的开发者。7.3 注册本地 MCP ServerClaude Code 支持通过配置文件注册 MCP Server。以项目级配置为例在项目根目录下的 MCP 配置文件中添加{ mcpServers: { agent-tool-server: { command: python, args: [D:/code/mcp_server.py], env: {} } } }路径要写绝对路径。启动 Claude Code 后它会自动拉起mcp_server.py然后就能在对话中直接调用get_current_time、add_numbers这些工具。这里要强调一点如果你的 MCP Server 使用 stdio 通信必须保证 Claude Code 的工作目录和 Python 环境正确。最常见的报错是 “command not found” 或 “No module named mcp”排查时优先看 Python 路径和虚拟环境是否激活。8. 接口 API 与批量任务设计8.1 把 Agent 封装成 HTTP 接口本地验证通过后下一步往往是把 Agent 能力开放给其他系统。用 FastAPI 封装一层即可。from fastapi import FastAPI from pydantic import BaseModel app FastAPI() class QueryRequest(BaseModel): text: str app.post(/agent) async def run_agent(request: QueryRequest): result executor.invoke({input: request.text}) return {output: result[output]}启动服务uvicorn main:app --host 127.0.0.1 --port 8000调用测试curl -X POST http://127.0.0.1:8000/agent \ -H Content-Type: application/json \ -d {text: 现在几点}接口可以跑通后就可以接到自己的业务系统里例如客服助手、文档处理流水线。8.2 批量任务与队列设计批量任务不适合同步阻塞接口。更合理的做法是接收任务后先生成任务 ID。把任务写入队列由后台 Worker 消费。Agent 执行结果写入数据库或文件。客户端通过任务 ID 查询状态。如果项目规模不大用 Redis 列表或简单文件队列就能实现。关键是要记录任务状态pending、running、success、failed。批量跑一批新闻摘要、合同信息抽取时这个设计能避免接口超时和内存堆积。8.3 接口安全性接口服务不要直接暴露到公网。如果必须对外提供建议加 API Key 校验、请求频率限制和超时时间。Agent 工具如果涉及文件操作或数据库写操作接口层要做权限校验防止任意调用。9. 资源占用与性能观察9.1 MCP Server 进程资源MCP Server 使用 stdio 通信时没有额外端口子进程由 Agent 进程直接拉起。以 Python 实现的 FastMCP Server 为例进程启动后占用内存通常较小可能只有几十兆级别具体取决于工具逻辑和依赖包大小。如果工具里加载了机器学习模型资源占用会明显上升。观察方式# Linux / macOS ps -o pid,rss,comm -p mcp_server_pid # Windows PowerShell Get-Process -Name python | Select-Object Id, WorkingSet649.2 性能瓶颈在哪这套链路里真正的性能瓶颈不在 MCP 框架而在两个地方。第一是模型推理 API 的响应时间。模型需要先生成工具调用请求工具执行后还要把结果再送回模型生成最终回答一次完整任务至少需要两轮模型请求。第二是工具的耗时。如果某个工具是同步 HTTP 请求而目标接口响应很慢整个 Agent 循环都会被卡住。建议给每个工具调用加超时时间避免单个工具拖死整条链路。9.3 如何降低延迟精简工具描述让模型更容易快速判断是否需要调用工具。把常用的查询结果做缓存减少重复调用。使用异步工具执行但要注意 LangChain Agent 对异步工具的支持情况。批量任务优先走队列而不是同步并行堆积在线程里。10. 常见问题与排查方法下面这张表整理了 LangChain 接 MCP 过程中最常见的几类问题。问题现象可能原因排查方式解决方案启动 MCP Server 提示 No module named mcp虚拟环境未激活或依赖未安装执行pip list检查 mcp 包激活虚拟环境后重新安装依赖Claude Code 提示 command not found配置里的 Python 路径不对检查 MCP 配置文件中的 command 字段写入 Python 绝对路径LangChain 加载工具为空MCP Server 没有注册工具在load_mcp_tools前打印 session 状态检查mcp.tool()装饰器是否生效Agent 不调用工具直接回答模型不支持工具调用或模型名不对把模型切换为支持工具调用的版本确认deepseek-chat支持 function callingtools 执行很慢工具内部是同步阻塞请求查看工具日志耗时给工具加超时或改成异步执行端口被占用使用 HTTP 传输时多个服务同时启动查看端口占用更换端口或改回 stdio 传输CLI Code 连不上 DeepSeek环境变量未生效在 Claude Code 终端执行env查看变量重新 export 后重启 Claude Code批量任务部分失败没有重试机制查看 Worker 日志增加失败重试记录错误信息另外要提醒的是langchain-mcp-adapters版本更新较快不同版本的导出函数名和初始化参数可能不同。遇到ImportError时优先查看安装版本的官方文档而不是死记旧代码。11. 最佳实践与使用建议从原理到落地一套可维护的 MCP Agent 架构应该遵循以下几条建议。第一工具粒度要适中。一个工具只做一件事不要写“万能工具”。工具描述里写清楚工具能做什么、什么时候用、参数是什么、返回什么。模型是靠描述来判断调用的描述越明确Agent 越不容易跑偏。第二环境隔离要严格。MCP Server 依赖 Python 包Agent 代码也依赖 Python 包不要图省事共用全局环境。项目里保留requirements.txt或pyproject.toml换机器时能一键重建。第三目录管理要规范。建议按下面的结构组织项目agent-project/ ├── mcp_server.py # MCP Server 入口 ├── main.py # FastAPI 接口层 ├── agent.py # LangChain / LangGraph 逻辑 ├── requirements.txt ├── configs/ │ └── mcp_config.json # Claude Code 等客户端配置 ├── inputs/ # 输入素材 └── outputs/ # 工具执行结果第四涉及敏感数据的 Agent必须在入口做权限校验。如果 Agent 要读取数据库、修改文件、调用外部 API应当遵循最小权限原则。人脸、声音、版权素材等数据参与处理前确认已经获得合法授权不要在未授权数据上做生成、采样或批处理。第五模型调用不是免费的。批量任务开始前先用 2 到 3 条测试数据验证效果和 token 消耗确认成本可以接受后再全量跑。加一个 token 计数和费用预警能避免月底收到意外账单。第六整个 Agent 链路里异常处理要放到边界位置MCP Server 内部、代理回调节点、HTTP 接口层分别做异常捕获。否则工具抛错时模型有可能把错误信息误解成业务结果。12. 总结与下一步这套链路里最值得你先跑通的是最小的 MCP Server 加 LangChain Agent整个流程只需要两个 Python 文件和一次 API 调用。先把“模型生成工具调用 - 执行工具 - 返回结果”这个循环跑通再扩展成 LangGraph 的多路由编排。最容易踩的坑集中在环境上Python 虚拟环境没激活、MCP Server 路径没写对、chatdeepseek等模型名配置错误。这三个问题解决后后面的链路会顺利很多。下一步建议从两个方向深入一是把 MCP Server 工具扩展到真实的文件检索、数据库查询或 HTTP 请求让 Agent 处理实际业务二是研究 LangGraph 的条件路由和子图设计把复杂的多 Agent 协作拆成可控的节点。LangChain 和 MCP 的组合不会停留在 demo 阶段工具就是 Agent 的双手这套协议已经能支撑起真实的生产任务了。