2.1 资源(Resources)深度解析

资源(Resources)是MCP的核心概念之一，它们提供了一种标准化的方式来为大语言模型提供上下文信息。在本章中，我们将深入探讨资源的设计、实现和最佳实践。

资源类型和URI模式

MCP中的资源通过URI(统一资源标识符)进行标识和访问。资源URI通常采用以下格式：

scheme://path/to/resource

常见的资源方案(schemes)

以下是一些常见的资源方案：

file://: 访问文件系统中的文件
data://: 访问内存或数据库中的数据
http(s)://: 访问通过HTTP(S)协议可获取的资源
db://: 访问数据库中的内容
user://: 访问用户相关的信息
doc://: 访问文档或知识库
config://: 访问配置信息
custom://: 自定义方案，用于特定应用场景

方案名称可以是任何有效标识符，但建议使用能清晰表达资源性质的方案名称。

URI路径部分

URI的路径部分可以包含：

静态路径：file:///home/user/document.txt
分层路径：doc://knowledge/science/physics
查询参数：data://search?query=example&limit=10

创建静态和动态资源

在MCP中，资源可以是静态的(内容固定)或动态的(内容根据请求生成)。

静态资源

静态资源的内容在服务器启动时确定，不会随请求变化。

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("静态资源示例")

@mcp.resource("static://greeting")
def get_greeting() -> str:
    """返回一个静态问候语"""
    return "Hello, world!"

@mcp.resource("static://version")
def get_version() -> str:
    """返回应用版本信息"""
    return "v1.0.0"

动态资源

动态资源的内容会根据请求参数生成。

import datetime

@mcp.resource("dynamic://time")
def get_current_time() -> str:
    """返回当前时间"""
    return datetime.datetime.now().isoformat()

@mcp.resource("weather://{city}")
def get_weather(city: str) -> str:
    """返回指定城市的天气信息"""
    # 在实际应用中，可能会调用外部API
    weather_data = {
        "北京": "晴天，25°C",
        "上海": "多云，28°C",
        "广州": "小雨，30°C"
    }
    return weather_data.get(city, f"没有{city}的天气信息")

参数化资源路径

参数化资源路径允许通过URI传递参数，这对构建灵活的资源API非常有用。

基本参数化

最简单的参数化使用花括号{}来定义参数：

@mcp.resource("user://{user_id}")
def get_user_profile(user_id: str) -> str:
    """获取用户资料"""
    # 从数据库或其他存储中获取用户信息
    return f"用户{user_id}的资料..."

@mcp.resource("doc://{category}/{document_id}")
def get_document(category: str, document_id: str) -> str:
    """获取指定类别和ID的文档"""
    return f"{category}类别中ID为{document_id}的文档内容..."

类型化参数

参数可以具有特定类型：

@mcp.resource("product://{product_id:int}")
def get_product(product_id: int) -> str:
    """获取产品信息"""
    return f"产品{product_id}的详细信息..."

@mcp.resource("report://{year:int}/{month:int}")
def get_monthly_report(year: int, month: int) -> str:
    """获取指定月份的报告"""
    return f"{year}年{month}月的月度报告..."

可选参数和默认值

资源路径也可以包含可选参数：

@mcp.resource("search://{query}/{page:int?}")
def search_results(query: str, page: int = 1) -> str:
    """搜索结果，带可选页码参数"""
    return f"搜索'{query}'的结果，第{page}页"

资源内容类型和MIME类型

MCP资源可以返回不同类型的内容，通过MIME类型指定。

文本内容

文本是最常见的资源内容类型：

@mcp.resource("text://example")
def get_text() -> str:
    """返回纯文本内容"""
    return "这是一段纯文本内容"

默认情况下，字符串返回值的MIME类型为text/plain。

结构化数据

对于结构化数据，可以返回字典或列表，它们将被自动转换为JSON：

@mcp.resource("data://users")
def get_users() -> dict:
    """返回用户列表（JSON格式）"""
    return {
        "users": [
            {"id": 1, "name": "张三"},
            {"id": 2, "name": "李四"},
            {"id": 3, "name": "王五"}
        ]
    }

