1. 什么是 Resource？ #

• 什么是资源？

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

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

• 如何处理资源模板参数。

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

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

本节将系统介绍 MCP 的 Resource（资源）机制，并结合 fastmcp 服务端的主要实现，讲解如何定义、注册和读取静态资源与模板资源，并解析其底层结构与调用流程。

在 FastMCP 服务中，资源（Resource）面向的是类似 config://settings 这样的 URI 标识，每个资源可以注册为“静态资源”（无参数 URI，如 config://settings）或“模板资源”（带参数 URI，如 echo://{msg}）。注册方式如：

@mcp.resource("config://settings")
def get_settings():
    return "system=fastmcp\n"

@mcp.resource("echo://{msg}")
def echo(msg):
    return f"echo: {msg}"

• 静态资源注册时，生成 _Resource 对象，函数参数必须为空。
• 模板资源注册时，生成 _ResourceTemplate 对象，函数参数与 {} 内的模板变量保持一致。

所有资源最终都注册到：

• self._resources[uri]：存放静态资源。
• self._resource_templates[uri_template]：存放模板资源。

客户端读取资源一般使用 session.read_resource(uri)，流程如下：

1. 客户端发起 resources/read 请求，带上目标 URI；
2. 服务端优先查找静态资源，找到则运行回调函数获取内容，返回 TextResourceContents 结构体；
3. 静态资源未找到，则遍历注册的资源模板，匹配 URI，并以提取的参数调用模板函数，支持文本和二进制两种内容返回：
   • 如果返回文本型，封装为 TextResourceContents；
   • 如果返回 bytes，则 base64 编码后作为 BlobResourceContents。
4. 若均未找到对应资源，则返回资源不存在的错误。

key 结构体

实现中涉及的主要响应结构体包含如下：

• ListResourcesResult：资源列表。
• Resource：{uri, name, mime_type} 等字段。
• ReadResourceResult：包含 contents 字段，为一组 TextResourceContents 或 BlobResourceContents。
• TextResourceContents / BlobResourceContents：分别表示文本型/二进制型资源块内容。

具体 Pydantic 结构体见下方代码定义片段。

调用流程总览

fastmcp 服务端核心处理逻辑：

• resources/list：遍历 self._resources，生成结果列表。
• resources/templates/list：遍历 self._resource_templates，生成模板列表。
• resources/read：结合 URI 匹配静态与动态模板资源，返回内容结构体（支持文本和二进制）。

这样，MCP Resource 机制让所有资源以结构化、统一、易扩展的自描述协议供 LLM/客户端即时获取。

npx @modelcontextprotocol/inspector uv --directory D:/aprepare/mcp-starter run resources_server.py

2. resources_client.py #

resources_client.py

# 导入os模块，用于文件路径操作
import os
# 从pydantic库导入AnyUrl类型，用于资源URI的类型校验
from pydantic import AnyUrl
# 从mcp_lite导入ClientSession、StdioServerParameters和types
from mcp_lite import (
    ClientSession,
    StdioServerParameters,
    types,
)
# 从mcp_lite.client.stdio模块导入stdio_client工厂方法
from mcp_lite.client.stdio import stdio_client


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",
        args=[server_path],
        env={},
    )
    # 建立到服务器的stdio连接
    with stdio_client(server_params) as (read, write):
        # 创建客户端会话并初始化
        session = ClientSession(read, write)
        session.initialize()
        # 列出已注册的静态资源（不带占位符的资源）
        resources = session.list_resources()
        print("[Resources]", [r.uri for r in resources.resources])
        # 列出资源模板（带占位符的资源）
        templates = session.list_resource_templates()
        print(
            "[ResourceTemplates]",
            [t.uri_template for t in templates.resource_templates],
        )
        # 读取静态资源 config://settings
        result_config = session.read_resource(AnyUrl("config://settings"))
        texts_config = []
        for block in result_config.contents:
            if isinstance(block, types.TextResourceContents):
                texts_config.append(block.text)
        print("[Read config://settings]", " | ".join(texts_config))
        # 读取模板资源 greeting://Alice
        result_greet = session.read_resource(AnyUrl("greeting://Alice"))
        texts_greet = []
        for block in result_greet.contents:
            if isinstance(block, types.TextResourceContents):
                texts_greet.append(block.text)
        print("[Read greeting://Alice]", " | ".join(texts_greet))
        # 读取模板资源 user://001 与 user://003
        for uid in ["001", "003"]:
            result_user = session.read_resource(AnyUrl(f"user://{uid}"))
            texts_user = []
            for block in result_user.contents:
                if isinstance(block, types.TextResourceContents):
                    texts_user.append(block.text)
            print(f"[Read user://{uid}]", " | ".join(texts_user))


if __name__ == "__main__":
    main()

官方代码

# 导入os模块，用于文件路径操作
import os
import asyncio
# 从pydantic库导入AnyUrl类型，用于资源URI的类型校验
from pydantic import AnyUrl
# 从mcp_lite导入ClientSession、StdioServerParameters和types
from mcp import (
    ClientSession,
    StdioServerParameters,
    types,
)
# 从mcp_lite.client.stdio模块导入stdio_client工厂方法
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",
        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()
            print("[Resources]", [r.uri for r in resources.resources])
            # 列出资源模板（带占位符的资源）
            templates = await session.list_resource_templates()
            print("[ResourceTemplates]",[t.uriTemplate for t in templates.resourceTemplates])
            result_config = await session.read_resource(AnyUrl("config://settings"))
            texts_config = []
            for block in result_config.contents:
                if isinstance(block, types.TextResourceContents):
                    texts_config.append(block.text)
            print("[Read config://settings]", " | ".join(texts_config))
            # 读取模板资源 greeting://Alice
            result_greet = await session.read_resource(AnyUrl("greeting://Alice"))
            texts_greet = []
            for block in result_greet.contents:
                if isinstance(block, types.TextResourceContents):
                    texts_greet.append(block.text)
            print("[Read greeting://Alice]", " | ".join(texts_greet))
            # 读取模板资源 user://001 与 user://003
            for uid in ["001", "003"]:
                result_user = await session.read_resource(AnyUrl(f"user://{uid}"))
                texts_user = []
                for block in result_user.contents:
                    if isinstance(block, types.TextResourceContents):
                        texts_user.append(block.text)
                print(f"[Read user://{uid}]", " | ".join(texts_user))


