ai
  • index
  • 1.首页
  • 2.介绍
  • 3.架构概览
  • 4.服务器概念
  • 5.客户端概念
  • 6.版本控制
  • 7.连接到远程MCP服务器
  • 8.连接到本地MCP服务器
  • json_rpc
  • 9.构建一个MCP服务器
  • 10.检查员
  • 11.构建一个MCP客户端
  • 14.架构
  • 15.基础协议概述
  • 16.生命周期
  • 17.传输
  • 18.授权
  • 19.安全最佳实践
  • 20.取消
  • 21.Ping
  • 22.进展
  • 23.Roots
  • 24.采样
  • 25.启发
  • 26.服务器特性
  • 27.提示词
  • 28.资源
  • 29.工具
  • 30.完成
  • 31.日志记录
  • 32.分页
  • 33.架构参考
  • URI模板
  • 12.实现
  • http.server
  • 动态客户端注册协议
  • 受保护资源元数据
  • 授权服务器元数据
  • JWKS
  • PKCE
  • PyJWT
  • secrets
  • watchfiles
  • 实现authorization
  • 实现cancel
  • 实现completion
  • 实现logging
  • 实现pagination
  • 实现process
  • 实现transport
  • psutil
  • pytz
  • zoneinfo
  • contextlib
  • Starlette
  • mcp.1.starter
  • mcp.2.Resource
  • mcp.3.structured_output
  • mcp.4.prompts
  • mcp.5.context
  • mcp.6.streamable
  • mcp.7.lowlevel
  • mcp.8.Completion
  • mcp.9.Elicitation
  • mcp.10.oauth
  • mcp.11.integration
  • mcp.12.best
  • mysql-mcp
  • databases
  • uvicorn
  • asynccontextmanager
  • AsyncExitStack
  • streamable
  • aiohttp
  • publish
  • email
  • schedule
  • twine
  • 1.教学文档总览
  • 2.教师使用指南
  • 3.教学系统快速参考
  • 4.新生入门指南
  • 5.学生使用指南
  • 1. 什么是 Resource?
  • 2. 服务器
  • 3. 客户端
  • 4.AnyUrl
    • 4.1. 基本结构
    • 4.2. MCP 中的常见格式
  • 4. 运行与验证

1. 什么是 Resource? #

本章主要介绍如何使用 MCP 的 Resource 机制,包括定义静态资源和模板资源,以及客户端如何读取资源。你将学会:

  • 如何使用 @mcp.resource() 注册静态资源;
  • 如何使用 @mcp.resource() 注册带参数的模板资源;
  • 如何读取静态资源和模板资源;
  • 如何处理资源模板参数。

  • Resource 可类比为“只读的 GET 接口”,用于把数据提供给客户端/LLM 的上下文。

  • 在 FastMCP 中,通过 `@mcp.resource("scheme://...")` 装饰器注册资源:
    • 没有 {} 占位符的 URI 称为“静态资源”,函数不接收参数。
    • 带 {param} 占位符的 URI 称为“模板资源”,函数参数必须与占位符名称一一对应,否则会报错。
  • 读取资源由客户端发起 resources/read,服务器返回文本或二进制内容。

2. 服务器 #

在 C:\mcp-project 下新建 resources_server.py:

# 导入 FastMCP 高层服务器类
from mcp.server.fastmcp import FastMCP
import json

# 创建 FastMCP 实例,命名为 Resources Demo
mcp = FastMCP(name="Resources MCP Server")


# 1) 静态资源:URI 中无参数占位符,函数也不应有参数
# 使用 @mcp.resource 装饰器注册资源,指定 URI 和 mime_type
@mcp.resource("config://settings", mime_type="application/json")
# 定义静态资源处理函数,不接收参数
def get_settings() -> str:
    """返回应用的 JSON 配置文本。"""
    # 返回 JSON 字符串
    return json.dumps(
        {"theme": "dark", "language": "zh-CN", "debug": False},
        ensure_ascii=False,
        indent=2,
    )


# 2) 模板资源:URI 中含有 {name} 占位符,函数参数必须完全一致(name)
# 使用 @mcp.resource 装饰器注册带参数的资源
@mcp.resource("greeting://{name}", mime_type="text/plain")
# 定义模板资源处理函数,参数名与 URI 占位符一致
def greeting(name: str) -> str:
    """根据传入的 name 返回问候语。"""
    # 返回个性化问候语
    return f"Hello, {name}! 欢迎使用 MCP 资源。"


# 3) 多个参数时必须全部匹配
# 注册另一个带参数的资源,URI 占位符为 {id}
@mcp.resource("user://{id}", mime_type="text/plain")
# 定义用户信息资源处理函数,参数 id 必须与 URI 占位符一致
def user_profile(id: str) -> str:
    """返回一个示例用户信息(字符串)。"""
    # 真实业务中可从数据库/文件系统中查询
    if id == "001":
        return "用户 001:Alice,角色:admin"
    if id == "002":
        return "用户 002:Bob,角色:viewer"
    # 未找到用户时返回提示
    return f"未找到用户 {id}"


# 主入口:以 stdio 方式运行服务器
if __name__ == "__main__":
    mcp.run(transport="stdio")

要点:

  • greeting://{name} 的函数必须是 def greeting(name: str) -> str。
  • 如果 URI 占位符与函数参数不一致,FastMCP 会在注册时抛错,便于尽早发现问题。