返回的字典或列表会被序列化为JSON，MIME类型为application/json。

二进制数据

对于二进制数据，可以使用bytes类型：

@mcp.resource("image://logo")
def get_logo() -> bytes:
    """返回logo图片数据"""
    with open("logo.png", "rb") as f:
        return f.read()

二进制数据的默认MIME类型为application/octet-stream。

自定义MIME类型

可以通过注解或额外返回值指定MIME类型：

@mcp.resource("custom://data")
def get_custom_data() -> tuple:
    """返回自定义MIME类型的数据"""
    data = "<svg>...</svg>"
    mime_type = "image/svg+xml"
    return data, mime_type

资源错误处理

资源处理过程中可能会遇到各种错误，MCP提供了标准化的错误处理机制。

捕获和返回错误

最简单的方式是在资源函数中捕获异常并返回适当的错误信息：

@mcp.resource("error://demo/{id}")
def get_with_error_handling(id: str) -> str:
    """演示错误处理的资源"""
    try:
        # 可能引发异常的代码
        if id == "invalid":
            raise ValueError("无效的ID")
        
        # 正常处理...
        return f"资源ID {id} 的内容"
    
    except ValueError as e:
        return f"错误: {e}"
    except Exception as e:
        return f"未知错误: {e}"

使用专用异常类

MCP提供了专用的异常类，用于表示不同类型的资源错误：

from mcp.errors import ResourceNotFoundError, ResourcePermissionError

@mcp.resource("protected://{resource_id}")
def get_protected_resource(resource_id: str) -> str:
    """获取受保护的资源"""
    
    # 检查资源是否存在
    if resource_id not in available_resources:
        raise ResourceNotFoundError(f"资源{resource_id}不存在")
    
    # 检查权限
    if not has_permission(resource_id):
        raise ResourcePermissionError(f"没有访问资源{resource_id}的权限")
    
    # 正常处理...
    return available_resources[resource_id]

状态码和错误元数据

对于更复杂的错误处理，可以返回包含状态码和错误详情的结构：

from mcp.types import ErrorContent

@mcp.resource("advanced://error")
def get_with_advanced_error() -> any:
    """演示高级错误处理"""
    try:
        # 业务逻辑...
        raise Exception("示例错误")
    
    except Exception as e:
        error = ErrorContent(
            status_code=500,  # HTTP风格的状态码
            reason="Internal Server Error",
            detail=str(e),
            help_text="请联系管理员解决此问题"
        )
        return error

进阶示例：文件系统浏览器

下面是一个更复杂的实例，实现一个文件系统浏览器，展示了如何将上述概念应用到实际场景：

import os
import json
from typing import Dict, List, Optional
from mcp.server.fastmcp import FastMCP
from mcp.errors import ResourceNotFoundError

mcp = FastMCP("文件系统浏览器")

# 定义基础目录，出于安全考虑限制访问范围
BASE_DIR = "/path/to/allowed/directory"

@mcp.resource("fs://list/{path:?}")
def list_directory(path: str = "") -> str:
    """列出目录内容
    
    参数:
        path: 相对于基础目录的路径，默认为根目录
    
    返回:
        目录内容的JSON格式字符串
    """
    full_path = os.path.normpath(os.path.join(BASE_DIR, path))
    
    # 安全检查：确保路径仍在BASE_DIR内
    if not full_path.startswith(BASE_DIR):
        raise ResourcePermissionError("无权访问该路径")
    
    if not os.path.exists(full_path):
        raise ResourceNotFoundError(f"路径不存在: {path}")
    
    if not os.path.isdir(full_path):
        raise ValueError(f"路径不是目录: {path}")
    
    # 获取目录内容
    entries = []
    with os.scandir(full_path) as it:
        for entry in it:
            entry_info = {
                "name": entry.name,
                "path": os.path.join(path, entry.name),
                "is_dir": entry.is_dir(),
                "size": entry.stat().st_size if not entry.is_dir() else None,
                "modified": entry.stat().st_mtime
            }
            entries.append(entry_info)
    
    # 排序：先目录后文件，按名称排序
    entries.sort(key=lambda e: (not e["is_dir"], e["name"].lower()))
    
    # 构建返回结果
    result = {
        "current_path": path,
        "parent_path": os.path.dirname(path) if path else None,
        "entries": entries
    }
    
    return json.dumps(result, indent=2)

