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

开源模型应用落地-模型上下文协议（MCP）-Resources-资源的使用逻辑

news 来源：原创 2025/6/23 3:39:00

一、前言

在大型语言模型与外部世界交互的探索中，如何高效、灵活地接入多样化数据始终是核心命题。MCP（Model Context Protocol）协议中的Resources 机制，正是为这一问题提供了优雅的解决方案。通过URI（统一资源标识符）的抽象设计，MCP服务器能够将文件系统、数据库、API响应甚至实时系统日志等数据转化为LLM可理解的上下文。这种“无副作用”的读取方式不仅简化了数据调用逻辑，更通过应用层控制权的让渡，赋予客户端对资源访问的自主性与灵活性。

本文将深入解析MCP资源的使用逻辑，从协议规范到实战场景，探讨如何通过这一机制释放大模型的无限潜能。

二、术语介绍

2.1.MCP

MCP协议（Model Context Protocol，模型上下文协议）是由Anthropic公司于2024年11月推出的开放标准协议，旨在为大型语言模型（LLM）与外部数据源、工具之间建立统一的通信接口。其核心作用在于解决AI模型因数据孤岛限制难以发挥潜力的问题，通过标准化交互方式提升AI应用的扩展性和实用性。

一、MCP协议的核心定义
1.1. 协议定位
MCP被类比为AI领域的“USB-C接口”，通过解耦LLM与外部系统的集成，实现一次开发即可动态连接多种服务（如数据库、API、本地文件系统等）。例如，开发者无需为每个新工具单独定制接口，只需遵循MCP标准即可实现AI与工具的即插即用。

1.2. 技术架构
采用客户端-服务器模型，包含三个核心组件：

MCP主机：运行AI模型的环境（如Claude桌面应用）；
MCP客户端：负责与服务器通信并转发请求；
MCP服务器：提供具体功能服务的轻量级程序（如文件管理或API调用）。

二、MCP协议的核心作用
2.1. 统一交互标准
标准化JSON-RPC 2.0协议支持本地（标准输入输出）与远程（HTTP/SSE）通信，兼容Python、Java等多种编程语言，降低开发复杂度。例如，本地通信保障医疗数据安全传输，远程通信支持云端服务扩展。

 2.2. 动态扩展能力
通过资源（Resource）、工具（Tool）、提示（Prompt）三大核心概念，AI模型可自动发现新工具并实时交互。例如连接GitHub时，AI可自动提交代码或修复BUG，无需人工预定义接口。

2.3. 安全与权限控制
内置权限验证机制，用户需手动批准敏感操作（如文件删除或邮件发送），确保数据隐私。在企业场景中，该特性尤为重要，如金融分析时需严格管控市场数据访问权限。

2.2.资源

代表服务器向客户端暴露的、可供读取的数据或内容原语。其主要作用是为大型语言模型提供交互所需的上下文信息。

具体特点包括：

1.数据类型多样性

Resources 可以涵盖多种结构化或非结构化数据，例如文件内容、数据库记录、API响应、实时系统数据、日志文件、屏幕截图等。

2.只读性

Resources 被设计为只读实体，客户端只能检索数据而无法修改，确保数据的安全性和一致性。

3.上下文与应用控制

它们作为LLM交互的上下文来源，同时由应用程序（服务器端）控制，即客户端仅能按预设规则访问特定资源。

4.协议功能支撑

Resources 是MCP协议三大核心能力之一（另两类为工具和提示），用于实现模型对本地或外部数据的定制化访问

三、前置条件

3.1.基础环境及前置条件

1. 操作系统：无限

3.2.MCP环境安装

参见：开源模型应用落地-模型上下文协议（MCP）-从数据孤岛到万物互联（一）_pycharm mcp-CSDN博客文章浏览阅读2.2w次，点赞57次，收藏50次。模型上下文协议（MCP）通过标准化的接口和交互语法，重新定义AI与物理世界的连接，提升人机协作的深度与广度。_pycharm mcphttps://charles.blog.csdn.net/article/details/147068018?spm=1011.2415.3001.5331

3.3.QWEN3推理环境准备

参见：

开源模型应用落地-qwen模型小试-Qwen3-8B-推理加速-vLLM（一）-CSDN博客文章浏览阅读8.9k次，点赞22次，收藏18次。解析Qwen3-8B与vLLM的技术协同逻辑，探讨其在复杂任务中的性能表现及落地价值。https://charles.blog.csdn.net/article/details/147718340?spm=1011.2415.3001.5331开源模型应用落地-qwen模型小试-Qwen3-8B-推理加速-vLLM-Docker（二）_docker 方式启动 vllm,运行qwen3-8b-fp8 参数详解-CSDN博客文章浏览阅读1.3w次，点赞25次，收藏19次。通过容器化技术整合高性能推理框架vLLM与Qwen3-8B模型，可实现资源高效利用、灵活部署及推理性能优化，为大模型本地化应用提供轻量化解决方案_docker 方式启动 vllm,运行qwen3-8b-fp8 参数详解https://charles.blog.csdn.net/article/details/147719905?spm=1011.2415.3001.5331

四、技术实现

4.1.资源准备

文件名：广州的名称衍变.txt

