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 装饰器，直接写在项目代码里。这种方式很灵活，但随着工具变多，痛点会越来越明显：

1. 复用难：每个项目都要重复写一遍天气查询、计算工具
2. 接入乱：想接第三方服务（查火车票、读数据库、操作 GitHub），每个都要单独写适配代码，接口格式千奇百怪
3. 跨语言难：别的团队用 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 里，让大模型自主决定什么时候调用这些工具。

核心知识点

1. MultiServerMCPClient 可以同时接入多个 MCP Server
2. 接入后的工具和本地 @tool 格式完全一致，对 LLM 透明
3. 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 报错	端口被占用	更换端口或关闭占用程序

---

九、下一步学习规划

到这里，Agent 的四大核心能力（工具调用、记忆、中间件、MCP 扩展）就全部学完了。接下来的学习方向：

1. 综合实战：把所有知识点整合，做一个带记忆、带审核、带 MCP 工具的完整智能客服
2. 最佳实践：工具设计原则、提示词优化、调试技巧
3. LangGraph 进阶：更复杂的工作流编排、多 Agent 协作

后续我会继续更新实战系列，一步步从入门到完整项目。

如果这篇文章对你有帮助，欢迎点赞收藏~ 有问题可以在评论区留言，我们一起交流进步！