MCP资源访问协议：标准化AI模型的数据源连接

引言

MCP的资源（Resources） 机制为AI模型提供了标准化访问外部数据源的能力。与工具调用不同，资源是数据导向的——它们让模型能够读取文件、查询数据库、浏览网页等，而不仅仅是执行操作。

本文将深入解析：

• 资源定义规范与URI Scheme设计
• resources/list 资源发现
• resources/read 资源读取
• resources/subscribe 资源订阅与变更通知
• 文件系统资源 vs API资源
• 资源模板（Resource Templates）

资源模型概述

资源与工具的区别

┌─────────────────────────────────────────────────────────────┐
│              MCP 资源模型                                    │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│   ┌─────────────────────────────────────────────────────┐  │
│   │                  资源类型                             │  │
│   ├───────────────┬───────────────────┬─────────────────┤  │
│   │               │                   │                 │  │
│   │  文件资源      │    API资源        │   数据库资源     │  │
│   │  file:///     │   api://          │   db://         │  │
│   │               │                   │                 │  │
│   │  本地文件系统   │   REST API数据    │   SQL查询结果    │  │
│   │               │                   │                 │  │
│   └───────────────┴───────────────────┴─────────────────┘  │
│                                                             │
└─────────────────────────────────────────────────────────────┘

资源定义规范

资源Schema

每个资源通过标准化的元数据描述：

{
  "resources": [
    {
      "uri": "file:///projects/my-app/README.md",
      "name": "项目README",
      "description": "My App项目的主README文档",
      "mimeType": "text/markdown",
      "size": 12345
    },
    {
      "uri": "file:///projects/my-app/src/main.ts",
      "name": "主入口文件",
      "description": "TypeScript主入口",
      "mimeType": "text/typescript",
      "size": 8192
    }
  ]
}

URI Scheme设计

MCP使用标准化的URI Scheme来标识不同类型的资源：

资源发现：resources/list

请求格式

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "resources/list"
}

带分页的请求：

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "resources/list",
  "params": {
    "cursor": "page_token_2"
  }
}

响应格式

{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "resources": [
      {
        "uri": "file:///docs/getting-started.md",
        "name": "快速入门指南",
        "mimeType": "text/markdown"
      },
      {
        "uri": "file:///docs/api-reference.md",
        "name": "API参考文档",
        "mimeType": "text/markdown"
      }
    ],
    "nextCursor": null
  }
}

Server端实现

server.setRequestHandler(ListResourcesRequestSchema, async () => {
  const resources = await scanResourceDirectory("/docs");
  return {
    resources: resources.map(r => ({
      uri: `file://${r.path}`,
      name: r.name,
      mimeType: detectMimeType(r.path),
      size: r.size,
    })),
    nextCursor: null,
  };
});

资源读取：resources/read

请求格式

{
  "jsonrpc": "2.0",
  "id": 2,
  "method": "resources/read",
  "params": {
    "uri": "file:///docs/getting-started.md"
  }
}

响应格式

{
  "jsonrpc": "2.0",
  "id": 2,
  "result": {
    "contents": [
      {
        "uri": "file:///docs/getting-started.md",
        "mimeType": "text/markdown",
        "text": "# 快速入门指南\n\n## 安装\n```bash\nnpm install my-app\n```\n..."
      }
    ]
  }
}

二进制资源使用Base64编码：

{
  "contents": [
    {
      "uri": "file:///images/logo.png",
      "mimeType": "image/png",
      "blob": "iVBORw0KGgoAAAANSUhEUgAA..."
    }
  ]
}

完整读取实现

import os
import base64
import mimetypes
from pathlib import Path

class FileResourceServer:
    def __init__(self, allowed_paths: list[str]):
        self.allowed_paths = [Path(p).resolve() for p in allowed_paths]

    def read_resource(self, uri: str) -> dict:
        """读取资源内容，支持文本和二进制"""
        path = self._uri_to_path(uri)

        # 安全检查：防止路径穿越
        resolved = path.resolve()
        if not any(
            str(resolved).startswith(str(base))
            for base in self.allowed_paths
        ):
            raise PermissionError(f"无权访问: {uri}")

        mime_type, _ = mimetypes.guess_type(str(path))
        if not mime_type:
            mime_type = "application/octet-stream"

        # 判断是文本还是二进制
        if mime_type.startswith("text/") or mime_type in [
            "application/json",
            "application/xml",
            "application/yaml",
        ]:
            with open(path, "r", encoding="utf-8") as f:
                content = f.read()
            return {
                "uri": uri,
                "mimeType": mime_type,
                "text": content,
            }
        else:
            with open(path, "rb") as f:
                content = base64.b64encode(f.read()).decode("utf-8")
            return {
                "uri": uri,
                "mimeType": mime_type,
                "blob": content,
            }

    def _uri_to_path(self, uri: str) -> Path:
        """将 file:// URI 转换为本地路径"""
        if not uri.startswith("file://"):
            raise ValueError(f"不支持的URI scheme: {uri}")
        path_str = uri[7:]  # 去掉 file://
        return Path(path_str)

