2025-8-25-1

资源、提示词与工具

资源

资源（Resources）是模型上下文协议（MCP）的核心基元，允许服务器暴露可以被客户端读取并用作 LLM 交互上下文的数据和内容。资源代表 MCP 服务器希望提供给客户端的任何类型的数据。这可以包括：

1. 文件内容
2. 数据库记录
3. API 响应
4. 实时系统数据
5. 截图和图像
6. 日志文件
7. 以及更多

每个资源都由唯一的 URI（统一资源标识符）标识，并且可以包含文本或二进制数据。协议和路径结构由 MCP 服务器实现定义。服务器可以定义自己的自定义 URI 方案。

[协议]://[主机]/[路径]
[protocol]://[host]/[path]
# 例子：
# file:///home/user/documents/report.pdf
# postgres://database/customers/schema
# screen://localhost/display1

资源类型

资源主要包含两种类型的内容。

	文本资源	二进制资源
说明	文本资源包含 UTF-8 编码的文本数据。	二进制资源包含以 base64 编码的原始二进制数据。
适用于	源代码、配置文件、日志文件、JSON/XML数据、纯文本	图像、音频文件、视频文件、PDF、其他非文本格式

资源发现&读取(面向Client端)

Client端通过以下两种主要方法发现可用资源：

1. 直接资源：服务器通过 resources/list 端点暴露一个具体的资源列表。 每个资源包括：

   {
     uri: string;           // 资源的唯一标识符
     name: string;          // 人类可读的名称
     description?: string;  // 可选的描述
     mimeType?: string;     // 可选的 MIME 类型
   }

2. 资源模板：对于动态资源，服务器可以暴露 URI 模板，客户端可以使用这些模板来构造有效的资源 URI：

   {
     uriTemplate: string;   // 遵循 RFC 6570 的 URI 模板
     name: string;          // 此类型的人类可读名称
     description?: string;  // 可选的描述
     mimeType?: string;     // 所有匹配资源的的可选 MIME 类型
   }

要读取资源，客户端可以使用资源 URI 发出 resources/read 请求。服务器将返回资源内容列表：

{
  contents: [
    {
      uri: string;        // 资源的 URI
      mimeType?: string;  // 可选的 MIME 类型

      // 二选一:
      text?: string;      // 适用于文本资源
      blob?: string;      // 适用于二进制资源（base64 编码）
    }
  ]
}

服务器可能会在对一个 resources/read 请求的响应中返回多个资源。 例如，当读取目录时，可以使用此方法返回目录中的文件列表。

资源更新

两种机制支持资源的实时更新：

1. 列表更改：当可用资源列表更改时，服务器可以通过 notifications/resources/list_changed 通知客户端。
2. 内容更改：客户端可以订阅特定资源的更新：
   1. 客户端使用资源 URI 发送 resources/subscribe
   2. 当资源更改时，服务器发送 notifications/resources/updated
   3. 客户端可以使用 resources/read 获取最新内容
   4. 客户端可以使用 resources/unsubscribe 取消订阅

示例

app = Server("example-server")

@app.list_resources()
async def list_resources() -> list[types.Resource]:
    return [
        types.Resource(
            uri="file:///logs/app.log",
            name="Application Logs",
            mimeType="text/plain"
            # 最好给出description
        )
    ]

@app.read_resource()
async def read_resource(uri: AnyUrl) -> str:
    if str(uri) == "file:///logs/app.log":
        log_contents = await read_log_file()
        return log_contents

    raise ValueError("Resource not found")

# Start server
async with stdio_server() as streams:
    await app.run(
        streams[0],
        streams[1],
        app.create_initialization_options()
    )

客户端使用资源的正常流程是：

1. 先调用 list_resources 拿到 “资源目录”，知道 “有什么资源、地址是什么”；
2. 再用目录中的 uri 调用 read_resource，获取具体内容。

• list_resources：告诉客户端 “有哪些资源”（类似图书馆的 “藏书目录”）。
• read_resource：让客户端 “读取某个具体资源的内容”（类似根据目录找到书后，翻开书读内容）。比如获取数据的日记文件、数据库记录。

提示词

MCP 提示词把 “重复的指令、标准化的流程、需要的参数” 提前定义成模板，用户调用模板时只需填少量参数（比如 “代码内容”“编程语言”），就能快速生成 LLM 能理解的完整交互内容。

工具

工具定义结构

{
  name: string;          // 工具的唯一标识符
  description?: string;  // 人类可读的描述
  inputSchema: {         // 工具参数的 JSON Schema
    type: "object",
    properties: { ... }  // 工具特定的参数
  }
}

工具的发现与调用：

1. list_tools()
2. call_tools()

app = Server("example-server")

@app.list_tools()  # MCP Python SDK 的装饰器：注册“工具列表”接口
async def list_tools() -> list[types.Tool]:  # 异步函数，返回 Tool 类型的列表
    # -> list[types.Tool]：类型注解，明确返回值是 types.Tool（MCP Python SDK 定义的工具类）的列表，确保类型安全。
    return [
        types.Tool(
            name="calculate_sum",
            description="Add two numbers together",
            inputSchema={
                "type": "object",
                "properties": {
                    "a": {"type": "number"},
                    "b": {"type": "number"}
                },
                "required": ["a", "b"]
            }
        )
    ]

