1.分页 #

• 什么是分页？

本次更改实现了 MCP 协议下的分页（Pagination）机制，分为客户端和服务端两部分代码，提供完整的分页交互演示。

分页是一种将大量数据按“页”分批返回给客户端的机制，避免一次性加载全部内容造成性能和资源压力。常见于需要列举大量数据（如文件、资源、条目等）的场景。

服务端（pagination_server.py）

• 服务端通过 resources/list 注册了带游标的分页处理函数。
• 服务端预设了 20 条模拟资源，每页返回 5 条（通过 PAGE_SIZE 设置）。
• 每次请求根据传入的游标(cursor)决定返回哪一批资源，并返回 next_cursor，指示客户端如何获取下一页数据。
• 当所有资源都已返回时，next_cursor 置为 None，客户端即停止请求。

客户端（pagination_client.py）

• 客户端以标准输入输出方式（stdio）启动服务端子进程，并与之异步通信。
• 循环调用 resources/list，首轮不传游标获取首页，收到 server 返回的 next_cursor 后继续请求下一页。
• 每次分页请求后，客户端会显示本页所有资源的 URI 和名称。
• 当 next_cursor 不存在（已拉取完所有数据）时，客户端退出循环，并汇总输出总共获取的资源条数。

重点

• 游标（Cursor）：每页请求时传递上一页 server 返回的 next_cursor，仅首页不传；服务端用它确定应返回的资源起始位置。
• 分页大小：服务端控制每页最多返回多少条数据。
• 兼容性：示例在打印时兼容 next_cursor 出现不同命名格式，确保健壮性。

通过这套分页机制，你能高效获取大批量分布在多页的数据，并且方便客户端控制数据获取的粒度（即翻页或“加载更多”），适用于云资源管理、文件系统等大数据集的异步访问场景。

2. pagination_client.py #

pagination_client.py

# MCP 分页功能客户端说明文档字符串
"""MCP 分页功能客户端

分页（Pagination）是 MCP 对可能返回大量结果的列表操作采用的分批返回机制。
服务器每次只返回一「页」数据，客户端如需更多，则用游标继续请求下一页，
避免一次性加载过多数据。

本客户端循环调用 resources/list，首次不传 cursor，后续用返回的 next_cursor
请求下一页，直到无更多数据。
"""

# 导入os模块，用于文件路径操作
import os
# 导入sys模块，用于标准输入输出配置
import sys

# 导入anyio库，用于异步运行
import anyio

# 从mcp包导入ClientSession、StdioServerParameters和stdio_client
from mcp import ClientSession, StdioServerParameters, stdio_client
# 从mcp.types导入分页请求参数类型
from mcp.types import PaginatedRequestParams

# 获取当前文件的目录
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",
)


# 定义异步主函数
async def main() -> None:
    # 创建StdioServerParameters对象，指定python解释器与服务器脚本
    server_params = StdioServerParameters(
        command=VENV_PYTHON,
        args=[os.path.join(BASE_DIR, "pagination_server.py")],
        env={},
        cwd=BASE_DIR,
    )

    # 以stdio方式与服务器建立异步连接
    async with stdio_client(server_params) as (read, write):
        # 创建客户端会话对象
        async with ClientSession(read, write) as session:
            # 初始化会话
            await session.initialize()
            # 控制台打印初始化完成
            print("会话已初始化")

            # 分页获取资源相关变量初始化（cursor、页码、资源列表）
            cursor = None
            page_num = 0
            all_resources = []

            # 循环分页请求资源，直到无更多数据
            while True:
                # 页码自增
                page_num += 1
                # 如果cursor存在则构造参数，否则为None（首页不传cursor）
                params = PaginatedRequestParams(cursor=cursor) if cursor else None
                # 异步请求资源列表
                result = await session.list_resources(params=params)

                # 获取当前页所有资源
                resources = result.resources
                # 累加到总资源列表中
                all_resources.extend(resources)

                # 打印当前页码和资源数
                print(f"\n第 {page_num} 页（共 {len(resources)} 条）:")
                # 遍历当前页资源并打印uri和名称
                for r in resources:
                    print(f"  - {r.uri} ({r.name})")

                # 检查是否有下一页游标（兼容不同返回格式的键名）
                next_cursor = getattr(result, "next_cursor", None) or getattr(
                    result, "nextCursor", None
                )
                # 若无下一页，跳出循环
                if next_cursor is None:
                    break
                # 更新cursor准备下次请求
                cursor = next_cursor

            # 打印总共获取的资源数量，演示完成
            print(f"\n共获取 {len(all_resources)} 条资源，分页演示完成")


# 程序入口，确保编码兼容并运行主函数
if __name__ == "__main__":
    # 判断sys.stdout对象是否有reconfigure方法（python3.7+）
    if hasattr(sys.stdout, "reconfigure"):
        # 设置stdout编码为utf-8
        sys.stdout.reconfigure(encoding="utf-8")
        # 设置stdin编码为utf-8
        sys.stdin.reconfigure(encoding="utf-8")
    # 使用anyio运行异步主函数
    anyio.run(main)

3. pagination_server.py #

pagination_server.py