if __name__ == "__main__":
    asyncio.run(main())

3. resources_server.py #

resources_server.py

# 导入 FastMCP 高层服务器类
from mcp_lite.server.fastmcp import FastMCP
# 导入 json 模块，用于序列化配置数据
import json
# 导入 sys 模块，用于访问标准流
import sys
# 创建 FastMCP 实例，命名为 Resources MCP Server
mcp = FastMCP(name="Resources MCP Server")

# 通过 @mcp.resource 装饰器注册静态资源，URI 为 config://settings，指定 MIME 类型为 application/json
@mcp.resource("config://settings", mime_type="application/json")
# 定义静态资源处理函数，不带参数，返回配置的 JSON 字符串
def get_settings() -> str:
    # 返回应用的 JSON 配置文本，包含主题、语言和调试开关
    return json.dumps(
        {"theme": "dark", "language": "zh-CN", "debug": False},
        ensure_ascii=False,
        indent=2,
    )

# 注册模板资源，URI 为 greeting://{name}，支持动态参数 name，MIME 类型为 text/plain
@mcp.resource("greeting://{name}", mime_type="text/plain")
# 定义模板资源处理函数，参数 name 必须和 URI 占位符一致
def greeting(name: str) -> str:
    # 根据传入的 name 返回个性化问候语
    return f"Hello, {name}! 欢迎使用 MCP 资源。"

# 注册另一个模板资源，URI 为 user://{id}，MIME 类型为 text/plain
@mcp.resource("user://{id}", mime_type="text/plain")
# 定义用户信息资源处理函数，参数 id 必须和 URI 占位符一致
def user_profile(id: str) -> str:
    # 判断 id 是否为 001，如果是则返回 Alice 的信息
    if id == "001":
        return "用户 001：Alice，角色：admin"
    # 判断 id 是否为 002，如果是则返回 Bob 的信息
    if id == "002":
        return "用户 002：Bob，角色：viewer"
    # 其他情况返回未找到用户的提示信息
    return f"未找到用户 {id}"

# 如果当前模块作为主程序运行
if __name__ == "__main__":
    # 检查 sys.stdout 是否支持 reconfigure 方法（用于设置编码）
    if hasattr(sys.stdout, 'reconfigure'):
        # "PYTHONIOENCODING": "utf-8"
        # 设置标准输出编码为 utf-8，避免中文乱码
        sys.stdout.reconfigure(encoding='utf-8')
        # 设置标准输入编码为 utf-8
        sys.stdin.reconfigure(encoding='utf-8')
    # 使用 stdio 作为通信方式启动 MCP 服务器
    mcp.run(transport="stdio")

官方代码

# 导入 FastMCP 高层服务器类
from mcp.server.fastmcp import FastMCP
# 导入 json 模块，用于序列化配置数据
import json
# 导入 sys 模块，用于访问标准流
import sys
# 创建 FastMCP 实例，命名为 Resources MCP Server
mcp = FastMCP(name="Resources MCP Server")

# 通过 @mcp.resource 装饰器注册静态资源，URI 为 config://settings，指定 MIME 类型为 application/json
@mcp.resource("config://settings", mime_type="application/json")
# 定义静态资源处理函数，不带参数，返回配置的 JSON 字符串
async def get_settings() -> str:
    # 返回应用的 JSON 配置文本，包含主题、语言和调试开关
    return json.dumps(
        {"theme": "dark", "language": "zh-CN", "debug": False},
        ensure_ascii=False,
        indent=2,
    )

# 注册模板资源，URI 为 greeting://{name}，支持动态参数 name，MIME 类型为 text/plain
@mcp.resource("greeting://{name}", mime_type="text/plain")
# 定义模板资源处理函数，参数 name 必须和 URI 占位符一致
async def greeting(name: str) -> str:
    # 根据传入的 name 返回个性化问候语
    return f"Hello, {name}! 欢迎使用 MCP 资源。"

# 注册另一个模板资源，URI 为 user://{id}，MIME 类型为 text/plain
@mcp.resource("user://{id}", mime_type="text/plain")
# 定义用户信息资源处理函数，参数 id 必须和 URI 占位符一致
async def user_profile(id: str) -> str:
    # 判断 id 是否为 001，如果是则返回 Alice 的信息
    if id == "001":
        return "用户 001：Alice，角色：admin"
    # 判断 id 是否为 002，如果是则返回 Bob 的信息
    if id == "002":
        return "用户 002：Bob，角色：viewer"
    # 其他情况返回未找到用户的提示信息
    return f"未找到用户 {id}"

# 如果当前模块作为主程序运行
if __name__ == "__main__":
    # 检查 sys.stdout 是否支持 reconfigure 方法（用于设置编码）
    if hasattr(sys.stdout, 'reconfigure'):
        # "PYTHONIOENCODING": "utf-8"
        # 设置标准输出编码为 utf-8，避免中文乱码
        sys.stdout.reconfigure(encoding='utf-8')
        # 设置标准输入编码为 utf-8
        sys.stdin.reconfigure(encoding='utf-8')
    # 使用 stdio 作为通信方式启动 MCP 服务器
    mcp.run(transport="stdio")

4. init.py #

mcp_lite/init.py

# 导入标准输入输出服务器参数类
from mcp_lite.client.stdio import StdioServerParameters
# 导入客户端会话类
from mcp_lite.client.session import ClientSession
# 导入类型模块
+from mcp_lite import types

