Jean's Blog

一个专注软件测试开发技术的个人博客

0%

LangChain集成MCP

发表于 分类于
学习LangChain学习笔记第十四讲

MCP介绍

  • MCP（Model Context Protocol）：开源标准，定义 AI 模型如何与外部工具、资源通信，实现标准化连接，类似 AI 领域的 “通用接口”（类比 USB‑C）。
  • LangChain 与 MCP：LangChain 提供 langchain‑mcp‑adapters 工具库，用于在 LangChain 生态中快速接入、使用 MCP 协议的外部工具与服务。
  • 核心价值：简化 AI 应用与外部系统的集成，无需重复开发定制接口，提升开发效率与扩展性。

MCP 是 AI 与外部工具的统一通信标准，LangChain 通过专用适配器让开发者能便捷接入 MCP 生态，快速扩展 AI 能力。

传输类型

传输类型 核心说明 适用场景 特点
Stdio 本地子进程通信，通过标准输入 / 输出传输 本地工具、本地调试、轻量场景 轻量、快速、无网络开销；仅本机通信
Streamable HTTP 基于 HTTP 的流式传输，支持边传边处理 远程服务、分布式部署、实时交互 支持流式响应、会话管理、跨网络
Server-Sent Events (SSE) 基于 HTTP 的单向服务器推送 前端、轻量远程通信 单向推流、简单易用；已逐步被 Streamable HTTP 替代

SSE vs Stream HTTP的简明对比

项目 Streamable HTTP（流式 HTTP） SSE（服务器发送事件）
核心口诀 我问你答 你主动告诉我
通信方向 双向（客户端↔服务端） 单向（服务端→客户端）
连接方式 短连接复用、多次请求 长连接持续推送
工作流程 客户端发请求 → 服务端流式返回 → 结束断开 建立长连接 → 服务端持续推送
通俗场景 查天气、查股票、API 交互 股价提醒、消息通知、实时监控
适用场景 常规 API、多次交互、数据传输 实时推送、持续监听、主动通知
核心特点 双向交互、灵活适配、无状态 单向推送、自动重连、长连接

生活化场景对比

场景 Streamable HTTP SSE
查天气 你主动打电话问天气 天气 App 每天自动发天气通知
股票行情 你手动刷新看最新价格 股价变动时自动推送提醒
应用使用 你手动刷新获取新内容 新消息自动弹出来

优缺点总结

Streamable HTTP

  • ✅ 灵活、双向交互、适合大多数 API
  • ❌ 不能主动推送，需要客户端主动请求

SSE

  • ✅ 真正的实时推送、自动重连、长连接
  • ❌ 仅单向推送、浏览器兼容性要求高

选择建议

  • 需要实时监控、日志、弹幕、主动通知 → 用 SSE
  • 普通 API 调用、数据传输、不确定场景 → 用 Streamable HTTP

你主动问 → 用 Streamable HTTP；服务端主动推 → 用 SSE。

自定义MCP服务

创建自己的 MCP 服务，可以使用mcp库，用简单方式定义工具，并让客户端作为服务运行。

安装

1
2
3
4
pip install mcp

# 或使用uv
uv add mcp

参考一下实定义MCP工具服务器的交互

  • stdio传输

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    # 导入 MCP 框架
    from mcp import FastMCP

    # 创建一个 MCP 服务，名字叫 Math
    mcp = FastMCP("Math")

    # 定义第一个工具：两个数相加
    @mcp.tool()
    def add_two_numbers(a: int, b: int) -> int:
        return a + b

    # 定义第二个工具：两个数相乘
    @mcp.tool()
    def multiply_two_numbers(a: int, b: int) -> int:
        return a * b

    # 启动服务，使用本地 stdio 传输（只能本机调用）
    if __name__ == "__main__":
        mcp.run_transport("stdio")
    • FastMCP：创建一个 AI 能用的工具服务
    • @mcp.tool()：把函数变成AI 可调用的工具
    • stdio：本地进程通信，只在本机跑，不上网
    • 作用：让 AI 可以调用你的 加法 和 乘法

  • streamable HTTP传输

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    # 导入 MCP
    from mcp import FastMCP

    # 创建天气服务
    mcp = FastMCP("Weather")

    # 定义工具：获取天气
    @mcp.tool()
    def get_weather(city: str) -> str:
        # 假装去查天气
        return f"Sunny in {city}"

    # 启动流式 HTTP 服务（可远程访问）
    if __name__ == "__main__":
        mcp.run_transport("streamable-http")
    • streamable-http：启动可远程访问的 HTTP 流式服务
    • 别人 / AI 可以通过网络调用你的 get_weather 工具
    • 适合做远程服务、API、AI 插件

