从零搭建MCP资源服务器:用Qwen3-8B模型构建智能历史问答系统(附完整代码)
从零搭建MCP资源服务器:用Qwen3-8B模型构建智能历史问答系统(附完整代码)
最近在折腾大模型应用落地的朋友,估计都遇到过同一个头疼的问题:模型本身能力很强,但一涉及到具体业务,比如查询公司内部文档、分析特定领域数据,或者像我今天要做的——构建一个历史知识问答系统,就发现模型对“外部世界”的了解几乎为零。它就像一个博学但足不出户的学者,书房里的书倒背如流,可书房外的档案柜、数据库、实时更新的文档,对它而言都是盲区。传统的做法要么是把所有资料都塞进提示词(上下文窗口爆炸),要么是写一堆定制化的API接口(维护成本飙升),整个过程既笨重又脆弱。
直到我深入研究了MCP(Model Context Protocol,模型上下文协议),尤其是其Resources(资源) 机制,才找到了一个优雅的解决方案。它不像某些方案那样试图把整个海洋灌进模型,而是为模型装上了一套标准化的“探针”和“数据管道”。简单来说,你可以把任何数据源——本地文本、数据库表、API接口甚至系统日志——封装成一个带有唯一URI的只读资源。模型通过协议发现这些资源,并在需要时按需、安全地读取,将获取的信息作为上下文来生成回答。这彻底改变了模型与数据交互的方式,从“一次性灌输”变成了“按需精准调用”。
今天,我就手把手带你用当下热门的Qwen3-8B开源模型,结合vLLM推理加速,从头构建一个智能历史问答系统。我们将以“广州名称演变”这段历史文本为例,将其封装为MCP资源,并实现一个端到端的问答流程。你会发现,借助MCP,让大模型“读懂”并活用你的私有数据,变得前所未有的清晰和可控。
1. 理解核心:MCP协议与Resources机制的精髓
在开始敲代码之前,我们得先搞明白MCP协议,特别是Resources机制,到底解决了什么根本问题。这绝非又一个晦涩的技术标准,而是AI应用工程化道路上的一块关键拼图。
大模型应用的“数据之痛” 通常体现在两个方面:一是上下文限制,动辄数十万字的专业文档无法全部放入提示词;二是集成复杂度,每对接一种新数据源(数据库、知识库、CRM系统),都需要为模型单独开发适配层,代码迅速变得臃肿且难以复用。MCP协议的出现,正是为了标准化模型与外部数据、工具之间的通信接口。你可以把它想象成AI世界的USB-C:一个统一的、双向的通信标准。任何符合MCP标准的“设备”(数据源或工具),都可以被模型“即插即用”。
在这个协议中,Resources(资源) 扮演着“只读数据源”的角色。它是MCP三大核心能力(另外两个是Tools工具和Prompts提示)中最基础、也最常用的一环。一个Resource本质上是一个带有描述信息的URI。这个URI可以指向:
- 一个本地文件 (
file://knowledge.txt) - 一个数据库查询结果 (
db://history/records?id=123) - 一个HTTP API的响应 (
https://api.example.com/data) - 甚至是一段实时生成的系统状态信息
它的核心设计哲学是 “无副作用读取” 。模型(通过客户端)可以请求读取资源的内容,但绝不能修改或删除它。这确保了数据源的安全性和一致性。同时,资源的暴露完全由服务器端控制。作为开发者,你决定哪些数据可以成为资源、如何命名、如何描述。模型客户端只能访问你明确公开的那些资源列表。
提示:MCP Resources 与 RAG(检索增强生成)中的向量检索并不冲突,而是互补。Resources 更适合提供确定的、结构化的上下文(如一份产品说明书、一段法律条文),而向量检索更适合从海量非结构化文档中寻找相关片段。在实际系统中,两者可以结合使用。
为了更直观地理解MCP Resources在架构中的位置,我们将其与传统定制集成方式做个对比:
| 特性维度 | 传统定制集成 | 基于MCP Resources的集成 |
|---|---|---|
| 数据接入方式 | 为每个数据源编写硬编码的适配器函数 | 将数据源统一封装为标准Resource,通过URI标识 |
| 模型交互 | 需要将数据预处理后拼接进提示词,或调用特定函数 | 模型自动发现可用资源,按需发起读取请求 |
| 安全性 | 依赖每个适配器自身的权限控制,难以统一审计 | 服务器端集中控制资源暴露,客户端仅能读取 |
| 可扩展性 | 新增数据源需修改核心逻辑,耦合度高 | 新增数据源只需添加新的Resource定义,与核心逻辑解耦 |
| 维护成本 | 高,每个数据源的变更都可能影响整体 | 低,资源定义独立,易于管理和更新 |
这种设计带来的最大好处是控制权的清晰分离。数据提供者(服务器)专注于如何高效、安全地提供数据片段;模型使用者(客户端)专注于如何利用这些数据片段解决实际问题。两者通过一个轻量级的JSON-RPC 2.0协议进行通信,无论是本地进程间通信还是远程HTTP/SSE,都能保持一致的接口。
2. 环境搭建:Qwen3-8B与vLLM推理后端部署
工欲善其事,必先利其器。我们的智能问答系统需要一个强大的“大脑”——Qwen3-8B模型,以及一个高效的“神经传导系统”——vLLM推理服务器。这部分我们会在本地完成部署,确保整个流程的可控性。
首先,我强烈建议使用Conda或venv创建一个独立的Python环境,避免包依赖冲突。这里以Conda为例:
# 创建并激活一个名为mcp-qwen的Python 3.10环境
conda create -n mcp-qwen python=3.10 -y
conda activate mcp-qwen
接下来,安装模型推理的核心:vLLM。vLLM是一个专为LLM设计的高吞吐量、内存高效的服务引擎,其PagedAttention技术能极大优化显存使用。
# 安装vLLM,这里选择安装包含CUDA支持的版本(确保你的机器有NVIDIA GPU)
pip install vllm
# 同时安装OpenAI兼容的客户端库,方便我们以标准API方式调用
pip install openai
现在,我们来启动vLLM服务,加载Qwen3-8B-Instruct模型。--served-model-name参数可以自定义模型在API中的名称,--api-key设置为EMPTY是因为我们本地测试无需验证,--max-model-len则根据你的GPU显存和需求调整。
# 在终端中运行以下命令启动vLLM服务器
python -m vllm.entrypoints.openai.api_server \
--model Qwen/Qwen3-8B-Instruct \
--served-model-name Qwen3-8B-Instruct \
--api-key EMPTY \
--port 9000 \
--max-model-len 8192
命令执行后,如果一切顺利,你会看到类似下面的输出,表示服务已在http://localhost:9000启动:
INFO 07-15 14:30:22 llm_engine.py:152] Initializing an LLM engine (v0.6.2) with config: model='Qwen/Qwen3-8B-Instruct', ...
INFO 07-15 14:30:30 model_runner.py:237] CUDA device: 0, name: NVIDIA GeForce RTX 4090
INFO 07-15 14:30:45 api_server.py:779] Started server process [12345]
INFO 07-15 14:30:45 api_server.py:784] Waiting for application startup.
INFO 07-15 14:30:45 api_server.py:799] Application startup complete.
INFO 07-15 14:30:45 api_server.py:804] Uvicorn running on http://0.0.0.0:9000 (Press CTRL+C to quit)
为了验证服务是否正常,我们可以打开另一个终端,用简单的cURL命令测试一下:
curl http://localhost:9000/v1/models
如果返回一个包含我们模型信息的JSON,比如{"object":"list","data":[{"id":"Qwen3-8B-Instruct", ...}]},那就恭喜你,推理后端已经就绪了。
注意:首次运行会从Hugging Face下载模型,耗时取决于你的网络。确保有足够的磁盘空间(约20GB)和GPU显存(Qwen3-8B FP16约需16GB)。如果显存不足,可以考虑使用
--quantization awq或gptq等量化选项,或者使用--dtype float8等参数。
3. 构建MCP资源服务器:封装历史知识
我们的“大脑”已经启动,现在需要为它构建一个专属的“资料室”——MCP资源服务器。这个服务器的唯一职责,就是管理我们的历史知识文本,并以MCP协议规定的格式对外提供。
首先,安装必要的Python库。我们将使用官方推荐的mcp库,它提供了高级的FastMCP类来简化开发。同时,由于需要异步读取文件,我们使用aiofiles。
pip install mcp aiofiles
接下来,创建我们的知识文件。在项目目录下,新建一个data文件夹,并在其中创建广州的名称衍变.txt,内容就是前面提供的关于广州历史名称由来的文本。
现在,是核心部分:编写MCP服务器代码。创建一个名为mcp_server.py的文件。
# -*- coding: utf-8 -*-
import asyncio
from mcp.server.fastmcp import FastMCP
import aiofiles
import os
# 创建一个FastMCP服务器实例,命名为"HistoryKnowledgeServer",并指定服务端口
mcp = FastMCP("HistoryKnowledgeServer", port=9999)
# 定义资源文件的路径
KNOWLEDGE_FILE_PATH = os.path.join(os.path.dirname(__file__), "data", "广州的名称衍变.txt")
@mcp.resource(
uri="file://guangzhou_history", # 资源的唯一标识符URI
name="guangzhou_history", # 资源名称,客户端通过此名称调用
description="获取关于广州城市名称历史演变的详细文本资料。内容涵盖从楚庭、南武到羊城、穗城等别名的由来,以及‘广州’行政名称的起源。", # 对资源的清晰描述,帮助模型理解何时调用
mime_type="text/plain" # 资源的媒体类型
)
async def get_guangzhou_history():
"""
异步读取并返回广州历史知识文件的内容。
这个函数被装饰为MCP资源,当客户端请求对应URI时触发。
"""
try:
# 使用aiofiles异步打开文件,避免阻塞事件循环
async with aiofiles.open(KNOWLEDGE_FILE_PATH, mode='r', encoding='utf-8') as f:
content = await f.read()
# 可选:记录日志,便于调试
print(f"[MCP Server] 资源 'guangzhou_history' 被请求,返回内容长度:{len(content)} 字符")
return content
except FileNotFoundError:
error_msg = f"错误:知识文件未找到于路径 {KNOWLEDGE_FILE_PATH}"
print(error_msg)
return error_msg
except Exception as e:
error_msg = f"读取文件时发生错误:{e}"
print(error_msg)
return error_msg
if __name__ == "__main__":
# 启动MCP服务器,使用SSE (Server-Sent Events)作为传输协议
# SSE适合需要服务器向客户端推送更新的场景,这里是标准用法
print(f"[MCP Server] 历史知识资源服务器启动中...")
print(f"[MCP Server] 资源URI: file://guangzhou_history")
print(f"[MCP Server] 服务地址: http://localhost:9999")
mcp.run(transport="sse")
这段代码的精髓在于@mcp.resource装饰器。它把一个普通的异步函数get_guangzhou_history“注册”为了一个MCP资源。当MCP客户端(也就是后面我们的问答程序)查询可用资源列表时,服务器会告知客户端:“我这里有这么一个资源,它的URI是file://guangzhou_history,名字叫guangzhou_history,描述是...,内容是纯文本。”客户端随后就可以根据这个描述,在需要的时候请求读取该资源的内容。
运行这个服务器:
python mcp_server.py
看到[MCP Server] 历史知识资源服务器启动中...的日志,说明你的资源服务器已经在http://localhost:9999待命了。它现在就像一座图书馆,安静地等待着有需求的读者(模型)来查阅那本名为《广州的名称衍变》的“书”。
4. 开发MCP客户端与智能问答逻辑
资源服务器准备好了,接下来要构建一个“聪明的读者”——MCP客户端。这个客户端需要完成三件事:1. 连接MCP服务器并发现资源;2. 连接Qwen3-8B模型;3. 根据用户问题,智能地决定是否以及如何调用资源,并生成最终答案。
创建一个名为mcp_client_qa.py的文件。这段代码稍长,但逻辑是清晰的,我会分段解释。
# -*- coding: utf-8 -*-
import asyncio
import json
from openai import OpenAI
from mcp.client.sse import sse_client
from mcp import ClientSession, StdioServerParameters
from contextlib import AsyncExitStack
from typing import Dict, Any
class HistoryQAClient:
"""
智能历史问答客户端。
整合MCP资源服务器与Qwen大模型,实现基于外部知识的问答。
"""
def __init__(self, llm_api_base: str, mcp_server_url: str, llm_model: str = "Qwen3-8B-Instruct"):
"""
初始化客户端。
:param llm_api_base: vLLM OpenAI API服务器地址
:param mcp_server_url: MCP资源服务器SSE地址
:param llm_model: 使用的模型名称
"""
# 初始化OpenAI客户端,指向我们本地启动的vLLM服务
self.llm_client = OpenAI(api_key="EMPTY", base_url=llm_api_base)
self.llm_model = llm_model
self.mcp_server_url = mcp_server_url
# AsyncExitStack用于优雅地管理多个异步上下文管理器
self.exit_stack = AsyncExitStack()
# 缓存从MCP服务器获取的资源元信息
self.available_resources: Dict[str, Dict[str, Any]] = {}
async def initialize_mcp_session(self):
"""初始化与MCP服务器的连接,并获取可用的资源列表。"""
# 建立SSE连接
read_stream, write_stream = await self.exit_stack.enter_async_context(
sse_client(self.mcp_server_url)
)
# 创建MCP会话
self.session: ClientSession = await self.exit_stack.enter_async_context(
ClientSession(read_stream, write_stream, params=StdioServerParameters())
)
# 执行初始化握手
await self.session.initialize()
# 列出服务器上所有可用的资源
list_response = await self.session.list_resources()
print(f"[Client] 发现 {len(list_response.resources)} 个MCP资源:")
for resource in list_response.resources:
print(f" - 名称: {resource.name}, URI: {resource.uri}, 描述: {resource.description}")
# 缓存资源信息,key为资源名称
self.available_resources[resource.name] = {
"uri": resource.uri,
"description": resource.description,
"mime_type": resource.mimeType,
}
def _build_tools_for_llm(self):
"""
将MCP资源转换为LLM可识别的工具(Tools)定义。
LLM通过工具调用的方式来表达它想要读取哪个资源。
"""
tools = []
for name, info in self.available_resources.items():
tool_def = {
"type": "function",
"function": {
"name": name,
"description": info["description"],
# 目前我们的资源读取不需要参数,所以schema为空
"parameters": {"type": "object", "properties": {}}
}
}
tools.append(tool_def)
return tools
async def ask_question(self, user_query: str) -> str:
"""
核心问答流程。
1. 将用户问题发送给LLM,LLM判断是否需要调用资源。
2. 如果LLM决定调用,则通过MCP会话读取资源内容。
3. 将资源内容作为上下文,再次请求LLM生成最终回答。
"""
# 第一步:构建初始对话消息
messages = [{"role": "user", "content": user_query}]
# 获取工具定义并请求LLM
tools = self._build_tools_for_llm()
print(f"[Client] 向模型发送问题: '{user_query}'")
print(f"[Client] 提供的工具: {[t['function']['name'] for t in tools]}")
first_response = self.llm_client.chat.completions.create(
model=self.llm_model,
messages=messages,
tools=tools,
tool_choice="auto", # 让模型自行决定是否调用工具
stream=False,
)
assistant_message = first_response.choices[0].message
print(f"[Client] 模型初次回复状态: {assistant_message.finish_reason}")
# 第二步:检查模型是否要求调用工具(即读取资源)
if assistant_message.tool_calls:
# 将模型的回复(包含工具调用请求)添加到对话历史
messages.append(assistant_message.model_dump())
tool_call = assistant_message.tool_calls[0]
tool_name = tool_call.function.name
tool_call_id = tool_call.id
print(f"[Client] 模型请求调用工具: '{tool_name}'")
# 第三步:通过MCP会话读取对应资源
if tool_name in self.available_resources:
resource_uri = self.available_resources[tool_name]["uri"]
print(f"[Client] 正在读取资源: {resource_uri}")
read_response = await self.session.read_resource(resource_uri)
resource_content = read_response.contents[0].text
print(f"[Client] 获取到资源内容 (前200字符): {resource_content[:200]}...")
# 将读取到的资源内容以“工具返回”的形式加入对话历史
messages.append({
"role": "tool",
"tool_call_id": tool_call_id,
"name": tool_name,
"content": resource_content
})
else:
# 理论上不会发生,因为工具列表来自资源列表
error_msg = f"请求的工具 '{tool_name}' 不存在。"
messages.append({
"role": "tool",
"tool_call_id": tool_call_id,
"name": tool_name,
"content": error_msg
})
print(f"[Client] 警告: {error_msg}")
# 第四步:将包含资源上下文的完整对话历史再次发送给LLM,生成最终答案
print(f"[Client] 请求模型基于上下文生成最终答案...")
final_response = self.llm_client.chat.completions.create(
model=self.llm_model,
messages=messages,
stream=False, # 非流式,一次性获取完整回答
)
final_answer = final_response.choices[0].message.content
else:
# 如果模型没有调用工具,则直接使用初次回复作为答案
final_answer = assistant_message.content
return final_answer
async def run_qa_loop(self):
"""运行一个简单的交互式问答循环。"""
await self.initialize_mcp_session()
print("\n" + "="*50)
print("智能历史问答系统已就绪。")
print("输入您关于广州历史名称的问题(输入'quit'或'退出'结束)")
print("="*50)
while True:
try:
user_input = input("\n您的问题: ").strip()
if user_input.lower() in ['quit', '退出', 'exit']:
print("感谢使用,再见!")
break
if not user_input:
continue
answer = await self.ask_question(user_input)
print(f"\n[系统回答]:\n{answer}")
except KeyboardInterrupt:
print("\n程序被中断。")
break
except Exception as e:
print(f"\n处理问题时发生错误: {e}")
async def cleanup(self):
"""清理资源,关闭连接。"""
await self.exit_stack.aclose()
async def main():
# 配置参数
LLM_API_BASE = "http://localhost:9000/v1" # vLLM OpenAI API 地址
MCP_SERVER_URL = "http://localhost:9999/sse" # MCP 服务器SSE端点
client = HistoryQAClient(llm_api_base=LLM_API_BASE, mcp_server_url=MCP_SERVER_URL)
try:
await client.run_qa_loop()
finally:
await client.cleanup()
if __name__ == "__main__":
asyncio.run(main())
这个客户端的运作流程,完美体现了MCP的核心价值:动态上下文获取。模型不再是基于训练时的静态知识回答问题,而是在推理时,实时地、按需地从外部数据源(我们的资源服务器)获取最新、最相关的上下文。整个过程是自动化的,模型根据对问题的理解,自主决定是否需要调用guangzhou_history这个工具(资源)来获取信息。
5. 系统联调与效果验证
现在,让我们点燃引擎,看看整个系统如何协同工作。请确保三个服务按顺序启动:
- 终端1 - vLLM模型服务:已在运行,监听9000端口。
- 终端2 - MCP资源服务器:运行
python mcp_server.py,监听9999端口。 - 终端3 - MCP客户端问答程序:运行
python mcp_client_qa.py。
在客户端终端,你会看到初始化成功的提示。现在,尝试提出几个问题:
您的问题: 广州为什么又叫羊城?
观察客户端终端的日志输出,你会看到类似下面的过程:
[Client] 向模型发送问题: '广州为什么又叫羊城?'
[Client] 提供的工具: ['guangzhou_history']
[Client] 模型初次回复状态: tool_calls
[Client] 模型请求调用工具: 'guangzhou_history'
[Client] 正在读取资源: file://guangzhou_history
[MCP Server] 资源 'guangzhou_history' 被请求,返回内容长度:xxxx 字符
[Client] 获取到资源内容 (前200字符): 楚庭是传说中广州最早的名称。关于“楚庭”之得名,有不同的说法:一说周夷王八年(公元前878年),南海臣服于楚国,因而建造楚庭。一说公元前600多年的春秋时期,楚成王熊恽平南方夷越之乱,在南海建楚庭。有研究者解释称,楚庭(又作楚亭)...
[Client] 请求模型基于上下文生成最终答案...
[系统回答]:
广州被称为“羊城”,源于一个古老的民间神话传说。相传在周朝时期,有五位仙人骑着五只口衔稻穗的仙羊降临广州(当时可能被称为“楚庭”),将稻穗赠予当地人民,并祝福此地永无饥荒。这个“五羊衔谷”或“五仙乘羊赠穗”的故事最早记载于晋代学者所著的《广州记》中。此后,“羊城”和“穗城”便成为广州广为流传的别称,并在唐代以后的诗词文学中频繁出现,成为广州富有浪漫色彩和文化底蕴的城市象征。
整个交互清晰可见:模型接收到问题后,识别出自己需要关于广州别名的具体历史知识,于是自动发起了对guangzhou_history资源的调用请求。客户端代理这个请求,从MCP服务器获取到完整的文本内容,并将其作为新的上下文提供给模型。模型最终结合问题与精准的上下文,生成了详实、准确的回答。
你可以继续测试更多问题,体验系统如何灵活运用这份知识:
- “楚庭这个名字是怎么来的?”
- “‘广州’这个行政区划名字最早出现在什么时候?”
- “南武和广州是什么关系?”
每一个问题,模型都会动态地决定是否需要查阅“资料”,从而确保回答既基于我们提供的权威文本,又经过了模型的自然语言理解和组织。这种模式,比简单地将全文放入提示词(可能超出上下文长度),或者训练一个专门的模型(成本高、不灵活),要高效和实用得多。
6. 进阶探索与生产化考量
我们成功搭建了一个可运行的Demo,但要从原型走向生产,还有几个关键点需要考虑和优化。这部分没有标准答案,更多是根据实际场景进行权衡和设计。
1. 资源管理的规模化 当前我们只有一个文本资源。实际项目中,资源可能是成百上千的。你需要设计一个高效的资源管理机制。
- 动态资源发现:可以让MCP服务器根据目录扫描、数据库查询等动态生成资源列表,而不是硬编码在代码里。
- 资源分类与标签:为资源添加元数据(如
category: history,tags: [“guangzhou”, “地名”]),帮助模型更精确地判断何时调用哪个资源。 - 资源内容预处理:对于长文档,可以在服务器端进行分块、摘要或提取关键信息,再作为资源提供,避免返回过多无关内容。
2. 与RAG技术的结合 MCP Resources 提供的是精确的、基于URI的数据获取,而RAG(检索增强生成)擅长从海量文档中检索相关片段。两者可以强强联合。
- 方案A(资源作为精读库):用RAG从海量文档库中检索出最相关的几个文档名或ID,然后将这些ID转化为对应的MCP Resource URI进行精确读取。这相当于先“检索目录”,再“精读章节”。
- 方案B(资源作为实时数据源):将实时更新的数据库、API或日志流封装为MCP资源。当用户问到最新数据时,模型通过MCP获取实时信息,再结合RAG从静态知识库获取的背景信息进行回答。
3. 性能与安全性优化
- 缓存策略:对于不常变动的资源内容,可以在客户端或服务器层增加缓存,避免重复读取和网络开销。
- 权限控制:MCP协议本身支持权限提示。在生产环境中,应为不同资源设置不同权限级别,并在客户端请求时进行认证和授权。例如,涉及敏感信息的资源需要用户明确确认后才能访问。
- 错误处理与降级:完善客户端和服务器的错误处理机制。当MCP服务器不可用时,客户端应能优雅降级,例如尝试使用本地缓存或直接提示用户“资料暂不可用”。
4. 扩展到更多工具类型 本文聚焦于Resources(只读),但MCP还有Tools(工具) 和Prompts(提示)。你可以很容易地将一个查询数据库的函数、一个调用天气API的服务,甚至一个执行代码解释器的功能,封装成MCP Tool。这样,你的AI助手不仅能“查资料”,还能“做事情”,能力边界被极大地扩展了。
最后,别忘了监控和日志。记录下模型每次调用了哪个资源、请求的内容是什么、生成的回答质量如何。这些数据对于迭代优化你的资源定义、工具描述乃至整个系统流程,都是无比宝贵的。
回过头看,我们通过MCP协议,用不到300行代码,就构建了一个能理解并运用特定领域知识的智能问答系统。它架构清晰、职责分明:数据提供者(MCP Server)、推理引擎(vLLM+Qwen)、协调中枢(MCP Client)各司其职。这种解耦带来的灵活性是巨大的——明天你想把知识库从文本文件换成MySQL,只需修改MCP Server的资源实现;想把模型从Qwen换成GLM,只需更改客户端的API地址;甚至想增加一个计算器工具,也只需在Server上新增一个Tool定义。
这种“标准化接口”的思想,正是现代软件工程的核心。MCP为AI应用带来的,正是这样一套亟需的接口标准。它让大模型真正开始以一种可预测、可管理、可扩展的方式,与我们的数字世界连接。
更多推荐




所有评论(0)