一、LangChain MCP适配器简介

最近，LangChain发布了一个新的开源项目langchain-mcp-adapters[1]，该项目为开发者提供了一种便捷的方式，将MCP (Model Control Protocol) 服务器无缝集成到LangChain生态系统中。

MCP是一种通信协议，允许大语言模型与外部工具和服务进行交互。而LangChain则是构建LLM应用的流行框架。langchain-mcp-adapters的出现，极大地简化了两者之间的集成过程，让开发者可以更专注于业务逻辑的实现，而非底层通信细节。

二、LangChain的ReAct Agent简介

在深入代码实现之前，有必要了解本案例中使用的核心技术——ReAct Agent。

ReAct Agent

ReAct (Reasoning and Acting) 是一种AI代理架构，它将大语言模型的推理能力与工具调用能力结合在一起。简单来说，ReAct允许AI模型：

1. 思考问题 (Reasoning)
2. 决定使用什么工具 (Planning)
3. 执行工具调用 (Acting)
4. 观察结果并更新思考 (Observation)

在LangChain中，create_react_agent函数提供了一种便捷方式来创建这样的代理。它接收三个关键参数：

• model：用于推理的语言模型
• tools：可供代理使用的工具列表
• prompt：引导代理行为的系统提示

这种方式构建的代理能够分析用户问题，规划并执行多步骤工具调用，最终整合信息为用户提供完整答案。

三、代码实现

下面，我们通过一个完整的天气信息助手案例，展示如何使用langchain-mcp-adapters和ReAct Agent构建智能应用。

1. 项目结构设计

本项目采用面向对象的设计思路，主要包含三个核心类：

• Configuration：负责环境变量管理和配置验证
• MCPServer：处理MCP服务器的连接与工具管理
• MCPClient：整合DeepSeek模型与MCP服务，提供用户交互接口

2. 核心代码实现

首先导入必要的库：

import asyncio
import logging
import os
from contextlib import AsyncExitStack
from typing import List, Optional

from dotenv import load_dotenv
from langchain_core.messages import SystemMessage 
from langchain_core.tools import BaseTool
from langchain_mcp_adapters.tools import load_mcp_tools
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

配置类用于管理环境变量：

class Configuration:
    """配置管理类，负责管理和验证环境变量"""
    
    def__init__(self) -> None:
        """初始化配置并加载环境变量"""
        self.load_env()
        self._validate_env()
        
    @staticmethod
    defload_env() -> None:
        """从.env文件加载环境变量"""
        load_dotenv()
        
    def_validate_env(self) -> None:
        """验证必需的环境变量是否存在"""
        required_vars = ["DEEPSEEK_API_KEY"]
        missing_vars = [var for var in required_vars ifnot os.getenv(var)]
        if missing_vars:
            raise ValueError(f"缺少必需的环境变量: {', '.join(missing_vars)}")
    
    @property
    defapi_key(self) -> str:
        """获取 DeepSeek API 密钥"""
        return os.getenv("DEEPSEEK_API_KEY", "")
    
    @property
    defbase_url(self) -> str:
        """获取 DeepSeek API 基础 URL"""
        return os.getenv("DEEPSEEK_BASE_URL", "https://api.deepseek.com")
    
    @property
    defmodel(self) -> str:
        """获取 DeepSeek 模型名称"""
        return os.getenv("DEEPSEEK_MODEL", "deepseek-chat")

MCP服务器类处理与天气服务的通信：

class MCPServer:
    """MCP 服务器管理类，处理服务器连接和工具执行"""
    def__init__(self, server_path: str) -> None:
        """
        初始化服务器管理器
        
        Args:
            server_path: 服务器脚本路径
        """
        self.server_path = server_path
        self.session: Optional[ClientSession] = None
        self.exit_stack = AsyncExitStack()
        self._cleanup_lock = asyncio.Lock()

    asyncdefinitialize(self) -> None:
        """初始化服务器连接，包含重试机制"""
        max_retries = 3
        retry_delay = 1.0
        
        for attempt inrange(max_retries):
            try:
                ifnot os.path.exists(self.server_path):
                    raise FileNotFoundError(f"找不到服务器文件: {self.server_path}")
                
                server_params = StdioServerParameters(
                    command='python',
                    args=[self.server_path],
                    env=None
                )
                
                stdio_transport = awaitself.exit_stack.enter_async_context(
                    stdio_client(server_params)
                )
                stdio, write = stdio_transport
                
                self.session = awaitself.exit_stack.enter_async_context(
                    ClientSession(stdio, write)
                )
                awaitself.session.initialize()
                logger.info("成功连接到 MCP 服务器")
                break
                
            except Exception as e:
                logger.error(f"第 {attempt + 1}/{max_retries} 次尝试失败: {str(e)}")
                if attempt < max_retries - 1:
                    await asyncio.sleep(retry_delay)
                else:
                    raise

    asyncdeflist_tools(self) -> List[BaseTool]:
        """获取服务器提供的可用工具列表"""
        ifnotself.session:
            raise RuntimeError("服务器未初始化")
        # LangChain方式获取可用工具列表
        tools = await load_mcp_tools(self.session)
        logger.info(f"成功加载工具: {[tool.name for tool in tools]}")
        return tools

    asyncdefcleanup(self) -> None:
        """清理服务器资源"""
        asyncwithself._cleanup_lock:
            try:
                awaitself.exit_stack.aclose()
                self.session = None
                logger.info("服务器资源清理完成")
            except Exception as e:
                logger.error(f"清理过程中出错: {str(e)}")

