当前位置：首页 > backend >正文

【动手学MCP从0到1】2.3 MCP中的Resource和Resource Template服务构建步骤详解

backend 2025/6/27 22:12:16

MCP中的Resource

1 资源定义
2 资源类型
3. 文本类型资源
- 3.1 服务端
- 3.2 客户端
- - 3.2.1 基础框架
  - 3.2.2 完善run函数
4. 二进制类型资源
- 4.1 服务端
- 4.2 客户端
5. Resource Template
- 5.1 格式
- 5.2 服务端全部代码
- 5.3 客户端
- - 5.3.1 客户端调试过程
  - 5.3.2 客户端全部代码

1 资源定义

资源是用于给客户端提供资源数据的，比如文本、图片、文件等。资源的定义使用如下结构:

[protocol]://[host]/[path]

比如：（最前面的名称可以是任意表述，但是不要影响大模型的识别，就可以取相对应的单词拼写即可）

file://home/user/documents/report.pdf
postgres://database/customers/schema
screen://localhost/display1

2 资源类型

资源包括以下两种类型:

1.文本类型:源代码、配置文件、日志文件、JSON/XML文件、纯文本
2.二进制类型:图像、PDF、音频文件、视频文件、其他非文本格式文件

3. 文本类型资源

3.1 服务端

我们可以通过 @mcp.resource快速定义一个资源（实例化后采用 @app.resource）。示例代码如下

安装aiofiles库：（用于处理异步条件下文件的读取，提高并发量，提高效率）

pip install aiofiles

在项目中，新建一个文件夹，命名为Resource_mcp，然后里面分别创建两个py文件为：server.py和client.py。涉及到资源的处理，还需要创建一个data文件夹，用于存放资源（data文件夹和py文件同一路径下，文件夹下是一个SMU.txt，里面的内容是上海海事大学基本的介绍信息）。文件系统架构如下

在这里插入图片描述

然后打开server.py文件，进行服务端的创建，采用sse协议进行通信，方便进行调试。先进行基础架构的搭建，通过对技术文档的查阅来了解函数具体的使用，如下

from mcp.server.fastmcp import FastMCPapp = FastMCP()@app.resource()if __name__ == '__main__':app.run(transport='sse')

通过按住ctrl键点击resource函数，进入函数的使用文档，如下：

在这里插入图片描述

核心参数是uri，然后name、description和mime_type都是可选参数，技术文档中也给出了具体的使用示例。

对于资源的加载方式如下：（单词加载文件操作，可以使用async with进行上下文管理）

"""
-------------------------------------------------------------------------------
@Project : MCP projects
File    : server.py
Time    : 2025-06-04 16:25
author  : musen
Email   : xianl828@163.com
-------------------------------------------------------------------------------
"""
import aiofiles
from mcp.server.fastmcp import FastMCPapp = FastMCP()@app.resource(uri="file://data/SMU.txt",name="SMU",description="获取上海海事大学的相关信息",mime_type="text/plain",
)
async def SMU_resource():# 打开文件获取数据，采用异步方式处理async with aiofiles.open("data/SMU.txt", mode="r", encoding="utf-8") as fp:content = await fp.read()return contentif __name__ == '__main__':app.run(transport='sse')

3.2 客户端

3.2.1 基础框架

根据之前创建的第一个MCP应用的客户端，直接把架构拿过来，放置在client.py文件中，如下

"""
-------------------------------------------------------------------------------
@Project : MCP projects
File    : client.py
Time    : 2025-06-04 16:25
author  : musen
Email   : xianl828@163.com
-------------------------------------------------------------------------------
"""
import asyncio
from mcp import ClientSession
from mcp.client.sse import sse_client
from openai import OpenAI
from contextlib import AsyncExitStackclass MCPClient:def __init__(self, server_path="./server.py"):self.deepseek = OpenAI(api_key="sk-5d307e0a45254xxxxx4575ff9",base_url="https://api.deepseek.com",)self.exit_stack = AsyncExitStack()async def run(self, query):passasync def aclose(self):await self.exit_stack.aclose()async def main():client = MCPClient(server_path="./server.py")try:await client.run("帮我查询一下上海海事大学的信息")finally:await client.aclose()if __name__ == "__main__":asyncio.run(main())

