langchain + MCP：如虎添翼

MCP技术毋须多言了，上半年火的一塌糊涂，现在进入冷静期了。

langchain 本身就很方便的集成进程内的工具，但是加上 MCP的功能，就如虎添翼，可以充分利用网上上万的MCP的服务。

langchain 自从上个月融资了1.25亿美元之后，资金充足，也更加有动力推进产品的演化，相继发布了langchain/langgraph 1.0的版本。 langchain 1.0中统一了agent的创建，使用create_agent代替之前的create_tool_calling_agent、create_react_agent、create_json_agent、create_xml_agent等。

这篇文章介绍 2（MCP的两种模式sse、stdio） x 2 (经典的langchain agent和1.0最新版create_agent两个模式)一共4个例子，介绍了langchain如何使用MCP 工具丰富其功能。

先决条件

在运行这些示例之前，请确保您已安装必要的库。您通常可以使用 pip 进行安装：

pip install langchain langchain-openai langchain-classic langchain-mcp-adapters mcp

此外，请确保您的环境中已安装 python3。

共同组件

这两个示例都共享几个核心组件和一个共同目标：使用 LangChain Agent 通过 MCP 服务器提供的工具来回答一个简单的数学问题。

• LLM 配置：两个示例都使用 ChatOpenAI，配置为 deepseek-v3 模型，温度为 0。
  这是我使用百度云上提供的deepseek服务，你要是使用deepseek官方的服务，需要修改模型为deepseek-chat。
  我已经把KEY和调用地址配置在环境变量中了,所以在代码中不用显示指定：

export OPENAI_API_KEY=bce-v3/abcsfsfdskgergerthntjrweeuidfu8324refbif3
export OPENAI_API_BASE=https://qianfan.baidubce.com/v2

创建模型对象：

llm = ChatOpenAI(model="deepseek-v3", temperature=0)

• Agent 提示：ChatPromptTemplate 用于定义 Agent 的角色并构建对话。它包括一个系统消息、可选的聊天历史记录、人类输入以及 Agent 暂存区（用于规划其行动）的占位符。

  prompt = ChatPromptTemplate.from_messages(
      [
          ("system", "你是一个可以使用工具的得力助手。 தயவுசெய்து கருவிகளைப் பயன்படுத்தவும்."),
          MessagesPlaceholder("chat_history", optional=True),
          ("human", "{input}"),
          MessagesPlaceholder("agent_scratchpad"),
      ]
  )

• Agent 创建和执行：create_tool_calling_agent 函数用于构建能够使用工具的 Agent ，AgentExecutor 用于运行 Agent 。

  agent = create_tool_calling_agent(llm, tools, prompt)
  agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True) # SSE 示例中 verbose=False

• 任务： Agent 被调用，输入为“123 + 456 等于多少？”。

MCP 服务

Stdio服务

import asyncio
from mcp.server import FastMCP

# 创建一个服务器实例
server = FastMCP(name="math_server", log_level="ERROR")

# 定义并注册 add 工具
@server.tool()
def add(a: int, b: int) -> int:
    """将两个整数相加。"""
    return a + b

async def main():
    await server.run_stdio_async()

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

sse 服务

import asyncio
from typing import Annotated
from mcp.server import FastMCP

# 创建一个服务器实例
server = FastMCP(name="math_server", instructions="一个可以做加法的简单数学服务器。", log_level="ERROR")

# 定义并注册 add 工具
@server.tool()
def add(
    a: Annotated[int, "第一个整数"],
    b: Annotated[int, "第二个整数"]
) -> int:
    """将两个整数相加。"""
    return a + b

async def main():
    await server.run_sse_async()

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

示例 1：example_1_mcp_tool_stdio.py（标准 I/O 通信）

此示例演示了如何设置一个 MCP 服务器，该服务器通过标准输入和输出流与客户端通信。这适用于本地、单进程交互，其中服务器可以作为子进程生成。

目的

example_1_mcp_tool_stdio.py 脚本展示了如何：

1. 定义一个通过 stdio 暴露工具的 MCP 服务器。
2. 创建一个 stdio_client 来连接到此服务器。
3. 将服务器提供的工具加载到 LangChain Agent 中。
4. 使用 Agent 解决需要加载工具的问题。

关键组件和代码

import asyncio
import os
from pathlib import Path

from langchain_openai import ChatOpenAI
from langchain_classic.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_mcp_adapters.tools import load_mcp_tools
from mcp.client.stdio import stdio_client
from mcp import ClientSession, StdioServerParameters

