LangChain MCP 适配器

该库提供了一个轻量级包装器，使 Anthropic 模型上下文协议 (MCP) 工具与 LangChain 和 LangGraph 兼容。

[!note] 此库的 JavaScript/TypeScript 版本也可在 langchainjs 中找到。

功能

• 🛠️ 将 MCP 工具转换为可与 LangGraph 代理一起使用的 LangChain 工具。
• 📦 一个客户端实现，允许您连接到多个 MCP 服务器并从中加载工具。

安装

pip install langchain-mcp-adapters

快速入门

以下是一个使用 MCP 工具与 LangGraph 代理的简单示例。

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

export OPENAI_API_KEY=<your_api_key>

服务器

首先，让我们创建一个可以对数字进行加法和乘法运算的 MCP 服务器。

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

mcp = FastMCP("Math")

@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")

客户端

# 创建用于 stdio 连接的服务器参数
from mcp import ClientSession、StdioServerParameters 和 mcp.client.stdio 的 stdio_client

从 langchain_mcp_adapters.tools 导入 load_mcp_tools，从 langchain.agents 导入 create_agent

server_params = StdioServerParameters(
    command="python",
    # 确保更新为 math_server.py 文件的完整绝对路径
    args=["/path/to/math_server.py"],
)

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)

        # 创建并运行代理
        agent = create_agent("openai:gpt-4.1", tools)
        agent_response = await agent.ainvoke({"messages": "3 加 5 再乘以 12 是多少？"})

多个 MCP 服务器

该库还允许您连接到多个 MCP 服务器，并从中加载工具：

服务器

# math_server.py
...

# weather_server.py
from typing import List
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("Weather")

@mcp.tool()
async def get_weather(location: str) -> str:
    """获取某个地点的天气状况"""
    return "纽约总是阳光明媚"

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

python weather_server.py

客户端

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

client = MultiServerMCPClient(
    {
        "math": {
            "command": "python",
            # 确保更新为 math_server.py 文件的完整绝对路径
            "args": ["/path/to/math_server.py"],
            "transport": "stdio",
        },
        "weather": {
            # 确保您的天气服务器在端口 8000 上启动
            "url": "http://localhost:8000/mcp",
            "transport": "http",
        }
    }
)
tools = await client.get_tools()
agent = create_agent("openai:gpt-4.1", tools)
math_response = await agent.ainvoke({"messages": "3 加 5 再乘以 12 是多少？"})
weather_response = await agent.ainvoke({"messages": "纽约的天气如何？"})

[!note] 上面的示例将在每次调用工具时启动一个新的 MCP ClientSession。如果您希望为特定服务器显式地启动会话，可以这样做：

from langchain_mcp_adapters.tools 导入 load_mcp_tools

client = MultiServerMCPClient({...})
async with client.session("math") as session:
    tools = await load_mcp_tools(session)

可流式传输的 HTTP

MCP 现在支持 可流式传输的 HTTP 传输方式。

要启动一个 示例 可流式传输的 HTTP 服务器，请运行以下命令：

cd examples/servers/streamable-http-stateless/
uv run mcp-simple-streamablehttp-stateless --port 3000

或者，您也可以直接使用 FastMCP（如上文示例所示）。

要将其与 Python MCP SDK 的 streamablehttp_client 一起使用：

# 使用 examples/servers/streamable-http-stateless 中的服务器

from mcp import ClientSession 和 mcp.client.streamable_http 的 streamablehttp_client

从 langchain.agents 导入 create_agent，从 langchain_mcp_adapters.tools 导入 load_mcp_tools

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)
        agent = create_agent("openai:gpt-4.1", tools)
        math_response = await agent.ainvoke({"messages": "3 加 5 再乘以 12 是多少？"})

使用 MultiServerMCPClient：

# 使用 examples/servers/streamable-http-stateless 中的服务器
from langchain_mcp_adapters.client 导入 MultiServerMCPClient，从 langchain.agents 导入 create_agent

client = MultiServerMCPClient(
    {
        "math": {
            "transport": "http",
            "url": "http://localhost:3000/mcp"
        },
    }
)
tools = await client.get_tools()
agent = create_agent("openai:gpt-4.1", tools)
math_response = await agent.ainvoke({"messages": "3 加 5 再乘以 12 是多少？"})

传递运行时头信息

在连接到 MCP 服务器时，您可以使用连接配置中的 headers 字段包含自定义头信息（例如用于身份验证或跟踪）。此功能适用于以下传输方式：

• sse
• http（或 streamable_http）

示例：使用 MultiServerMCPClient 传递头信息

from langchain_mcp_adapters.client 导入 MultiServerMCPClient，从 langchain.agents 导入 create_agent

client = MultiServerMCPClient(
    {
        "weather": {
            "transport": "http",
            "url": "http://localhost:8000/mcp",
            "headers": {
                "Authorization": "Bearer YOUR_TOKEN",
                "X-Custom-Header": "custom-value"
            },
        }
    }
)
tools = await client.get_tools()
agent = create_agent("openai:gpt-4.1", tools)
response = await agent.ainvoke({"messages": "纽约的天气如何？"})