@app.call_tool()  # MCP Python SDK 的装饰器：注册“工具调用”接口
async def call_tool(
    name: str,  # 客户端要调用的工具名称（如“calculate_sum”）
    arguments: dict  # 客户端传的参数对象（如 {"a": 1, "b": 2}）
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
    if name == "calculate_sum": # 判断工具名是否为 calculate_sum
        a = arguments["a"]
        b = arguments["b"]
        result = a + b
        # 返回结果：按 MCP 格式包装成 TextContent 实例（文本类型的内容）
        return [types.TextContent(type="text", text=str(result))]
    # 工具不存在时抛出错误
    raise ValueError(f"Tool not found: {name}")

工具示例

系统操作

与本地系统交互的工具：客户端（如大模型、远程服务）直接调用 本地服务器的 shell 命令，实现对本地系统的控制（如查看文件、启动进程、安装软件）。

能够实际使用的场景：

1. 客户端想 “查看服务器 /logs 目录下的文件”，调用时传：
   {"command": "ls", "args": ["-l", "/logs"]}
2. 服务器执行 ls -l /logs 命令，返回结果（如文件列表）给客户端；
3. 风险提示：这是 高权限工具（能执行系统命令），必须加严格的访问控制（如仅管理员可调用），防止恶意操作（如 rm -rf / 删除系统文件）。

{
  "name": "execute_command",  // 工具唯一名称：明确是“执行命令”
  "description": "运行 shell 命令",  // 功能描述：客户端一看就知道能“跑 shell 命令”
  "inputSchema": {  // 参数约束：告诉客户端“该传什么格式的参数”
    "type": "object",  // 参数整体是一个对象
    "properties": {  // 对象包含两个属性（参数）
      "command": { "type": "string" },  // 必传？非必传？看 required（这里没写，默认非必传，但实际用中是核心）
      // 说明：要执行的 shell 命令本体（如 "ls" 查看目录、"python" 启动Python解释器）
      "args": { "type": "array", "items": { "type": "string" } }
      // 说明：命令的参数列表（数组，每个元素是字符串），如 "ls -l /home" 中，args 是 ["-l", "/home"]
    }
    // 注意：这里缺少 required 字段，实际生产中应加 "required": ["command"]（命令必须传）
  }
}

API 集成

包装外部 API 的工具：将 外部第三方 API（GitHub API）包装成 MCP 工具，让客户端无需直接对接 GitHub API（不用处理鉴权、请求格式），通过 MCP 就能调用 GitHub 功能。

{
  "name": "github_create_issue",  // 工具唯一名称：明确是“创建 GitHub Issue”
  "description": "创建 GitHub issue",  // 功能描述：客户端知道能“在 GitHub 上建 Issue”
  "inputSchema": {
    "type": "object",
    "properties": {
      "title": { "type": "string" },  // Issue 的标题（如 "修复登录页面按钮不显示问题"）
      "body": { "type": "string" },   // Issue 的详细内容（如 "点击登录按钮后无响应，浏览器控制台报错：xxx"）
      "labels": { "type": "array", "items": { "type": "string" } }
      // Issue 的标签（数组，如 ["bug", "frontend"]，用于分类 Issue）
    }
    // 实际生产中应加 "required": ["title", "body"]（标题和内容必传，标签可选）
  }
}

实际用法场景：

客户端（如开发工具）想 “在仓库modelcontextprotocol/python-sdk上创建 Bug Issue”，调用时传：

{
  "title": "修复 calculate_sum 工具参数校验bug",
  "body": "当传非数字参数时，工具未报错，直接返回 NaN",
  "labels": ["bug", "tools"]
}

服务器收到请求后，会：

1. 用预先配置的 GitHub Token 进行鉴权（客户端不用管鉴权）；
2. 按 GitHub API 格式组装请求（如 POST https://api.github.com/repos/modelcontextprotocol/python-sdk/issues）；
3. 将 GitHub 返回的 Issue 链接（如 https://github.com/.../issues/123）返回给客户端；

这样的话，客户端无需了解 GitHub API 的细节（鉴权方式、请求 URL、格式），只需按 MCP 工具参数传值，降低集成成本。

数据处理

转换或分析数据的工具：对 本地或服务器可访问的 CSV 文件 进行标准化数据处理（求和、求平均、计数），让客户端无需自己解析 CSV、写计算逻辑，直接获取分析结果。

{
  "name": "analyze_csv",  // 工具唯一名称：明确是“分析 CSV 文件”
  "description": "分析 CSV 文件",  // 功能描述：客户端知道能“处理 CSV 数据”
  "inputSchema": {
    "type": "object",
    "properties": {
      "filepath": { "type": "string" },  // CSV 文件在服务器上的路径（如 "/data/sales.csv"）
      "operations": {  // 要执行的分析操作（数组，只能选枚举值）
        "type": "array",
        "items": {
          "enum": ["sum", "average", "count"]  // 允许的操作：求和、求平均、计数
        }
      }
    }
    // 实际生产中应加 "required": ["filepath", "operations"]（文件路径和操作必传）
  }
}

实际用法场景：服务器上有 sales.csv（内容：month,sales\n1,100\n2,200\n3,300），客户端想 “计算销售额的总和、平均值”，调用时传：
{"filepath": "/data/sales.csv", "operations": ["sum", "average"]}

服务器执行逻辑：

1. 读取 sales.csv 文件，解析 sales 列的数据（[100, 200, 300]）；
2. 执行 sum 操作：100+200+300=600；
3. 执行 average 操作：600/3=200；
4. 返回结果：{"sum": 600, "average": 200}；

此工具的用途：客户端不用自己写 “读取 CSV→解析数据→计算” 的代码，尤其适合非技术用户或大模型（只需指定文件和操作，不用处理数据逻辑）。

传输

MCP 协议中 “基于内存对象流的异步传输层管理器”：负责管理客户端与服务器之间的消息收发生命周期（启动消息处理、提供发送通道、自动清理资源）。

传输层中概念

参数	比喻说明
read_stream（读流）	相当于 “收件箱”，用来接收对方发来的消息（如客户端收到服务器的工具调用结果）；
write_stream（写流）	相当于 “发件箱”，用来发送自己的消息（如客户端给服务器发 “调用工具” 的请求）；
process_messages（消息处理函数）	相当于 “快递分拣员”，收到消息后拆解、处理（如服务器收到工具调用请求后，触发对应的工具逻辑）；

下面的示例代码相当于 “收发站管理员”，负责开门（启动分拣员）、提供发件箱、关门时清理（关闭箱子、辞退分拣员）。

@contextmanager # @contextmanager：Python 的上下文管理器装饰器（这里是异步版本，需搭配 async），作用是让函数支持 async with 语法，自动处理 “进入上下文” 和 “退出上下文” 的逻辑（比如进入时启动任务，退出时清理资源）。
async def create_transport(
    read_stream: MemoryObjectReceiveStream[JSONRPCMessage | Exception],
    write_stream: MemoryObjectSendStream[JSONRPCMessage]
):
    """
    这是 MCP 的传输层接口，两个参数的作用：
        read_stream：用于读取 “进来的消息”（对方发来的）；
        write_stream：用于写入 “出去的消息”（自己要发的）。
    """
    async with anyio.create_task_group() as tg:
        try:
            # Start processing messages
            tg.start_soon(lambda: process_messages(read_stream))

            # Send messages
            async with write_stream:
                yield write_stream

        except Exception as exc:
            # Handle errors
            raise exc
        finally:
            # Clean up
            tg.cancel_scope.cancel()
            await write_stream.aclose()
            await read_stream.aclose()

双向通信

适配 ASGI 场景的 MCP 自定义传输层实现（基于 AnyIO 内存对象流）：核心作用是为客户端与服务器搭建 “双向通信通道”，同时管理消息处理任务的生命周期和资源清理。它比你之前看到的create_transport更贴近实际 Web 场景（因包含 ASGI 标准的Scope/Receive/Send参数），但本质仍是 “消息收发的管家”。

@contextmanager
async def example_transport(scope: Scope, receive: Receive, send: Send):
    try:
        # 参数是ASGI（异步服务器网关接口） 的标准参数（常见于 FastAPI、Starlette 等异步 Web 框架），作用是：
		# - Scope：请求上下文（如客户端 IP、请求路径）；
		# - Receive：接收客户端消息的函数；
		# - Send：向客户端发送消息的函数；
        # 在 ASGI 服务中，为 MCP 协议提供 “内存级双向通信能力”
        read_stream_writer, read_stream = anyio.create_memory_object_stream(0)
        write_stream, write_stream_reader = anyio.create_memory_object_stream(0)

        async def message_handler():
            try:
                async with read_stream_writer:
                    # Message handling logic
                    pass
            except Exception as exc:
                logger.error(f"Failed to handle message: {exc}")
                raise exc

        async with anyio.create_task_group() as tg:
            tg.start_soon(message_handler)
            try:
                # Yield streams for communication
                yield read_stream, write_stream
            except Exception as exc:
                logger.error(f"Transport error: {exc}")
                raise exc
            finally:
                tg.cancel_scope.cancel()
                await write_stream.aclose()
                await read_stream.aclose()
    except Exception as exc:
        logger.error(f"Failed to initialize transport: {exc}")
        raise exc

对比两组代码

对比维度	双向通信example_transport	单项通信create_transport
适配场景	ASGI Web服务（含Scope/Receive/Send参数）	通用本地场景（无ASGI依赖）
流的数量	两组流（双向通信，收/发分离更清晰）	一组流（读/写端来自同一组）
消息处理	单独定义message_handler函数（逻辑更独立）	直接调用process_messages（逻辑耦合）
错误日志	多层日志（初始化/消息处理/传输错误）	无日志（需手动添加）

核心共性：都基于AnyIO内存流，都用上下文管理器+任务组管理生命周期，本质都是MCP传输层的实现。