3.7 LangChain MCP Adapters 深度解析

来源: langchain-mcp-adapters GitHub | LangChain MCP 官方文档整理日期: 2025-12-22

概述

LangChain MCP Adapters 是一个轻量级的 Python 库，用于将 Anthropic 的 Model Context Protocol (MCP) 工具无缝集成到 LangChain 和 LangGraph 生态系统中。

💡 通俗理解: 如果把 AI 应用比作一台电脑，MCP 就像是 USB 接口标准，而 LangChain MCP Adapters 就是将各种 MCP "设备"（工具服务器）转换成 LangChain 能识别的"驱动程序"。

核心功能

功能	说明
工具转换	自动将 MCP 工具转换为 LangChain/LangGraph 兼容格式
多服务器管理	同时连接和管理多个 MCP 服务器
传输协议支持	支持 stdio、HTTP、Streamable HTTP 等多种传输方式
拦截器机制	提供中间件模式，实现认证、日志、重试等功能

---

1. MCP 协议架构基础

在深入 LangChain MCP Adapters 之前，我们需要理解 MCP 的核心架构。

1.1 三大组件

┌─────────────────────────────────────────────────────────────────┐
│                        MCP Host（宿主应用）                       │
│                  如：Claude Desktop、VS Code、自定义应用          │
│                                                                 │
│  ┌─────────────────┐  ┌─────────────────┐  ┌─────────────────┐  │
│  │   MCP Client    │  │   MCP Client    │  │   MCP Client    │  │
│  │   (客户端 1)    │  │   (客户端 2)    │  │   (客户端 3)    │  │
│  └────────┬────────┘  └────────┬────────┘  └────────┬────────┘  │
└───────────┼────────────────────┼────────────────────┼───────────┘
            │                    │                    │
            ▼                    ▼                    ▼
    ┌───────────────┐    ┌───────────────┐    ┌───────────────┐
    │  MCP Server   │    │  MCP Server   │    │  MCP Server   │
    │  (数学服务)   │    │  (天气服务)   │    │  (文件服务)   │
    └───────────────┘    └───────────────┘    └───────────────┘

组件	英文	职责
宿主应用	MCP Host	运行 AI 应用的主程序，负责管理 MCP 客户端
客户端	MCP Client	与 MCP 服务器建立一对一连接，发送请求并接收响应
服务器	MCP Server	提供工具、资源和提示词模板，执行实际操作

1.2 协议分层

┌─────────────────────────────────────┐
│       数据层 (JSON-RPC 2.0)         │
│  • 生命周期管理 (初始化/关闭)        │
│  • 原语交互 (工具/资源/提示词)       │
│  • 实时通知                         │
└─────────────────────────────────────┘
              ↕ (抽象)
┌─────────────────────────────────────┐
│          传输层 (Transport)         │
│  • stdio (本地进程通信)              │
│  • Streamable HTTP (远程服务)        │
└─────────────────────────────────────┘

1.3 MCP 原语 (Primitives)

MCP 服务器可以暴露三种类型的原语：

原语类型	英文	用途	示例
工具	Tools	可执行的函数，LLM 可以调用	数据库查询、API 调用、文件操作
资源	Resources	提供上下文数据的来源	文件内容、数据库记录、API 响应
提示词	Prompts	可重用的交互模板	系统提示词、少样本示例

---

2. 安装与配置

2.1 基础安装

pip install langchain-mcp-adapters

2.2 完整安装（推荐）

pip install langchain-mcp-adapters langgraph "langchain[openai]"

2.3 设置环境变量

# OpenAI API（如果使用 OpenAI 模型）
export OPENAI_API_KEY=<your_api_key>

# Anthropic API（如果使用 Claude 模型）
export ANTHROPIC_API_KEY=<your_api_key>

---

3. 核心使用模式

3.1 创建自定义 MCP 服务器

使用 FastMCP 库创建一个简单的数学运算服务器：

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

# 创建 MCP 服务器实例
mcp = FastMCP("Math")

@mcp.tool()
def add(a: int, b: int) -> int:
    """
    加法运算

    Args:
        a: 第一个加数
        b: 第二个加数

    Returns:
        两数之和
    """
    return a + b

