1. 本章目标 #
第 8 章把工具定义在你自己的进程里：一个 @tool 函数，和 Agent 跑在同一份 Python 代码中。这在「查订单、做计算、读配置」时最合适。
真实项目里还有另一类工具：
- 工具已经由别的团队写成了独立服务，不能把源码贴进你的 Agent
- 同一套能力要同时给 Cursor、Claude Desktop、你的 LangChain Agent 用
- 工具需要单独部署、单独鉴权，不能和模型进程绑死
这类场景的标准接法是 MCP（Model Context Protocol）。它是一套开放协议：服务器用统一方式把工具、资源和提示词暴露给 LLM 应用。LangChain 这边用 langchain-mcp-adapters 把 MCP 工具转成第 8 章那种 StructuredTool，再照常交给 create_agent。
本章目标：
自己写一台 MCP 服务器，用适配器拉成 LangChain 工具，接到 Agent 上。 顺便分清几条容易卡住的边界：stdio / HTTP、无状态 / 有状态、错误回灌。
学完你应能：
- 用 FastMCP 写出带工具 / 资源 / 提示词的本地服务器
- 用
MultiServerMCPClient.get_tools()把 MCP 工具转成 LangChain 工具并本地ainvoke单测 - 把工具列表交给
create_agent，读懂模型 ↔ MCP 的消息轨迹 - 在 stdio 和 HTTP 两种传输之间做选择
- 说清默认「每次调用新开会话」和
client.session()的差别 - 处理工具执行失败（默认回灌，而不是把 Agent 打崩）
参考文档：
建议阅读顺序： §2 建立心智模型 → §4～§6 亲手跑通「写服务器 → 拉工具 → 接 Agent」（本章最该动手的一段）→ §8、§9 弄清无状态默认和错误回灌 → §7 / §10～§13 按需。踩坑清单 §14 可随时翻。
先预告几条和直觉相反的实测结论（读到对应小节会有完整证据）：
| 你可能以为 | 实际情况 | 见 |
|---|---|---|
MCP 工具和第 8 章一样能 tool.invoke(...) |
只有异步实现，invoke 直接 NotImplementedError；Agent 也必须 ainvoke |
§5.2 |
async with MultiServerMCPClient(...) 会帮你管理连接 |
0.1.0 起就禁止当上下文管理器，会 NotImplementedError |
§8.3 |
command="python" 最省事 |
子进程可能用到系统 Python，找不到 mcp。要用 sys.executable |
§5.1 |
每次 ainvoke 都复用同一个服务器进程 |
默认无状态：每次调用新建 MCP 会话；stdio 下就是再拉起一个子进程 | §8 |
工具里 raise ValueError 会把 Agent 打崩 |
默认 handle_tool_errors=True，错误文本回给模型 |
§9 |
返回值就是 Python 的 8 |
适配器给出的是 content block 列表 [{'type': 'text', 'text': '8', ...}] |
§5.2 |
2. MCP 是什么 #
2.1. 一句话 #
MCP 是 LLM 应用和外部能力之间的 USB 协议。
U 盘不管插到哪台电脑，文件系统都长一个样。MCP 也一样：服务器用同一种方式声明「我有哪些工具、怎么调」，Cursor、Claude Desktop、LangChain Agent 都能连——不必为每个宿主各写一套插件。
它标准化的是三件事：
- 有什么：工具、资源、提示词模板
- 怎么说话：stdio（子进程管道）或 HTTP
- 一次调用长什么样：名字 + JSON 参数 → 文本 / 结构化结果 / 错误
LangChain 没有重新发明 MCP。适配器只做翻译：MCP 工具 → 第 8 章的 LangChain 工具。翻完之后，选型、填参、回灌 ToolMessage，仍是 create_agent 的活。
2.2. 和 @tool 怎么分工 #
第 8 章 @tool |
本章 MCP | |
|---|---|---|
| 代码在哪 | 和 Agent 同一进程 | 独立进程 / 独立服务 |
| 谁执行 | 你的函数直接跑 | 适配器发协议请求，服务器跑 |
| 适合 | 业务函数、无独立部署需求 | 要复用、要隔离、已经是独立服务 |
| 给模型看的 | 工具名 + 描述 + schema | 同样三样，翻译之后模型分不出来源 |
判断口诀：
工具只服务这一个 Agent，就
@tool。 工具要给多个宿主用，或必须隔离部署，再上 MCP。
能 @tool 解决的，不要为了「用了 MCP」而用 MCP。官方在 LangGraph API 场景里也写过：stdio 本来就是给「用户机器上的本地应用」用的。Web 服务里，先问自己是不是一个普通函数就够了。
2.3. 服务器能提供的三样东西 #
MCP 服务器不只是「远程函数」。协议里有三类能力；入门先分清：谁给模型调，谁给程序读。
| 类型 | 干什么 | LangChain 侧变成什么 | 谁发起 |
|---|---|---|---|
| Tools | 可执行动作：查库、运算、调 API | StructuredTool，可交给 Agent |
模型决定调不调 |
| Resources | 只读数据：文件、配置、文档 | Blob（文本或二进制） |
你的代码读，不是模型自己读 |
| Prompts | 预置提示词模板 | HumanMessage / AIMessage 列表 |
你的代码取，再塞进对话 |
常见误解：把 Resource 当成工具。资源没有「参数 + 执行」这套语义，Agent 默认也看不到它们。要让模型按需取数，仍然得包成 Tool。
本章 §4～§6、§9、§11～§13 都围着 Tools 转——接到 Agent 的主路径在这里。Resources / Prompts 在 §10 各看一个最小例子，知道入口即可。
2.4. 两种传输 #
客户端和服务器之间怎么传字节，叫 transport（传输）。入门记住两种就够：
stdio（本地子进程）
客户端 fork 出服务器进程
双方用 stdin / stdout 交换 JSON-RPC
适合：本机脚本、桌面应用、教程
HTTP（streamable-http）
服务器已经在某个 URL 上听着
客户端发 HTTP 请求
适合：远程服务、多客户端共享、要带鉴权头langchain-mcp-adapters 里 "http"、"streamable_http"、"streamable-http" 是同一种传输的别名。更老的 sse 已被 MCP 规范弃用，新代码不要用。
stdio 和 HTTP 有个硬差别：服务器进程的寿命绑在这条连接上。即便如此，MultiServerMCPClient 默认仍是「每次工具调用新建会话」——stdio 下就是每次再拉起一个子进程。§8 专门拆这件事。
3. 安装与环境 #
在第 8、9 章的依赖之外，本章只多装一个适配器。它会把官方 mcp SDK 一并拉下来；FastMCP 就在这个 SDK 里，不必再装 fastmcp 包。
pip install langchain-mcp-adapters装好后应能 import 这几样：
# 验证三样关键入口都能导入，缺哪个就回头检查是否装到了当前 venv
from mcp.server.fastmcp import FastMCP
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.agents import create_agent
print("ok")输出：
ok跑 Agent 示例前，项目根目录 .env 里要有 DEEPSEEK_API_KEY（和第 2 章相同）。只做「拉工具 + 本地 ainvoke」的小节不需要密钥。
Windows 特别注意： 后面所有 stdio 示例的 command 都写成 sys.executable（当前解释器的绝对路径），不要写 "python"。
MCP 拉起子进程时只继承 PATH 等少量环境变量——"python" 可能指到系统安装，而不是 .venv。子进程里 import mcp 会失败，报错很长，却很难想到是「用错了 Python」。
4. 写一个最小 MCP 服务器 #
先把服务器写出来。没有服务器，客户端连空气。
4.1. math_server.py #
下面这份文件就是一台完整的 MCP 服务器：三个工具、一份资源、一个提示词模板。保存后，后面所有客户端都假定它和自己在同一目录里。
# 过滤 FastMCP 启动时 pydantic_settings 的无关告警，避免干扰演示输出
import warnings
# 指定只忽略 pydantic_settings 这个模块发出的告警
warnings.filterwarnings("ignore", module="pydantic_settings")
# FastMCP 是 MCP 官方 Python SDK 提供的「快速写服务器」入口
from mcp.server.fastmcp import FastMCP
# 建一个名叫 Math 的 MCP 服务器；log_level 调到 ERROR，避免把内部请求日志打到终端
mcp = FastMCP("Math", log_level="ERROR")
# @mcp.tool() 把函数注册成 MCP 工具，模型以后看到的就是函数名 + docstring + 参数类型
@mcp.tool()
# a、b 都必须是整数；返回值也是整数
def add(a: int, b: int) -> int:
# 这段 docstring 会变成工具描述，模型靠它决定「什么时候该调 add」
"""把两个整数相加。"""
# 真正的业务逻辑：普通 Python 加法
return a + b
# 再注册一个乘法工具，演示同一台服务器可以挂多个工具
@mcp.tool()
def multiply(a: int, b: int) -> int:
"""把两个整数相乘。"""
return a * b
# 除法工具专门用来演示「工具内部抛错」时适配器怎么处理
@mcp.tool()
def divide(a: int, b: int) -> float:
"""整数相除。b 为 0 时会报错。"""
# 主动检查非法输入，抛出的异常会被 MCP 标成 isError=True
if b == 0:
raise ValueError("除数不能为 0")
return a / b
# 资源不是给模型「调用」的，而是给客户端「读取」的只读数据
@mcp.resource("resource://handbook")
def handbook() -> str:
"""一份给客户端读的制度摘要。"""
return "差旅报销需在返程 7 日内提交。"
# Prompt 是服务器上预置的提示词模板，客户端按名字取，再塞进对话
@mcp.prompt()
def summarize(topic: str) -> str:
"""生成一段要求模型做摘要的用户消息。"""
return f"请用三句话总结「{topic}」。"
# 被客户端用 stdio 拉起时，从这里进入事件循环，通过标准输入/输出说话
if __name__ == "__main__":
# transport="stdio" 表示：我是子进程，父进程用 stdin/stdout 跟我通信
mcp.run(transport="stdio")不要手动 python math_server.py 来「启动」它。stdio 服务器是被客户端拉起的：客户端指定 command + args，子进程从 stdin 读协议消息。你在终端里单独跑它，它只会挂着等标准输入，看起来像卡死。
4.2. 这段代码在干什么 #
对照第 8 章，注册工具的方式几乎一样，只是装饰器换成了：
| 第 8 章 | 本章服务器 |
|---|---|
@tool |
`@mcp.tool()` |
| 函数名 → 工具名 | 相同 |
| docstring → 描述 | 相同 |
| 类型注解 → 参数 schema | 相同 |
`@mcp.tool()的括号不能省，写成@mcp.tool会立刻TypeError`。
divide 在 b == 0 时 raise ValueError，不是 return "错误：..."。对 MCP 来说这是一次工具执行失败（协议里的 isError=True）。适配器默认会把它翻成错误文本送回模型；§9 再对比两种开关。
`@mcp.resource和@mcp.prompt先登记在服务器上，§10 再用客户端 API 去读。它们不会自动出现在get_tools()` 的列表里。
5. 拉工具、先单测 #
和第 8 章同一条纪律：先本地调用工具，再交给模型。 否则「服务器没起来」和「模型选错工具」会糊成同一个症状。
5.1. load_tools.py #
把 math_server.py 和本文件放在同一目录后直接运行。流程是：拉起服务器子进程 → 列出工具 → 不经过模型，本地算 3+5 和 8×12。
# 标准库：跑异步主函数
import asyncio
# 标准库：当前 Python 解释器的绝对路径（不要写死 "python"）
import sys
# 标准库：拼服务器脚本的路径
from pathlib import Path
# 把多个 MCP 服务器收成一套 LangChain 工具的客户端
from langchain_mcp_adapters.client import MultiServerMCPClient
async def main() -> None:
# 服务器脚本和本文件放在同一目录；resolve() 得到绝对路径，stdio 子进程要求绝对路径更稳
server = Path(__file__).resolve().parent / "math_server.py"
# 找不到就直接退出，避免连上一个不存在的文件后报一长串底层错误
if not server.exists():
raise SystemExit("请先把 math_server.py 保存到本文件同一目录")
# connections 的键是你给这台服务器起的名字，后面 session("math") 用的就是它
client = MultiServerMCPClient(
{
"math": {
# stdio：客户端拉起一个子进程，用标准输入/输出对话
"transport": "stdio",
# 必须用当前 venv 的解释器，否则子进程可能找不到 mcp 包
"command": sys.executable,
"args": [str(server)],
}
}
)
# get_tools 会先握手、再列出工具，然后把每个 MCP 工具转成 LangChain StructuredTool
tools = await client.get_tools()
# 打印工具名，确认服务器上的三个函数都过来了
print("工具名:", [t.name for t in tools])
# 类型一律是 StructuredTool，和第 8 章 @tool 得到的是同一类东西
print("类型:", type(tools[0]).__name__)
# 按名字取出加法工具，后面单测用
add = next(t for t in tools if t.name == "add")
# MCP 转出来的工具只有异步实现，必须 ainvoke，invoke 会 NotImplementedError
add_result = await add.ainvoke({"a": 3, "b": 5})
print("add(3, 5) =", add_result)
# 乘法同理，参数仍然是字典，键必须和服务器函数参数名一致
multiply = next(t for t in tools if t.name == "multiply")
mul_result = await multiply.ainvoke({"a": 8, "b": 12})
print("multiply(8, 12) =", mul_result)
# Windows / 脚本入口：用 asyncio.run 驱动整个异步流程
if __name__ == "__main__":
asyncio.run(main())输出（id 每次运行都会变，看 text 字段）：
工具名: ['add', 'multiply', 'divide']
类型: StructuredTool
add(3, 5) = [{'type': 'text', 'text': '8', 'id': 'lc_19ac4c93-fd22-4e31-9b10-d7fa48e673b2'}]
multiply(8, 12) = [{'type': 'text', 'text': '96', 'id': 'lc_c2c12238-f4d7-4c8d-b8db-3c0b00a78a16'}]三个要点：
- 工具名就是服务器上的函数名，三个都过来了。
- 类型是
StructuredTool，所以后面create_agent(tools=tools)和第 8 章是同一套 API。 - 返回值不是整数
8，而是 LangChain 的标准 content block。模型看到的是文本"8"；自己写断言时取result[0]["text"]，不要assert result == 8。
MultiServerMCPClient 的构造函数只是记下连接配置，这一步还不会拉起子进程。真正握手发生在 await client.get_tools()。之后每一次 ainvoke，默认还会再开一条新会话（stdio 下就是再启动一个服务器进程）。所以这个短脚本其实起停了多次 math_server.py——这是故意的。§8 再对比「复用同一条连接」。
5.2. 为什么必须 ainvoke #
适配器把 MCP 调用做成了 StructuredTool 的 coroutine，没有同步实现。实测如下：
add.invoke({"a": 3, "b": 5})
# NotImplementedError: StructuredTool does not support sync invocation.同样，agent.invoke(...) 执行到工具那一步也会踩上这个错误。本章所有调用都走 asyncio.run + ainvoke。
参数仍然和第 8 章一样：传字典，键 = 服务器函数的参数名。
5.3. 模型到底收到了什么：schema.py #
翻完之后，模型看到的 JSON 和第 8 章本地工具是同一套结构。用 convert_to_openai_tool 看原文：
# 标准库：把工具 schema 格式化成可读 JSON
import json
# 标准库：异步入口
import asyncio
# 标准库：当前解释器路径
import sys
# 标准库：定位服务器脚本
from pathlib import Path
# 多服务器 MCP 客户端
from langchain_mcp_adapters.client import MultiServerMCPClient
# 和第 8 章相同：看模型真正收到的 JSON
from langchain_core.utils.function_calling import convert_to_openai_tool
async def main() -> None:
# 服务器脚本必须是绝对路径
server = Path(__file__).resolve().parent / "math_server.py"
# 缺文件就明确提示，不要让底层 stdio 报错
if not server.exists():
raise SystemExit("请先把 math_server.py 保存到本文件同一目录")
# 只连一台 stdio 服务器
client = MultiServerMCPClient(
{
"math": {
"transport": "stdio",
"command": sys.executable,
"args": [str(server)],
}
}
)
# 拉工具
tools = await client.get_tools()
# 取出 add，专门看它的说明书
add = next(t for t in tools if t.name == "add")
# LangChain 简化视图：参数名、类型，通常没有 description
print("args:", add.args)
# 发给模型 API 的原文，结构和第 8 章本地 @tool 一样
schema = convert_to_openai_tool(add)
print(json.dumps(schema, ensure_ascii=False, indent=2))
if __name__ == "__main__":
asyncio.run(main())输出：
args: {'a': {'title': 'A', 'type': 'integer'}, 'b': {'title': 'B', 'type': 'integer'}}
{
"type": "function",
"function": {
"name": "add",
"description": "把两个整数相加。",
"parameters": {
"properties": {
"a": {
"type": "integer"
},
"b": {
"type": "integer"
}
},
"required": [
"a",
"b"
],
"type": "object"
}
}
}和第 8 章 §3.3 同一个坑：参数 a、b 没有 description，模型只知道它们是整数。工具描述来自 docstring「把两个整数相加。」——所以服务器函数的 docstring 仍然要写清「何时用」，不能只写「做什么」。
想给每个参数加说明，在服务器侧用带 Field(description=...) 的类型注解（Pydantic）；不要指望 docstring 里的 Args: 段自动拆开。
6. 接到 create_agent #
单测通过后，工具列表可以直接交给第 2、9 章的 create_agent。模型不知道、也不需要知道这些工具来自 MCP。
6.1. agent.py #
问一句「3 加 5 再乘 12」，看它会不会先 add 再 multiply——而不是心算。
# 标准库：异步入口
import asyncio
# 标准库：当前解释器
import sys
# 标准库：定位服务器脚本
from pathlib import Path
# 创建 Agent 的官方入口，和第 2 / 9 章相同
from langchain.agents import create_agent
# 从 .env 读取 DEEPSEEK_API_KEY
from dotenv import load_dotenv
# MCP 适配器客户端
from langchain_mcp_adapters.client import MultiServerMCPClient
# 覆盖已有环境变量，保证用的是项目根目录 .env 里的密钥
load_dotenv(override=True)
async def main() -> None:
# 服务器脚本绝对路径
server = Path(__file__).resolve().parent / "math_server.py"
if not server.exists():
raise SystemExit("请先把 math_server.py 保存到本文件同一目录")
# 连上本地数学服务器
client = MultiServerMCPClient(
{
"math": {
"transport": "stdio",
"command": sys.executable,
"args": [str(server)],
}
}
)
# 拉到的 tools 可以直接塞给 create_agent，用法和第 8 章完全一样
tools = await client.get_tools()
# 模型字符串和前几章保持一致，方便对照
agent = create_agent(
"deepseek:deepseek-v4-flash",
tools,
# 明确要求必须调工具，避免模型心算跳过 MCP
system_prompt="你是简洁的中文助手。算术必须调用工具，不要心算。",
)
# 注意：MCP 工具不支持同步 invoke，Agent 也必须 ainvoke
result = await agent.ainvoke(
{
"messages": [
{
"role": "user",
"content": "3 加 5 再乘 12 等于多少？",
}
]
}
)
# 把完整轨迹打出来：哪一步调了哪个工具、参数是什么
for i, message in enumerate(result["messages"]):
# 消息类型名：HumanMessage / AIMessage / ToolMessage
kind = type(message).__name__
# 若模型发起了工具调用，把名字和参数附在后面
calls = ""
if getattr(message, "tool_calls", None):
calls = " " + str([(c["name"], c["args"]) for c in message.tool_calls])
# 正文可能是 str，也可能是 MCP 转过来的 content block 列表
content = message.content
print(f"[{i}] {kind}{calls}: {content!r}")
if __name__ == "__main__":
asyncio.run(main())输出（最终一句的措辞每次可能不同，轨迹结构应对齐）：
[0] HumanMessage: '3 加 5 再乘 12 等于多少？'
[1] AIMessage [('add', {'a': 3, 'b': 5})]: ''
[2] ToolMessage: [{'type': 'text', 'text': '8', 'id': '...'}]
[3] AIMessage [('multiply', {'a': 8, 'b': 12})]: ''
[4] ToolMessage: [{'type': 'text', 'text': '96', 'id': '...'}]
[5] AIMessage: '(3 + 5) × 12 = 96。'6.2. 读这条轨迹 #
- 模型没有把
3+5×12一次算完，而是按「先加后乘」拆成两次工具调用——这正是我们要的：算术走 MCP，不走训练记忆。 - 第二次
multiply的a是8，来自第一次工具结果，不是用户原文里的3。说明ToolMessage已经回灌成功。 ToolMessage.content仍是 content block 列表。模型能读text字段；你自己解析轨迹时，不要假设它一定是str。
流程图：
sequenceDiagram
participant U as 用户
participant A as create_agent
participant M as 模型
participant T as StructuredTool
participant S as math_server.py
U->>A: 3 加 5 再乘 12
A->>M: 带工具 schema 的消息
M-->>A: tool_calls add(3,5)
A->>T: ainvoke
T->>S: MCP CallTool
S-->>T: 8
T-->>A: ToolMessage
A->>M: 带上 8
M-->>A: tool_calls multiply(8,12)
A->>S: 再调 MCP
S-->>A: 96
M-->>U: 等于 96
7. HTTP 传输 #
stdio 适合本机脚本。工具已经作为 HTTP 服务部署，或你想连别人的公开服务器时，换 transport="http"。
7.1. http.py：连官方文档的公开 MCP #
LangChain 文档自己挂了一台公开 MCP：https://docs.langchain.com/mcp，不用 API Key。这是练 HTTP 最省事的路径——不必先在本地起服务。
# 标准库：异步入口
import asyncio
# 多服务器客户端，HTTP 和 stdio 用同一套 API
from langchain_mcp_adapters.client import MultiServerMCPClient
async def main() -> None:
# 官方文档自己就挂了一台公开 MCP 服务器，不用 API Key，适合练 HTTP 传输
client = MultiServerMCPClient(
{
"docs": {
# http 和 streamable-http、streamable_http 是同一种传输的别名
"transport": "http",
# 公开地址，不需要先在本地起服务
"url": "https://docs.langchain.com/mcp",
}
}
)
# 和 stdio 一样，先握手再拉工具
tools = await client.get_tools()
# 打印工具名，确认 HTTP 通路是通的
print("工具数量:", len(tools))
for tool in tools:
print(f"- {tool.name}: {tool.description}")
if __name__ == "__main__":
asyncio.run(main())输出（工具名以当时文档站为准，数量和描述可能会变）：
- search_docs_by_lang_chain：在 Docs by LangChain 知识库中搜索相关信息、代码示例、API 参考和指南。当需要解答关于 Docs by LangChain 的问题、查找特定文档、理解功能工作原理，或定位实现细节时，请使用此工具。搜索会返回带标题的上下文内容，以及指向文档页面的直接链接。如果需要某个具体页面的完整内容，请使用 query_docs_filesystem 工具，对该页面路径执行 `head` 或 `cat`（在搜索返回的路径后追加 `.mdx`——例如 `head -200 /api-reference/create-customer.mdx`）。
- query_docs_filesystem_docs_by_lang_chain：对一个虚拟的内存文件系统执行只读的类 shell 查询。该文件系统以 `/` 为根目录，**仅包含** Docs by LangChain 的文档页面和 OpenAPI 规范。这**不是**真实机器上的 shell——不会在用户电脑、服务器主机或任何网络上执行任何操作。该文件系统是由文档分块支撑的沙箱。
读取文档页面的方式如下：没有单独的「获取页面」工具。要读取某个页面，请将其 `.mdx` 路径（例如 `/quickstart.mdx`、`/api-reference/create-customer.mdx`）传给 `head` 或 `cat`。若要在文档中进行精确关键词或正则匹配搜索，请使用 `rg`。若要了解文档结构，请使用 `tree` 或 `ls`。
**工作流程：** 对于「如何认证」「速率限制」这类宽泛或概念性查询，请先使用搜索工具。当你需要精确的关键词/正则匹配、结构探索，或按路径读取某个页面的完整内容时，请使用本工具。
支持的命令：rg（ripgrep）、grep、find、tree、ls、cat、head、tail、stat、wc、sort、uniq、cut、sed、awk、jq，以及基本的文本工具。禁止写入、禁止联网、禁止进程控制。对任意命令运行 `--help` 可查看用法。
每次调用都是**无状态的**：工作目录始终重置为 `/`，shell 变量、别名和历史记录不会在多次调用之间保留。如果需要在子目录中操作，请在同一次调用中用 `&&` 串联命令，或传入绝对路径（例如 `cd /api-reference && ls` 或 `ls /api-reference`）。**不要**假定某次调用中的 `cd` 会影响下一次调用。
示例：
- `tree / -L 2` — 查看顶层目录结构
- `rg -il "rate limit" /` — 查找所有提到 “rate limit” 的文件
- `rg -C 3 "apiKey" /api-reference/` — 显示每次匹配及其前后各 3 行上下文
- `head -80 /quickstart.mdx` — 读取某个页面的前 80 行
- `head -80 /quickstart.mdx /installation.mdx /guides/first-deploy.mdx` — 一次调用读取多个页面
- `cat /api-reference/create-customer.mdx` — 需要完整内容时读取整个页面
- `cat /openapi/spec.json | jq '.paths | keys'` — 列出 OpenAPI 端点
每次调用的输出上限为 30KB。对大文件优先使用有针对性的 `rg -C` 或 `head -N`，避免宽泛的 `cat`。若只需读取大文件中的相关部分，请使用 `rg -C 3 "pattern" /path/file.mdx`。尽可能将多次文件读取合并到同一次 `head` 或 `cat` 调用中。
在回复用户时引用页面，请将文件系统路径转换为 URL 路径：去掉 `.mdx` 扩展名。例如，`/quickstart.mdx` 变为 `/quickstart`，`/api-reference/overview.mdx` 变为 `/api-reference/overview`。
- submit_feedback：向文档团队报告文档站点的问题，以便修复。当文档页面有误、过时、表述不清、内容不完整或示例损坏时使用。此反馈针对文档内容本身——不是产品支持请求，也不是关于本工具或助手的反馈。和 stdio 的唯一差别在连接配置：一边是 command + args，一边是 url。get_tools() / ainvoke / create_agent 完全一样。
7.2. 自己起一台 HTTP 服务器 #
本地调试可以用 FastMCP 的 streamable-http：默认监听 127.0.0.1:8000，路径 /mcp。
# 过滤无关告警
import warnings
# 只忽略 pydantic_settings
warnings.filterwarnings("ignore", module="pydantic_settings")
# 导入 FastMCP
from mcp.server.fastmcp import FastMCP
# port 可改；streamable_http_path 默认就是 /mcp
mcp = FastMCP("Weather", log_level="ERROR", port=8000)
# HTTP 服务器上的工具也可以是 async 函数
@mcp.tool()
async def get_weather(location: str) -> str:
"""获取指定地点的天气（演示用，返回固定文本）。"""
return f"{location} 今天晴，气温 25°C。"
if __name__ == "__main__":
# 注意：这里是 streamable-http，不是 stdio
mcp.run(transport="streamable-http")另开一个终端跑它，它会一直听着。客户端这样连即可：
# 标准库：异步
import asyncio
# 适配器客户端
from langchain_mcp_adapters.client import MultiServerMCPClient
async def main() -> None:
# 连本机刚刚起的 HTTP 服务器
client = MultiServerMCPClient(
{
"weather": {
"transport": "http",
"url": "http://127.0.0.1:8000/mcp",
}
}
)
# 拉工具并本地调用，确认 HTTP 通路
tools = await client.get_tools()
weather = next(t for t in tools if t.name == "get_weather")
print(await weather.ainvoke({"location": "北京"}))
if __name__ == "__main__":
asyncio.run(main())入门阶段用 §7.1 的公开服务器就够了。自己起 HTTP 时记住：URL 要写到 /mcp 这一层，不要只写到 http://127.0.0.1:8000。
7.3. HTTP 请求头 #
需要带 Token 时，在连接配置里加 headers。这只对 HTTP / SSE 有效；stdio 没有 HTTP 头这回事。
# 标准库：异步
import asyncio
# 适配器客户端
from langchain_mcp_adapters.client import MultiServerMCPClient
async def main() -> None:
# 连公开文档 MCP，顺便演示 headers 字段怎么写
client = MultiServerMCPClient(
{
"docs": {
"transport": "http",
"url": "https://docs.langchain.com/mcp",
# 每次 HTTP 请求都会带上这些头；真实项目里放 Bearer Token
"headers": {
"Authorization": "Bearer YOUR_TOKEN",
"X-Custom-Header": "demo",
},
}
}
)
# 公开服务器不校验这个 Token，这里只确认加了头之后仍然能拉到工具
tools = await client.get_tools()
print("仍能拉到", len(tools), "个工具")
if __name__ == "__main__":
asyncio.run(main())更复杂的 OAuth / 自定义 httpx.Auth 属于生产鉴权；官方 MCP SDK 有现成实现，入门不必先做。要按工具动态改头（例如每个工具换不同 Token），走 §11 的拦截器 request.override(headers=...)。
8. 无状态默认与显式会话 #
这是适配器里最容易踩错的一点。
8.1. 默认：无状态 #
MultiServerMCPClient 默认无状态：每次工具调用都新建一条 MCP 会话，用完就拆。stdio 下，「新建会话」= 再启动一个服务器子进程。
所以：
- 服务器里若用全局变量记「上一次的中间结果」，下一次调用大概率看不见
- 工具很重、启动很慢时，每次调用都要再付一次启动成本
- 好处是隔离干净：一个调用崩了，不影响下一个
普通加减法、没有会话状态的 HTTP API，用默认值即可。
8.2. session.py：需要时再握住一条连接 #
服务器要在多次调用之间保留上下文，或者你想避免反复拉起进程，就用 client.session("服务器名") 握住一条显式会话，再 load_mcp_tools(session)。
# 标准库：异步入口
import asyncio
# 标准库：当前解释器
import sys
# 标准库：定位服务器
from pathlib import Path
# 客户端 + 在已有 session 上加载工具
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_mcp_adapters.tools import load_mcp_tools
async def main() -> None:
# 仍然先定位 math_server.py
server = Path(__file__).resolve().parent / "math_server.py"
if not server.exists():
raise SystemExit("请先把 math_server.py 保存到本文件同一目录")
# 构造客户端，这一步还不拉起进程
client = MultiServerMCPClient(
{
"math": {
"transport": "stdio",
"command": sys.executable,
"args": [str(server)],
}
}
)
# ----- 默认：无状态。get_tools 握手一次；之后每次 ainvoke 再开新会话 -----
tools = await client.get_tools()
add = next(t for t in tools if t.name == "add")
print("无状态第一次:", await add.ainvoke({"a": 1, "b": 1}))
print("无状态第二次:", await add.ainvoke({"a": 2, "b": 2}))
# ----- 显式 session：子进程在 with 块内一直活着，多次调用复用同一条连接 -----
async with client.session("math") as session:
# 必须把这个 session 传给 load_mcp_tools，工具才会绑在这条连接上
sess_tools = await load_mcp_tools(session)
sess_add = next(t for t in sess_tools if t.name == "add")
print("同一会话第一次:", await sess_add.ainvoke({"a": 3, "b": 3}))
print("同一会话第二次:", await sess_add.ainvoke({"a": 4, "b": 4}))
if __name__ == "__main__":
asyncio.run(main())输出（block 的 id 会变，算术结果应对齐）：
无状态第一次: [{'type': 'text', 'text': '2', ...}]
无状态第二次: [{'type': 'text', 'text': '4', ...}]
同一会话第一次: [{'type': 'text', 'text': '6', ...}]
同一会话第二次: [{'type': 'text', 'text': '8', ...}]两种方式都能算出正确答案。差别不在返回值，而在连接寿命：async with client.session("math") 期间，stdio 子进程只活一份；出了 with 块才退出。
注意：get_tools() 拿来的工具不会自动绑到后来打开的 session 上。要复用连接，必须用 load_mcp_tools(session) 再拉一份。
9. 工具出错时发生什么 #
第 8 章本地 @tool 若直接 raise，agent.invoke() 会把异常甩到你脸上。MCP 这边默认更温和。
9.1. errors.py #
divide(1, 0) 在服务器里会 raise ValueError("除数不能为 0")。下面对比两个开关：
# 标准库：异步入口
import asyncio
# 标准库：当前解释器
import sys
# 标准库：定位服务器
from pathlib import Path
from langchain_mcp_adapters.client import MultiServerMCPClient
async def main() -> None:
server = Path(__file__).resolve().parent / "math_server.py"
if not server.exists():
raise SystemExit("请先把 math_server.py 保存到本文件同一目录")
# 连接配置抽出来，下面两个客户端共用
conn = {
"math": {
"transport": "stdio",
"command": sys.executable,
"args": [str(server)],
}
}
# 默认 handle_tool_errors=True：执行失败变成错误文本，不崩
client_ok = MultiServerMCPClient(conn)
tools_ok = await client_ok.get_tools()
divide_ok = next(t for t in tools_ok if t.name == "divide")
print("默认（回灌错误）:", await divide_ok.ainvoke({"a": 1, "b": 0}))
# handle_tool_errors=False：恢复成抛异常，和旧版适配器一样
client_raise = MultiServerMCPClient(conn, handle_tool_errors=False)
tools_raise = await client_raise.get_tools()
divide_raise = next(t for t in tools_raise if t.name == "divide")
try:
await divide_raise.ainvoke({"a": 1, "b": 0})
except Exception as exc:
print("关闭回灌后:", type(exc).__name__)
print(" 信息:", exc)
if __name__ == "__main__":
asyncio.run(main())输出：
默认（回灌错误）: [{'type': 'text', 'text': 'Error executing tool divide: 除数不能为 0', ...}]
关闭回灌后: _MCPToolExecutionError
信息: Error executing tool divide: 除数不能为 09.2. 怎么选 #
handle_tool_errors |
工具 isError=True 时 |
适合 |
|---|---|---|
True（默认，≥0.3.0） |
变成错误文本 / 失败的 ToolMessage，Agent 能读到并改参重试 |
日常 Agent |
False |
抛 _MCPToolExecutionError（ToolException 子类） |
你想在外层自己 catch |
它只覆盖 MCP 工具执行失败。传输断了、会话握手失败、内容类型转换失败，不论开关怎么开都会抛——那些不是「业务上这次没查到」，而是「协议层坏了」。
和本地工具对齐的口诀：
能预期的业务错误，让模型看见。 协议 / 网络失败，让它响亮地炸——不要假装成一次普通工具结果。
10. Resources 与 Prompts #
Tools 是给模型调的。Resources / Prompts 这两样是给你的程序调的：读一份静态资料，或取一段预置提示词，再决定要不要塞进对话。
10.1. resources.py #
# 标准库：异步入口
import asyncio
# 标准库：当前解释器
import sys
# 标准库：定位服务器
from pathlib import Path
from langchain_mcp_adapters.client import MultiServerMCPClient
async def main() -> None:
server = Path(__file__).resolve().parent / "math_server.py"
if not server.exists():
raise SystemExit("请先把 math_server.py 保存到本文件同一目录")
client = MultiServerMCPClient(
{
"math": {
"transport": "stdio",
"command": sys.executable,
"args": [str(server)],
}
}
)
# 按服务器名拉资源；不传 uris 就读取该服务器列出的全部静态资源
blobs = await client.get_resources("math")
print("资源数量:", len(blobs))
for blob in blobs:
# metadata["uri"] 是资源地址；as_string() 取出文本内容
print("URI:", blob.metadata["uri"])
print("MIME:", blob.mimetype)
print("内容:", blob.as_string())
# 按名字取 prompt，arguments 对应服务器函数的参数
messages = await client.get_prompt(
"math",
"summarize",
arguments={"topic": "MCP"},
)
for message in messages:
print(f"{type(message).__name__}: {message.content}")
if __name__ == "__main__":
asyncio.run(main())输出：
资源数量: 1
URI: resource://handbook
MIME: text/plain
内容: 差旅报销需在返程 7 日内提交。
HumanMessage: 请用三句话总结「MCP」。资源变成了 LangChain 的 Blob：as_string() 读文本，二进制则是 bytes。也可以 get_resources("math", uris=["resource://handbook"])，只取指定 URI。
带路径参数的动态资源（URI 模板）不会出现在「列出全部」里，必须自己知道 URI 再读——这是 MCP SDK 的行为，不是适配器漏了。
Prompt 取回来已经是消息对象，可以直接放进 agent.ainvoke({"messages": messages + [用户这句]})。它不会自动当系统提示用，得你自己拼。
需要模型「自己决定读哪份资料」时，仍然要包成 Tool——不要指望 Agent 去调 get_resources。
11. 拦截器入门 #
MCP 服务器跑在别的进程里，看不见 LangGraph 的 store、context、Agent 状态。拦截器挂在「适配器真正发调用」的前后，让你在自己这边改参数、打日志、补 HTTP 头，或直接短路返回。
官方文档后半还有运行时注入、Store、Command 跳转、进度回调、elicitation（服务器反过来问用户要输入）等。那些是生产向能力；入门先把「洋葱模型 + override」跑通即可。
11.1. interceptor.py #
下面挂两个拦截器：外层打日志，内层在 add 执行前把 a 乘 2。传入 add(2, 3)，真正算的是 4 + 3 = 7。
# 标准库：异步入口
import asyncio
# 标准库：当前解释器
import sys
# 标准库：定位服务器
from pathlib import Path
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_mcp_adapters.interceptors import MCPToolCallRequest
async def logging_interceptor(request: MCPToolCallRequest, handler):
"""调用前后打日志，演示拦截器最基本的「洋葱」写法。"""
# 调用前能看到工具名和参数
print(f"调用前: {request.name} {request.args}")
# 必须把控制权交给 handler，否则工具根本不会执行
result = await handler(request)
# 调用后能看到 MCP SDK 的原始结果类型
print(f"调用后: {type(result).__name__}")
return result
async def double_a_interceptor(request: MCPToolCallRequest, handler):
"""演示 request.override：在真正执行前改参数。"""
# 只对 add 动手，其他工具原样放行
if request.name == "add" and "a" in request.args:
# override 是不可变更新，返回新请求，不改原来那份
request = request.override(args={**request.args, "a": request.args["a"] * 2})
print("已把 a 改成", request.args["a"])
return await handler(request)
async def main() -> None:
server = Path(__file__).resolve().parent / "math_server.py"
if not server.exists():
raise SystemExit("请先把 math_server.py 保存到本文件同一目录")
client = MultiServerMCPClient(
{
"math": {
"transport": "stdio",
"command": sys.executable,
"args": [str(server)],
}
},
# 列表里越靠前越外层：先 logging，再改参数，最后才真正调工具
tool_interceptors=[logging_interceptor, double_a_interceptor],
)
tools = await client.get_tools()
add = next(t for t in tools if t.name == "add")
# 传入 a=2, b=3；拦截器会把 a 改成 4，所以结果应是 7
print("结果:", await add.ainvoke({"a": 2, "b": 3}))
if __name__ == "__main__":
asyncio.run(main())输出：
调用前: add {'a': 2, 'b': 3}
已把 a 改成 4
调用后: CallToolResult
结果: [{'type': 'text', 'text': '7', ...}]按输出顺序就能记住洋葱：
logging 进 → double_a 改参 → 真正调工具 → double_a 出 → logging 出tool_interceptors 列表里第一个是最外层。这和第 9 章 middleware 的洋葱是同一套心智。
三条入门规则：
- 签名固定：
async def (request, handler)，必须await handler(...)或自己返回结果。 - 改请求用
request.override(...)，不要原地改。可改name/args/headers；server_name和runtime是只读上下文。 - 外层看到的是内层改参之前的请求——所以上面「调用前」打印的还是
a=2。
12. 多服务器与工具名前缀 #
MultiServerMCPClient 一次可以连多台服务器；get_tools() 把它们的工具拼成一张列表。stdio 和 HTTP 可以混在同一个字典里。
# 标准库：异步入口
import asyncio
# 标准库：当前解释器
import sys
# 标准库：定位本地服务器脚本
from pathlib import Path
from langchain_mcp_adapters.client import MultiServerMCPClient
async def main() -> None:
# 本地数学服务器，和前面几节相同
math_server = Path(__file__).resolve().parent / "math_server.py"
if not math_server.exists():
raise SystemExit("请先把 math_server.py 保存到本文件同一目录")
# 一个客户端同时连 stdio 和 HTTP 两台服务器
client = MultiServerMCPClient(
{
"math": {
"transport": "stdio",
"command": sys.executable,
"args": [str(math_server)],
},
"docs": {
"transport": "http",
"url": "https://docs.langchain.com/mcp",
},
}
)
# 两台服务器的工具会被拼成一张列表
tools = await client.get_tools()
print("合并后的工具:", [t.name for t in tools])
if __name__ == "__main__":
asyncio.run(main())两台服务器都有名叫 search 的工具时，后加载的会盖掉先加载的（模型只看见一个 search）。打开 tool_name_prefix=True 后，名字变成 math_search、docs_search。
12.1. 9_prefix.py #
# 标准库：异步入口
import asyncio
# 标准库：当前解释器
import sys
# 标准库：定位服务器
from pathlib import Path
from langchain_mcp_adapters.client import MultiServerMCPClient
async def main() -> None:
server = Path(__file__).resolve().parent / "math_server.py"
if not server.exists():
raise SystemExit("请先把 math_server.py 保存到本文件同一目录")
# 默认同名工具会撞车；打开前缀后变成 math_add、weather_add
client = MultiServerMCPClient(
{
"math": {
"transport": "stdio",
"command": sys.executable,
"args": [str(server)],
}
},
tool_name_prefix=True,
)
tools = await client.get_tools()
print("带前缀的工具名:", [t.name for t in tools])
add = next(t for t in tools if t.name == "math_add")
print("math_add(1, 2) =", await add.ainvoke({"a": 1, "b": 2}))
if __name__ == "__main__":
asyncio.run(main())输出：
带前缀的工具名: ['math_add', 'math_multiply', 'math_divide']
math_add(1, 2) = [{'type': 'text', 'text': '3', ...}]只连一台服务器时，前缀不是必须的。一开始就要拼多台、又不能保证工具名不撞，打开更安全。打开之后，系统提示里也要写 math_add 这种全名，否则模型和你自己都会调错名字。
13. 可复用模板：10_template.py #
把「连服务器 → 拉工具 → 建 Agent → ainvoke」收成一个函数；以后换服务器路径即可。
# 标准库：异步入口
import asyncio
# 标准库：当前解释器
import sys
# 标准库：定位服务器脚本
from pathlib import Path
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain_mcp_adapters.client import MultiServerMCPClient
# 覆盖已有环境变量，和第 2 章相同
load_dotenv(override=True)
async def build_mcp_agent(server_path: str):
"""连上指定的 stdio MCP 服务器，返回一个已经挂好工具的 Agent。"""
# 配置只含一台服务器；要加 HTTP 再往字典里塞键
client = MultiServerMCPClient(
{
"math": {
"transport": "stdio",
"command": sys.executable,
"args": [str(Path(server_path).resolve())],
}
}
)
# 拉工具，得到的是 StructuredTool 列表
tools = await client.get_tools()
# 和第 2 章相同的 create_agent，只是 tools 来自 MCP
agent = create_agent(
"deepseek:deepseek-v4-flash",
tools,
system_prompt="你是简洁的中文助手。需要计算时必须调用工具。",
)
return agent
async def main() -> None:
# 仍然要求 math_server.py 在同目录
server = Path(__file__).resolve().parent / "math_server.py"
if not server.exists():
raise SystemExit("请先把 math_server.py 保存到本文件同一目录")
# 先建 Agent 再问
agent = await build_mcp_agent(str(server))
result = await agent.ainvoke(
{"messages": [{"role": "user", "content": "8 乘 7 等于多少？"}]}
)
# 只打印最终回答；完整轨迹见 §6
print(result["messages"][-1].content)
if __name__ == "__main__":
asyncio.run(main())输出（措辞可能不同，得数应是 56）：
8 乘 7 等于 **56**。要加第二台 HTTP 服务器，往 connections 字典里再塞一个键即可；get_tools() 会自动合并。
14. 踩坑清单 #
带 ⚠ 的是静默失败，或「报错很长、原因却很土」的一类。
连接与进程
command用sys.executable，不要"python"。否则 Windows 上很容易连到系统 Python，子进程里import mcp失败args里的脚本路径用绝对路径（Path.resolve()）- ⚠ 不要手动跑
python math_server.py来配合 stdio 客户端。stdio 服务器由客户端拉起；你先起一份，客户端还会再起一份 async with MultiServerMCPClient已被禁止，改用client.session(...)或直接get_tools()- HTTP 的 URL 要包含路径，FastMCP 默认是
/mcp
调用方式
- 只用
ainvoke/asyncio.run。tool.invoke和agent.invoke都会NotImplementedError - ⚠ 返回值是 content block 列表，不是裸
8。断言写result[0]["text"] - 参数传字典，键与服务器函数参数名一致
状态与多服务器
- ⚠ 默认每次调用新会话。服务器若靠内存保存会话状态，会丢。要跨调用保留状态，用
client.session+load_mcp_tools(session) - 多服务器工具同名会覆盖，打开
tool_name_prefix=True
错误
- 默认业务错误会回灌，Agent 不崩；想自己 catch，再设
handle_tool_errors=False - 传输失败永远抛，和这个开关无关
- 适配器
<0.3.0时，业务错误会以ToolException打上来，行为和现在不同
和本地工具的边界
- 能
@tool解决的，不要上 MCP - Resource / Prompt 不会自动出现在 Agent 的工具列表里
- 中文工具名仍然不合法（第 8 章的
^[a-zA-Z0-9_-]+$同样适用）
15. 练习 #
每题都先本地 ainvoke 单测，再考虑接模型。
- 改服务器再拉一次：给
math_server.py加subtract(a: int, b: int)，确认get_tools()里出现新名字，并本地算出10-3=7。 - 复现同步调用失败：对
add调invoke（不是ainvoke），把完整异常记下来。 - 接 Agent：问「先把 6 和 7 相乘，再加 5」，打印完整
messages，确认工具顺序和参数。 - HTTP：跑
http.py，任选一个文档工具看它的convert_to_openai_tool输出。 - 错误回灌：让 Agent 去
divide(1, 0)，对比默认开关和handle_tool_errors=False时，ainvoke会不会把异常抛给你。 - 前缀：连数学服务器 + 文档 HTTP 服务器，打开
tool_name_prefix=True，打印全部工具名。 - 拦截器：写一个拦截器，把所有
add的结果日志打到终端（调用后打印CallToolResult）。确认add(1, 1)仍返回2。
16. 本章小结 #
- MCP 是 LLM 应用和外部能力之间的协议。LangChain 用
langchain-mcp-adapters把它翻成第 8 章那种工具——翻完就还是create_agent。 - 自己写服务器用 SDK 自带的 FastMCP：`@mcp.tool()
/@mcp.resource/@mcp.prompt()`。stdio 服务器不要手动前台启动，交给客户端拉起。 - 客户端核心就三步：
MultiServerMCPClient(配置)→await client.get_tools()→await agent.ainvoke(...)。 - stdio 用
sys.executable+ 脚本绝对路径；HTTP 用url（记得带/mcp）和可选的headers。 - 默认无状态，每次调用新会话；要握住连接，用
async with client.session("name")再load_mcp_tools(session)。客户端本身不能async with。 - MCP 工具只有异步实现，必须
ainvoke。返回值是 content block，不是裸的 Python 值。 - 业务错误默认回灌给模型（
>=0.3.0）；协议 / 网络失败始终抛出。 - 拦截器是洋葱，第一个最外层；
request.override改参或改头。入门够用；Store / elicitation / OAuth 等生产向能力，需要时再查官方页。
工具已经接到 Agent 之后，若要把多个专家拆开、按步骤交接，或按需加载说明书，见 第 18 章 Multi-agent。