# 设置该模块导出的内容
+__all__ = ["StdioServerParameters", "ClientSession", "types"]

5. session.py #

mcp_lite/client/session.py

import sys
# 导入 SessionMessage 类
from mcp_lite.message import SessionMessage
# 从 mcp_lite.types 模块导入相关类型和常量
from mcp_lite.types import (
+   InitializeRequestParams,
+   InitializeRequest,
+   LATEST_PROTOCOL_VERSION,
+   ClientCapabilities,
+   Implementation,
+   InitializedNotification,
+   InitializeResult,
+   JSONRPCRequest,
+   JSONRPCResponse,
+   JSONRPCError,
+   JSONRPCNotification,
+   ListToolsRequest,
+   ListToolsResult,
+   CallToolResult,
+   CallToolRequest,
+   CallToolRequestParams,
+   ListResourcesRequest,
+   ListResourcesResult,
+   ListResourceTemplatesRequest,
+   ListResourceTemplatesResult,
+   ReadResourceRequest,
+   ReadResourceRequestParams,
+   ReadResourceResult,
)
# 定义 MCPError 异常类，用于表示协议相关错误
class MCPError(Exception):
    # 初始化方法，接收错误码、错误信息和可选的数据
    def __init__(self, code, message, data=None):
        # 将错误码、错误信息、附加数据赋值为实例属性
        self.code, self.message, self.data = code, message, data
        # 调用父类 Exception 的初始化，只传递错误信息
        super().__init__(message)

    # 类方法，利用 JSONRPCError 对象创建 MCPError 实例
    @classmethod
    def from_jsonrpc(cls, err):
        # 从 JSONRPCError.error 对象中取出 code/message/data 创建 MCPError
        return cls(err.error.code, err.error.message, err.error.data)
# 定义 ClientSession 类，用于描述客户端会话
class ClientSession:
    # 构造方法，接收读取流和写入流
    def __init__(self, read_stream, write_stream):
        # 赋值读取流和写入流
        self._read, self._write = read_stream, write_stream
        # 初始化请求 id 计数器为 0
        self._req_id = 0
    # 内部方法，发送请求并同步等待响应
    def _request(self, req):
        # 当前请求 id
        rid = self._req_id
        # 自增请求 id，确保下次唯一
        self._req_id += 1
        # 对请求对象进行序列化，转为 dict
        d = req.model_dump(by_alias=True, mode="json", exclude_none=True)
        # 构建 JSONRPCRequest，再封装入 SessionMessage，通过写入流发送
        jreq = JSONRPCRequest(jsonrpc="2.0", id=rid, method=d["method"], params=d.get("params"))
        # 打印请求信息到标准错误流
        print("[Client] Request:", jreq.model_dump_json(by_alias=True, exclude_unset=True), file=sys.stderr)
        # 将请求对象封装成 SessionMessage 后发送
        self._write.send(SessionMessage(message=jreq))
        # 循环等待响应
        while True:
            # 从读取流取出一条消息
            msg = self._read.get()
            # 若取到 None，说明连接断开，抛出异常
            if msg is None:
                raise MCPError(-32000, "Connection closed")
            # 若消息类型不是 SessionMessage，忽略继续等
            if not isinstance(msg, SessionMessage):
                continue
            # 取消息内容
            m = msg.message
            # 若为 Response 或 Error，且 id 匹配
            if isinstance(m, (JSONRPCResponse, JSONRPCError)) and getattr(m, "id", None) == rid:
                print("[Client] Response:", m.model_dump_json(by_alias=True, exclude_unset=True), file=sys.stderr)
                # 若为错误，抛出 MCPError 异常
                if isinstance(m, JSONRPCError):
                    raise MCPError.from_jsonrpc(m)
                # 为正常响应则返回结果数据
                return m.result
    # 内部方法，发送通知消息
    def _notify(self, n):
        # 通知对象序列化为 dict
        d = n.model_dump(by_alias=True, mode="json", exclude_none=True)
        # 封装为 JSONRPCNotification，再封装成 SessionMessage 后发出
        self._write.send(SessionMessage(
            message=JSONRPCNotification(
                jsonrpc="2.0",
                method=d["method"],
                params=d.get("params"),
            )
        ))            
    # 初始化过程，发送初始化请求，收到响应后再发初始化完成通知
    def initialize(self):
        # 构造初始化请求，并发送等待返回
        r = self._request(InitializeRequest(
            params=InitializeRequestParams(
                protocol_version=LATEST_PROTOCOL_VERSION,         # 协议版本号
                capabilities=ClientCapabilities(),                # 客户端能力
                client_info=Implementation(name="mcp_lite", version="0.1.0"),  # 客户端信息
            )
        ))
        # 发送"初始化完成"通知
        self._notify(InitializedNotification())
        # 用 InitializeResult 验证响应并返回
        return InitializeResult.model_validate(r, by_name=False)
    def list_tools(self):
        # 发送 ListToolsRequest 请求，使用 ListToolsResult 校验并返回
        return ListToolsResult.model_validate(self._request(ListToolsRequest()), by_name=False)   
    # 调用指定工具方法，传入工具名和参数
    def call_tool(self, name, arguments=None):
        # 构造 CallToolRequest 并发送，请求参数包含工具名和参数
        return CallToolResult.model_validate(
+           self._request(CallToolRequest(
+               params=CallToolRequestParams(name=name, arguments=arguments or {})
+           )),
+           by_name=False,
+       )

    # 列出已注册的静态资源
+   def list_resources(self):
        # 调用 _request 方法发送 ListResourcesRequest 请求
        # 使用 ListResourcesResult 的 model_validate 进行结果校验和结构化
+       return ListResourcesResult.model_validate(
+           self._request(ListResourcesRequest()),
+           by_name=False,
+       )

    # 列出资源模板（带占位符的资源）