@mcp.tool()
def multiply(a: int, b: int) -> int:
    """
    乘法运算

    Args:
        a: 被乘数
        b: 乘数

    Returns:
        两数之积
    """
    return a * b

@mcp.tool()
def divide(a: float, b: float) -> float:
    """
    除法运算

    Args:
        a: 被除数
        b: 除数

    Returns:
        商

    Raises:
        ValueError: 当除数为零时
    """
    if b == 0:
        raise ValueError("除数不能为零")
    return a / b

if __name__ == "__main__":
    mcp.run(transport="stdio")

3.2 单服务器连接（stdio 传输）

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_agent

async def main():
    # 配置服务器参数
    server_params = StdioServerParameters(
        command="python",
        args=["math_server.py"],
    )

    # 建立连接并加载工具
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # 初始化会话
            await session.initialize()

            # 加载 MCP 工具并转换为 LangChain 格式
            tools = await load_mcp_tools(session)

            # 创建 Agent
            agent = create_agent("openai:gpt-4.1", tools)

            # 执行查询
            response = await agent.ainvoke({
                "messages": "计算 (3 + 5) × 12 的结果是多少？"
            })

            print(response)

asyncio.run(main())

3.3 多服务器连接（MultiServerMCPClient）

from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.agents import create_agent

async def main():
    # 配置多个 MCP 服务器
    client = MultiServerMCPClient({
        # 本地数学服务（stdio 传输）
        "math": {
            "command": "python",
            "args": ["math_server.py"],
            "transport": "stdio",
        },
        # 远程天气服务（HTTP 传输）
        "weather": {
            "url": "http://localhost:8000/mcp",
            "transport": "http",
        },
        # 带认证的数据服务
        "database": {
            "url": "https://api.example.com/mcp",
            "transport": "http",
            "headers": {
                "Authorization": "Bearer YOUR_TOKEN",
                "X-API-Version": "v2"
            },
        }
    })

    # 获取所有服务器的工具
    tools = await client.get_tools()

    # 创建 Agent
    agent = create_agent("anthropic:claude-sonnet-4-5-20250929", tools)

    # 执行复杂查询
    response = await agent.ainvoke({
        "messages": "查询北京今天的天气，然后计算华氏温度"
    })

    print(response)

asyncio.run(main())

---

4. 传输协议详解

4.1 传输类型对比

传输类型	适用场景	特点
stdio	本地进程	高性能，无网络开销，进程生命周期与客户端绑定
http	远程服务	支持认证头，无状态，每次调用创建新连接
streamable_http	现代远程服务	替代 SSE，支持流式响应

⚠️ 注意: 从 MCP 协议版本 2025-03-26 开始，SSE 作为独立传输方式已被弃用，推荐使用 Streamable HTTP。

4.2 Streamable HTTP 示例

from mcp.client.streamable_http import streamablehttp_client
from mcp import ClientSession
from langchain_mcp_adapters.tools import load_mcp_tools

async def connect_streamable():
    async with streamablehttp_client("http://localhost:3000/mcp") as (read, write, _):
        async with ClientSession(read, write) as session:
            await session.initialize()
            tools = await load_mcp_tools(session)
            return tools

4.3 HTTP 认证配置

import httpx
from langchain_mcp_adapters.client import MultiServerMCPClient

# 方式一：使用 headers
client = MultiServerMCPClient({
    "api_server": {
        "transport": "http",
        "url": "https://api.example.com/mcp",
        "headers": {
            "Authorization": "Bearer sk-xxx",
            "X-Custom-Header": "custom-value"
        },
    }
})

# 方式二：使用 httpx.Auth（更灵活）
class CustomAuth(httpx.Auth):
    def auth_flow(self, request):
        request.headers["Authorization"] = f"Bearer {self.get_token()}"
        yield request

    def get_token(self):
        # 动态获取 token 的逻辑
        return "dynamic-token"

client = MultiServerMCPClient({
    "api_server": {
        "transport": "http",
        "url": "https://api.example.com/mcp",
        "auth": CustomAuth(),
    }
})

---

5. 状态管理

