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())

想入门 AI 大模型却找不到清晰方向？备考大厂 AI 岗还在四处搜集零散资料？别再浪费时间啦！2025 年 AI 大模型全套学习资料已整理完毕，从学习路线到面试真题，从工具教程到行业报告，一站式覆盖你的所有需求，现在全部免费分享！

👇👇扫码免费领取全部内容👇👇

​

一、学习必备：100+本大模型电子书+26 份行业报告 + 600+ 套技术PPT，帮你看透 AI 趋势

想了解大模型的行业动态、商业落地案例？大模型电子书？这份资料帮你站在 “行业高度” 学 AI：

1. 100+本大模型方向电子书

2. 26 份行业研究报告：覆盖多领域实践与趋势

报告包含阿里、DeepSeek 等权威机构发布的核心内容，涵盖：

• 职业趋势：《AI + 职业趋势报告》《中国 AI 人才粮仓模型解析》；
• 商业落地：《生成式 AI 商业落地白皮书》《AI Agent 应用落地技术白皮书》；
• 领域细分：《AGI 在金融领域的应用报告》《AI GC 实践案例集》；
• 行业监测：《2024 年中国大模型季度监测报告》《2025 年中国技术市场发展趋势》。

3. 600+套技术大会 PPT：听行业大咖讲实战

PPT 整理自 2024-2025 年热门技术大会，包含百度、腾讯、字节等企业的一线实践：

• 安全方向：《端侧大模型的安全建设》《大模型驱动安全升级（腾讯代码安全实践）》；
• 产品与创新：《大模型产品如何创新与创收》《AI 时代的新范式：构建 AI 产品》；
• 多模态与 Agent：《Step-Video 开源模型（视频生成进展）》《Agentic RAG 的现在与未来》；
• 工程落地：《从原型到生产：AgentOps 加速字节 AI 应用落地》《智能代码助手 CodeFuse 的架构设计》。

二、求职必看：大厂 AI 岗面试 “弹药库”，300 + 真题 + 107 道面经直接抱走

想冲字节、腾讯、阿里、蔚来等大厂 AI 岗？这份面试资料帮你提前 “押题”，拒绝临场慌！

1. 107 道大厂面经：覆盖 Prompt、RAG、大模型应用工程师等热门岗位

面经整理自 2021-2025 年真实面试场景，包含 TPlink、字节、腾讯、蔚来、虾皮、中兴、科大讯飞、京东等企业的高频考题，每道题都附带思路解析：

2. 102 道 AI 大模型真题：直击大模型核心考点

针对大模型专属考题，从概念到实践全面覆盖，帮你理清底层逻辑：

3. 97 道 LLMs 真题：聚焦大型语言模型高频问题

专门拆解 LLMs 的核心痛点与解决方案，比如让很多人头疼的 “复读机问题”：

三、路线必明： AI 大模型学习路线图，1 张图理清核心内容

刚接触 AI 大模型，不知道该从哪学起？这份「AI大模型 学习路线图」直接帮你划重点，不用再盲目摸索！

路线图涵盖 5 大核心板块，从基础到进阶层层递进：一步步带你从入门到进阶，从理论到实战。

L1阶段:启航篇丨极速破界AI新时代

L1阶段：了解大模型的基础知识，以及大模型在各个行业的应用和分析，学习理解大模型的核心原理、关键技术以及大模型应用场景。

L2阶段：攻坚篇丨RAG开发实战工坊

L2阶段：AI大模型RAG应用开发工程，主要学习RAG检索增强生成：包括Naive RAG、Advanced-RAG以及RAG性能评估，还有GraphRAG在内的多个RAG热门项目的分析。

L3阶段：跃迁篇丨Agent智能体架构设计

L3阶段：大模型Agent应用架构进阶实现，主要学习LangChain、 LIamaIndex框架，也会学习到AutoGPT、 MetaGPT等多Agent系统，打造Agent智能体。

L4阶段：精进篇丨模型微调与私有化部署

L4阶段：大模型的微调和私有化部署，更加深入的探讨Transformer架构，学习大模型的微调技术，利用DeepSpeed、Lamam Factory等工具快速进行模型微调，并通过Ollama、vLLM等推理部署框架，实现模型的快速部署。

L5阶段：专题集丨特训篇 【录播课】

四、资料领取：全套内容免费抱走，学 AI 不用再找第二份

不管你是 0 基础想入门 AI 大模型，还是有基础想冲刺大厂、了解行业趋势，这份资料都能满足你！
现在只需按照提示操作，就能免费领取：

👇👇扫码免费领取全部内容👇👇

​

2025 年想抓住 AI 大模型的风口？别犹豫，这份免费资料就是你的 “起跑线”！