【干货】LangChain与MCP完美结合：4种模式解锁AI Agent新能力，小白必学收藏！

荔枝学Python

1296人浏览 · 2025-12-12 13:58:00

荔枝学Python · 2025-12-12 13:58:00 发布

简介

本文详细介绍了如何将MCP技术与LangChain框架集成，通过4个实战示例展示了两种MCP通信模式（SSE和Stdio）与两种LangChain Agent创建方式（经典方式和1.0版本的create_agent函数）的组合应用。文章提供了完整的代码实现和部署指南，帮助开发者利用丰富的MCP服务扩展AI Agent功能，构建更强大、更灵活的应用程序。

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/abcsfsfdskgergerthntjrweeuidfu8324refbif3export OPENAI_API_BASE=https://qianfan.baidubce.com/v2

创建模型对象：

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

Agent 提示：ChatPromptTemplate 用于定义 Agent 的角色并构建对话。它包括一个系统消息、可选的聊天历史记录、人类输入以及 Agent 暂存区（用于规划其行动）的占位符。```plaintext
prompt = ChatPromptTemplate.from_messages( [ (“system”, “你是一个可以使用工具的得力助手。 தயவுசெய்து கருவிகளைப் பயன்படுத்தவும்.”), MessagesPlaceholder(“chat_history”, optional=True), (“human”, “{input}”), MessagesPlaceholder(“agent_scratchpad”), ])
Agent 创建和执行：create_tool_calling_agent 函数用于构建能够使用工具的 Agent ，AgentExecutor 用于运行 Agent 。```plaintext
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 asynciofrom mcp.server import FastMCP# 创建一个服务器实例server = FastMCP(name="math_server", log_level="ERROR")# 定义并注册 add 工具@server.tool()def add(a: int, b: int) -> int:    """将两个整数相加。"""    return a + basync def main():    await server.run_stdio_async()if __name__ == "__main__":    asyncio.run(main())

sse 服务

import asynciofrom typing import Annotatedfrom 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 + basync 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 脚本展示了如何：

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

关键组件和代码

import asyncioimport osfrom pathlib import Pathfrom langchain_openai import ChatOpenAIfrom langchain_classic.agents import AgentExecutor, create_tool_calling_agentfrom langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholderfrom langchain_mcp_adapters.tools import load_mcp_toolsfrom mcp.client.stdio import stdio_clientfrom mcp import ClientSession, StdioServerParametersasync 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 脚本展示了如何：

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

关键组件和代码演练

import asyncioimport osfrom pathlib import Pathimport sysfrom langchain_openai import ChatOpenAIfrom langchain_classic.agents import AgentExecutor, create_tool_calling_agentfrom langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholderfrom langchain_mcp_adapters.client import MultiServerMCPClientasync 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 asyncioimport osfrom pathlib import Pathimport sysfrom langchain_openai import ChatOpenAIfrom langchain.agents import create_agentfrom langchain_mcp_adapters.client import MultiServerMCPClientasync 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 asyncioimport osfrom pathlib import Pathfrom langchain_openai import ChatOpenAIfrom langchain.agents import create_agentfrom langchain_mcp_adapters.tools import load_mcp_toolsfrom mcp.client.stdio import stdio_clientfrom mcp import ClientSession, StdioServerParametersasync 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())

读者福利：如果大家对大模型感兴趣，这套大模型学习资料一定对你有用

对于0基础小白入门：