楚庭是传说中广州最早的名称。关于“楚庭”之得名，有不同的说法：一说周夷王八年（公元前878年），南海臣服于楚国，因而建造楚庭。一说公元前600多年的春秋时期，楚成王熊恽平南方夷越之乱，在南海建楚庭。有研究者解释称，楚庭（又作楚亭）最早或是宫室，或是市集的意思，后来才慢慢被用作广州的代称。清代，人们将楚庭看作广州最早的名字，并建牌坊纪念。
“南武”也是记载中广州早期的名称。“南武”之得名也有多个说法，一说认为此名本为越国旧有地名，楚国灭越后，由公师隅在今天的广州所在地建城，并沿用了故国的名称；一说认为赵佗号为南越武帝，故将南海郡改称南武郡，以郡名为城市命名。不过，关于楚庭、南武的说法。
广州众多别名中，“羊城”“穗城”最广为人知。这两个别名源于五仙乘羊赠穗的美丽神话。相传周朝时，五个仙人骑五羊送来稻穗。五羊衔谷的传说最早见于晋人所著的《广州记》，书中称“五羊衔谷萃于楚庭”。在后来的古籍中，故事情节发展为五位仙人手里拿着一茎六出的谷穗，骑着五羊而来，并祝福此方土地永无饥荒。唐代以来，“五羊”“羊城”逐渐成为广州城的代称，诗人常用“羊城”代指广州，如高适写有“海对羊城阔，山连象郡高”之句，殷尧藩有“遐荒迢递五羊城，归兴浓消客里情”之句。
“广州”作为一个行政区域概念，距今已经有1796年历史，“广州”二字在历史上的地域含义因时而异。西汉年间，长江以南被划分为三个州：扬州、荆州、交州。交州属下设立九郡，广州就属于交州南海郡。汉武帝平定南越国后，颁布圣旨称：初开粤地，宜广布恩信，意思是希望地方官员以怀柔政策管理此地。州府所在地因此定名为“广信”（一说今广西梧州，一说今广东封开），“广州”之“广”因此得名。　“广州”这一地名出现在三国年间。226年，孙权考虑到岭南地区过于辽阔，不易管辖，又因交州刺史士燮势力太大，遂决定把交州一拆为二，史称“交广分治”，“广州”作为一个地名从此出现，合浦以西为交州，合浦以东为广州。广州辖区范围相当于今广东、广西之大部。

4.2.MCP Server实现

# -*-  coding:utf-8 -*-
from mcp.server.fastmcp import FastMCP
import aiofilesDEAULT_TRANSPORT = "sse"# Create an MCP server
mcp = FastMCP("Demo", port=9999)@mcp.resource(uri="file://knowledge.txt",name="knowledge",description="获取广州名称的衍变历史信息",mime_type="text/plain"
)
async def knowledge_base():"""获取知了课堂的相关介绍信息"""async with aiofiles.open(r"E:\data\广州的名称衍变.txt", mode="r", encoding='utf-8') as fp:content = await fp.read()print(f'获取的内容：{content}')return contentif __name__ == "__main__":# Initialize and run the serverprint(f"start mcp server,transport: {DEAULT_TRANSPORT}...")  # 记录服务启动日志mcp.run(transport=DEAULT_TRANSPORT)

需要安装aiofiles依赖：uv pip install aiofiles

调用：

直接在pycharm中运行

4.3.MCP Client实现

# -*- coding:utf-8 -*-
import asynciofrom openai import OpenAI
from mcp.client.sse import sse_client
from mcp import ClientSession
from contextlib import AsyncExitStackclass MCPClient:def __init__(self, api_key: str, llm_server_url: str, mcp_server_url):self.client = OpenAI(api_key=api_key, base_url=llm_server_url)models = self.client.models.list()self.model = models.data[0].idself.mcp_server_url = mcp_server_urlself.exit_stack = AsyncExitStack()self.resources = {}async def run(self, query):read_stream, write_stream = await self.exit_stack.enter_async_context(sse_client(self.mcp_server_url))session: ClientSession = await self.exit_stack.enter_async_context(ClientSession(read_stream, write_stream))# 初始化await session.initialize()# 获取可用的资源response = await session.list_resources()resources = response.resourcestools = []for resource in resources:uri = resource.uriname = resource.namedescription = resource.descriptionmime_type = resource.mimeTypetools.append({"type": "function","function": {"name": name,"description": description,"input_schema": None}})self.resources[name] = {"uri": uri,"name": name,"description": description,"mime_type": mime_type,}# 定义消息messages = [{"role": "user","content": query}]output = self.client.chat.completions.create(messages=messages,model=self.model,tools=tools,stream=False)print(f"首次调用响应: {output}")choice = output.choices[0]if choice.finish_reason == 'tool_calls':messages.append(output.choices[0].message.model_dump())print(f"更新消息历史: {messages}")tool_call = choice.message.tool_calls[0]tool_id = tool_call.idfunction = tool_call.functionfunction_name = function.namefunction_uri = self.resources[function_name]["uri"]print(f'function_name: {function_name},function_uri: {function_uri}')response = await session.read_resource(function_uri)result = response.contents[0].textprint(f"获取到内容: {result}")messages.append({"role": "tool","content": result,"tool_call_id": tool_id,"name": function_name})# 第二次调用：流式响应生成最终回答output = self.client.chat.completions.create(messages=messages,model=self.model,stream=True)print("\nAI助手：")for chunk in output:if chunk.choices and chunk.choices[0].delta.content:print(chunk.choices[0].delta.content, end='', flush=True)async def aclose(self):await self.exit_stack.aclose()async def main():# 环境配置openai_api_key = "EMPTY"openai_api_base = "http://localhost:9000/v1"mcp_server_url = "http://localhost:9999/sse"client = MCPClient(api_key=openai_api_key,llm_server_url=openai_api_base,mcp_server_url=mcp_server_url)try:await client.run("帮我查一下广州这个名称的由来")finally:await client.aclose()if __name__ == '__main__':asyncio.run(main())