MCP 资源管理

资源（Resources）是 MCP 中向 AI 模型提供上下文数据的重要机制。与工具不同，资源是被动提供的数据源，客户端可以读取但不能直接修改。

什么是资源？

资源是服务器向客户端提供的可读数据，包括：

• 文件内容
• 数据库查询结果
• API 响应
• 配置数据
• 系统状态信息
• 任意结构化或二进制数据

资源的设计理念是让 AI 模型能够访问和理解外部上下文，而不需要主动执行操作。

资源基础

资源结构

每个资源定义包含以下字段：

from mcp.types import Resource, ResourceTemplate

# 静态资源（已知 URI）
Resource(
    uri="file:///project/README.md",     # 唯一标识符（必需）
    name="项目说明",                      # 显示名称（必需）
    description="项目的 README 文档",    # 详细描述（可选）
    mimeType="text/markdown",            # MIME 类型（可选）
    size=1024                            # 大小（字节，可选）
)

# 资源模板（动态 URI）
ResourceTemplate(
    uriTemplate="file:///project/{filename}",  # URI 模板
    name="项目文件",
    description="读取项目中的任意文件",
    mimeType="text/plain"
)

URI 设计

资源 URI 是资源的唯一标识符，建议使用清晰的命名方案：

方案	用途	示例
file://	文件系统资源	file:///home/user/doc.txt
http:// 或 https://	Web 资源	https://api.example.com/data
config://	配置资源	config://app/settings
db://	数据库资源	db://users/123
自定义方案	应用特定资源	myapp://resource/id

MIME 类型

正确设置 MIME 类型有助于客户端理解和处理资源：

类型	MIME
纯文本	text/plain
Markdown	text/markdown
JSON	application/json
HTML	text/html
PDF	application/pdf
PNG	image/png
JPEG	image/jpeg

定义资源

使用底层 Server 类

from mcp.server import Server
from mcp.types import Resource, TextContent
import json

app = Server("file-server")

@app.list_resources()
async def list_resources() -> list[Resource]:
    """返回可用资源列表"""
    return [
        Resource(
            uri="config://app/settings",
            name="应用配置",
            description="当前应用的配置信息",
            mimeType="application/json"
        ),
        Resource(
            uri="file:///README.md",
            name="项目说明",
            description="项目 README 文档",
            mimeType="text/markdown"
        )
    ]

@app.read_resource()
async def read_resource(uri: str) -> list[TextContent]:
    """读取资源内容"""
    if uri == "config://app/settings":
        settings = load_settings()
        return [TextContent(
            type="text",
            text=json.dumps(settings, ensure_ascii=False, indent=2)
        )]
    
    if uri.startswith("file:///"):
        file_path = uri[7:]  # 移除 "file://" 前缀
        content = read_file_safely(file_path)
        return [TextContent(type="text", text=content)]
    
    raise ValueError(f"Unknown resource: {uri}")

使用 FastMCP

FastMCP 提供了更简洁的资源定义方式：

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("my-server")

@mcp.resource("config://app/settings")
def get_settings() -> str:
    """获取应用配置"""
    return json.dumps(load_settings(), indent=2)

@mcp.resource("file:///README.md")
def get_readme() -> str:
    """获取 README 内容"""
    return Path("README.md").read_text()

# 动态资源模板
@mcp.resource("db://users/{user_id}")
def get_user(user_id: str) -> str:
    """获取用户信息
    
    Args:
        user_id: 用户 ID
    """
    user = database.get_user(user_id)
    return json.dumps(user, indent=2)

动态资源模板

资源模板允许客户端使用参数化 URI 访问资源：

from mcp.types import ResourceTemplate

@app.list_resources()
async def list_resources() -> list[Resource | ResourceTemplate]:
    return [
        # 静态资源
        Resource(
            uri="config://version",
            name="版本信息",
            mimeType="application/json"
        ),
        # 动态资源模板
        ResourceTemplate(
            uriTemplate="db://users/{user_id}",
            name="用户信息",
            description="根据 ID 获取用户详细信息",
            mimeType="application/json"
        ),
        ResourceTemplate(
            uriTemplate="api://repos/{owner}/{repo}",
            name="仓库信息",
            description="获取 GitHub 仓库信息",
            mimeType="application/json"
        )
    ]

