导航菜单

  • 1.什么是MCP
  • 2.MCP架构
  • 3.MCP服务器
  • 4.MCP客户端
  • 5.版本控制
  • 6.连接MCP服务器
  • 7.SDKs
  • 8.Inspector
  • 9.规范
  • 10.架构
  • 11.协议
  • 12.生命周期
  • 13.工具
  • 14.资源
  • 15.提示
  • 16.日志
  • 17.进度
  • 18.传输
  • 19.补全
  • 20.引导
  • 21.采样
  • 22.任务
  • 23.取消
  • 24.Ping
  • 25.根
  • 26.分页
  • 27.授权
  • 28.初始化
  • 29.工具
  • 30.资源
  • 31.结构化输出
  • 32.提示词
  • 33.上下文
  • 34.StreamableHTTP
  • 35.参数补全
  • 36.引导
  • 37.采样
  • 38.LowLevel
  • 39.任务
  • 40.取消
  • 41.ping
  • 42.根
  • 43.分页
  • 44.授权
  • 45.授权
  • Keycloak
  • asyncio
  • contextlib
  • httpx
  • pathlib
  • pydantic
  • queue
  • starlette
  • subprocess
  • threading
  • uvicorn
  • JSON-RPC
  • z
  • 1.根
  • 2. root_client.py
  • 3. root_server.py
  • 4.工作流程
    • 4.1 整体流程
    • 4.2 服务端 root_server.py 讲解
      • 4.2.1. 核心结构
    • 4.3 关键点
    • 4.4 请求方向
    • 4.5 客户端 root_client.py 讲解
      • 4.5.1. 核心流程
      • 4.5.2. 关键点
      • 4.5.3. 回调实现
    • 4.6 时序图
    • 4.7 消息格式(JSON-RPC)
    • 4.8 Roots 能力声明
    • 4.9 运行方式

1.根 #

  • 什么是根?

根(Roots)是 MCP 协议中用于定义客户端允许服务器访问的文件(或目录)范围的机制。通过暴露根目录,客户端告知服务器哪些目录是“可访问的”,从而设定了服务器的操作边界。这既能保护用户主机的文件安全,又为自动化或远程工具提供了明确的操作区域。

在此次更改中:

  1. 客户端(root_client.py)

    • 客户端定义了 list_roots_callback 回调,该回调向服务器提供了“项目目录”和“用户主目录”的 URI。
    • 客户端通过 ClientSession 注册该回调,声明支持 roots 能力。
    • 客户端入口会调用服务端的 get_roots 工具,触发 roots/list 请求流程,服务器收到该请求后,其返回内容会在客户端展示。
  2. 服务端(root_server.py)

    • 服务端通过 get_roots 工具,主动向客户端发送 roots/list 请求以获取暴露的根目录。
    • 服务端收到根目录列表后,将其格式化为带注释的清单返回给调用方。
    • 支持在服务端调用或集成时查看客户端开放的所有根目录,便于权限控制与调试。
  3. 意义与作用

    • 有效隔离、限制服务器访问主机文件系统的范围,最大限度减少了误操作与安全风险。
    • 支持多根目录灵活配置,便于共享多个项目或自定义工作区域。

简而言之,根(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/call

4.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 获取根目录并返回格式化结果。

← 上一节 41.ping 下一节 43.分页 →

访问验证

请输入访问令牌

Token不正确,请重新输入