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#fragment4.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