+   def list_resource_templates(self):
        # 调用 _request 方法发送 ListResourceTemplatesRequest 请求
        # 使用 ListResourceTemplatesResult 的 model_validate 进行结果校验和结构化
+       return ListResourceTemplatesResult.model_validate(
+           self._request(ListResourceTemplatesRequest()),
+           by_name=False,
+       )

    # 读取指定 URI 的资源内容
+   def read_resource(self, uri):
        # uri 可为 str 或 AnyUrl 类型
        # 构造 ReadResourceRequestParams 并发送 ReadResourceRequest 请求
        # 使用 ReadResourceResult 的 model_validate 进行结果校验和结构化
+       return ReadResourceResult.model_validate(
+           self._request(ReadResourceRequest(params=ReadResourceRequestParams(uri=str(uri)))),
+           by_name=False,
+       )

6. fastmcp.py #

mcp_lite/server/fastmcp.py

# 导入 sys 模块，用于标准输入输出操作
import sys
# 导入 base64 模块，用于二进制资源编码
+import base64
# 导入 re 模块，用于 URI 模板匹配
+import re
# 从 urllib.parse 导入 unquote，用于 URL 解码
+from urllib.parse import unquote
# 从 mcp_lite.message 模块导入 SessionMessage 类
from mcp_lite.message import SessionMessage
# 从 mcp_lite.server 模块导入 stdio 子模块
from mcp_lite.server import stdio
# 导入 inspect 用于函数签名分析
import inspect
# 导入 asyncio 用于异步操作
import asyncio
# 从 mcp_lite.types 模块导入所有相关类型和类
+from mcp_lite.types import (                          # 从 mcp_lite.types 模块导入以下类型和类
+   JSONRPCRequest,                                   # JSONRPC 请求结构体
+   JSONRPCError,                                     # JSONRPC 错误响应结构体
+   ErrorData,                                        # 错误数据结构体
+   InitializeResult,                                 # 初始化响应结构体
+   LATEST_PROTOCOL_VERSION,                          # 最新协议版本常量
+   ToolsCapability,                                  # 工具能力描述结构体
+   ResourcesCapability,                              # 资源能力描述结构体
+   ServerCapabilities,                               # 服务器能力结构体
+   Implementation,                                   # 实现信息结构体
+   JSONRPCResponse,                                  # JSONRPC 响应结构体
+   ListToolsResult,                                  # 工具列表响应结构体
+   Tool,                                             # 工具描述结构体
+   CallToolRequestParams,                            # 调用工具请求参数结构体
+   CallToolResult,                                   # 工具调用响应结构体
+   Resource,                                         # 资源元数据结构体
+   ResourceTemplate,                                 # 资源模板结构体
+   ListResourcesResult,                              # 资源列表响应结构体
+   ListResourceTemplatesResult,                      # 资源模板列表响应结构体
+   ReadResourceRequestParams,                        # 读资源请求参数结构体
+   ReadResourceResult,                               # 读资源响应结构体
+   TextResourceContents,                             # 文本型资源内容结构体
+   BlobResourceContents,                             # 二进制型资源内容结构体
)
# 内部函数：根据函数定义提取参数信息并生成 schema
def _schema(fn):
    # 获取函数签名
   sig = inspect.signature(fn)
    # 用于存储参数属性的字典
   props = {}
    # 用于存储必需参数名的列表
   req = []
    # 遍历所有参数
   for n, p in sig.parameters.items():
        # 跳过以 "_" 开头的参数
       if n.startswith("_"):
           continue
        # 所有参数类型都设为 string，title 为参数名
       props[n] = {"type": "string", "title": n}
        # 如果参数没有默认值，则为必需参数
       if p.default is inspect.Parameter.empty:
           req.append(n)
    # 返回对象类型 schema，包括属性和必需项目
   return {"type": "object", "properties": props, "required": req}
# 内部类：用于封装工具函数
class _Tool:
    # 初始化方法，接收目标函数、工具名和描述
   def __init__(self, fn, name=None, desc=None):
        # 保存原始函数引用
       self.fn = fn
        # 工具名称，优先采用传参，否则用函数名
       self.name = name or fn.__name__
        # 工具描述，优先采用传参，否则用函数 docstring
       self.desc = desc or (fn.__doc__ or "").strip()
        # 自动生成入参 schema
       self.schema = _schema(fn)
        # 判断函数是否为异步（协程）函数
       self.async_fn = asyncio.iscoroutinefunction(fn)

    # 转换为 Tool 类型的对象
   def to_tool(self):
        # 返回 Tool 类型实例
       return Tool(name=self.name, description=self.desc or None, input_schema=self.schema)

    # 执行工具函数，参数为字典类型
   def run(self, args):
        # 如果是异步函数则使用 asyncio.run 执行，否则同步调用
       if self.async_fn:
           return asyncio.run(self.fn(**args))
       return self.fn(**args)


# 内部类：用于封装静态资源
+class _Resource:
    # 构造方法，保存资源 URI、回调函数和 MIME 类型，检查函数是否异步
+   def __init__(self, uri, fn, mime_type="text/plain"):
        # 保存资源 URI
+       self.uri = uri
        # 保存资源内容生成函数
+       self.fn = fn
        # 保存资源 MIME 类型
+       self.mime_type = mime_type
        # 检查并保存函数是否为异步函数
+       self.async_fn = asyncio.iscoroutinefunction(fn)

    # 执行资源内容生成函数，支持同步/异步
+   def run(self):
        # 若为异步函数则通过 asyncio 运行
+       if self.async_fn:
+           return asyncio.run(self.fn())
        # 否则直接同步调用
+       return self.fn()


# 内部类：用于封装模板资源
+class _ResourceTemplate:
    # 构造方法，保存 URI 模板、处理函数、MIME 类型，并提取模板参数名
+   def __init__(self, uri_template, fn, mime_type="text/plain"):
        # 保存 URI 模板
+       self.uri_template = uri_template
        # 保存用于动态生成资源内容的函数
