1.根 #
根(Roots)是 MCP 协议中用于定义客户端允许服务器访问的文件(或目录)范围的机制。通过暴露根目录,客户端告知服务器哪些目录是“可访问的”,从而设定了服务器的操作边界。这既能保护用户主机的文件安全,又为自动化或远程工具提供了明确的操作区域。
在此次更改中:
客户端(root_client.py)
- 客户端定义了
list_roots_callback回调,该回调向服务器提供了“项目目录”和“用户主目录”的 URI。 - 客户端通过
ClientSession注册该回调,声明支持 roots 能力。 - 客户端入口会调用服务端的
get_roots工具,触发 roots/list 请求流程,服务器收到该请求后,其返回内容会在客户端展示。
- 客户端定义了
服务端(root_server.py)
- 服务端通过
get_roots工具,主动向客户端发送roots/list请求以获取暴露的根目录。 - 服务端收到根目录列表后,将其格式化为带注释的清单返回给调用方。
- 支持在服务端调用或集成时查看客户端开放的所有根目录,便于权限控制与调试。
- 服务端通过
意义与作用
- 有效隔离、限制服务器访问主机文件系统的范围,最大限度减少了误操作与安全风险。
- 支持多根目录灵活配置,便于共享多个项目或自定义工作区域。
简而言之,根(Roots)机制通过客户端-服务端协商,安全、灵活地实现了远程操作服务的“文件边界”能力,为后续的安全文件操作和多工具协作打下了基础。
2. root_client.py #
root_client.py
# MCP Roots 功能客户端,定义边界并向服务器暴露可访问目录
"""MCP Roots 功能客户端
根(Roots)定义了 MCP 服务器在文件系统中的操作边界。
客户端向服务器暴露「可访问的目录」,服务器据此知道能读写哪些路径。
本客户端通过 list_roots_callback 提供根目录列表。当服务端调用
get_roots 工具并发送 roots/list 请求时,此回调返回暴露的目录。
"""
# 导入操作系统模块
import os
# 导入系统相关模块
import sys
# 导入路径管理模块
from pathlib import Path
# 导入异步IO库 anyio
import anyio
# 从 mcp 包中导入客户端会话、服务器参数、标准IO客户端
from mcp import ClientSession, StdioServerParameters, stdio_client
# 导入根目录类型定义
from mcp.types import ListRootsResult, Root
# 获取当前文件所在目录的绝对路径
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
# 根据操作系统类型拼接虚拟环境中的 python 解释器路径
VENV_PYTHON = os.path.join(
BASE_DIR, ".venv", "Scripts" if os.name == "nt" else "bin",
"python.exe" if os.name == "nt" else "python",
)
# 定义 roots/list 请求时的回调函数,返回暴露的根目录
async def list_roots_callback(_context) -> ListRootsResult:
"""当服务端请求 roots/list 时,返回客户端暴露的可访问目录。"""
# 将当前项目目录转为 file:// URI
project_uri = Path(BASE_DIR).as_uri()
# 获取用户主目录的 file:// URI
home_uri = Path.home().as_uri()
# 返回根目录列表
return ListRootsResult(
roots=[
Root(uri=project_uri, name="项目目录"),
Root(uri=home_uri, name="用户主目录"),
]
)
# 主异步入口函数
async def main() -> None:
# 构造服务端启动参数
server_params = StdioServerParameters(
command=VENV_PYTHON,
args=[os.path.join(BASE_DIR, "root_server.py")],
env={},
cwd=BASE_DIR,
)
# 启动 stdio 客户端,连接服务端
async with stdio_client(server_params) as (read, write):
# 创建 MCP 客户端会话并注册 roots 回调
async with ClientSession(
read, write, list_roots_callback=list_roots_callback
) as session:
# 执行初始化流程并声明 roots 能力
await session.initialize()
print("会话已初始化(已声明 roots 能力)")
# 获取全部可用工具列表并打印
tools = await session.list_tools()
print(f"可用工具: {[t.name for t in tools.tools]}")
# 调用 get_roots,服务端会向客户端发送 roots/list,触发 list_roots_callback
print("\n调用 get_roots(服务端将请求 roots/list)...")
result = await session.call_tool("get_roots", {})
if result.content:
# 打印根目录展示文本
print(result.content[0].text)
else:
# 无结果提示
print("无结果")
print("\nRoots 功能演示完成")
# 脚本主入口
if __name__ == "__main__":
# 如果终端支持 reconfigure,则设定编码为 utf-8,避免中文乱码
if hasattr(sys.stdout, "reconfigure"):
sys.stdout.reconfigure(encoding="utf-8")
sys.stdin.reconfigure(encoding="utf-8")
# 启动异步主循环,执行 main
anyio.run(main)
3. root_server.py #
root_server.py
# MCP Roots 功能服务端
"""
根(Roots)定义了 MCP 服务器在文件系统中的操作边界。
客户端向服务器暴露「可访问的目录」,服务器据此知道能读写哪些路径。
本服务端提供 get_roots 工具:向客户端发送 roots/list 请求,
获取客户端暴露的根目录列表,并返回给调用方。
"""
# 导入sys模块,用于设置输入输出编码
import sys
# 导入anyio用于异步IO操作
import anyio
# 导入mcp的类型定义
import mcp.types as types
# 导入MCP服务端主类
from mcp.server import Server
# 导入stdio_server用于标准输入输出的通信
from mcp.server.stdio import stdio_server
# 创建名为"root-server"的MCP服务端实例
server = Server("root-server")
# 注册list_tools处理器,声明工具列表
@server.list_tools()
async def handle_list_tools() -> types.ListToolsResult:
# 返回包含get_roots工具的工具列表
return types.ListToolsResult(
tools=[
types.Tool(
name="get_roots", # 工具名:get_roots
description="向客户端请求根目录列表(roots/list),返回客户端暴露的可访问目录。", # 工具描述
inputSchema={"type": "object", "properties": {}}, # 输入参数定义为空对象
)
]
)
# 注册call_tool处理器,用于调用工具
@server.call_tool()
async def handle_call_tool(name: str, arguments: dict | None) -> types.CallToolResult:
# 如果工具名不是get_roots,返回错误信息
if name != "get_roots":
return types.CallToolResult(
content=[types.TextContent(type="text", text=f"未知工具: {name}")],
isError=True,
)
try:
# 通过session.list_roots()向客户端发送roots/list请求
result = await server.request_context.session.list_roots()
lines = []
# 遍历返回的根目录列表,格式化输出
for r in result.roots:
name_part = f" ({r.name})" if r.name else ""
lines.append(f"- {r.uri}{name_part}")
text = "客户端暴露的根目录:\n" + "\n".join(lines) if lines else "无根目录"
# 返回根目录结果
return types.CallToolResult(
content=[types.TextContent(type="text", text=text)]
)
except Exception as e:
# 捕获异常,返回错误信息
return types.CallToolResult(
content=[types.TextContent(type="text", text=f"获取 roots 失败: {e}")],
isError=True,
)
# 定义主异步入口函数
async def main() -> None:
# 启动stdio_server上下文,获取读写流
async with stdio_server() as (read_stream, write_stream):
# 运行MCP服务端主循环,监听并处理请求
await server.run(
read_stream,
write_stream,
server.create_initialization_options(),
)
# 如果当前脚本作为主程序入口则执行
if __name__ == "__main__":
# 如果标准输出支持reconfigure,则设置编码为utf-8,防止中文乱码
if hasattr(sys.stdout, "reconfigure"):
sys.stdout.reconfigure(encoding="utf-8")
sys.stdin.reconfigure(encoding="utf-8")
# 启动anyio异步主循环,运行main
anyio.run(main)
4.工作流程 #
4.1 整体流程 #
Roots(根) 用来定义 MCP 服务器在文件系统中的操作范围。客户端通过根目录列表告诉服务器可以访问哪些路径,服务器据此决定能读写哪些目录。
本示例中:客户端 通过 list_roots_callback 提供根目录,服务端 在 get_roots 工具中调用 session.list_roots(),向客户端发送 roots/list 请求,获取这些根目录。
4.2 服务端 root_server.py 讲解 #
4.2.1. 核心结构 #
server = Server("root-server")
@server.list_tools() # 声明 get_roots 工具
@server.call_tool() # 处理 tools/call4.3 关键点 #
| 部分 | 说明 |
|---|---|
| get_roots 工具 | 供客户端调用,内部向客户端请求根目录 |
server.request_context.session |
当前请求的会话,用于发送服务端→客户端的请求 |
session.list_roots() |
发送 roots/list 请求,等待客户端返回 ListRootsResult |
| Root 结构 | uri(file:// URI)、name(可选显示名) |
4.4 请求方向 #
Roots 的请求方向是 服务端 → 客户端:服务端主动发起 roots/list,客户端通过回调返回根目录列表。
4.5 客户端 root_client.py 讲解 #
4.5.1. 核心流程 #
# 1. 注册 list_roots_callback
async with ClientSession(read, write, list_roots_callback=list_roots_callback) as session:
# 2. 初始化时声明 roots 能力
await session.initialize() # capabilities.roots 非空
# 3. 调用 get_roots,触发服务端发送 roots/list
result = await session.call_tool("get_roots", {})4.5.2. 关键点 #
| 部分 | 说明 |
|---|---|
| list_roots_callback | 收到 roots/list 时被调用,返回 ListRootsResult |
| Root(uri, name) | uri 使用 Path.as_uri() 转为 file:// 格式 |
| roots 能力 | 提供 callback 后,InitializeResult 中会声明 capabilities.roots |
4.5.3. 回调实现 #
async def list_roots_callback(_context) -> ListRootsResult:
"""当服务端请求 roots/list 时,返回客户端暴露的可访问目录。"""
# 将当前项目目录转为 file:// URI
project_uri = Path(BASE_DIR).as_uri()
# 获取用户主目录的 file:// URI
home_uri = Path.home().as_uri()
# 返回根目录列表
return ListRootsResult(
roots=[
Root(uri=project_uri, name="项目目录"),
Root(uri=home_uri, name="用户主目录"),
]
)4.6 时序图 #
sequenceDiagram
participant C as 客户端<br/>root_client.py
participant S as 服务端<br/>root_server.py
Note over C,S: 1. 建立连接与初始化
C->>S: stdio_client 启动子进程 root_server.py
S->>S: stdio_server() 绑定 stdin/stdout
C->>S: initialize (capabilities.roots)
S->>C: InitializeResult
C->>C: print("会话已初始化(已声明 roots 能力)")
Note over C,S: 2. 列出工具
C->>S: tools/list
S->>C: ListToolsResult (get_roots)
C->>C: print("可用工具: ['get_roots']")
Note over C,S: 3. 调用 get_roots 工具
C->>S: tools/call (name=get_roots)
S->>S: handle_call_tool("get_roots")
Note over C,S: 4. 服务端请求 roots(服务端→客户端)
S->>C: roots/list
C->>C: list_roots_callback() 执行
C->>C: Path(BASE_DIR).as_uri(), Path.home().as_uri()
C->>S: ListRootsResult (roots: [项目目录, 用户主目录])
Note over C,S: 5. 服务端返回工具结果
S->>S: 格式化 roots 为文本
S->>C: CallToolResult (客户端暴露的根目录:...)
C->>C: print(result.content[0].text)
Note over C,S: 6. 结束
C->>C: print("Roots 功能演示完成")
4.7 消息格式(JSON-RPC) #
roots/list 请求(服务端 → 客户端):
{"jsonrpc": "2.0", "id": 3, "method": "roots/list", "params": null}roots/list 响应(客户端 → 服务端):
{
"jsonrpc": "2.0",
"id": 3,
"result": {
"roots": [
{"uri": "file:///D:/forever/docs/mcpsdk2", "name": "项目目录"},
{"uri": "file:///C:/Users/Administrator", "name": "用户主目录"}
]
}
}4.8 Roots 能力声明 #
客户端在 initialize 时,若提供 list_roots_callback,会在 ClientCapabilities 中声明:
"capabilities": {
"roots": {"listChanged": true}
}服务端据此知道可以发送 roots/list 请求。
4.9 运行方式 #
uv run root_client.py客户端会启动服务端、初始化会话、调用 get_roots,服务端通过 roots/list 获取根目录并返回格式化结果。