stdio：本地工具，本机用，轻量

streamable-http：远程服务，网络用，可公开

使用MCP工具

使用 langchain-mcp-adapters 库，让 LangChain 框架能够调用任意多个 MCP 服务上的工具。

简单说：LangChain + MCP 适配器 = 让 AI 能用所有 MCP 工具。

安装

1
2
3
4
pip install langchain-mcp-adapters

# 或uv安装
uv add langchain-mcp-adapters

代码示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.agents import create_agent

# 初始化多服务器客户端，配置两个不同的服务：数学(math)和天气(weather)
client = MultiServerMCPClient(
    {
        "math": {
            "transport": "stdio",  # 本地子进程通信
            "command": "python",
            # 指向你的 math_server.py 文件的绝对路径
            "args": ["/path/to/math_server.py"],
        },
        "weather": {
            "transport": "streamable_http",  # 基于 HTTP 的远程服务器
            # 确保你的天气服务器运行在 8000 端口
            "url": "http://localhost:8000/mcp",
        }
    }
)

# 获取所有配置服务器中的工具列表
tools = await client.get_tools()

# 创建 Agent，并将获取到的工具注入
agent = create_agent(
    "claude-3-5-sonnet-20241022", # 注意：原图代码中日期为20250929，此处建议使用实际可用版本
    tools
)

# 调用 Agent 执行数学计算任务
math_response = await agent.ainvoke(
    {"messages": [{"role": "user", "content": "what's (3 + 5) x 12?"}]}
)

# 调用 Agent 执行天气查询任务
weather_response = await agent.ainvoke(
    {"messages": [{"role": "user", "content": "what is the weather in nyc?"}]}
)

这段代码展示了 MCP 的核心优势：互操作性。它允许 AI Agent 像插拔 U 盘一样，同时接入多个异构的数据源或功能模块。

关键组件说明：

  • MultiServerMCPClient: 这是整个架构的“集线器”。它负责管理与多个 MCP 服务器的连接。在配置中，你可以混合使用不同的传输协议（如本地的 stdio 或远程的 http）。
  • 传输协议 (Transport):
    • stdio: 适用于本地脚本，通过标准输入输出与 Python 进程直接通信，安全性高且延迟低。
    • streamable_http: 适用于远程服务，允许 Agent 调用部署在云端或其他服务器上的工具。
  • 工具发现 (get_tools): 这是一个异步操作，它会自动遍历所有配置的服务器，查询它们支持的功能（Tools），并将其统一封装为 LangChain 可识别的格式。
  • 无状态性 (Stateless): 图片底部的提示非常重要。MultiServerMCPClient 默认是无状态的。这意味着每次执行工具调用时，它都会创建一个新的 ClientSession，执行完后立即清理。这保证了资源的释放，但也意味着如果你需要跨调用维持某种连接状态，需要额外的处理。

上面介绍了无状态的工具使用，下面在介绍下有状态工具使用

1
2
3
4
5
6
7
8
9
10
from langchain_mcp_adapters.tools import load_mcp_tools

# 假设 client 已经按照之前的方法初始化
client = MultiServerMCPClient({...})

# 使用 async with 语句为特定的服务器（如 "math"）创建一个持久会话
async with client.session("math") as session:
    # 在该会话上下文中加载工具
    tools = await load_mcp_tools(session)
    # 在此代码块内，session 会一直保持连接状态，直到退出 block

这段代码解决的核心问题是：如何在多次工具调用之间记住“我是谁”或“我之前做了什么”。

关键概念说明：

  • client.session("math"): 这是核心方法。它显式地为名为 "math" 的服务器开启一个连接通道。
  • async with (上下文管理器): 使用异步上下文管理器可以确保连接的生命周期管理。当程序进入 with 块时，连接建立；当程序执行完毕退出块时，连接会自动安全关闭。
  • load_mcp_tools(session): 与之前的 client.get_tools() 不同，这里是将工具直接绑定到这个特定的、活动的 session 上。

有状态的工具有什么用呢

在大多数简单的 API 调用中，无状态就足够了。但在以下场景中，必须使用 client.session()：

场景 描述
认证与登录 某些服务器需要先调用 login 工具，后续操作必须在同一个 Session 中才能识别身份。
长连接资源 比如操作一个打开的数据库游标、文件流或正在运行的仿真模拟器。
性能优化 如果工具调用极其频繁，重复创建/销毁 Session 会产生大量的握手开销。保持 Session 可以显著降低延迟。

总结对比

  • 默认模式 (无状态)：client.get_tools() -> 简单、自动、适合独立任务（如查天气）。
  • 持久模式 (有状态)：async with client.session(...) -> 灵活、可控、适合复杂且有上下文关联的交互任务。