客户端类整合模型与MCP服务：

class MCPClient:
    """MCP 客户端实现，集成了 DeepSeek API"""
    
    def__init__(self, config: Configuration) -> None:
        """
        初始化 MCP 客户端
        
        Args:
            config: 配置对象
        """
        self.config = config
        self.server: Optional[MCPServer] = None
        self.llm_client = ChatOpenAI(
            api_key=config.api_key,
            base_url=config.base_url,
            model=config.model
        )
        
    asyncdefinitialize(self) -> None:
        """初始化客户端并连接到服务器"""
        server_path = os.path.join(
            os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
            "server",
            "weather_server.py"
        )
        self.server = MCPServer(server_path)
        awaitself.server.initialize()
        
    asyncdefprocess_query(self, query: str):
        """
        处理用户查询，集成工具调用，支持多轮工具交互

        Args:
            query: 用户查询字符串

        Returns:
            处理后的响应结果
        """
        ifnotself.server:
            raise RuntimeError("客户端未初始化")

        # 创建提示模板
        prompt = SystemMessage(content="""你是一个专注于天气信息的助手...（详细系统提示内容）""")
        
        # 获取工具
        tools = awaitself.server.list_tools()

        # 创建ReAct Agent
        logger.info("正在创建agent...")
        agent = create_react_agent(
            model=self.llm_client,  
            tools=tools,
            prompt=prompt
        )
        logger.info("Agent创建成功")

        # 发送查询
        logger.info("正在发送天气查询...")
        agent_response = await agent.ainvoke({
            "messages": query
        })

        # 返回响应
        return agent_response

3. 核心技术要点解析

工具加载与集成

通过langchain-mcp-adapters提供的load_mcp_tools函数，我们可以将MCP服务器中的工具转换为LangChain可识别的BaseTool对象：

tools = await load_mcp_tools(self.session)

这行代码是整个集成的关键，它自动处理了MCP工具到LangChain工具的转换过程，使我们不必手动实现适配逻辑。

ReAct Agent的创建

使用create_react_agent函数创建代理：

agent = create_react_agent(
    model=self.llm_client,  
    tools=tools,
    prompt=prompt
)

这个函数接收三个关键参数：

• model：我们使用的DeepSeek模型
• tools：从MCP服务器加载的工具列表
• prompt：引导代理行为的系统提示

创建的agent对象能够接受用户查询，分析需求，规划并执行工具调用，最终返回整合后的结果。

四、运行效果展示

下面是一个实际使用案例：

用户查询：

最近一周郑州有没有高温或大风预警？周末适合户外活动吗？

系统日志：

2025-05-04 19:10:36,485 - INFO - 成功加载工具: ['get_weather_warning', 'get_daily_forecast']
2025-05-04 19:10:36,485 - INFO - 正在创建agent...
2025-05-04 19:10:36,494 - INFO - Agent创建成功
2025-05-04 19:10:36,494 - INFO - 正在发送天气查询...

Agent响应：

### 郑州近期天气情况：

1. **天气预警**：
   - **大风蓝色预警**：河南省气象台于2025年5月4日发布大风蓝色预警，预计5月5日至6日郑州及周边地区将出现大风天气，阵风可达8-9级，局部10-11级，并伴有扬沙或浮尘。请注意防范。

2. **未来一周天气预报**：
   - **5月4日（今天）**：多云，最高27°C，最低18°C。
   - **5月5日（明天）**：多云转晴，最高31°C，最低17°C，风力较大（西风1-3级）。
   - **...（其他日期天气）**
   - **5月10日（周六）**：晴转多云，最高30°C，最低17°C。

### 周末户外活动建议：
- **周六（5月10日）**：天气晴朗，最高气温30°C，风力较小（西风1-3级），适合户外活动。但请注意防晒和补水。
- **周日（5月11日）**：未提供具体预报，建议关注后续更新。

**注意**：目前有大风预警，尤其是5月5日至6日风力较大，建议避免高空或水上活动，并注意防风防尘。周末天气较为稳定，适合安排户外活动。

从上述案例可以看出，系统成功地：

1. 识别出用户询问的是天气预警和周末户外活动适宜性
2. 调用get_weather_warning工具获取预警信息
3. 调用get_daily_forecast工具获取未来天气预报
4. 整合信息，分析天气状况，并给出周末户外活动建议

五、技术优势与应用场景

技术优势

1. 简化的MCP集成：langchain-mcp-adapters大幅简化了MCP与LangChain的集成流程，开发者无需手动处理协议转换。
2. 灵活的Agent架构：ReAct模式使模型能够自主规划工具调用路径，处理复杂多步骤任务。
3. 可扩展性：基于此架构，可以轻松添加新的工具和功能，而无需大幅修改核心代码。
4. 可靠的异常处理：代码中包含完善的错误处理和重试机制，提高了系统的稳定性。