5.1 无状态模式（默认）

MultiServerMCPClient 默认采用无状态模式：

# 每次工具调用都会：
# 1. 创建新的 MCP ClientSession
# 2. 执行工具
# 3. 清理并关闭会话
client = MultiServerMCPClient({...})
tools = await client.get_tools()
# 调用工具时自动管理会话

5.2 有状态会话

当服务器需要在多次调用间保持上下文时，使用显式会话：

from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_mcp_adapters.tools import load_mcp_tools
from langchain.agents import create_agent

async def stateful_example():
    client = MultiServerMCPClient({
        "conversational_db": {
            "command": "python",
            "args": ["db_server.py"],
            "transport": "stdio",
        }
    })

    # 使用显式会话保持状态
    async with client.session("conversational_db") as session:
        tools = await load_mcp_tools(session)
        agent = create_agent("openai:gpt-4.1", tools)

        # 多次交互共享同一会话状态
        await agent.ainvoke({"messages": "打开数据库连接"})
        await agent.ainvoke({"messages": "执行查询 SELECT * FROM users"})
        await agent.ainvoke({"messages": "关闭连接"})

---

6. 高级功能：拦截器 (Interceptors)

拦截器是 LangChain MCP Adapters 的强大特性，提供类似中间件的功能，可以在工具执行前后进行处理。

6.1 拦截器架构

请求 → [拦截器1: before] → [拦截器2: before] → 工具执行 → [拦截器2: after] → [拦截器1: after] → 响应

💡 洋葱模型: 多个拦截器按照"先进后出"的顺序执行，类似 Express.js 的中间件机制。

6.2 注入用户上下文

from langchain_mcp_adapters.client import MultiServerMCPClient, MCPToolCallRequest

async def inject_user_context(request: MCPToolCallRequest, handler):
    """在每次工具调用中注入用户信息"""
    runtime = request.runtime
    user_id = runtime.context.user_id

    # 修改请求参数，添加用户 ID
    modified_request = request.override(
        args={**request.args, "user_id": user_id}
    )

    return await handler(modified_request)

client = MultiServerMCPClient(
    {...},
    tool_interceptors=[inject_user_context],
)

6.3 认证检查拦截器

from langchain_core.messages import ToolMessage

async def require_authentication(request: MCPToolCallRequest, handler):
    """敏感操作需要认证"""
    state = request.runtime.state
    is_authenticated = state.get("authenticated", False)

    # 定义需要认证的操作列表
    sensitive_tools = ["delete_file", "execute_sql", "send_email"]

    if request.name in sensitive_tools and not is_authenticated:
        return ToolMessage(
            content="❌ 此操作需要先进行身份认证。",
            tool_call_id=request.runtime.tool_call_id,
        )

    return await handler(request)

6.4 重试与错误处理

import asyncio

async def retry_interceptor(request: MCPToolCallRequest, handler, max_retries=3):
    """自动重试失败的工具调用"""
    for attempt in range(max_retries):
        try:
            return await handler(request)
        except Exception as e:
            if attempt < max_retries - 1:
                # 指数退避
                wait_time = 1.0 * (2 ** attempt)
                print(f"工具调用失败，{wait_time}秒后重试... (尝试 {attempt + 1}/{max_retries})")
                await asyncio.sleep(wait_time)
            else:
                raise RuntimeError(f"工具调用失败，已达最大重试次数: {e}")

6.5 动态添加认证头

async def auth_header_interceptor(request: MCPToolCallRequest, handler):
    """根据工具名称动态添加认证头"""
    token = await get_token_for_tool(request.name)

    modified_request = request.override(
        headers={"Authorization": f"Bearer {token}"}
    )

    return await handler(modified_request)

async def get_token_for_tool(tool_name: str) -> str:
    """根据工具获取对应的认证令牌"""
    token_map = {
        "github_api": "ghp_xxx",
        "slack_api": "xoxb-xxx",
        "database": "db_token_xxx",
    }
    return token_map.get(tool_name, "default_token")

6.6 状态更新与流程控制

from langgraph.types import Command

