LangChain Agent MCP 协议实战：工具即插即用

风吹夏回

227人浏览 · 2026-07-13 10:52:11

风吹夏回 · 2026-07-13 10:52:11 发布

LangChain Agent 进阶实战：MCP 协议从零到一，手把手教你实现工具即插即用

作者：风吹夏回
承接上一篇 Agent 入门教程，本篇带你吃透 MCP 协议，让你的 Agent 能力轻松扩展

写在前面

接着上一篇 Agent 入门的进度，今天我们正式学习 MCP（模型上下文协议）。最开始我以为这是个很复杂的底层协议，真的动手跑下来才发现，它的核心写法和我们之前用的 @tool 几乎一模一样，但却能解决「工具复用难、第三方接入乱」的大问题。

这篇文章依然以初学者视角记录完整实战过程，从概念理解到本地 Server 编写，再到接入 LangChain Agent，全程附可直接运行的代码，跟着做就能跑通。

本次学习技术栈：

大模型：火山方舟豆包 doubao-seed-2.0-lite
框架：LangChain + LangGraph + MCP SDK
运行模式：先 Stdio 本地模式，再 Streamable HTTP 远程模式

一、先搞懂：MCP 到底是来解决什么问题的？

在上一篇的学习中，我们定义工具的方式是 @tool 装饰器，直接写在项目代码里。这种方式很灵活，但随着工具变多，痛点会越来越明显：

复用难：每个项目都要重复写一遍天气查询、计算工具
接入乱：想接第三方服务（查火车票、读数据库、操作 GitHub），每个都要单独写适配代码，接口格式千奇百怪
跨语言难：别的团队用 Go 写的工具，我们 Python 项目没法直接用

MCP（Model Context Protocol）就是 AI 领域的「USB-C 标准」：

所有工具服务都按照同一套协议暴露能力
所有 AI 应用都按照同一套方式接入
不管工具是什么语言、部署在哪里，接入代码完全一致

一句话总结：MCP 统一了大模型和外部工具的通信标准，让工具可以「即插即用」。

MCP 核心三角色

角色	职责	对应我们的场景
MCP Host	AI 应用宿主	我们的 LangChain Agent 程序
MCP Client	通信客户端	LangChain 官方提供的 MCP 适配器
MCP Server	工具服务提供者	独立运行的工具服务（自己写/第三方）

两种主流传输方式

初学者建议先从 Stdio 入手，最轻量无依赖：

方式	通信原理	适用场景
Stdio	通过进程标准输入输出通信	本地开发调试，工具和程序在同一台机器
Streamable HTTP	HTTP 流式传输	生产环境，工具部署在远程服务器

二、环境准备

先安装核心依赖，建议在之前的虚拟环境里直接安装：

pip install mcp langchain-mcp-adapters

三、第一步：手写一个本地 MCP Server

我们先从最简单的 Stdio 模式开始，写一个本地工具服务。你会发现写法和 @tool 几乎没有区别。

新建文件 mcp_server.py：

# mcp_server.py
from mcp.server.fastmcp import FastMCP

# 创建 MCP Server 实例，自定义服务名称
mcp = FastMCP("我的本地工具集")


@mcp.tool()
def add(a: int, b: int) -> int:
    """计算两个整数的和
    Args:
        a: 第一个整数
        b: 第二个整数
    """
    return a + b


@mcp.tool()
def multiply(a: int, b: int) -> int:
    """计算两个整数的乘积
    Args:
        a: 第一个整数
        b: 第二个整数
    """
    return a * b


@mcp.tool()
def get_weather(city: str) -> str:
    """查询指定城市的实时天气
    Args:
        city: 城市名称，如北京、上海
    """
    return f"{city}今日天气：晴转多云，24-31℃，东南风2级"


if __name__ == "__main__":
    # 以 Stdio 模式启动服务
    mcp.run(transport="stdio")

💡 初学者观察点：除了装饰器从 @tool 变成 @mcp.tool()，函数写法、docstring 规范和之前完全一致。这就是标准化的好处，学习成本极低。

四、第二步：先测试 Server 通不通