3. 客户端 #

在 C:\mcp-project 下新建 test_client_resources.py:

# 导入异步IO库,用于支持异步编程
import asyncio

# 导入os模块,用于文件路径操作
import os

# 从pydantic库导入AnyUrl类型,用于资源URI的类型校验
from pydantic import AnyUrl

# 从mcp包导入ClientSession、StdioServerParameters和types,用于客户端会话和类型定义
from mcp import (
    ClientSession,
    StdioServerParameters,
    types,
)

# 从mcp.client.stdio模块导入stdio_client工厂方法,用于创建stdio客户端
from mcp.client.stdio import stdio_client


# 定义主异步函数
async def main() -> None:
    # 获取当前文件的绝对路径所在目录
    base_dir = os.path.dirname(os.path.abspath(__file__))
    # 拼接得到服务器脚本的完整路径
    server_path = os.path.join(base_dir, "resources_server.py")

    # 配置以stdio方式启动服务器的参数
    server_params = StdioServerParameters(
        command="python",  # 指定启动命令为python
        args=[server_path],  # 启动参数为服务器脚本路径
        env={},  # 环境变量为空
    )

    # 建立到服务器的stdio连接
    async with stdio_client(server_params) as (read, write):
        # 创建客户端会话并初始化
        async with ClientSession(read, write) as session:
            # 执行会话初始化(握手)
            await session.initialize()

            # 列出已注册的静态资源(不带占位符的资源)
            resources = await session.list_resources()
            # 打印所有静态资源的URI
            print("[Resources]", [r.uri for r in resources.resources])

            # 列出资源模板(带占位符的资源)
            templates = await session.list_resource_templates()
            # 打印所有资源模板的URI模板
            print(
                "[ResourceTemplates]",
                [t.uriTemplate for t in templates.resourceTemplates],
            )

            # 读取静态资源 config://settings
            result_config = await session.read_resource(AnyUrl("config://settings"))
            # 用于存放读取到的文本内容
            texts_config = []
            # 遍历资源内容块
            for block in result_config.contents:
                # 判断内容块类型是否为TextResourceContents
                if isinstance(block, types.TextResourceContents):
                    # 提取文本内容并添加到列表
                    texts_config.append(block.text)
            # 打印 config://settings 的读取结果
            print("[Read config://settings]", " | ".join(texts_config))

            # 读取模板资源 greeting://Alice(将{name}占位符替换为Alice)
            result_greet = await session.read_resource(AnyUrl("greeting://Alice"))
            # 用于存放读取到的文本内容
            texts_greet = []
            # 遍历资源内容块
            for block in result_greet.contents:
                # 判断内容块类型是否为TextResourceContents
                if isinstance(block, types.TextResourceContents):
                    # 提取文本内容并添加到列表
                    texts_greet.append(block.text)
            # 打印 greeting://Alice 的读取结果
            print("[Read greeting://Alice]", " | ".join(texts_greet))

            # 读取模板资源 user://001 与 user://003(分别测试命中和未命中情况)
            for uid in ["001", "003"]:
                # 读取 user://{uid} 资源
                result_user = await session.read_resource(AnyUrl(f"user://{uid}"))
                # 用于存放读取到的文本内容
                texts_user = []
                # 遍历资源内容块
                for block in result_user.contents:
                    # 判断内容块类型是否为TextResourceContents
                    if isinstance(block, types.TextResourceContents):
                        # 提取文本内容并添加到列表
                        texts_user.append(block.text)
                # 打印 user://{uid} 的读取结果
                print(f"[Read user://{uid}]", " | ".join(texts_user))


# 判断当前模块是否为主模块入口
if __name__ == "__main__":
    # 在事件循环中运行主异步函数
    asyncio.run(main())

说明:

  • list_resources() 返回当前可直接读取的“静态资源”集合。
  • list_resource_templates() 返回“模板资源”集合,展示带占位符的 URI 模板。
  • read_resource(AnyUrl(...)) 读取指定 URI 对应内容;对于模板资源,需将占位符替换为实际值(如 greeting://Alice)。

4.AnyUrl #

AnyUrl 是 MCP 协议中用于标识和访问资源的统一资源标识符(URI)系统。它提供了一种标准化的方式来引用各种类型的数据和功能。

4.1. 基本结构 #

scheme://authority/path?query#fragment

4.2. MCP 中的常见格式 #

# 文件系统资源
"file:///path/to/file.txt"
"file://C:/Users/username/documents/file.txt"

# 网络资源
"https://api.example.com/data"
"http://localhost:8000/api/users"

# 数据库资源
"postgresql://localhost:5432/mydb"
"mongodb://localhost:27017/collection"

# 自定义协议
"mcp://info"                    # MCP 内部资源
"example://data"                # 示例资源
"custom://myapp/users"          # 自定义应用资源

4. 运行与验证 #

cd C:\mcp-project
call .venv\Scripts\activate
python test_client_resources.py

预期输出:

[Resources] [AnyUrl('config://settings')]
[ResourceTemplates] ['greeting://{name}', 'user://{id}']
[Read config://settings] {
  "theme": "dark",
  "language": "zh-CN",
  "debug": false
}
[Read greeting://Alice] Hello, Alice! 欢迎使用 MCP 资源。
[Read user://001] 用户 001:Alice,角色:admin
[Read user://003] 未找到用户 003

访问验证

请输入访问令牌

Token不正确,请重新输入