async def handle_task_completion(request: MCPToolCallRequest, handler):
    """任务完成后更新状态并跳转"""
    result = await handler(request)

    if request.name == "submit_order":
        # 返回 Command 对象来更新状态和控制流程
        return Command(
            update={"task_status": "completed", "order_id": result.content},
            goto="summary_agent",  # 跳转到摘要处理节点
        )

    return result

---

7. 回调与事件处理

7.1 进度通知

from langchain_mcp_adapters.callbacks import Callbacks

async def on_progress(progress, total, message, context):
    """处理进度更新"""
    if total:
        percent = progress / total * 100
        print(f"[{context.server_name}] 进度: {percent:.1f}% - {message}")
    else:
        print(f"[{context.server_name}] 进度: {progress} - {message}")

client = MultiServerMCPClient(
    {...},
    callbacks=Callbacks(on_progress=on_progress)
)

7.2 日志处理

async def on_logging_message(params, context):
    """处理服务器日志"""
    level_emoji = {
        "debug": "🐛",
        "info": "ℹ️",
        "warning": "⚠️",
        "error": "❌"
    }
    emoji = level_emoji.get(params.level, "📝")
    print(f"{emoji} [{context.server_name}] {params.level.upper()}: {params.data}")

client = MultiServerMCPClient(
    {...},
    callbacks=Callbacks(on_logging_message=on_logging_message)
)

7.3 交互式输入 (Elicitation)

MCP 服务器可以在执行过程中请求用户输入：

# 服务器端 (FastMCP)
from mcp.server.fastmcp import FastMCP, Context
from pydantic import BaseModel

mcp = FastMCP("UserProfile")

class UserDetails(BaseModel):
    email: str
    age: int
    preferences: list[str]

@mcp.tool()
async def create_profile(name: str, ctx: Context) -> str:
    """创建用户档案"""
    # 请求用户提供详细信息
    result = await ctx.elicit(
        message=f"请为用户 {name} 提供详细信息：",
        schema=UserDetails,
    )

    if result.action == "accept":
        return f"档案创建成功: {name} - {result.data}"
    elif result.action == "decline":
        return "用户拒绝提供信息"
    else:  # cancel
        return "操作已取消"

# 客户端处理
from langchain_mcp_adapters.callbacks import Callbacks, ElicitResult

async def on_elicitation(mcp_context, params, context):
    """处理服务器的交互式请求"""
    print(f"服务器请求输入: {params.message}")

    # 可以从 UI 获取用户输入，这里演示自动响应
    return ElicitResult(
        action="accept",
        content={
            "email": "user@example.com",
            "age": 25,
            "preferences": ["tech", "music"]
        }
    )

client = MultiServerMCPClient(
    {...},
    callbacks=Callbacks(on_elicitation=on_elicitation),
)

---

8. LangGraph 集成

8.1 基础 StateGraph 模式

from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import ToolNode, tools_condition
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_openai import ChatOpenAI

async def create_mcp_graph():
    # 初始化 MCP 客户端
    client = MultiServerMCPClient({
        "math": {
            "command": "python",
            "args": ["math_server.py"],
            "transport": "stdio",
        }
    })

    # 获取工具
    tools = await client.get_tools()

    # 创建模型
    model = ChatOpenAI(model="gpt-4").bind_tools(tools)

    # 定义节点函数
    def call_model(state: MessagesState):
        response = model.invoke(state["messages"])
        return {"messages": [response]}

    # 构建图
    builder = StateGraph(MessagesState)
    builder.add_node("call_model", call_model)
    builder.add_node("tools", ToolNode(tools))

    # 添加边
    builder.add_edge(START, "call_model")
    builder.add_conditional_edges("call_model", tools_condition)
    builder.add_edge("tools", "call_model")

    return builder.compile()

8.2 部署到 LangGraph API Server

graph.py:

from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.agents import create_agent

async def make_graph():
    """创建可部署的 Agent 图"""
    client = MultiServerMCPClient({
        "weather": {
            "url": "http://localhost:8000/mcp",
            "transport": "http",
        },
        "calendar": {
            "url": "http://localhost:8001/mcp",
            "transport": "http",
        }
    })

    tools = await client.get_tools()
    return create_agent("openai:gpt-4.1", tools)

