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

简介

本文详细介绍了如何将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基础小白入门：