只有 sse 和 http 传输方式支持运行时头信息。这些头信息会随每次向 MCP 服务器发出的 HTTP 请求一起传递。

与 LangGraph StateGraph 配合使用

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

from langchain.chat_models import init_chat_model
model = init_chat_model("openai:gpt-4.1")

client = MultiServerMCPClient(
    {
        "math": {
            "command": "python",
            # 请确保更新为 math_server.py 文件的完整绝对路径
            "args": ["./examples/math_server.py"],
            "transport": "stdio",
        },
        "weather": {
            # 请确保在端口 8000 上启动天气服务器
            "url": "http://localhost:8000/mcp",
            "transport": "http",
        }
    }
)
tools = await client.get_tools()

def call_model(state: MessagesState):
    response = model.bind_tools(tools).invoke(state["messages"])
    return {"messages": response}

builder = StateGraph(MessagesState)
builder.add_node(call_model)
builder.add_node(ToolNode(tools))
builder.add_edge(START, "call_model")
builder.add_conditional_edges(
    "call_model",
    tools_condition,
)
builder.add_edge("tools", "call_model")
graph = builder.compile()
math_response = await graph.ainvoke({"messages": "what's (3 + 5) x 12?"})
weather_response = await graph.ainvoke({"messages": "what is the weather in nyc?"})

与 LangGraph API 服务器配合使用

[!TIP] 请查看这篇指南，了解如何开始使用 LangGraph API 服务器。

如果您希望在 LangGraph API 服务器中运行使用 MCP 工具的 LangGraph 代理，可以采用以下设置：

# graph.py
from contextlib import asynccontextmanager
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.agents import create_agent

async def make_graph():
    client = MultiServerMCPClient(
        {
            "weather": {
                # 请确保在端口 8000 上启动天气服务器
                "url": "http://localhost:8000/mcp",
                "transport": "http",
            },
            # 注意：MCP 的 stdio 传输方式主要是为支持用户本地运行的应用程序而设计的。
            # 在 Web 服务器环境中使用 stdio 之前，请评估是否有更合适的解决方案。
            # 例如，您是否真的需要 MCP？或者仅仅使用简单的 `@tool` 就足够了？
            "math": {
                "command": "python",
                # 请确保更新为 math_server.py 文件的完整绝对路径
                "args": ["/path/to/math_server.py"],
                "transport": "stdio",
            },
        }
    )
    tools = await client.get_tools()
    agent = create_agent("openai:gpt-4.1", tools)
    return agent

在您的 langgraph.json 中，请确保将 make_graph 指定为您的图入口点：

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

LangChain MCP Adapters 快速上手指南

langchain-mcp-adapters 是一个轻量级封装库，旨在将 Anthropic Model Context Protocol (MCP) 工具无缝集成到 LangChain 和 LangGraph 生态中。通过它，你可以轻松连接多个 MCP 服务器，并将暴露的工具转换为 LangChain Agent 可调用的标准工具。

环境准备

在开始之前，请确保你的开发环境满足以下要求：

• 操作系统：Linux, macOS 或 Windows
• Python 版本：建议 Python 3.10 或更高版本
• 前置依赖：
  • 已安装 pip 包管理工具
  • 拥有有效的 LLM API Key（本指南以 OpenAI 为例）

安装步骤

使用 pip 安装核心库及必要的依赖项。国内开发者若遇到下载速度慢的问题，推荐使用清华或阿里镜像源。

# 使用默认源安装
pip install langchain-mcp-adapters langgraph "langchain[openai]"

# 或者使用国内镜像源加速安装（推荐）
pip install langchain-mcp-adapters langgraph "langchain[openai]" -i https://pypi.tuna.tsinghua.edu.cn/simple

安装完成后，请在终端设置你的 API Key：

export OPENAI_API_KEY=<your_api_key>

基本使用

以下示例演示了如何创建一个本地 MCP 数学服务，并通过 LangGraph Agent 调用该服务进行计算。

1. 创建 MCP 服务端

新建一个文件 math_server.py，定义两个简单的数学工具（加法和乘法）：

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

mcp = FastMCP("Math")

@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two numbers"""
    return a + b

@mcp.tool()
def multiply(a: int, b: int) -> int:
    """Multiply two numbers"""
    return a * b

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

2. 创建客户端并运行 Agent

新建一个客户端脚本，连接上述服务器并执行任务。请将代码中的 /path/to/math_server.py 替换为你实际的绝对路径。

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",
        # 重要：请更新为 math_server.py 的完整绝对路径
        args=["/path/to/math_server.py"],
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # 初始化连接
            await session.initialize()

            # 加载 MCP 工具
            tools = await load_mcp_tools(session)

            # 创建 Agent 并执行任务
            agent = create_agent("openai:gpt-4.1", tools)
            response = await agent.ainvoke({"messages": "what's (3 + 5) x 12?"})
            
            print(response)

if __name__ == "__main__":
    asyncio.run(main())

运行客户端脚本后，Agent 将自动调用 MCP 服务器中的 add 和 multiply 工具来计算 (3 + 5) * 12 的结果。