news 2026/8/31 21:23:00

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

作者头像

张小明

前端开发工程师

1.2k 24
文章封面图
MCP Agent实战:从零编写Server到LangChain/LangGraph工具调用集成

做 Agent 这个事,最难的不是把大模型的对话接起来,而是让模型在推理过程中真正能操作外部系统。MCP(Model 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 -> LLM(DeepSeek API)
MCP 角色Host(Agent/Claude Code)、Client(LangChain 适配层)、Server(工具服务)
工具类型本地计算、时间查询、文件读写、HTTP 请求、数据库等,统一封装为标准工具
LLM 接入方式DeepSeek API,OpenAI 兼容协议,可直接使用 ChatOpenAI
Agent 编排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 循环,具体流程是:

  1. 用户输入问题。
  2. 模型推理,判断是否需要工具。
  3. 如果需要,输出工具名和参数。
  4. Agent 框架执行工具,拿到结果。
  5. 结果作为新的消息回传给模型。
  6. 模型继续推理,直到不再调用工具,输出最终答案。

MCP 的价值在于第 3 步和第 4 步之间:工具不再是写死在代码里的 Python 函数,而是由独立进程通过协议暴露出来的服务。这样工具可以跨项目复用,也可以由不同团队分别维护。

2.3 为什么 LangChain 要接 MCP

LangChain 自己有@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 SDK,langchain-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(transport="stdio")

这段代码有两个要点。

第一,@mcp.tool()装饰器会自动扫描函数签名,把函数名和 docstring 作为工具描述传给客户端。docstring 不是可选项,模型会依据描述判断什么时候调用这个工具,描述越清晰,调用准确率越高。

第二,transport="stdio"表示通过标准输入输出通信。这种方式适合本地 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( command="python", args=["mcp_server.py"], env=None, ) 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_timeadd_numbers变成了 LangChain 的 StructuredTool,args_schema也自动生成。这一层是整条链路的核心:MCP Server 负责工具执行,LangChain 框架负责模型调用和工具调度。

5.2 配置 DeepSeek 作为 Agent 的 LLM

DeepSeek 提供 OpenAI 兼容的 API,所以不需要写自定义封装,直接用langchain-openaiChatOpenAI即可。

from langchain_openai import ChatOpenAI model = ChatOpenAI( model="deepseek-chat", api_key="sk-xxxxxxxxxxxx", base_url="https://api.deepseek.com", temperature=0, )

有三个参数要特别注意。

  • 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(agent=agent, tools=tools, verbose=True) result = executor.invoke({"input": "现在几点?顺便算一下 12.5 加 7.3 等于多少"}) print(result["output"])

执行时主要看两个信息。

  • 模型是否输出了正确的工具调用结构。
  • 工具结果是否被正确返回给模型,并生成最终回答。

如果执行成功,verbose=True会打印出每一步工具调用信息。这是验证 Agent 链路最直观的方式。

6. 用 LangGraph 编排多工具 Agent

6.1 为什么从 AgentExecutor 升级到 LangGraph

AgentExecutor 适合快速验证,但业务逻辑复杂后,就会出现几个问题:分支条件不好控制,工具调用失败重试逻辑不透明,状态管理不灵活。LangGraph 把 Agent 流程建模成一张有向图,节点是处理步骤,边是状态流转,开发人员可以精确控制“什么时候调用工具、调用哪个工具、失败后怎么办”。

6.2 用create_react_agent快速起步

LangGraph 提供了预置的 React Agent,直接用模型和工具列表构造。

from langgraph.prebuilt import create_react_agent graph_agent = create_react_agent(model=model, tools=tools) 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_URL="https://api.deepseek.com/anthropic" export ANTHROPIC_AUTH_TOKEN="sk-xxxxxxxxxxxx"

配置完成后,启动 Claude Code 时,它会使用 DeepSeek 模型处理对话和工具调度。这种做法适合想用本地命令行 Agent 但不想额外搭建服务层的开发者。

7.3 注册本地 MCP Server

Claude Code 支持通过配置文件注册 MCP Server。以项目级配置为例,在项目根目录下的 MCP 配置文件中添加:

{ "mcpServers": { "agent-tool-server": { "command": "python", "args": ["D:/code/mcp_server.py"], "env": {} } } }

路径要写绝对路径。启动 Claude Code 后,它会自动拉起mcp_server.py,然后就能在对话中直接调用get_current_timeadd_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 列表或简单文件队列就能实现。关键是要记录任务状态:pendingrunningsuccessfailed。批量跑一批新闻摘要、合同信息抽取时,这个设计能避免接口超时和内存堆积。

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, WorkingSet64

9.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 calling
tools 执行很慢工具内部是同步阻塞请求查看工具日志耗时给工具加超时,或改成异步执行
端口被占用使用 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.txtpyproject.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 的双手,这套协议已经能支撑起真实的生产任务了。

版权声明: 本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若内容造成侵权/违法违规/事实不符,请联系邮箱:809451989@qq.com进行投诉反馈,一经查实,立即删除!
网站建设 2026/8/31 21:17:57

Boost电路Matlab/Simulink仿真实战:从原理到闭环PI控制

Boost电路&#xff0c;也就是升压斩波电路&#xff0c;是电力电子领域最基础也最常用的拓扑之一。它不复杂&#xff0c;但电感电流连续模式&#xff08;CCM&#xff09;、断续模式&#xff08;DCM&#xff09;、占空比极限、带载能力这些问题&#xff0c;光看公式推导很难建立直…

作者头像 李华
网站建设 2026/8/31 21:17:32

管道组合 + 结构化输出:让 Claude 和 grep 并肩作战

管道组合 结构化输出&#xff1a;让 Claude 和 grep 并肩作战 系列第 2 篇 读完 6 分钟 前置&#xff1a;第 1 篇 Headless 入门 上篇说了 claude -p 单次调用。但这只是"问一句&#xff0c;答一句"。真正的威力在于把 Claude Code 串进 Unix 工具链——让它输出机…

作者头像 李华
网站建设 2026/8/31 21:15:19

日收最新泛目录程序,泛站群程序,2265下载站模板

此站群只适合做百度搜索引擎&#xff0c;其他引擎全部关闭。适用所有行业关键字&#xff0c;全站泛解析泛目录玩法。新D58SEO内核部分域名能做到日收&#xff0c;域名的挑选技巧&#xff0c;卖我程序我都会教&#xff0c;盗版的只有程序1&#xff1a;此为官方版本二开&#xff…

作者头像 李华
网站建设 2026/8/31 21:14:44

从零开始理解博弈搜索:AI下棋的决策密码

从零开始理解博弈搜索&#xff1a;AI下棋的决策密码当你和AI下棋时&#xff0c;它到底在“想”什么&#xff1f;一、先搞清楚三个前提&#xff1a;什么样的游戏能用这套方法&#xff1f; 在讲具体算法之前&#xff0c;得先弄清楚一个前提——极小化极大算法不是万能的&#xff…

作者头像 李华