@app.read_resource()
async def read_resource(uri: str) -> list[TextContent]:
    # 解析 URI 并提取参数
    if uri.startswith("db://users/"):
        user_id = uri.split("/")[-1]
        user = db.query_user(user_id)
        return [TextContent(
            type="text",
            text=json.dumps(user, ensure_ascii=False)
        )]
    
    if uri.startswith("api://repos/"):
        parts = uri.split("/")
        owner, repo = parts[-2], parts[-1]
        repo_info = await fetch_repo_info(owner, repo)
        return [TextContent(
            type="text",
            text=json.dumps(repo_info, ensure_ascii=False)
        )]
    
    raise ValueError(f"Unknown resource: {uri}")

资源订阅与更新

资源订阅机制允许客户端在资源内容变化时收到通知，这对于动态数据源非常有用。

声明订阅能力

服务器需要声明支持订阅：

from mcp.types import ServerCapabilities

app = Server(
    "my-server",
    options=ServerCapabilities(
        resources={
            "subscribe": True,      # 支持订阅
            "listChanged": True     # 支持列表变更通知
        }
    )
)

实现订阅处理

# 存储订阅状态
subscriptions: dict[str, set[str]] = {}  # uri -> set of client_ids

@app.subscribe()
async def subscribe_resource(uri: str, client_id: str):
    """客户端订阅资源"""
    if uri not in subscriptions:
        subscriptions[uri] = set()
    subscriptions[uri].add(client_id)
    
    # 启动文件监听等
    if uri.startswith("file://") and len(subscriptions[uri]) == 1:
        start_file_watcher(uri)

@app.unsubscribe()
async def unsubscribe_resource(uri: str, client_id: str):
    """客户端取消订阅"""
    if uri in subscriptions:
        subscriptions[uri].discard(client_id)
        if not subscriptions[uri]:
            stop_file_watcher(uri)
            del subscriptions[uri]

发送更新通知

当资源内容变化时，服务器发送通知：

async def notify_resource_update(uri: str):
    """通知所有订阅该资源的客户端"""
    await session.send_notification(
        method="notifications/resources/updated",
        params={"uri": uri}
    )

# 在文件变化时调用
def on_file_changed(file_path: str):
    uri = f"file://{file_path}"
    asyncio.create_task(notify_resource_update(uri))

资源列表变更通知

当可用资源列表发生变化时：

async def notify_list_changed():
    """通知资源列表已变更"""
    await session.send_notification(
        method="notifications/resources/list_changed"
    )

二进制资源

资源可以返回二进制数据，如图像、PDF 等：

from mcp.types import BlobContent, ImageContent
import base64

@app.read_resource()
async def read_resource(uri: str):
    if uri.startswith("image://"):
        image_path = uri[8:]
        with open(image_path, "rb") as f:
            image_data = base64.b64encode(f.read()).decode()
        
        return [ImageContent(
            type="image",
            data=image_data,
            mimeType="image/png"
        )]
    
    if uri.startswith("binary://"):
        file_path = uri[9:]
        with open(file_path, "rb") as f:
            blob_data = base64.b64encode(f.read()).decode()
        
        return [BlobContent(
            type="blob",
            data=blob_data,
            mimeType="application/octet-stream"
        )]

资源缓存

对于频繁访问的资源，实现缓存可以提高性能：

from functools import lru_cache
from datetime import datetime, timedelta

class ResourceCache:
    def __init__(self, ttl_seconds: int = 60):
        self.ttl = timedelta(seconds=ttl_seconds)
        self.cache: dict[str, tuple[str, datetime]] = {}
    
    def get(self, uri: str) -> str | None:
        if uri in self.cache:
            content, timestamp = self.cache[uri]
            if datetime.now() - timestamp < self.ttl:
                return content
            del self.cache[uri]
        return None
    
    def set(self, uri: str, content: str):
        self.cache[uri] = (content, datetime.now())
    
    def invalidate(self, uri: str):
        self.cache.pop(uri, None)

cache = ResourceCache(ttl_seconds=300)