后续重点就在于完善run函数中的内容，至此，基础架构就搭建完成

3.2.2 完善run函数

此时采用sse协议通信，对于run函数中的第一步设置网络通讯参数就不需要单独再进行制定了，而是直接给出具体的监听端口地址即可，完善后的代码如下

async def run(self, query):# 1.创建read_stream、write_stream数据读写流read_stream,write_stream = await self.exit_stack.enter_async_context(sse_client("http://127.0.0.1:8000/sse"))# 2. 创建session,指定具体类型对象，方便后续进行代码补全识别session:ClientSession = await self.exit_stack.enter_async_context(ClientSession(read_stream, write_stream))# 3. 初始化await session.initialize()# 4. 获取服务端提供的所有的资源response = await session.list_resources()print(response)

基于sse协议通信的前4步和第一个mcp项目中的步骤基本一致。需要说明此处有两点注意事项：

①在第二步时候，创建session对象时候，建议添加一个对象类型说明，也就是这里的:ClientSession。方便后面敲代码时候的补全识别操作。
②然后之前是获取所有的tool，这里是获取所有的资源，对应的是list_resources().

执行客户端代码，输出结果如下：（可以正常读取服务端的资源）
在这里插入图片描述

进一步对资源进行操作，和之前的工具类似。由于都是列表对象结构，可通过遍历循环获取里面具体的内容信息。

class MCPClient:def __init__(self, server_path="./server.py"):self.resources = {}   #初始化一个资源字典，用于装信息# 4. 获取服务端提供的所有的资源
responses = (await session.list_resources()).resources
for resource in responses:uri = resource.uriname = resource.namedescription = resource.descriptionmime_type = resource.mimeTypeself.resources[name] = {"uri": uri,"name": name,"description": description,"mime_type": mime_type,}

代码中，根据上面的输出结果，responses需要进进一步获取，所以这里使用了 (await session.list_resources()).resources的方式。此外，之前创还能一个空的列表用作Tool的容器，这里是创建一个空的字典用来装资源信息，注意区别，然后为了方便在类中操作，将资源容器创建放在了类的初始化函数中了。

第五步，类比于工具操作，将资源转化为大模型能够调用的形式（Function Calling），新建一个fontions列表，用于存放数据（这个就是前面步骤创建的一个tools列表，用于大模型识别的对象）

注： mcp里面有tool，然后大模型里面也有tool。这里起名fontions其实就是大模型中的tools，为了区别前面mcp中tool工具，注意区分。

functions = []
# 4. 获取服务端提供的所有的资源
responses = (await session.list_resources()).resources
for resource in responses:uri = resource.uriname = resource.namedescription = resource.descriptionmime_type = resource.mimeTypeself.resources[name] = {"uri": uri,"name": name,"description": description,"mime_type": mime_type,}#Function Calling函数格式functions.append({"type":"function","function":{"name":name,"description":description,#资源中没有input_schema信息，为保持格式一致，设置为None"input_schema":None}})
# 创建消息发送给大模型
messages = [{"role":"user","content":query
}]
deepseek_respones = self.deepseek.chat.completions.create(messages=messages,model="deepseek-chat",tools=functions   #这里就是functions列表本身就是大模型中的tools，只是为了避免混淆，换个名称
)print(deepseek_respones)

客户端代码执行结果如下

在这里插入图片描述

然后就是根据进程选择进行读取资源（之前是利用进程读取工具）。注意根据上面输出的结果进行读取资源。代码如下

model_choice = deepseek_respones.choices[0]
if model_choice.finish_reason == "tool_calls":model_messages = model_choice.messagetool_calls = model_messages.tool_callsfor tool_call in tool_calls:tool_call_id = tool_call.idfunction = tool_call.functionfunction_arguments = function.argumentsfunction_name = function.nameuri = self.resources[function_name]["uri"]# 执行调用，response是服务端返回的response = await  session.read_resource(uri)print(response)