初学者必做：不要直接接入 Agent，先单独测 Server！
很多人一上来就接 Agent，出了问题不知道是 Server 写错了还是 Agent 配置错了。先写一个纯客户端测试，确认服务没问题。

新建 test_mcp_client.py：

# test_mcp_client.py
import asyncio
import sys
import os
from mcp.client.stdio import stdio_client
from mcp import ClientSession, StdioServerParameters


async def main():
    # 获取 server 脚本的绝对路径（避免路径找不到的坑）
    current_dir = os.path.dirname(os.path.abspath(__file__))
    server_path = os.path.join(current_dir, "mcp_server.py")

    # 配置 Server 启动参数
    server_params = StdioServerParameters(
        command=sys.executable,  # 用当前虚拟环境的 Python 解释器
        args=[server_path],
    )

    # 建立连接
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # 第1步：握手初始化
            await session.initialize()
            print("✅ MCP Server 连接成功")

            # 第2步：获取全部工具列表
            tools = await session.list_tools()
            print(f"\n📦 可用工具列表：")
            for tool in tools.tools:
                print(f"  - {tool.name}：{tool.description}")

            # 第3步：调用工具测试
            print("\n🔧 测试调用 add 工具：10 + 20")
            result = await session.call_tool("add", {"a": 10, "b": 20})
            print(f"   返回结果：{result.content}")

            print("\n🔧 测试调用 get_weather 工具：长沙")
            result2 = await session.call_tool("get_weather", {"city": "长沙"})
            print(f"   返回结果：{result2.content}")


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

运行这个脚本，如果能正常打印工具列表和调用结果，说明你的 MCP Server 已经完全跑通了。

⚠️ 踩坑提醒：Server 脚本路径一定要写对，建议用 os.path.abspath 计算绝对路径，否则相对路径很容易找不到文件。

五、第三步：接入 LangChain Agent

Server 没问题之后，我们把它接入到熟悉的 Agent 里，让大模型自主决定什么时候调用这些工具。

核心知识点

MultiServerMCPClient 可以同时接入多个 MCP Server
接入后的工具和本地 @tool 格式完全一致，对 LLM 透明
MCP 客户端是异步的，所以 Agent 调用要用 ainvoke

新建 agent_with_mcp.py，完整可运行代码：

# agent_with_mcp.py
import asyncio
import os
import sys
from dotenv import load_dotenv, find_dotenv
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
from langchain_mcp_adapters.client import MultiServerMCPClient

# 加载环境变量
dotenv_path = find_dotenv()
load_dotenv(dotenv_path, override=True)

api_key = os.getenv("DEEPSEEK_API_KEY")
base_url = os.getenv("DEEPSEEK_BASE_URL")

# 初始化大模型
llm = ChatOpenAI(
    model="doubao-seed-2.0-lite",
    api_key=api_key,
    base_url=base_url,
    temperature=0.7,
)


# 本地工具：演示和MCP工具混合使用
@tool
def say_hello(name: str) -> str:
    """向指定的人打招呼
    Args:
        name: 对方的名字
    """
    return f"你好呀，{name}！欢迎体验 MCP 工具扩展能力"