应用场景

1. 智能客服系统：集成企业内部API，构建能够回答产品、订单、物流等问题的智能客服。
2. 个人助理：连接日历、邮件、任务管理等工具，创建全功能个人助理。
3. 数据分析助手：集成数据库查询、数据处理工具，构建交互式数据分析助手。
4. 企业知识管理：连接企业内部知识库、文档系统，构建智能知识检索与问答系统。

六、总结与展望

通过本文的实践，我们不仅展示了如何利用langchain-mcp-adapters简化MCP服务的集成，更展示了如何基于ReAct模式构建智能代理系统。这种方法相比传统的固定流程工具调用，更加灵活和智能，能够根据用户需求动态规划执行路径。

未来，随着大语言模型能力的提升和工具生态的丰富，基于此架构的应用将变得更加强大。企业可以基于此框架，快速构建各类垂直领域的智能助手，提升服务效率和用户体验。

如何学习大模型 AI ？

由于新岗位的生产效率，要优于被取代岗位的生产效率，所以实际上整个社会的生产效率是提升的。

但是具体到个人，只能说是：

“最先掌握AI的人，将会比较晚掌握AI的人有竞争优势”。

这句话，放在计算机、互联网、移动互联网的开局时期，都是一样的道理。

我在一线互联网企业工作十余年里，指导过不少同行后辈。帮助很多人得到了学习和成长。

我意识到有很多经验和知识值得分享给大家，也可以通过我们的能力和经验解答大家在人工智能学习中的很多困惑，所以在工作繁忙的情况下还是坚持各种整理和分享。但苦于知识传播途径有限，很多互联网行业朋友无法获得正确的资料得到学习提升，故此将并将重要的AI大模型资料包括AI大模型入门学习思维导图、精品AI大模型学习书籍手册、视频教程、实战学习等录播视频免费分享出来。

第一阶段（10天）：初阶应用

该阶段让大家对大模型 AI有一个最前沿的认识，对大模型 AI 的理解超过 95% 的人，可以在相关讨论时发表高级、不跟风、又接地气的见解，别人只会和 AI 聊天，而你能调教 AI，并能用代码将大模型和业务衔接。

• 大模型 AI 能干什么？
• 大模型是怎样获得「智能」的？
• 用好 AI 的核心心法
• 大模型应用业务架构
• 大模型应用技术架构
• 代码示例：向 GPT-3.5 灌入新知识
• 提示工程的意义和核心思想
• Prompt 典型构成
• 指令调优方法论
• 思维链和思维树
• Prompt 攻击和防范
• …

第二阶段（30天）：高阶应用

该阶段我们正式进入大模型 AI 进阶实战学习，学会构造私有知识库，扩展 AI 的能力。快速开发一个完整的基于 agent 对话机器人。掌握功能最强的大模型开发框架，抓住最新的技术进展，适合 Python 和 JavaScript 程序员。

• 为什么要做 RAG
• 搭建一个简单的 ChatPDF
• 检索的基础概念
• 什么是向量表示（Embeddings）
• 向量数据库与向量检索
• 基于向量检索的 RAG
• 搭建 RAG 系统的扩展知识
• 混合检索与 RAG-Fusion 简介
• 向量模型本地部署
• …

第三阶段（30天）：模型训练

恭喜你，如果学到这里，你基本可以找到一份大模型 AI相关的工作，自己也能训练 GPT 了！通过微调，训练自己的垂直大模型，能独立训练开源多模态大模型，掌握更多技术方案。

到此为止，大概2个月的时间。你已经成为了一名“AI小子”。那么你还想往下探索吗？

• 为什么要做 RAG
• 什么是模型
• 什么是模型训练
• 求解器 & 损失函数简介
• 小实验2：手写一个简单的神经网络并训练它
• 什么是训练/预训练/微调/轻量化微调
• Transformer结构简介
• 轻量化微调
• 实验数据集的构建
• …

第四阶段（20天）：商业闭环

对全球大模型从性能、吞吐量、成本等方面有一定的认知，可以在云端和本地等多种环境下部署大模型，找到适合自己的项目/创业方向，做一名被 AI 武装的产品经理。

• 硬件选型
• 带你了解全球大模型
• 使用国产大模型服务
• 搭建 OpenAI 代理
• 热身：基于阿里云 PAI 部署 Stable Diffusion
• 在本地计算机运行大模型
• 大模型的私有化部署
• 基于 vLLM 部署大模型
• 案例：如何优雅地在阿里云私有部署开源大模型
• 部署一套开源 LLM 项目
• 内容安全
• 互联网信息服务算法备案
• …

学习是一个过程，只要学习就会有挑战。天道酬勤，你越努力，就会成为越优秀的自己。

如果你能在15天内完成所有的任务，那你堪称天才。然而，如果你能完成 60-70% 的内容，你就已经开始具备成为一名大模型 AI 的正确特征了。

这份完整版的大模型 AI 学习资料已经上传CSDN，朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】