执行客户端代码，输出结果如下（之前加载工具是使用session.call_tool，这里加载资源是用的是session.read_resource，注意区别，然后两者传入的参数是不同的。特意强调，Function Calling格式中是没有uri参数，因此为了全局使用，特意在类的初始化中定义messages容器，就是为了方便这时候传入uri参数。）
在这里插入图片描述

进一步，将数据交给大模型，让大模型进行总结输出，如下（和第一个mcp服务创建一致）

model_choice = deepseek_respones.choices[0]
# 如果大模型的回复是tool_calls，那么我们就要执行调用工具的代码
if model_choice.finish_reason == 'tool_calls':# 为了让大模型能够更加精准的回复，需要将大模型返回回来的message也添加到messages中model_message = model_choice.message# message.model_dump：pydantic库提供的方法，model_message是pydantic的BaseModel的子类对象# model_dump是将Model对象上的属性转换为字典messages.append(model_message.model_dump())tool_calls = model_message.tool_callsfor tool_call in tool_calls:tool_call_id = tool_call.idfunction = tool_call.functionfunction_arguments = function.argumentsfunction_name = function.nameuri = self.resources[function_name]["uri"]# 执行调用，response是服务端返回的response = await session.read_resource(uri)result = response.contents[0].text# 把result丢给大模型，让大模型生成最终的结果messages.append({"role": "tool","content": result,"tool_call_id": tool_call_id})model_response = self.deepseek.chat.completions.create(model="deepseek-chat",messages=messages)print(model_response.choices[0].message.content)

代码执行后输出结果如下
在这里插入图片描述

而对比之前的SMU文件中的信息，回复的消息更加完善，经过大模型加工后，回复的消息更充实。

4. 二进制类型资源

以获取桌面壁纸图片为例，在Resource_mcp文件夹中新建两个py文件，分别命名为client_img.py和server_img.py。

4.1 服务端

其中server_img.py中的代码如下

"""
"""
import aiofiles
from mcp.server.fastmcp import FastMCPapp = FastMCP()@app.resource(uri="image://MCP.png",name="desktop_bg",description="获取电脑桌面壁纸",mime_type="image/png",
)
async def Desktop_bg():# 打开文件获取数据，采用异步方式处理async with aiofiles.open("C:/Users/pc/desktop/MCP.png", mode="rb") as fp:content = await fp.read()return contentif __name__ == '__main__':app.run(transport='sse')

代码区别就是在与mode中将r改为rb，MCP.png这个图片需要放置在桌面

4.2 客户端