+       self.fn = fn
        # 保存资源 MIME 类型
+       self.mime_type = mime_type
        # 检查函数是否为异步函数
+       self.async_fn = asyncio.iscoroutinefunction(fn)
        # 获取函数签名
+       sig = inspect.signature(fn)
        # 提取所有非下划线开头的参数名，作为模板变量名
+       self.param_names = [n for n in sig.parameters if not n.startswith("_")]

    # 匹配实际 URI 是否符合模板，并返回变量字典
+   def matches(self, uri):
        # 使用正则表达式将 URI 模板转换成命名捕获组的正则模式
+       pattern = self.uri_template.replace("{", "(?P<").replace("}", ">[^/]+)")
        # 尝试用给定的 URI 匹配该模式
+       m = re.match(f"^{pattern}$", uri)
        # 若匹配成功则返回捕获的参数字典，进行 URL 解码
+       if m:
+           return {k: unquote(v) for k, v in m.groupdict().items()}
        # 匹配失败返回 None
+       return None

    # 执行模板资源内容生成函数，支持异步/同步
+   def run(self, args):
        # 若为异步函数，通过 asyncio 运行
+       if self.async_fn:
+           return asyncio.run(self.fn(**args))
        # 否则直接同步调用
+       return self.fn(**args)


# 主服务类 FastMCP
class FastMCP:
    # 初始化方法，支持自定义服务器名称
    def __init__(self, name="mcp-server"):
        # 保存服务器名称
        self.name = name
        # 初始化工具池为字典
        self._tools = {}
        # 初始化静态资源池
+       self._resources = {}
        # 初始化资源模板池
+       self._resource_templates = {}

    # 工具注册装饰器，支持自定义工具名和描述
    def tool(self, name=None, description=None):
        # 装饰器实际函数
       def deco(fn):
            # 创建 _Tool 实例
           t = _Tool(fn, name, description)
            # 注册进工具池
           self._tools[t.name] = t
            # 返回原始函数
           return fn
        # 返回装饰器
+      return deco

    # 资源注册装饰器，支持静态资源和模板资源
+   def resource(self, uri, mime_type="text/plain"):
+       def deco(fn):
+           if "{" in uri and "}" in uri:
                # 模板资源
+               template = _ResourceTemplate(uri, fn, mime_type)
+               self._resource_templates[uri] = template
+           else:
                # 静态资源
+               res = _Resource(uri, fn, mime_type)
+               self._resources[uri] = res
+           return fn
+       return deco

    # 实际 JSONRPC 方法分发与业务处理
    def _handle(self, req):
        # 拆分 method、params、id
        method, params, rid = req.method, req.params or {}, req.id
        # 处理初始化请求
        if method == "initialize":
            # 组装初始化响应，包括协议版本、能力与服务信息
+           caps = ServerCapabilities(
+               tools=ToolsCapability(),
+               resources=ResourcesCapability(),
+           )
            r = InitializeResult(
                protocol_version=LATEST_PROTOCOL_VERSION,
+               capabilities=caps,
                server_info=Implementation(name=self.name, version="0.1.0"),
            )
            # 返回 JSONRPCResponse 包装的结果
            return JSONRPCResponse(jsonrpc="2.0", id=rid, result=r.model_dump(by_alias=True, exclude_none=True))
         # 处理工具列表请求
        if method == "tools/list":
            # 构建工具列表结果
           r = ListToolsResult(tools=[t.to_tool() for t in self._tools.values()])
            # 返回 JSONRPCResponse 包装的结果
           return JSONRPCResponse(jsonrpc="2.0", id=rid, result=r.model_dump(by_alias=True, exclude_none=True))     
          # 处理工具调用请求
        if method == "tools/call":
            # 参数校验及实例化为 CallToolRequestParams
           p = CallToolRequestParams.model_validate(params, by_name=False)
            # 查找对应工具
           t = self._tools.get(p.name)
            # 找不到工具时返回参数错误
           if not t:
               return JSONRPCError(jsonrpc="2.0", id=rid, error=ErrorData(code=-32602, message=f"Unknown tool: {p.name}"))
           try:
                # 执行工具函数
               out = t.run(p.arguments or {})
                # 如果结果已经是 CallToolResult，直接拆取 content
               if isinstance(out, CallToolResult):
                   c = out.content
                # 返回字符串时，封装为文本类型内容列表
               elif isinstance(out, str):
                   c = [{"type": "text", "text": out}]
                # 结果为 None 时，返回空内容列表
               elif out is None:
                   c = []
                # 其它类型结果统一转为字符串文本
               else:
                   c = [{"type": "text", "text": str(out)}]
                # 构造 CallToolResult 作为最终响应体
               r = CallToolResult(content=c)
           except Exception as e:
                # 工具执行出错时，返回错误内容、标记 is_error
               r = CallToolResult(content=[{"type": "text", "text": str(e)}], is_error=True)
            # 返回 JSONRPCResponse 包装的工具执行结果
+          return JSONRPCResponse(jsonrpc="2.0", id=rid, result=r.model_dump(by_alias=True, exclude_none=True))

        # 处理 resources/list 请求
        # 判断请求方法是否为 "resources/list"
+       if method == "resources/list":
            # 遍历所有静态资源字典，将每个资源封装为 Resource 对象，组成资源列表
+           resources = [
+               Resource(uri=u, name=u, mime_type=r.mime_type)
+               for u, r in self._resources.items()
+           ]
            # 用资源列表生成 ListResourcesResult 响应体
+           r = ListResourcesResult(resources=resources)
            # 将响应体封装为 JSONRPCResponse 并返回
+           return JSONRPCResponse(jsonrpc="2.0", id=rid, result=r.model_dump(by_alias=True, exclude_none=True))

        # 处理 resources/templates/list 请求
        # 判断请求方法是否为 "resources/templates/list"