# MCP 分页功能服务端说明文档字符串
"""MCP 分页功能服务端

分页（Pagination）是 MCP 对可能返回大量结果的列表操作采用的分批返回机制。
服务器每次只返回一「页」数据，客户端如需更多，则用游标继续请求下一页，
避免一次性加载过多数据。

本服务端对 resources/list 实现分页：模拟 20 条资源，每页 5 条，
通过 cursor 游标返回下一页。
"""

# 导入标准输入输出和系统编码相关模块
import sys

# 导入异步IO库 anyio
import anyio
# 导入mcp.types类型定义为 types
import mcp.types as types
# 从mcp.server导入Server类
from mcp.server import Server
# 从mcp.server.stdio导入stdio_server(用于标准流通信)
from mcp.server.stdio import stdio_server

# 定义每页返回的资源数量
PAGE_SIZE = 5
# 模拟20条资源数据，URI格式为 resource://items/item-xx
ITEMS = [f"resource://items/item-{i:02d}" for i in range(1, 21)]

# 创建 Server 实例，服务名为 pagination-server
server = Server("pagination-server")

# 注册 list_tools 处理函数，声明本服务端可用的工具
@server.list_tools()
async def handle_list_tools() -> types.ListToolsResult:
    # 返回工具列表（本例为空）
    return types.ListToolsResult(tools=[])

# 注册 resources/list 处理函数，实现分页
@server.list_resources()
async def handle_list_resources(req: types.ListResourcesRequest) -> types.ListResourcesResult:
    # 支持分页的资源列表。cursor 为上一页返回的 next_cursor，无则从0开始
    cursor = req.params.cursor if req.params else None
    # 计算起始下标，若无游标则从0，否则从指定游标位置
    start = 0 if cursor is None else int(cursor)
    # 计算本页应返回的结束下标
    end = start + PAGE_SIZE

    # 构造本页返回的资源对象列表
    page_items = [
        types.Resource(
            uri=uri,
            name=uri.split("/")[-1],
            description=f"资源 {uri}",
        )
        for uri in ITEMS[start:end]
    ]

    # 如果还有未返回数据，则返回下次分页游标，否则置为None
    next_cursor = str(end) if end < len(ITEMS) else None
    # 构造分页结果返回给客户端
    return types.ListResourcesResult(resources=page_items, next_cursor=next_cursor)

# 程序主入口
async def main() -> None:
    # 标准输入输出方式启动服务端
    async with stdio_server() as (read_stream, write_stream):
        # 运行Server，处理MCP请求
        await server.run(
            read_stream,
            write_stream,
            server.create_initialization_options(),
        )

# 仅当作为主程序入口时才运行main
if __name__ == "__main__":
    # 如果stdout支持reconfigure（python>=3.7），则设置编码为utf-8
    if hasattr(sys.stdout, "reconfigure"):
        sys.stdout.reconfigure(encoding="utf-8")
        sys.stdin.reconfigure(encoding="utf-8")
    # 启动异步主循环、运行main
    anyio.run(main)

4.工作流程 #

4.1 整体流程 #

分页（Pagination） 是 MCP 对可能返回大量结果的列表操作采用的分批返回机制。服务端每次只返回一页数据，客户端如需更多，则用游标（cursor）继续请求下一页，避免一次性加载过多数据。

本示例中：服务端 模拟 20 条资源，每页 5 条；客户端 循环调用 resources/list，首次不传 cursor，后续用 next_cursor 请求下一页，直到 next_cursor 为 None。

4.2 服务端 pagination_server.py 讲解 #

4.2.1 核心结构 #

PAGE_SIZE = 5
ITEMS = [f"resource://items/item-{i:02d}" for i in range(1, 21)]  # 20 条

@server.list_resources()
async def handle_list_resources(req: types.ListResourcesRequest) -> types.ListResourcesResult:

4.2.. 分页逻辑 #

4.2.3. 游标含义 #

本示例用偏移量作为游标："0" → "5" → "10" → "15" → "20"。MCP 规范要求游标为不透明字符串，客户端不应解析其内容。

4.3 客户端 pagination_client.py 讲解 #

4.3.1. 核心流程 #

cursor = None
while True:
    params = PaginatedRequestParams(cursor=cursor) if cursor else None
    result = await session.list_resources(params=params)

    # 处理当前页
    resources = result.resources
    all_resources.extend(resources)

    # 检查是否有下一页
    next_cursor = getattr(result, "next_cursor", None) or getattr(result, "nextCursor", None)
    if next_cursor is None:
        break
    cursor = next_cursor

4.3.2. 关键点 #

4.4 时序图 #

4.5 消息格式（JSON-RPC） #

resources/list 请求（第 1 页）：

{"jsonrpc": "2.0", "id": 2, "method": "resources/list", "params": null}

resources/list 请求（第 2 页）：

{"jsonrpc": "2.0", "id": 3, "method": "resources/list", "params": {"cursor": "5"}}

resources/list 响应：

{
  "jsonrpc": "2.0",
  "id": 3,
  "result": {
    "resources": [
      {"uri": "resource://items/item-06", "name": "item-06", "description": "..."},
      ...
    ],
    "nextCursor": "10"
  }
}

最后一页时 nextCursor 为 null 或不存在。

4.6 支持分页的 MCP 接口 #

4.7 运行方式 #

uv run pagination_client.py

客户端会依次请求 4 页，每页 5 条，共 20 条资源。