async def run_mcp_tool_example():
    # ... (LLM 和提示设置，如共同组件中所述) ...

    # 1. 定义 MCP 数学服务器的路径
    # 此行构建了 stdio 数学服务器脚本的绝对路径。
    # `Path(__file__).parent` 获取当前脚本的目录。
    mcp_server_path = Path(__file__).parent / "mcp_math_server_stdio.py"

    # 2. 为 MCP 服务器设置 stdio 客户端参数
    # StdioServerParameters 指定如何运行 MCP 服务器。
    # `command="python3"` 指示解释器。
    # `args=[str(mcp_server_path)]` 提供要作为参数执行的脚本。
    server_params = StdioServerParameters(
        command="python3",
        args=[str(mcp_server_path)],
    )

    # 3. 建立 stdio 客户端会话
    # `stdio_client(server_params)` 创建一个异步上下文管理器，
    # 它生成服务器进程并提供用于通信的读/写流。
    async with stdio_client(server_params) as (read, write):
        # 4. 创建 MCP 客户端会话
        # `ClientSession` 通过提供的读/写流管理 MCP 协议。
        async with ClientSession(read, write) as session:
            # 5. 初始化连接
            # 此步骤对于客户端和服务器建立 MCP 连接至关重要。
            await session.initialize()
            # 6. 从 MCP 服务器获取工具
            # `load_mcp_tools(session)` 从 MCP 服务器获取工具定义
            # 并将它们转换为 LangChain 兼容的工具对象。
            tools = await load_mcp_tools(session)

            # ... ( Agent 创建和执行，如共同组件中所述) ...

            print("正在调用 agent 回答一个数学问题...")
            response = await agent_executor.ainvoke(
                {"input": "123 + 456 等于多少？", "chat_history": []}
            )
            print(f"Agent 回答: {response['output']}")

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

如何运行

要运行此示例，只需执行 Python 脚本：

python3 example_1_mcp_tool_stdio.py

您将看到 Agent 的思考过程和最终答案打印到控制台。

示例 2：example_1_mcp_tool_sse.py（服务器发送事件通信）

此示例演示了如何集成一个 MCP 服务器，该服务器通过 HTTP 上的服务器发送事件（SSE）暴露其工具。此方法更适用于工具服务器可能是独立网络服务的情况。

目的

example_1_mcp_tool_sse.py 脚本展示了如何：

1. 将 MCP SSE 服务器作为单独的进程启动。
2. 创建一个 MultiServerMCPClient 来连接到此基于 HTTP 的服务器。
3. 将服务器提供的工具加载到 LangChain Agent 中。
4. 使用 Agent 解决需要加载工具的问题。

关键组件和代码演练

import asyncio
import os
from pathlib import Path
import sys

from langchain_openai import ChatOpenAI
from langchain_classic.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_mcp_adapters.client import MultiServerMCPClient


async def run_mcp_tool_example():
    # ... (LLM 和提示设置，如共同组件中所述) ...

    # 1. 启动 MCP 数学服务器作为子进程
    # 这将 `mcp_math_server_sse.py` 脚本作为单独的 Python 进程启动。
    # `sys.executable` 确保使用正确的 Python 解释器。
    # `stdout=asyncio.subprocess.PIPE` 和 `stderr=asyncio.subprocess.PIPE`
    # 捕获子进程的输出，尽管此处未明确读取。
    mcp_server_path = Path(__file__).parent / "mcp_math_server_sse.py"
    server_process = await asyncio.create_subprocess_exec(
        sys.executable, str(mcp_server_path),
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )

    # 2. 等待服务器启动
    # 添加一个小的延迟，以使 SSE 服务器有时间初始化并开始监听。
    await asyncio.sleep(5)

    # 3. 为 MCP 服务器设置客户端
    # `MultiServerMCPClient` 用于基于 HTTP 的 MCP 服务器。
    # 它接受一个字典，其中键是服务器名称（例如，“math”），值
    # 指定传输类型（“sse”）和 SSE 端点的 URL。
    client = MultiServerMCPClient(
        {
            "math": {
                "transport": "sse",
                "url": "http://localhost:8000/sse",
            },
        }
    )

    # 4. 从 MCP 服务器获取工具
    # `client.get_tools()` 连接到指定的 SSE 端点，
    # 检索工具定义，并将其作为 LangChain 工具提供。
    tools = await client.get_tools()

    # ... ( Agent 创建和执行，如共同组件中所述) ...

    print("正在调用 agent 回答一个数学问题...")
    response = await agent_executor.ainvoke(
        {"input": "123 + 456 等于多少？", "chat_history": []}
    )
    print(f"Agent 回答: {response['output']}")

    # 5. 终止服务器进程
    # 通过终止已启动的子进程进行清理非常重要。
    server_process.terminate()
    await server_process.wait()


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

如何运行

要运行此示例，只需执行 Python 脚本：

python3 example_1_mcp_tool_sse.py

您将看到 Agent 的思考过程和最终答案打印到控制台。请注意，在此示例中，AgentExecutor 的 verbose 标志设置为 False，以避免 SSE 服务器子进程产生过多的输出。

比较：Stdio 与 SSE

• Stdio（标准 I/O）：

  • 简单性：更易于设置本地、单机通信。
  • 执行：MCP 服务器通常作为客户端应用程序的子进程生成。
  • 用例：适用于紧密耦合的组件，或者当您希望将工具服务器直接与应用程序捆绑时。