async def main():
    # 1. 配置 MCP Server 连接（支持同时接入多个）
    current_dir = os.path.dirname(os.path.abspath(__file__))
    server_path = os.path.join(current_dir, "mcp_server.py")

    client = MultiServerMCPClient({
        "my-local-tools": {
            "transport": "stdio",
            "command": sys.executable,
            "args": [server_path],
        }
        # 后续可以继续追加更多 Server
        # "remote-server": {
        #     "transport": "streamable-http",
        #     "url": "http://xxx/mcp"
        # }
    })

    # 2. 获取所有 MCP 工具
    mcp_tools = await client.get_tools()
    print(f"✅ 成功加载 {len(mcp_tools)} 个 MCP 工具")

    # 3. 合并本地工具 + MCP 工具
    all_tools = [say_hello] + mcp_tools

    # 4. 配置记忆（延续上一篇的知识点）
    checkpointer = InMemorySaver()
    config = {"configurable": {"thread_id": "mcp_demo_001"}}

    # 5. 创建 Agent
    agent = create_agent(
        model=llm,
        tools=all_tools,
        system_prompt="你是一个智能助手，可以使用计算工具和天气工具来帮助用户。请根据问题选择合适的工具，不需要工具时直接回答。",
        checkpointer=checkpointer,
    )

    # 6. 测试调用
    print("\n" + "=" * 60)
    print("测试1：计算 123 + 456 等于多少？")
    print("=" * 60)
    result = await agent.ainvoke(
        {"messages": [{"role": "user", "content": "计算 123 + 456 等于多少？"}]},
        config=config
    )
    print("Agent 回复：", result["messages"][-1].content)

    print("\n" + "=" * 60)
    print("测试2：武汉今天天气怎么样？")
    print("=" * 60)
    result2 = await agent.ainvoke(
        {"messages": [{"role": "user", "content": "武汉今天天气怎么样？"}]},
        config=config
    )
    print("Agent 回复：", result2["messages"][-1].content)

    print("\n" + "=" * 60)
    print("测试3：和小明打个招呼")
    print("=" * 60)
    result3 = await agent.ainvoke(
        {"messages": [{"role": "user", "content": "和小明打个招呼"}]},
        config=config
    )
    print("Agent 回复：", result3["messages"][-1].content)

    # 关闭客户端
    await client.close()


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

运行效果

Agent 会自主识别问题类型：

数学计算 → 调用 MCP 的 add 工具
天气查询 → 调用 MCP 的 get_weather 工具
打招呼 → 调用本地 say_hello 工具

整个过程中，大模型完全感知不到哪些是本地工具、哪些是 MCP 远程工具，统一决策、统一调用。

⚠️ 初学者必踩坑：MCP 客户端是异步的，必须用 agent.ainvoke()，直接用同步的 invoke() 会报错。

六、进阶：HTTP 模式 MCP Server

当工具需要部署在远程服务器、给多个 Agent 复用时，就用 HTTP 模式。Server 端代码几乎不用改，只改启动参数。

HTTP 版 Server 代码

新建 mcp_server_http.py：

# mcp_server_http.py
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("远程计算工具服务")

@mcp.tool()
def add(a: int, b: int) -> int:
    """计算两个整数的和"""
    return a + b

@mcp.tool()
def multiply(a: int, b: int) -> int:
    """计算两个整数的乘积"""
    return a * b

if __name__ == "__main__":
    # HTTP 模式启动，默认监听 127.0.0.1:8000
    mcp.run(transport="streamable-http")

运行后，服务地址为 http://127.0.0.1:8000/mcp。

Agent 接入 HTTP Server

只需要修改 MultiServerMCPClient 的配置，其他代码完全不动：

client = MultiServerMCPClient({
    "remote-calc-tools": {
        "transport": "streamable-http",
        "url": "http://127.0.0.1:8000/mcp"
    }
})

这就是标准化协议的威力：工具部署方式变了，业务代码一行都不用改。

七、选型建议：本地工具 vs MCP 工具

初学者不用什么都往 MCP 里塞，按场景选择：

对比维度	本地工具（@tool）	MCP 工具
定义位置	写在项目代码内	独立 Server 运行
适用场景	业务逻辑简单、项目专属	通用能力、跨项目复用、第三方服务
维护方式	和主项目一起维护	独立部署、独立迭代
上手难度	极低，写函数即可	中等，需要启动服务
生态复用	仅自身项目可用	所有支持 MCP 的应用都能接入

初学者学习建议：

项目早期、逻辑简单，直接用 @tool 最快
当工具需要被多个项目复用，或者想接入社区现成 MCP 服务时，再上 MCP
两者可以混合使用，同一个 Agent 同时挂载本地工具和 MCP 工具

八、今日踩坑总结

问题	原因	解决方案
Server 连接失败	相对路径写错，找不到脚本文件	用 `os.path.abspath` 计算绝对路径
接入 Agent 后报错	用了同步 `invoke` 调用异步客户端	改用 `ainvoke`，外层包 `asyncio.run`
工具不被调用	工具描述写得太模糊	完善 docstring，写清楚用途和参数格式
启动 HTTP Server 报错	端口被占用	更换端口或关闭占用程序