MCP中的资源
这里我们说下MCP中，什么是资源
在MCP之中，资源是其中的核心原语，用于将服务器端的某些数据暴露给客户端，供客户端上的大模型进行使用，添加到上下文之中。
而这个数据，可以是任意类型的文本或者二进制数据，包括文件，数据库记录，日志，音频，视频等。
对于资源，在主机上是由着唯一的URI确定的，遵循着通用化的协议
例如
file:///home/user/docs/report.pdf
postgres://db.example.com/customers/schema
screen://localhost/display
根据不同的协议，确定URI格式，内容
但是需要注意一点，就是不同的MCP的客户端处理资源的方式不同
比如常见的Claude Tare，需要让用户明确选择资源
其他的客户端会自主的选择资源
因此需要注意，在利用MCP协议暴露资源的时候，需要意识到MCP Host类型不同，使用资源的方式不同，如果需要明确由LLM来控制资源使用，应该将资源封装为工具而不是资源。
其次是在MCP中资源的具体使用方式
服务器首先要声明自己支持Resource暴露功能
{
“capabilities”: {
“resources”: {
“subscribe”: true,
“listChanged”: true
}
}
}
其中subscribe 是指的客户端是否支持订阅单个资源
listChanged指的是资源列表变化的时候，服务器端是否发送通知。
服务器端需要一个统一的数据结构来描述每个资源对象，
包含 URI 资源标识符，name，description，mineType，size
并且在服务器上，分为了静态资源和动态资源
|
{
“uri”: “file:///logs/app.log”, “name”: “Application Logs”, “description”: “实时日志文件”, “mimeType”: “text/plain” } |
|
{
“uriTemplate”: “logs://recent?timeframe={duration}”, “name”: “最近日志”, “description”: “按时长获取日志”, “mimeType”: “text/plain” } |
对于动态数据，根据其中的标准填入，让客户端进行构建对应的URI，从而获取到最近的资源。
对于服务器端的代码实现，则是可以通过装饰器和注册函数来实现资源定义接口
|
# 使用装饰器注册资源列表处理函数
@app.list_resources() async def list_resources() -> list[types.Resource]: “”” 实现资源列表API，返回服务器提供的资源列表 返回: list[types.Resource]: 资源对象列表 “”” return [ # 创建一个资源对象 types.Resource( uri=”file:///logs/app.log”, # 资源的唯一标识符 name=”Application Logs” # 资源的显示名称 mimeType=”text/plain” ) ] |
通过app.list_resources来让服务器端注册暴露资源的函数
对于客户端，则是通过发起不同的查询请求，获取到可用资源
比如常见的 resources/list 请求来获取到可用资源
|
{
“jsonrpc”: “2.0”, “id”: 1, “method”: “resources/list”, “params”: { “cursor”: “optional-cursor-value” } } |
对应的响应为：
|
{
“jsonrpc”: “2.0”, “id”: 1, “result”: { “resources”: [ { “uri”: “file:///project/src/main.rs”, “name”: “main.rs”, “description”: “Primary application entry point”, “mimeType”: “text/x-rust” } ], “nextCursor”: “next-page-cursor” } } |
通过resources/read 来指定URI，从而读取到文件内容。
客户端的请求格式如下
|
{
“jsonrpc”: “2.0”, “id”: 2, “method”: “resources/read”, “params”: { “uri”: “file:///logs/app.log”, } } |
服务器的响应为
|
{
“jsonrpc”: “2.0”, “id”: 2, “method”: “resources/read”, “params”: { “uri”: “file:///logs/app.log”, } } |
除此外还有这实时通知机制，让客户端保持对资源状态的感知
列表变更通知，当服务器资源进行增删改的时候，发出对应的通知
或者文件内容发生更新订阅，服务器端会通知对应的订阅者，方便获取到最新内容。
那么最终的交互流程为
同时我们可以结合代码来看，假设我们在项目目录下有一系列的文档
我们可以在服务器端通过函数进行资源暴露，代码如下
|
# 导入必要的库
import asyncio import os import mcp.types as types from mcp.server import Server from mcp.server.stdio import stdio_server # 创建一个MCP服务器实例，名称为”example-server” app = Server(“example-server”) DOC_DIR = “你的路径/mcp-in-action/05-resource-资源发现/server/medical_docs” # 注册资源列表 @app.list_resources() async def list_resources() -> list[types.Resource]: files = [f for f in os.listdir(DOC_DIR) if f.endswith(“.txt”)] return [ types.Resource( uri=f”file://{os.path.join(DOC_DIR, fname)}”, name=fname, description=”文档”, mimeType=”text/plain” ) for fname in files ] # 读取资源内容 @app.read_resource() async def read_resource(uri: str) -> str: path = uri.replace(“file://”, “”) with open(path, encoding=”utf-8″) as f: return f.read() async def main(): async with stdio_server() as streams: await app.run( streams[0], streams[1], app.create_initialization_options() ) if __name__ == “__main__”: asyncio.run(main()) |
之后客户端可以通过诸如stdio的协议进行读取。
|
import asyncio
import sys from mcp import ClientSession, StdioServerParameters from mcp.types import Notification, Resource from mcp.client.stdio import stdio_client async def main(): # 检查命令行参数 – 必须通过两个参数启动此程序，一个当前客户端，一个服务器程序 if len(sys.argv) < 2: print(“Usage: python simple_client.py <path_to_server_script>”) sys.exit(1) server_script = sys.argv[1] # 使用服务器的 Python 解释器来启动 Server，而不是当前客户端的Python环境 params = StdioServerParameters( command=”你的环境路径/bin/python3″, args=[server_script], env=None ) # 建立 stdio 传输并创建 Session async with stdio_client(params) as (reader, writer): async with ClientSession(reader, writer) as session: # 初始化握手 await session.initialize() # 发送初始化完成通知 notification = Notification( method=”notifications/initialized”, params={} ) await session.send_notification(notification) # 列出资源 response = await session.list_resources() print(“资源列表：”) # 打印返回格式 print(f”返回类型: {type(response)}”) print(f”返回内容: {response}”) # 简化处理资源列表 resources = getattr(response, ‘resources’, response) for res in resources: print(f”- URI: {res.uri}, Name: {res.name}”) if __name__ == “__main__”: asyncio.run(main()) |
通过发送通知并利用list_resources函数获取到了相关的资源。
最后我们总结下，
MCP支持其中服务端的资源向着客户端暴露数据和内容，作为LLM交互的上下文
并通过特定函数，诸如resources/list接口来提供URI使用。从而建立了合格的资源管理机制，来对外部数据进行引用