langgraph.json:

{
    "dependencies": ["."],
    "graphs": {
        "agent": "./graph.py:make_graph"
    },
    "env": ".env"
}

---

9. 多模态内容处理

MCP 工具可以返回图像、文本等多种类型的内容：

from langchain_core.messages import ToolMessage

async def process_multimodal_response(result):
    """处理包含多模态内容的响应"""
    for message in result["messages"]:
        if isinstance(message, ToolMessage):
            # 处理结构化内容
            if message.artifact:
                structured = message.artifact.get("structured_content")
                print(f"结构化数据: {structured}")

            # 处理多模态内容块
            for block in getattr(message, 'content_blocks', []):
                if block["type"] == "text":
                    print(f"文本: {block['text']}")
                elif block["type"] == "image":
                    print(f"图像 URL: {block.get('url')}")
                    # 可以下载或显示图像

---

10. 最佳实践

10.1 传输选择指南

场景	推荐传输	原因
本地开发/测试	stdio	简单快速，无需网络配置
生产环境远程服务	Streamable HTTP	现代标准，支持流式响应
需要认证的 API	HTTP + headers	灵活的认证配置

10.2 性能优化

# ✅ 推荐：复用客户端实例
client = MultiServerMCPClient({...})

async def handle_request(query):
    tools = await client.get_tools()  # 工具列表会被缓存
    agent = create_agent("openai:gpt-4.1", tools)
    return await agent.ainvoke({"messages": query})

# ❌ 避免：每次请求创建新客户端
async def handle_request_bad(query):
    client = MultiServerMCPClient({...})  # 每次都创建新连接
    # ...

10.3 错误处理模式

from langchain_mcp_adapters.client import MultiServerMCPClient

async def robust_mcp_call():
    client = MultiServerMCPClient({
        "api": {
            "url": "http://api.example.com/mcp",
            "transport": "http",
        }
    })

    try:
        tools = await client.get_tools()
        # 使用工具...
    except ConnectionError as e:
        print(f"无法连接到 MCP 服务器: {e}")
        # 实现降级策略
    except TimeoutError:
        print("请求超时，请检查网络连接")
    finally:
        # 确保清理资源
        await client.close()

10.4 安全考虑

# ✅ 使用环境变量存储敏感信息
import os

client = MultiServerMCPClient({
    "secure_api": {
        "url": os.getenv("MCP_API_URL"),
        "transport": "http",
        "headers": {
            "Authorization": f"Bearer {os.getenv('MCP_API_TOKEN')}"
        }
    }
})

# ⚠️ stdio 传输注意事项
# MCP 的 stdio 传输主要设计用于本地应用
# 在 Web 服务器中使用前，请评估是否有更合适的方案

---

11. 行业采用情况

自 2024 年 11 月 Anthropic 发布 MCP 以来，该协议得到了广泛采用：

时间	事件
2024.11	Anthropic 发布 MCP
2025.02	超过 1,000 个开源连接器发布
2025.03	OpenAI 正式采用 MCP
2025.12	MCP 被捐赠给 Linux 基金会下的 Agentic AI Foundation

主要采用者:

• AWS、GitHub、OpenAI
• Block (Square)、Apollo、Zed
• Replit、Codeium、Sourcegraph

---

总结

LangChain MCP Adapters 是连接 MCP 生态系统与 LangChain/LangGraph 的关键桥梁。通过本文档，你学习了：

1. ✅ MCP 协议的核心架构和组件
2. ✅ 如何安装和配置 LangChain MCP Adapters
3. ✅ 单服务器和多服务器的连接模式
4. ✅ 不同传输协议的选择和使用
5. ✅ 拦截器机制实现高级功能
6. ✅ 回调与事件处理
7. ✅ LangGraph 集成部署
8. ✅ 生产环境最佳实践

---

参考资料

• langchain-mcp-adapters GitHub
• LangChain MCP 官方文档
• LangGraph MCP 参考
• MCP 官方文档
• MCP 架构说明
• PyPI: langchain-mcp-adapters
• The Complete Guide to langchain-mcp-adapters