• SSE（服务器发送事件）：

  • 灵活性：允许 MCP 服务器作为独立的网络服务运行，可能在不同的机器上。
  • 可伸缩性：可以成为更大微服务架构的一部分。
  • 执行：客户端连接到正在运行的 HTTP 端点。服务器需要单独启动（如示例中通过 asyncio.create_subprocess_exec 所示）。
  • 用例：适用于分布式系统、基于 Web 的应用程序，或者当工具服务器需要被多个客户端访问时。

这两种方法都有效地允许 LangChain Agent 发现和利用使用多客户端协议定义的工具，从而为这些工具的部署和访问提供了灵活性。

这两个示例都是langchain经典的Agent开发模型，他们演示了langchain将 MCP 工具与 LangChain Agent 集成的强大功能和灵活性。无论它们是通过 stdio 还是 SSE 暴露。这使得健壮且可伸缩的 Agent 应用程序成为可能。

接下来我介绍如何使用 Langchain 1.0 中新的 create_agent 函数来构建 Agent ，并将其与 MCP 工具集成。与之前的示例不同，create_agent 提供了一种更简洁的方式来定义 Agent ，直接接受 LLM 模型、工具列表和系统提示，并返回一个可流式传输的图（graph）对象。

示例 3：使用 Langchain 1.0 create_agent 的 MCP SSE 工具集成

此示例展示了如何将 MCP SSE 服务器提供的工具与 Langchain 1.0 的 create_agent 函数结合使用。它演示了如何启动 SSE 服务器、获取工具，然后使用新的 Agent 创建和调用模式来解决数学问题。

import asyncio
import os
from pathlib import Path
import sys

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


async def run_mcp_sse_new_agent_example():
    server_process = None
    try:
        # 启动 MCP 数学服务器作为子进程
        mcp_server_path = Path(__file__).parent / "mcp_math_server_sse.py"
        server_process = await asyncio.create_subprocess_exec(
            sys.executable, str(mcp_server_path),
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )

        # 等待服务器启动
        await asyncio.sleep(5)

        # 为 MCP 服务器设置客户端
        client = MultiServerMCPClient(
            {
                "math": {
                    "transport": "sse",
                    "url": "http://localhost:8000/sse",
                },
            }
        )

        # 从 MCP 服务器获取工具
        tools = await client.get_tools()

        # 使用用户偏好配置 LLM
        llm = ChatOpenAI(model="deepseek-v3", temperature=0)
        # 创建一个 agent
        graph = create_agent(
            model=llm,
            tools=tools,
            system_prompt="你是一个可以使用工具的得力助手。 தயவுசெய்து கருவிகளைப் பயன்படுத்தவும்.",
        )

        print("正在调用 agent 回答一个数学问题...")
        inputs = {"messages": [{"role": "user", "content": "123 + 456 等于多少？"}]}
        async for chunk in graph.astream(inputs, stream_mode="updates"):
            print(chunk)

    finally:
        if server_process and server_process.returncode is None:
            print("\nTerminating server process...")
            server_process.terminate()
            await server_process.wait()
        elif server_process and server_process.returncode is not None:
            print(f"\nServer process exited with code: {server_process.returncode}")

        if server_process and server_process.stderr:
            stderr_output = await server_process.stderr.read()
            if stderr_output:
                print("\nServer stderr output:")
                print(stderr_output.decode())


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

示例 4：使用 Langchain 1.0 create_agent 的 MCP Stdio 工具集成

此示例演示了如何将 MCP Stdio 服务器提供的工具与 Langchain 1.0 的 create_agent 函数结合使用。它展示了如何通过 stdio 客户端连接到服务器、获取工具，然后使用新的 Agent 创建和调用模式来解决数学问题。

import asyncio
import os
from pathlib import Path

from langchain_openai import ChatOpenAI
from langchain.agents import create_agent
from langchain_mcp_adapters.tools import load_mcp_tools
from mcp.client.stdio import stdio_client
from mcp import ClientSession, StdioServerParameters

async def run_mcp_stdio_new_agent_example():
    # 定义 MCP 数学服务器的路径
    mcp_server_path = Path(__file__).parent / "mcp_math_server_stdio.py"

    # 为 MCP 服务器设置 stdio 客户端
    server_params = StdioServerParameters(
        command="python3",
        args=[str(mcp_server_path)],
    )

    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)

            # 使用用户偏好配置 LLM
            llm = ChatOpenAI(model="deepseek-v3", temperature=0)
            # 创建一个 agent
            graph = create_agent(
                model=llm,
                tools=tools,
                system_prompt="你是一个可以使用工具的得力助手。 தயவுசெய்து கருவிகளைப் பயன்படுத்தவும்.",
            )

            print("正在调用 agent 回答一个数学问题...")
            inputs = {"messages": [{"role": "user", "content": "123 + 456 等于多少？"}]}
            async for chunk in graph.astream(inputs, stream_mode="updates"):
                print(chunk)

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