+       if method == "resources/templates/list":
            # 遍历所有资源模板字典，将每个模板封装为 ResourceTemplate 对象，组成模板列表
+           templates = [
+               ResourceTemplate(uri_template=u, name=u, mime_type=t.mime_type)
+               for u, t in self._resource_templates.items()
+           ]
            # 用模板列表生成 ListResourceTemplatesResult 响应体
+           r = ListResourceTemplatesResult(resource_templates=templates)
            # 封装为 JSONRPCResponse 并返回
+           return JSONRPCResponse(jsonrpc="2.0", id=rid, result=r.model_dump(by_alias=True, exclude_none=True))

        # 处理 resources/read 请求
        # 判断请求方法是否为 "resources/read"
+       if method == "resources/read":
            # 用 ReadResourceRequestParams 校验并实例化参数
+           p = ReadResourceRequestParams.model_validate(params, by_name=False)
            # 将 uri 转为字符串
+           uri_str = str(p.uri)
+           try:
                # 先查找静态资源
+               if res := self._resources.get(uri_str):
                    # 调用静态资源的 run 方法获取输出内容
+                   out = res.run()
                    # 封装为 TextResourceContents（文本类型资源内容）
+                   content = TextResourceContents(uri=uri_str, text=str(out), mime_type=res.mime_type)
                    # 构造 ReadResourceResult 响应体
+                   r = ReadResourceResult(contents=[content])
                    # 封装为 JSONRPCResponse 并返回
+                   return JSONRPCResponse(jsonrpc="2.0", id=rid, result=r.model_dump(by_alias=True, exclude_none=True))
                # 如果未找到静态资源，则依次查找所有资源模板
+               for template in self._resource_templates.values():
                    # template.matches 返回匹配参数字典，匹配成功则进入分支
+                   if args := template.matches(uri_str):
                        # 执行模板资源的 run 方法，传入提取的参数
+                       out = template.run(args)
                        # 如果输出为 bytes 类型，封装为二进制内容
+                       if isinstance(out, bytes):
+                           content = BlobResourceContents(
+                               uri=uri_str,
+                               blob=base64.b64encode(out).decode(),
+                               mime_type=template.mime_type or "application/octet-stream",
+                           )
+                       else:
                            # 其它类型一律作为文本类型内容封装
+                           content = TextResourceContents(
+                               uri=uri_str,
+                               text=str(out),
+                               mime_type=template.mime_type or "text/plain",
+                           )
                        # 构造 ReadResourceResult 响应体
+                       r = ReadResourceResult(contents=[content])
                        # 封装为 JSONRPCResponse 并返回
+                       return JSONRPCResponse(jsonrpc="2.0", id=rid, result=r.model_dump(by_alias=True, exclude_none=True))
                # 静态资源和模板资源都未命中，返回资源不存在的错误
+               return JSONRPCError(jsonrpc="2.0", id=rid, error=ErrorData(code=-32602, message=f"Unknown resource: {uri_str}"))
+           except Exception as e:
                # 发生异常，返回内部错误（服务端错误，code -32603）
+               return JSONRPCError(jsonrpc="2.0", id=rid, error=ErrorData(code=-32603, message=str(e)))

        # 不支持的方法，返回 "方法未找到" 错误（code -32601）
        return JSONRPCError(jsonrpc="2.0", id=rid, error=ErrorData(code=-32601, message=f"Method not found: {method}"))    
    # 处理 SessionMessage 消息，主要路由 JSONRPCRequest
    def _handle_msg(self, msg):
        # 判断消息类型不是 SessionMessage 则忽略
        if not isinstance(msg, SessionMessage):
            return None
        # 获取消息体
        m = msg.message
        # 只处理 JSONRPCRequest 类型（通知类消息如 notifications/initialized 不响应）
+       if not isinstance(m, JSONRPCRequest) or getattr(m, "id", None) is None:
            return None
        print("[Server] Request:", m.model_dump_json(by_alias=True, exclude_unset=True), file=sys.stderr)    
        try:
            # 调用实际业务处理方法
            resp = self._handle(m)
            print("[Server] Response:", resp.model_dump_json(by_alias=True, exclude_unset=True), file=sys.stderr)
            return resp
        except Exception as e:
            # 捕获异常，返回标准 JSONRPCError（code -32603，内部错误）
            err = JSONRPCError(jsonrpc="2.0", id=m.id, error=ErrorData(code=-32603, message=str(e)))
            print("[Server] Response (error):", err.model_dump_json(by_alias=True, exclude_unset=True), file=sys.stderr)
            return err
    # 运行服务方法，默认采用 stdio 传输方式
    def run(self, transport="stdio"):
        # 仅支持 stdio，其他方式抛出异常
        if transport != "stdio":
            raise ValueError(f"unsupported transport: {transport}")
        # 启动 stdio server，当前对象 _handle_msg 作为消息处理回调
        stdio.stdio_server(self._handle_msg)    

7. types.py #

mcp_lite/types.py

# 从 pydantic 导入 BaseModel、ConfigDict、Field、TypeAdapter
from pydantic import BaseModel, ConfigDict, Field, TypeAdapter
# 导入 pydantic 的 to_camel 驼峰命名生成器
from pydantic.alias_generators import to_camel
# 导入 Any 和 Literal 类型注解
from typing import Any, Literal
# 定义 RequestId 类型，既可以是 int 也可以是 str
RequestId = int | str
# 定义 MCP 的基础模型类，支持驼峰命名和按名称填充
class MCPModel(BaseModel):
    # 指定 Pydantic 模型配置：使用驼峰形式命名和按名称填充
   model_config = ConfigDict(alias_generator=to_camel, populate_by_name=True)
# 定义客户端能力结构体
class ClientCapabilities(MCPModel):
    # 可选字段：客户端实验性能力扩展，键为 str，值为字典
   experimental: dict[str, dict[str, Any]] | None = None
