3个步骤掌握LangChain MCP Adapters：实现MCP与LangChain跨框架集成

LangChain MCP Adapters是一个轻量级协议转换工具集，它通过构建工具适配层实现了Anthropic Model Context Protocol (MCP)与LangChain/LangGraph生态的无缝对接。该项目核心价值在于解决异构系统间的工具互通问题，允许开发者将任意MCP兼容服务转化为LangChain可用工具，同时支持多服务器协同工作，为构建复杂智能代理提供灵活的工具集成方案。

理解核心价值：MCP与LangChain的桥梁

工具适配器是连接MCP服务与LangChain生态的关键组件，类似不同插头的转换接口，它能够将MCP协议定义的工具方法自动转换为LangChain Agent可识别的工具格式。这种转换不仅保留了原始工具的功能特性，还添加了符合LangChain规范的描述元数据，使AI代理能够正确理解和调用这些外部工具。

该架构主要包含三个核心部分：MCP服务器集群提供各类功能工具，客户端管理连接并执行协议转换，最终通过LangGraph Agent实现工具的智能调度。这种分层设计确保了系统的模块化和可扩展性，允许独立升级或替换任何组件而不影响整体功能。

思考：为什么需要专门的适配器而不是直接使用MCP客户端？

场景化应用：构建智能计算器代理

准备工作

首先克隆项目仓库并安装依赖：

git clone https://gitcode.com/gh_mirrors/la/langchain-mcp-adapters
cd langchain-mcp-adapters
pip install . langgraph langchain-openai

设置环境变量：

export OPENAI_API_KEY="你的API密钥"

搭建数学工具服务器

创建一个提供基础数学运算的MCP服务：

# math_server.py
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("数学工具服务")

@mcp.tool()
def add(a: int, b: int) -> int:
    """计算两个整数的和"""
    return a + b

@mcp.tool()
def multiply(a: int, b: int) -> int:
    """计算两个整数的乘积"""
    return a * b

if __name__ == "__main__":
    mcp.run(transport="stdio")  # 使用标准输入输出作为传输方式

启动服务器：

python math_server.py

开发智能代理客户端

创建一个能够使用数学工具的LangGraph代理：

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from langchain_mcp_adapters.tools import load_mcp_tools  # 加载MCP工具的核心函数
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI

# 初始化AI模型
model = ChatOpenAI(model="gpt-4o")

# 配置服务器连接参数
server_params = StdioServerParameters(
    command="python",
    args=["math_server.py"],
    transport="stdio"
)

# 连接服务器并创建代理
async def calculate_expression(question: str):
    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)  # 将MCP工具转换为LangChain格式
            agent = create_react_agent(model, tools)  # 创建具备工具使用能力的代理
            return await agent.ainvoke({"messages": question})

# 测试代理
result = await calculate_expression("计算(3 + 5) × 12的结果")
print(result)

这段代码实现了一个能够理解数学问题并调用相应工具的智能代理，通过适配器将MCP工具无缝集成到LangChain的代理系统中。

进阶实践：多服务器协同工作

在实际应用中，我们通常需要整合多种类型的工具服务。MultiServerMCPClient类提供了统一管理多个MCP服务器的能力，使代理能够同时使用来自不同服务器的工具。

配置多服务器连接

from langchain_mcp_adapters.client import MultiServerMCPClient

# 定义服务器配置字典
server_config = {
    "math": {
        "command": "python",
        "args": ["math_server.py"],
        "transport": "stdio"
    },
    "weather": {
        "url": "http://localhost:8000/sse",
        "transport": "sse"  # 使用服务器发送事件协议
    }
}

# 上下文管理器自动处理连接生命周期
async with MultiServerMCPClient(server_config) as client:
    # 获取所有服务器提供的工具
    all_tools = client.get_tools()
    # 创建能够使用所有工具的智能代理
    agent = create_react_agent(model, all_tools)
    
    # 测试多工具调用能力
    math_result = await agent.ainvoke({"messages": "3的4次方乘以5等于多少？"})
    weather_result = await agent.ainvoke({"messages": "北京今天的天气如何？"})
    
    print("数学计算结果:", math_result)
    print("天气查询结果:", weather_result)

这种多服务器架构允许系统按需扩展，每个服务器可以独立开发、部署和扩展，同时保持客户端接口的一致性。

思考：如何处理不同服务器提供同名工具时的命名冲突问题？

与LangGraph API服务器集成

将多服务器代理部署为API服务：

# graph.py
from contextlib import asynccontextmanager
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.prebuilt import create_react_agent
from langchain_anthropic import ChatAnthropic

# 使用Claude模型
model = ChatAnthropic(model="claude-3-5-sonnet-latest")

@asynccontextmanager
async def make_graph():
    # 在上下文管理器中创建多服务器客户端
    async with MultiServerMCPClient({
        "math": {"command": "python", "args": ["math_server.py"], "transport": "stdio"},
        "weather": {"url": "http://localhost:8000/sse", "transport": "sse"}
    }) as client:
        # 创建包含所有工具的代理
        agent = create_react_agent(model, client.get_tools())
        yield agent

在langgraph.json中配置入口点：

{
  "entrypoint": "graph:make_graph"
}

启动LangGraph服务器：

langgraph serve

现在你可以通过HTTP API访问这个具备多工具能力的智能代理服务了。

生态拓展：行业应用与扩展开发

行业应用案例

1. 智能客服系统

整合CRM工具服务器、知识库查询服务器和工单系统，构建能够自动解答客户问题、查询订单状态并创建支持工单的智能客服代理。

2. 数据分析助手

连接SQL查询服务器、Excel处理服务器和可视化工具，使非技术人员能够通过自然语言查询数据、生成报表并创建数据可视化。

3. 开发辅助工具

集成代码分析服务器、文档生成工具和测试框架，构建能够帮助开发者审查代码、生成API文档和创建单元测试的AI助手。

扩展开发指南

自定义工具适配器

要开发自定义适配器，需要实现BaseToolAdapter抽象类：

from langchain_mcp_adapters.adapters import BaseToolAdapter

class MyCustomAdapter(BaseToolAdapter):
    def convert(self, mcp_tool):
        # 实现自定义转换逻辑
        langchain_tool = YourToolClass(
            name=mcp_tool.name,
            func=self._wrap_function(mcp_tool.func),
            description=mcp_tool.description
        )
        return langchain_tool
    
    def _wrap_function(self, original_func):
        # 实现函数包装逻辑
        async def wrapped(*args, **kwargs):
            # 添加自定义预处理或后处理逻辑
            return await original_func(*args, **kwargs)
        return wrapped

注册自定义适配器：

from langchain_mcp_adapters.registry import register_adapter

register_adapter("custom", MyCustomAdapter)

贡献代码

1. Fork项目仓库
2. 创建特性分支：git checkout -b feature/amazing-feature
3. 提交更改：git commit -m 'Add some amazing feature'
4. 推送到分支：git push origin feature/amazing-feature
5. 打开Pull Request

工具注册系统的核心实现可以在[langchain_mcp_adapters/tools.py]中找到，建议在开发自定义适配器前先了解现有实现。

通过这种灵活的扩展机制，开发者可以针对特定领域需求定制工具转换逻辑，进一步扩展LangChain MCP Adapters的应用范围。