@mcp.resource("fs://file/{path}")
def get_file_content(path: str) -> tuple:
    """获取文件内容
    
    参数:
        path: 相对于基础目录的文件路径
    
    返回:
        文件内容和MIME类型
    """
    full_path = os.path.normpath(os.path.join(BASE_DIR, path))
    
    # 安全检查
    if not full_path.startswith(BASE_DIR):
        raise ResourcePermissionError("无权访问该文件")
    
    if not os.path.exists(full_path):
        raise ResourceNotFoundError(f"文件不存在: {path}")
    
    if not os.path.isfile(full_path):
        raise ValueError(f"路径不是文件: {path}")
    
    # 确定MIME类型
    # 简化版，实际应用中可使用mimetypes模块或更复杂的逻辑
    mime_type = "text/plain"
    extension = os.path.splitext(path)[1].lower()
    
    mime_map = {
        ".txt": "text/plain",
        ".html": "text/html",
        ".htm": "text/html",
        ".json": "application/json",
        ".xml": "application/xml",
        ".png": "image/png",
        ".jpg": "image/jpeg",
        ".jpeg": "image/jpeg",
        ".gif": "image/gif",
        ".pdf": "application/pdf"
    }
    
    mime_type = mime_map.get(extension, "application/octet-stream")
    
    # 文本文件直接读取文本内容
    text_mimes = ["text/plain", "text/html", "application/json", "application/xml"]
    
    if mime_type in text_mimes:
        try:
            with open(full_path, "r", encoding="utf-8") as f:
                content = f.read()
            return content, mime_type
        except UnicodeDecodeError:
            # 如果文本解码失败，退回到二进制模式
            mime_type = "application/octet-stream"
    
    # 二进制文件读取为bytes
    with open(full_path, "rb") as f:
        content = f.read()
    
    return content, mime_type

@mcp.resource("fs://info/{path}")
def get_file_info(path: str) -> str:
    """获取文件或目录的详细信息
    
    参数:
        path: 相对于基础目录的路径
    
    返回:
        文件或目录信息的JSON格式字符串
    """
    full_path = os.path.normpath(os.path.join(BASE_DIR, path))
    
    # 安全检查
    if not full_path.startswith(BASE_DIR):
        raise ResourcePermissionError("无权访问该路径")
    
    if not os.path.exists(full_path):
        raise ResourceNotFoundError(f"路径不存在: {path}")
    
    # 获取基本信息
    stat_info = os.stat(full_path)
    is_dir = os.path.isdir(full_path)
    
    info = {
        "name": os.path.basename(path) or path,
        "path": path,
        "is_dir": is_dir,
        "size": stat_info.st_size if not is_dir else None,
        "created": stat_info.st_ctime,
        "modified": stat_info.st_mtime,
        "accessed": stat_info.st_atime,
        "permissions": stat_info.st_mode & 0o777
    }
    
    # 对于目录，添加内容计数
    if is_dir:
        contents = os.listdir(full_path)
        info["item_count"] = len(contents)
        info["dir_count"] = sum(1 for item in contents if os.path.isdir(os.path.join(full_path, item)))
        info["file_count"] = sum(1 for item in contents if os.path.isfile(os.path.join(full_path, item)))
    
    return json.dumps(info, indent=2)

if __name__ == "__main__":
    mcp.run()

这个文件系统浏览器示例展示了:

参数化资源路径
错误处理
不同内容类型的处理
安全检查
复杂逻辑的资源实现

小结

在本章中，我们深入探讨了MCP资源的各个方面：

资源URI的设计和模式
静态和动态资源的创建
参数化资源路径的实现
资源内容类型和MIME类型
资源错误处理机制
通过文件系统浏览器示例，展示了复杂资源的实现

资源是MCP的核心组件之一，掌握其设计和实现技巧将使您能够构建出功能丰富、灵活且易于使用的MCP服务。在下一章中，我们将探讨MCP的另一个核心概念：工具(Tools)的高级用法。