@app.read_resource()
async def read_resource(uri: str) -> list[TextContent]:
    # 先检查缓存
    cached = cache.get(uri)
    if cached:
        return [TextContent(type="text", text=cached)]
    
    # 加载资源
    content = await load_resource(uri)
    cache.set(uri, content)
    
    return [TextContent(type="text", text=content)]

资源安全

路径遍历防护

验证文件路径，防止目录遍历攻击：

from pathlib import Path

ALLOWED_BASE_DIRS = [
    Path("/home/user/projects").resolve(),
    Path("/var/data").resolve(),
]

def is_path_allowed(path: Path) -> bool:
    """检查路径是否在允许的目录内"""
    resolved = path.resolve()
    return any(
        resolved.is_relative_to(base_dir)
        for base_dir in ALLOWED_BASE_DIRS
    )

@app.read_resource()
async def read_resource(uri: str) -> list[TextContent]:
    if uri.startswith("file://"):
        file_path = Path(uri[7:])
        
        if not is_path_allowed(file_path):
            raise ValueError("Access denied: path outside allowed directories")
        
        if not file_path.exists():
            raise ValueError(f"File not found: {uri}")
        
        return [TextContent(
            type="text",
            text=file_path.read_text(encoding="utf-8")
        )]

敏感数据过滤

在返回资源前过滤敏感信息：

SENSITIVE_PATTERNS = [
    r"password\s*=\s*['\"][^'\"]+['\"]",
    r"api_key\s*=\s*['\"][^'\"]+['\"]",
    r"secret\s*=\s*['\"][^'\"]+['\"]",
]

def filter_sensitive(content: str) -> str:
    """过滤敏感信息"""
    import re
    for pattern in SENSITIVE_PATTERNS:
        content = re.sub(pattern, "***REDACTED***", content, flags=re.IGNORECASE)
    return content

@app.read_resource()
async def read_resource(uri: str) -> list[TextContent]:
    content = await load_resource(uri)
    safe_content = filter_sensitive(content)
    return [TextContent(type="text", text=safe_content)]

资源与工具的区别

理解资源和工具的区别对于正确设计 MCP 服务器至关重要：

特性	资源 (Resources)	工具 (Tools)
方向	服务器 → 客户端（单向）	双向（请求-响应）
用途	提供上下文数据	执行操作
主动性	被动（客户端拉取）	主动（LLM 决定调用）
副作用	无（只读）	可能有（执行操作）
缓存	通常可缓存	每次执行不同
订阅	支持变更订阅	不支持

选择指南：

• 如果只是提供数据给 AI 参考 → 使用资源
• 如果需要执行操作或有副作用 → 使用工具
• 如果数据会动态变化且需要实时更新 → 使用资源+订阅

典型应用场景

1. 项目文档

@mcp.resource("project://structure")
def get_project_structure() -> str:
    """获取项目目录结构"""
    # 返回目录树结构
    return generate_tree(Path.cwd())

@mcp.resource("project://docs/{doc_name}")
def get_documentation(doc_name: str) -> str:
    """获取项目文档"""
    return Path(f"docs/{doc_name}.md").read_text()

2. 数据库 schema

@mcp.resource("db://schema")
def get_database_schema() -> str:
    """获取数据库结构"""
    tables = db.get_all_tables()
    schema = {}
    for table in tables:
        schema[table] = db.get_table_columns(table)
    return json.dumps(schema, indent=2)

3. 系统状态

@mcp.resource("system://status")
def get_system_status() -> str:
    """获取系统状态"""
    return json.dumps({
        "cpu_percent": psutil.cpu_percent(),
        "memory_percent": psutil.virtual_memory().percent,
        "disk_percent": psutil.disk_usage('/').percent,
        "uptime": get_uptime()
    })

最佳实践

1. 语义化 URI：使用有意义的 URI 命名方案，便于理解和管理
2. 正确设置 MIME 类型：帮助客户端正确处理资源
3. 实现合理的缓存策略：平衡实时性和性能
4. 严格的安全验证：防止路径遍历和数据泄露
5. 提供有意义的描述：帮助 AI 理解资源的用途
6. 适度使用订阅：只为真正需要实时更新的资源启用订阅

下一步

• 提示模板：预定义的工作流
• 采样机制：服务器发起的 LLM 调用
• 调试指南：资源和工具的调试技巧