资源模板（Resource Templates）

资源模板允许Server定义参数化的资源模式：

模板定义

{
  "resourceTemplates": [
    {
      "uriTemplate": "file:///projects/{projectId}/docs/{filename}",
      "name": "项目文档",
      "description": "按项目ID和文件名访问文档",
      "mimeType": "text/markdown"
    },
    {
      "uriTemplate": "api://weather/{city}/forecast",
      "name": "天气预报",
      "description": "按城市名获取天气预报",
      "mimeType": "application/json"
    }
  ]
}

模板匹配实现

import re
from typing import Optional

class ResourceTemplateMatcher:
    def __init__(self):
        self.templates = []

    def add_template(self, uri_template: str, handler: callable):
        """注册资源模板"""
        # 将 {param} 转换为正则组
        pattern = re.sub(
            r"\{(\w+)\}",
            r"(?P<\1>[^/]+)",
            re.escape(uri_template).replace("\\{", "{").replace("\\}", "}")
        )
        self.templates.append({
            "pattern": re.compile(f"^{pattern}$"),
            "uri_template": uri_template,
            "handler": handler,
        })

    def match(self, uri: str) -> Optional[dict]:
        """匹配URI到模板并提取参数"""
        for template in self.templates:
            match = template["pattern"].match(uri)
            if match:
                return {
                    "uri_template": template["uri_template"],
                    "params": match.groupdict(),
                    "handler": template["handler"],
                }
        return None

# 使用示例
matcher = ResourceTemplateMatcher()
matcher.add_template(
    "file:///projects/{projectId}/docs/{filename}",
    lambda params: read_project_doc(params["projectId"], params["filename"])
)

result = matcher.match("file:///projects/my-app/docs/readme.md")
# result.params = {"projectId": "my-app", "filename": "readme.md"}

资源订阅机制

订阅流程

Client                              Server
  │                                   │
  │── resources/subscribe ──────────► │
  │    { uri: "file:///config.json" } │
  │                                   │
  │◄── resources/subscribe response ►│
  │    { success: true }             │
  │                                   │
  │    [配置变更...]                  │
  │                                   │
  │◄── notifications/resources/ ─────│
  │     changed                       │
  │    { uri: "file:///config.json" } │
  │                                   │
  │── resources/read ───────────────► │
  │    { uri: "file:///config.json" } │
  │                                   │
  │◄── 最新内容 ─────────────────────│

订阅实现

// Server端：处理订阅
import { watch } from "fs";

server.setRequestHandler(SubscribeRequestSchema, async (request) => {
  const { uri } = request.params;
  const path = uriToPath(uri);

  // 使用文件系统监听
  const watcher = watch(path, (eventType) => {
    if (eventType === "change") {
      server.notification({
        method: "notifications/resources/changed",
        params: { uri },
      });
    }
  });

  // 存储watcher以备后续取消订阅
  subscribers.set(uri, watcher);

  return { success: true };
});

server.setRequestHandler(UnsubscribeRequestSchema, async (request) => {
  const { uri } = request.params;
  const watcher = subscribers.get(uri);
  if (watcher) {
    watcher.close();
    subscribers.delete(uri);
  }
  return { success: true };
});

安全考虑

路径安全检查

def validate_resource_access(uri: str, allowed_dirs: list[str]) -> bool:
    """验证资源访问权限"""
    from urllib.parse import urlparse

    parsed = urlparse(uri)

    # 只允许 file:// scheme
    if parsed.scheme not in ["file", "api", "db"]:
        return False

    if parsed.scheme == "file":
        path = parsed.path
        resolved = os.path.realpath(path)

        # 防止路径穿越攻击
        for allowed in allowed_dirs:
            allowed_real = os.path.realpath(allowed)
            if resolved.startswith(allowed_real):
                return True
        return False

    return True

最佳实践

1. 资源组织原则

2. 性能优化

// 资源缓存
const resourceCache = new Map<string, {
  content: ResourceContent;
  timestamp: number;
  ttl: number;
}>();

function getCachedResource(uri: string): ResourceContent | null {
  const cached = resourceCache.get(uri);
  if (cached && Date.now() - cached.timestamp < cached.ttl) {
    return cached.content;
  }
  return null;
}

function setResourceCache(uri: string, content: ResourceContent) {
  resourceCache.set(uri, {
    content,
    timestamp: Date.now(),
    ttl: 5000, // 5秒缓存
  });
}

总结

MCP的资源访问协议为AI模型提供了标准化的数据获取能力：

下一步学习建议：

• 工具调用协议
• MCP Server快速搭建
• MCP安全最佳实践

本文最后更新于 2024-07-03。