# 定义服务器或客户端的实现信息结构体
class Implementation(MCPModel):
    # 实现的名称
   name: str = ""
    # 实现的版本号
   version: str = ""
    # 可选字段：人类可读的标题
   title: str | None = None
    # 可选字段：实现的描述
   description: str | None = None    
# 定义初始化请求参数结构体
class InitializeRequestParams(MCPModel):
    # 协议版本号
   protocol_version: str = ""
    # 可选字段：客户端能力描述
   capabilities: ClientCapabilities = None
    # 可选字段：客户端实现信息
   client_info: Implementation = None
# 定义初始化请求结构体
class InitializeRequest(MCPModel):
    # 方法名，固定为 "initialize"
   method: Literal["initialize"] = "initialize"
    # 可选字段：参数，类型为 InitializeRequestParams
   params: InitializeRequestParams = None    
# 当前协议的最新版本号
LATEST_PROTOCOL_VERSION = "2024-11-05"
# 定义工具相关能力结构体
class ToolsCapability(MCPModel):
    # 工具列表是否发生变化，可选布尔类型
   list_changed: bool | None = None
# 定义资源相关能力结构体
+class ResourcesCapability(MCPModel):
    # 是否支持资源订阅
+  subscribe: bool | None = None
    # 资源列表是否变化通知
+  list_changed: bool | None = None
# 定义服务端能力描述结构体
class ServerCapabilities(MCPModel):
    # 可选字段：实验性能力扩展
   experimental: dict[str, dict[str, Any]] | None = None
    # 可选字段：工具能力
   tools: ToolsCapability | None = None
    # 可选字段：资源能力
+  resources: ResourcesCapability | None = None
# 定义初始化响应结构体
class InitializeResult(MCPModel):
    # 协议版本号
   protocol_version: str = ""
    # 可选字段：服务端能力描述
   capabilities: ServerCapabilities = None
    # 可选字段：服务端实现信息
   server_info: Implementation = None
    # 可选字段：初始化说明信息
   instructions: str | None = None
# 定义客户端初始化完成通知结构体
class InitializedNotification(MCPModel):
    # 方法名，固定为 "notifications/initialized"
   method: Literal["notifications/initialized"] = "notifications/initialized"
    # 可选字段：通知参数，可以为字典或 None
   params: dict[str, Any] | None = None    
# 定义 JSONRPC 请求的数据结构
class JSONRPCRequest(BaseModel):
    # jsonrpc 协议版本，固定为 "2.0"
    jsonrpc: Literal["2.0"] = "2.0"
    # 请求 ID，可以为 int 或 str 类型
    id: RequestId = None
    # 方法名称，字符串类型
    method: str = ""
    # 方法参数，为一个字典或 None
    params: dict[str, Any] | None = None

# 定义 JSONRPC 通知的数据结构（没有 id 字段）
class JSONRPCNotification(BaseModel):
    # jsonrpc 协议版本，固定为 "2.0"
    jsonrpc: Literal["2.0"] = "2.0"
    # 通知的方法名称
    method: str = ""
    # 通知参数，可以为字典或者 None
    params: dict[str, Any] | None = None

# 定义 JSONRPC 响应的数据结构
class JSONRPCResponse(BaseModel):
    # jsonrpc 协议版本，固定为 "2.0"
    jsonrpc: Literal["2.0"] = "2.0"
    # 响应的 ID，需要与请求 ID 匹配
    id: RequestId = None
    # 响应结果，可以为字典或 None
    result: dict[str, Any] = None

# 定义错误的数据结构
class ErrorData(BaseModel):
    # 错误码，默认值为 0
    code: int = 0
    # 错误信息，默认值为空字符串
    message: str = ""
    # 附加的错误数据，可以为任意类型或 None
    data: Any = None

# 定义 JSONRPC 错误消息的数据结构
class JSONRPCError(BaseModel):
    # jsonrpc 协议版本，固定为 "2.0"
    jsonrpc: Literal["2.0"] = "2.0"
    # 错误对应的请求 ID，可以为 None
    id: RequestId | None = None
    # 错误的详细信息，类型为 ErrorData
    error: ErrorData = None
# 定义所有 JSONRPC 消息的联合类型
JSONRPCMessage = JSONRPCRequest | JSONRPCNotification | JSONRPCResponse | JSONRPCError
# 定义 JSONRPC 消息适配器，用于类型自动推断和校验
jsonrpc_message_adapter = TypeAdapter(JSONRPCMessage)
# 定义工具描述数据结构体
class Tool(MCPModel):
    # 工具名称
   name: str = ""
    # 可选字段：工具描述
   description: str | None = None
    # 工具输入参数的 schema，默认为空字典
   input_schema: dict[str, Any] = Field(default_factory=dict)
    # 可选字段：工具输出 schema
   output_schema: dict[str, Any] | None = None
# 定义获取工具列表请求结构体
class ListToolsRequest(MCPModel):
    # 方法名，固定为 "tools/list"
   method: Literal["tools/list"] = "tools/list"
    # 可选字段：参数，可以为字典或 None
   params: dict[str, Any] | None = None

# 定义获取工具列表响应结构体
class ListToolsResult(MCPModel):
    # 工具列表，默认为空列表
   tools: list[Tool] = []
    # 可选字段：分页游标，可为 None
   next_cursor: str | None = None

# 定义资源元数据结构体
+class Resource(MCPModel):
    # 资源 URI
+  uri: str = ""
    # 可选：资源名称
+  name: str | None = None
    # 可选：人类可读标题
+  title: str | None = None
    # 可选：资源描述
+  description: str | None = None
    # 可选：MIME 类型
+  mime_type: str | None = None

# 定义资源模板结构体
+class ResourceTemplate(MCPModel):
    # URI 模板，如 greeting://{name}
+  uri_template: str = ""
    # 可选：模板名称
+  name: str | None = None
    # 可选：人类可读标题
+  title: str | None = None
    # 可选：模板描述
+  description: str | None = None
    # 可选：MIME 类型