客户端先用之前的代码进行跑，如下：（可以测试2次调用大模型，最后发现报错，BadRequestError: Error code: 400 - {‘error’: {‘message’: "This model’s maximum context length is 65536 tokens.However, you requested 342653 tokens (342653 in the messages, 0 in the completion)）

import base64
import aiofiles
import asyncio
from mcp import ClientSession
from mcp.client.sse import sse_client
from openai import OpenAI
from contextlib import AsyncExitStackclass MCPClient:def __init__(self, server_path="./server.py"):self.deepseek = OpenAI(api_key="sk-5d307e0xxxxxx15a6ce4575ff9",base_url="https://api.deepseek.com",)self.exit_stack = AsyncExitStack()self.resources = {}async def run(self, query):# 1.创建read_stream、write_stream数据读写流read_stream,write_stream = await self.exit_stack.enter_async_context(sse_client("http://127.0.0.1:8000/sse"))# 2. 创建session,指定具体类型对象，方便后续进行代码补全识别session:ClientSession = await self.exit_stack.enter_async_context(ClientSession(read_stream, write_stream))# 3. 初始化await session.initialize()functions = []# 4. 获取服务端提供的所有的资源responses = (await session.list_resources()).resourcesfor resource in responses:uri = resource.uriname = resource.namedescription = resource.descriptionmime_type = resource.mimeTypeself.resources[name] = {"uri": uri,"name": name,"description": description,"mime_type": mime_type,}#Function Calling函数格式functions.append({"type":"function","function":{"name":name,"description":description,#资源中没有input_schema信息，为保持格式一致，设置为None"input_schema":None}})# 创建消息发送给大模型messages = [{"role":"user","content":query}]deepseek_respones = self.deepseek.chat.completions.create(messages=messages,model="deepseek-chat",tools=functions)# model_choice = deepseek_respones.choices[0]# # 如果大模型的回复是tool_calls，那么我们就要执行调用工具的代码# if model_choice.finish_reason == 'tool_calls':#     # 为了让大模型能够更加精准的回复，需要将大模型返回回来的message也添加到messages中#     model_message = model_choice.message#     # message.model_dump：pydantic库提供的方法，model_message是pydantic的BaseModel的子类对象#     # model_dump是将Model对象上的属性转换为字典#     messages.append(model_message.model_dump())#     tool_calls = model_message.tool_calls#     for tool_call in tool_calls:#         tool_call_id = tool_call.id#         function = tool_call.function#         function_arguments = function.arguments#         function_name = function.name#         uri = self.resources[function_name]["uri"]#         # 执行调用，response是服务端返回的#         response = await session.read_resource(uri)#         print(response)#         result = response.contents[0].blob#         # 把result丢给大模型，让大模型生成最终的结果#         messages.append({#             "role": "tool",#             "content": result,#             "tool_call_id": tool_call_id#         })#         model_response = self.deepseek.chat.completions.create(#             model="deepseek-chat",#             messages=messages#         )#         print(model_response)#         async with aiofiles.open("data/desktop.png", mode="wb") as fp:#             await fp.write(base64.b64decode(model_response.contents[0].blob.encode('utf-8')))#         print("下载完毕！")#经过测试，不能使用大模型来两次，不然结果显示#BadRequestError: Error code: 400 - {'error': {'message': "This model's maximum context length is 65536 tokens.#However, you requested 342653 tokens (342653 in the messages, 0 in the completion).choice = deepseek_respones.choices[0]if choice.finish_reason == 'tool_calls':tool_call = choice.message.tool_calls[0]# print(tool_call)function = tool_call.functionfunction_name = function.namefunction_uri = self.resources[function_name]["uri"]result = await session.read_resource(function_uri)# print(result)async with aiofiles.open("data/desktop.png", mode="wb") as fp:await fp.write(base64.b64decode(result.contents[0].blob.encode('utf-8')))print("下载完毕！")async def aclose(self):await self.exit_stack.aclose()async def main():client = MCPClient(server_path="./server_img.py")try:await client.run("帮我获取一张桌面壁纸")finally:await client.aclose()if __name__ == "__main__":asyncio.run(main())

所以删除第二次调用，直接在第一次调用后，下载资源信息。之前文本类型资源对应的是text，而二进制图片这里对应的是blob，最终运行后输出的结果如下，然后在data文件夹下回多出一个desktop.pngd的图片，如下

在这里插入图片描述

5. Resource Template

5.1 格式

有些资源不是静态的，需要根据传入的参数动态返回，这时候就需要在资源路径中加入参数。比如

#根据用户id获取用户信息
users://{user id}#根据sku获取产品信息
product://{sku}

以个人信息获取为例，介绍Resource Template的服务应用。创建一个Resource_template_mcp文件夹，然后新建两个py文件，分别是server.py和client.py文件

5.2 服务端全部代码

server.pyw文件中的代码如下：（列举一个简单的使用use_id进行人员信息查询的服务。这里需要注意参数的描述，在一行时候需要使用换行符，然后格式需要一致，:param user_id: 参数具体描述)

from mcp.server.fastmcp import FastMCPapp = FastMCP("resource template mcp")@app.resource(uri="user://{user_id}",name="user_detail",description="根据参数user_id，返回用户的详情信息。\n:param user_id: 用户的id",mime_type="application/json"
)
async def user_detail(user_id: str):return {"user_id": user_id,"username": "张三",'gender': 'male',"university": "北京大学"}if __name__ == '__main__':app.run(transport="sse")

5.3 客户端

5.3.1 客户端调试过程

在run函数的前三步和之前一致，如下

from mcp.client.sse import sse_client
from mcp import ClientSession
from openai import OpenAI
from contextlib import AsyncExitStack
import asyncio
import jsonclass MCPClient:def __init__(self):self.deepseek = OpenAI(api_key="sk-5d307e0axxxxxx15a6ce4575ff9",base_url="https://api.deepseek.com")self.exit_stack = AsyncExitStack()self.resources = {}async def run(self, query):# 1. 创建read_stream、write_streamread_stream, write_stream = await self.exit_stack.enter_async_context(sse_client(url="http://127.0.0.1:8000/sse"))# 2. 创建session对象session: ClientSession = await self.exit_stack.enter_async_context(ClientSession(read_stream, write_stream))# 3. 初始化await session.initialize()functions = []

然后在第四步中，获取服务端提供的所有resources，这里使用的函数和方法与前面的操纵有差异，如下

# 获取服务端提供的所有resource
# diff: list_resource_templates
# diff: resourceTemplates
resources = (await session.list_resource_templates()).resourceTemplates
print(resources)

执行客服端的代码后，输出如下：（resources是个列表。参数也与原来的参数有所不同，因此后续获取的时候也要有所变化）
在这里插入图片描述

进一步针对resources进行循环，获取里面所有的参数，这里的uri也有所变化（上图红框中可以发现，变成了uriTemplate参数）

for resource in resources:# diff: uriTemplateuri = resource.uriTemplatename = resource.namedescription = resource.descriptionmime_type = resource.mimeTypeself.resources[name] = {"uri": uri,"name": name,"description": description,"mime_type": mime_type,}# Function Calling的函数格式functions.append({"type": "function","function": {"name": name,"description": description,# 资源类型，input_schema设置为None"input_schema": None}})# 创建消息发送给大模型
messages = [{"role": "user","content": query
}]
deepseek_response = self.deepseek.chat.completions.create(messages=messages,model="deepseek-chat",tools=functions
)
model_choice = deepseek_response.choices[0]
# 如果大模型的回复是tool_calls，那么我们就要执行调用工具的代码
if model_choice.finish_reason == 'tool_calls':# 为了让大模型能够更加精准的回复，需要将大模型返回回来的message也添加到messages中model_message = model_choice.message# message.model_dump：pydantic库提供的方法，model_message是pydantic的BaseModel的子类对象# model_dump是将Model对象上的属性转换为字典messages.append(model_message.model_dump())tool_calls = model_message.tool_callsfor tool_call in tool_calls:tool_call_id = tool_call.idfunction = tool_call.functionprint(function)

由上面代码可以看出，只有uri这个参数有变化，剩下到function获取的这里都没有变化。特别需要说明的是，在前面进行文本资源读取时候，里面的arguments参数是空的，而此时的function这里是有变化的，里面的arguments参数是有数据的，而且使用pycharm时，也可以发现前面的文本资源获取代码中的function_arguments赋值变量显示为灰色，说明没有被调用。
在这里插入图片描述

而在动态资源获取时，需要通过arguments参数进行传递，因此，读取资源时候需要结合这个参数，然后获取uri数据如下

function_arguments = function.arguments
function_name = function.name
uri = self.resources[function_name]["uri"]
print(uri,type(uri))

执行客户端，输出结果如下：（这里有个注意事项，就是直接获取到的function_arguments和uri其实都是str字符串数据类型）
在这里插入图片描述

而在后续的调用中，session.read_resource()函数中，需要传入的是字典数据类型（字符串数据类型传入会报错），修改如下

for tool_call in tool_calls:tool_call_id = tool_call.idfunction = tool_call.function# print(function)function_arguments = json.loads(function.arguments)function_name = function.nameuri = self.resources[function_name]["uri"]# print(function_arguments,type(function_arguments),uri,type(uri))print(function_arguments,type(function_arguments))print(uri,uri.format(**function_arguments))# diff：uri中包含参数，所以需要使用format方法，将大模型提取到的参数，生成完整的uriresponse = await session.read_resource(uri.format(**function_arguments))print(response)

执行结果输出如下：（通过json.loads()将字符串数据类型转化为字典数据类型，这里面的**的用法，有点解包的意思，就是将对应的键变成对应的值，从而完成格式化表达）
在这里插入图片描述

剩下的代码就没有变化了，就是把response的内容再一次丢给大模型，进行最终结果的返回，代码如下

result = response.contents[0].text
# 把result丢给大模型，让大模型生成最终的结果
messages.append({"role": "tool","content": result,"tool_call_id": tool_call_id
})
model_response = self.deepseek.chat.completions.create(model="deepseek-chat",messages=messages
)
print(model_response.choices[0].message.content)

执行代码，输出结果如下（结果可以正常输出）
在这里插入图片描述

5.3.2 客户端全部代码

from mcp.client.sse import sse_client
from mcp import ClientSession
from openai import OpenAI
from contextlib import AsyncExitStack
import asyncio
import jsonclass MCPClient:def __init__(self):self.deepseek = OpenAI(api_key="sk-5d307e0a4xxxxxce4575ff9",base_url="https://api.deepseek.com")self.exit_stack = AsyncExitStack()self.resources = {}async def run(self, query):# 1. 创建read_stream、write_streamread_stream, write_stream = await self.exit_stack.enter_async_context(sse_client(url="http://127.0.0.1:8000/sse"))# 2. 创建session对象session: ClientSession = await self.exit_stack.enter_async_context(ClientSession(read_stream, write_stream))# 3. 初始化await session.initialize()functions = []# 获取服务端提供的所有resource# diff: list_resource_templates# diff: resourceTemplatesresources = (await session.list_resource_templates()).resourceTemplatesprint(resources)for resource in resources:# diff: uriTemplateuri = resource.uriTemplatename = resource.namedescription = resource.descriptionmime_type = resource.mimeTypeself.resources[name] = {"uri": uri,"name": name,"description": description,"mime_type": mime_type,}# Function Calling的函数格式functions.append({"type": "function","function": {"name": name,"description": description,# 资源类型，input_schema设置为None"input_schema": None}})# 创建消息发送给大模型messages = [{"role": "user","content": query}]deepseek_response = self.deepseek.chat.completions.create(messages=messages,model="deepseek-chat",tools=functions)model_choice = deepseek_response.choices[0]# 如果大模型的回复是tool_calls，那么我们就要执行调用工具的代码if model_choice.finish_reason == 'tool_calls':# 为了让大模型能够更加精准的回复，需要将大模型返回回来的message也添加到messages中model_message = model_choice.message# message.model_dump：pydantic库提供的方法，model_message是pydantic的BaseModel的子类对象# model_dump是将Model对象上的属性转换为字典messages.append(model_message.model_dump())tool_calls = model_message.tool_callsfor tool_call in tool_calls:tool_call_id = tool_call.idfunction = tool_call.function# print(function)function_arguments = json.loads(function.arguments)function_name = function.nameuri = self.resources[function_name]["uri"]# diff：uri中包含参数，所以需要使用format方法，将大模型提取到的参数，生成完整的uriresponse = await session.read_resource(uri.format(**function_arguments))result = response.contents[0].text# 把result丢给大模型，让大模型生成最终的结果messages.append({"role": "tool","content": result,"tool_call_id": tool_call_id})model_response = self.deepseek.chat.completions.create(model="deepseek-chat",messages=messages)print(model_response.choices[0].message.content)async def aclose(self):await self.exit_stack.aclose()async def main():client = MCPClient()try:await client.run("帮我查找一下用户id为111的用户信息")finally:await client.aclose()if __name__ == '__main__':asyncio.run(main())