资源读取操作的实现是一种机制,它将静态统一资源标识符 (URI) 转换为大型语言模型 (LLM) 可访问的内容。资源列表提供了可用数据点的目录,而读取处理器则执行实际的数据获取。在同步环境下,此过程涉及接收请求、定位底层数据、正确格式化数据,并立即将其返回给客户端。资源读取生命周期当 LLM 选择要检查的资源时,客户端会发送一个使用 resources/read 方法的 JSON-RPC 请求。此请求包含模型希望访问的特定 URI。服务器必须将此 URI 路由到正确的内部处理器函数。同步读取的生命周期包含三个不同的阶段:路由: 服务器将传入的 URI 与已注册的模板或方案进行匹配。获取: 处理器函数执行从内存、文件系统或数据库获取数据的逻辑。序列化: 原始数据被封装在一个标准内容对象中,通常包含文本载荷和 MIME 类型。以下图表描绘了客户端发起读取请求时的数据流。digraph G { rankdir=TB; node [fontname="Helvetica", shape=box, style=filled, color="#dee2e6", fillcolor="#f8f9fa"]; edge [fontname="Helvetica", color="#adb5bd"]; subgraph cluster_0 { label = "客户端"; style = filled; color = "#e9ecef"; Client [label="MCP 客户端", fillcolor="#a5d8ff", color="#74c0fc"]; } subgraph cluster_1 { label = "服务器端"; style = filled; color = "#e9ecef"; Router [label="URI 路由器", fillcolor="#b197fc", color="#9775fa"]; Handler [label="读取处理器", fillcolor="#69db7c", color="#40c057"]; DataSource [label="数据源\n(数据库/文件)", fillcolor="#ffc9c9", color="#ff8787"]; } Client -> Router [label="resources/read(uri)"]; Router -> Handler [label="路由匹配"]; Handler -> DataSource [label="获取数据"]; DataSource -> Handler [label="原始字节"]; Handler -> Client [label="资源内容"]; }同步资源读取请求的数据流,说明了路由和数据获取步骤。注册读取处理器在 Python SDK 中,处理资源读取涉及使用装饰器修饰一个接受 URI 作为参数的函数。SDK 管理底层的 JSON-RPC 通信,让您专注于数据获取逻辑。处理器必须以特定格式返回内容。协议定义了一个 ReadResourceResult,其中包含一个内容项列表。通常,您将返回 TextContent,它包含 uri、text 正文和 mimeType。考虑一个旨在公开系统日志的服务器。URI 方案可能遵循 logs://system/{log_level} 模式。实现要求定义一个函数来解析此 URI 并相应地筛选日志数据。from mcp.server.fastmcp import FastMCP # 初始化服务器 mcp = FastMCP("LogServer") # 模拟数据存储 LOG_DATA = { "error": "模块 X 发生严重故障\n数据库连接丢失", "info": "服务在端口 8080 启动\n健康检查通过", "debug": "变量状态转储: {x: 1, y: 2}" } @mcp.resource("logs://system/{level}") def read_log(level: str) -> str: """ 读取特定严重级别的系统日志。 """ # 同步访问请求的数据 content = LOG_DATA.get(level) if content is None: # 在内容中返回错误消息通常更安全 # 比为简单查询抛出异常更好 return f"未找到级别为: {level} 的日志" return content在此示例中,模式 logs://system/{level} 自动从传入的 URI 中提取 level 变量。如果客户端请求 logs://system/error,SDK 会调用 read_log("error") 并自动将返回的字符串封装在有效的资源响应对象中。处理二进制和 JSON 数据虽然文本是 LLM 上下文最常见的格式,但资源通常需要表示结构化数据或二进制内容。对于结构化数据,最佳实践是在返回对象之前将其序列化为 JSON。这确保 LLM 收到一个语法正确的字符串,它可以轻松解析。您应将 MIME 类型设置为 application/json,以为模型提供关于内容结构的提示。import json @mcp.resource("users://{user_id}/profile") def get_user_profile(user_id: str) -> str: # 模拟数据库查询 user_data = { "id": user_id, "role": "admin", "last_login": "2023-10-27T10:00:00Z" } # 返回序列化的 JSON return json.dumps(user_data, indent=2)在 JSON 输出中提供缩进会稍微增加令牌计数,但显著提高了模型的可读性,有助于模型更准确地处理数据结构。动态 URI 匹配同步读取通常需要处理未在资源目录中明确列出的 URI。您可能明确列出 file://project/readme.md,但您可能希望支持使用通配符或模式读取目录中的任何文件。在实现处理器时,您必须确保 URI 的动态部分指向有效数据。这引入了一个安全考虑:路径遍历。当资源处理器接受一个作为文件路径或数据库标识符的动态参数时,验证是防止未经授权访问所必需的。以下图表阐明了处理动态 URI 时所需的决策逻辑。digraph Logic { rankdir=TB; node [fontname="Helvetica", shape=box, style=filled, color="#dee2e6", fillcolor="#f8f9fa"]; edge [fontname="Helvetica", color="#adb5bd"]; Input [label="传入 URI", fillcolor="#e7f5ff", color="#74c0fc"]; Parse [label="解析参数", fillcolor="#d0bfff", color="#9775fa"]; Validate [label="验证路径/ID", fillcolor="#ffec99", color="#fcc419"]; Fetch [label="读取数据", fillcolor="#b2f2bb", color="#40c057"]; Error [label="返回错误", fillcolor="#ffc9c9", color="#fa5252"]; Input -> Parse; Parse -> Validate; Validate -> Fetch [label="有效"]; Validate -> Error [label="无效/未经授权"]; }用于验证和处理动态资源参数的逻辑流。同步读取中的错误处理当资源无法读取时,服务器必须清楚地传达此故障。在 MCP 架构中,您在读取操作期间处理错误有两个主要选项:协议级错误: 抛出异常,SDK 会将其转换为 JSON-RPC 错误响应。这向客户端表示请求本身失败或格式不正确。内容级错误: 返回包含错误文本描述的资源。对于 LLM 交互,第二种方法通常更佳。如果模型请求了一个不存在的文件,收到 JSON-RPC 错误可能会导致工具使用链中断,或者模型虚构失败原因。返回文本结果,例如“错误:在路径 X 未找到文件”,允许模型将错误作为上下文读取,并可能通过请求不同路径来纠正自己的错误。@mcp.resource("file://{path}") def read_file(path: str) -> str: try: # 验证路径是否相对于安全目录 if ".." in path or path.startswith("/"): raise ValueError("访问被拒绝") with open(path, "r") as f: return f.read() except FileNotFoundError: return f"错误:文件 '{path}' 不存在。" except ValueError as e: return f"错误:{str(e)}" except Exception: return "错误:读取文件时发生内部错误。"性能同步读取会阻塞请求处理线程。当服务器正在读取文件或查询数据库时,如果实现是单线程或阻塞的,它将无法处理该特定连接上的其他消息。对于本地文件读取,延迟可以忽略不计。但是,如果您的资源从缓慢的外部 API 获取数据,延迟会直接增加用户等待响应的时间。如果数据获取时间预计会很长(例如,超过 500 毫秒),您应考虑将结果缓存在内存中。由于资源是通过特定 URI 请求的,这些 URI 是极佳的缓存键。$$T_{响应} = T_{网络} + T_{处理} + T_{查找}$$通过缓存最小化 $T_{查找}$ 确保 LLM 交互的上下文组装阶段保持流畅。简单的 Python 字典或 LRU(最近最少使用)缓存装饰器是对于不经常变化的资源来说的有效策略。