+  mime_type: str | None = None

# 定义 resources/list 请求结构体
+class ListResourcesRequest(MCPModel):
+  method: Literal["resources/list"] = "resources/list"
+  params: dict[str, Any] | None = None

# 定义 resources/list 响应结构体
+class ListResourcesResult(MCPModel):
+  resources: list[Resource] = []
+  next_cursor: str | None = None

# 定义 resources/templates/list 请求结构体
+class ListResourceTemplatesRequest(MCPModel):
+  method: Literal["resources/templates/list"] = "resources/templates/list"
+  params: dict[str, Any] | None = None

# 定义 resources/templates/list 响应结构体
+class ListResourceTemplatesResult(MCPModel):
+  resource_templates: list[ResourceTemplate] = []
+  next_cursor: str | None = None

# 定义 resources/read 请求参数结构体
+class ReadResourceRequestParams(MCPModel):
+  uri: str = ""

# 定义 resources/read 请求结构体
+class ReadResourceRequest(MCPModel):
+  method: Literal["resources/read"] = "resources/read"
+  params: ReadResourceRequestParams = None

# 定义资源内容基类（文本）
+class TextResourceContents(MCPModel):
+  uri: str = ""
+  mime_type: str | None = None
+  text: str = ""

# 定义资源内容基类（二进制）
+class BlobResourceContents(MCPModel):
+  uri: str = ""
+  mime_type: str | None = None
+  blob: str = ""  # base64 编码

# 定义 resources/read 响应结构体
+class ReadResourceResult(MCPModel):
+  contents: list[TextResourceContents | BlobResourceContents] = []

# 定义调用工具请求参数结构体
class CallToolRequestParams(MCPModel):
    # 工具名称
   name: str = ""
    # 可选字段：输入参数，为字典类型或 None
   arguments: dict[str, Any] | None = None        
# 定义调用工具请求结构体
class CallToolRequest(MCPModel):
    # 方法名，固定为 "tools/call"
   method: Literal["tools/call"] = "tools/call"
    # 可选字段：参数，类型为 CallToolRequestParams
   params: CallToolRequestParams = None
# 定义调用工具响应结构体
class CallToolResult(MCPModel):
    # 返回内容，为一个列表，默认为空
   content: list = Field(default_factory=list)
    # 可选字段：结构化内容，为字典或 None
   structured_content: dict[str, Any] | None = None
    # 是否为错误，布尔类型，默认 False
   is_error: bool = False       

8 工作流程 #

8.1 整体架构 #

资源功能在 mcp_lite 中实现 MCP 协议的 resources 能力，包括：

1. 静态资源：固定 URI（如 config://settings）
2. 模板资源：带占位符的 URI（如 user://{id}、greeting://{name}）

8.2 各模块修改说明 #

8.2.1. resources_client.py（客户端示例） #

• 使用 stdio_client 建立与服务器的 stdio 连接
• 创建 ClientSession 并 initialize()
• 调用 list_resources() 获取静态资源列表
• 调用 list_resource_templates() 获取模板列表
• 调用 read_resource(uri) 读取资源内容
• 通过 types.TextResourceContents 判断并提取文本内容

8.2.2. resources_server.py（服务端示例） #

• 使用 `@mcp.resource(uri, mime_type)` 注册资源
• 静态资源：config://settings，无参数
• 模板资源：greeting://{name}、user://{id}，参数与占位符对应
• 启动前通过 sys.stdout.reconfigure(encoding='utf-8') 保证中文不乱码

8.2.3. mcp_lite/client/session.py（客户端会话） #

新增三个方法：

8.3. mcp_lite/server/fastmcp.py（服务端核心） #

内部类：

• _Resource：静态资源，保存 uri、fn、mime_type，run() 无参调用
• _ResourceTemplate：模板资源，保存 uri_template、fn、mime_type，matches(uri) 用正则提取参数，run(args) 带参调用

装饰器：

@mcp.resource(uri, mime_type)
def handler(...): ...

• 若 uri 含 {...} → 注册为模板资源
• 否则 → 注册为静态资源

请求处理：

• resources/list：返回所有静态资源的 Resource 列表
• resources/templates/list：返回所有模板的 ResourceTemplate 列表
• resources/read：先查静态资源，再按模板匹配；支持文本和二进制（base64）

8.4 5. mcp_lite/types.py（类型定义） #

新增类型：

• ResourcesCapability：资源能力
• Resource / ResourceTemplate：资源与模板元数据
• ListResourcesRequest/Result、ListResourceTemplatesRequest/Result：列表请求/响应
• ReadResourceRequestParams、ReadResourceRequest：读资源请求
• TextResourceContents / BlobResourceContents：文本/二进制内容
• ReadResourceResult：读资源响应

8.5 资源读取流程 #

1. 客户端发送 resources/read，参数为 uri
2. 服务端：先查 _resources 静态资源
3. 若未命中 → 遍历 _resource_templates，用 matches(uri) 匹配
4. 匹配成功 → 用 run(args) 生成内容
5. 内容为 bytes → 返回 BlobResourceContents（base64）
6. 内容为 str → 返回 TextResourceContents

8.6 URI 模板匹配逻辑 #

模板 user://{id} 的匹配过程：

1. 正则：user://{id} → ^user://(?P<id>[^/]+)$
2. user://001 匹配成功，得到 {id: "001"}
3. 调用 user_profile(id="001") 生成内容

8.7 数据流概览 #

客户端 read_resource("user://001")
    → JSON-RPC: {"method":"resources/read","params":{"uri":"user://001"}}
    → stdio 传输
    → FastMCP._handle 解析
    → _ResourceTemplate.matches("user://001") → {id:"001"}
    → user_profile(id="001") → "用户 001：Alice，角色：admin"
    → TextResourceContents(text="...", mime_type="text/plain")
    → JSON-RPC Response
    → 客户端解析 contents，提取 text