科研资料云端整理助手：基于 MCP Server 的个人效率智能体实践

摘要

本文基于 Datawhale 《MCP 极简开发》学习项目，构建“科研资料云端整理助手”以提升科研人士的资料检索与整理效率。系统采用 MCP Server 作为中枢，结合 arXiv 数据源与 Markdown 笔记结构化能力，实现论文检索、学习计划生成与笔记归档的自动化流程。本文详细介绍系统架构、实验环境搭建、工具实现和调试方法，并提供完整教学手册，帮助读者从零搭建并复用该智能体。实验结果表明，通过标准化 MCP 工具组合，可以显著降低科研资料整理的重复劳动，为云计算导论课程的期末论文提供可复现案例。

关键词：MCP Server；科研资料整理；云计算；个人效率；智能体
Keywords: MCP Server; Research Data Management; Cloud Computing; Personal Productivity; Intelligent Agent

引言

云计算与分布式技术的发展，使得科研活动对云端资源与智能体接口的依赖愈发显著。面对多源数据、跨终端协同与个体效率需求，传统人工整理流程容易受限。Model Context Protocol（MCP）提供了统一的工具接口与跨客户端调用机制，为构建可复用的科研助手提供了可靠基础。

本研究拓展 mcp-lite-dev 项目，于第 10 章个人效率场景下，打造“科研资料云端整理助手”。该助手具备：

1. 多源检索：调用 arXiv API 聚合最新论文信息；
2. 学习计划：结合时间预算生成分阶段精读方案；
3. 云端笔记归档：对零散记录进行结构化整理并输出 Markdown。

在课程教学层面，我们借鉴第 6 章的“动手写一个 MCP Server”以及第 7 章的“开发进阶与调试方法”，将实验流程整理为一步一图的指导手册，确保读者能够快速复现并扩展。

正文：实验流程与教学手册

1. 环境准备与独立项目骨架

1. 初始化独立项目

   • 在 E:\research-assistant-mcp 新建项目目录；
   • 在该目录内执行 uv init --package（或复制本文提供的 pyproject.toml）；
   • 当前仓库的目录结构如下，供对照：

     E:\research-assistant-mcp\
     ├─ README.md
     ├─ pyproject.toml
     ├─ .gitignore
     ├─ .env          # 本地自建，存放 API Key，不纳入版本控制
     └─ src/
        └─ research_assistant_mcp/
           └─ server.py
     

2. 创建 pyproject.toml（示例）

   [build-system]
   requires = ["setuptools>=80.9.0", "wheel>=0.45.1"]
   build-backend = "setuptools.build_meta"
   
   [project]
   name = "research-assistant-mcp-huangshuai"
   version = "0.1.1"
   description = "科研资料云端整理助手（基于 MCP Server 与 DeepSeek 模型）"
   readme = "README.md"
   requires-python = ">=3.10"
   dependencies = [
       "httpx>=0.28.1",
       "mcp>=1.10.1",
       "openai>=1.93.3",
       "python-dotenv>=1.1.1",
       "textwrap3>=0.9.2"
   ]
   
    [project.scripts]
    research-assistant-mcp-huangshuai = "research_assistant_mcp.client:cli"
   
   [tool.setuptools]
   package-dir = {"" = "src"}
   
   [tool.setuptools.packages.find]
   where = ["src"]
   

3. 准备环境配置

   • 在项目根目录创建 .env，至少包含 OPENWEATHER_API_KEY、BASE_URL、MODEL、API_KEY 等键值，供后续扩展使用；
   • .gitignore 参考 learnMCP 根目录，确保 .venv/、dist/、.env 等被忽略；
   • README.md 简要说明安装步骤与运行命令。

4. 安装 uv 并创建虚拟环境（同 docs/ch06/ch06.md）

   1. pip install uv，uv --version 验证；
   2. 在 E:\research-assistant-mcp 根目录运行 uv venv；
   3. .\.venv\Scripts\activate（Windows）或 source .venv/bin/activate（macOS/Linux）；
   4. uv sync --python 3.13 安装依赖。

5. 加载环境变量

   • 在项目根目录的 .env 中配置 OPENWEATHER_API_KEY（可选）、BASE_URL、MODEL、API_KEY 等参数；
   • 即使暂时只在 MCP Server / Client 侧做本地测试，也建议提前保留这些键，便于后续与 LLM 集成。

2. 项目结构确认

research-assistant-mcp/
├─ pyproject.toml
├─ README.md
├─ .env
├─ .gitignore
├─ docs/
├─ src/
│  └─ research_assistant_mcp/
│     └─ server.py
└─ uv.lock

docs/ 目录用于存放教学材料，默认为空；若需撰写教程，可在此新增 tutorial.md 等文件。Server 入口脚本位于 src/research_assistant_mcp/server.py，命名与 pyproject.toml 中的包一致。

若将本教程嵌入 learnMCP 项目，可将 server.py 命名为 research_assistant.py 并放入 src/ch10/ 目录；但本章节默认你在 E:\research-assistant-mcp 目录中独立完成实验。

3. 代码讲解（结合 docs/ch10/ch10.md 与 src/research_assistant/server.py）

1. 服务初始化

   from mcp.server.fastmcp import FastMCP
   
   ARXIV_API_ENDPOINT = "https://export.arxiv.org/api/query"
   USER_AGENT = "research-assistant-mcp/0.1"
   DEFAULT_MAX_RESULTS = 5
   SUMMARY_MAX_LENGTH = 120
   
   mcp = FastMCP("ResearchAssistantServer")
   

2. 论文检索工具 search_papers

   • 异步请求 arXiv API，并解析 Atom Feed：

   async def fetch_arxiv_entries(topic: str, max_results: int) -> list[dict[str, str]]:
       params = {
           "search_query": f"all:{topic}",
           "start": 0,
           "max_results": max(min(max_results, 10), 1),
           "sortBy": "submittedDate",
           "sortOrder": "descending",
       }
       headers = {"User-Agent": USER_AGENT}
       async with httpx.AsyncClient(timeout=30.0, headers=headers, follow_redirects=True) as client:
           response = await client.get(ARXIV_API_ENDPOINT, params=params)
           response.raise_for_status()
       return parse_arxiv_feed(response.text)
   

   • 将结果格式化为 Markdown：

   def format_papers(entries: list[dict[str, str]]) -> str:
       if not entries:
           return "未找到相关论文，可尝试更换关键词。"
       parts = []
       for idx, item in enumerate(entries, start=1):
           published = item["published"][:10] if item["published"] else "未知日期"
           parts.append(
               textwrap.dedent(
                   f"""\
                   {idx}. **{item['title']}**  
                      - 作者：{item['authors']}  
                      - 日期：{published}  
                      - 摘要：{item['summary']}  
                      - 链接：{item['link']}
                   """
               ).strip()
           )
       return "\n\n".join(parts)
   

   • 注册 MCP 工具：

   @mcp.tool()
   async def search_papers(topic: str, max_results: int = DEFAULT_MAX_RESULTS) -> str:
       try:
           entries = await fetch_arxiv_entries(topic, max_results)
       except httpx.HTTPError as exc:
           return f"检索失败：{exc}"
       except ET.ParseError as exc:
           return f"解析 arXiv 数据时出错：{exc}"
       return format_papers(entries)
   

3. 阅读计划工具 build_reading_plan

   def build_time_blocks(available_hours: int, target_papers: int) -> list[tuple[str, float]]:
       available_hours = max(1, min(available_hours, 40))
       target_papers = max(1, min(target_papers, 10))
       per_paper = available_hours / target_papers
       return [
           ("快速浏览与筛选", round(per_paper * 0.3, 1)),
           ("精读与批注", round(per_paper * 0.5, 1)),
           ("总结与回顾", round(per_paper * 0.2, 1)),
       ]
   

   @mcp.tool()
   def build_reading_plan(topic: str, available_hours: int = 6, target_papers: int = 3) -> str:
       return format_reading_plan(topic, available_hours, target_papers)
   

   • format_reading_plan 负责输出 Markdown，包括主题、时间预算、阶段安排和执行建议。

4. 笔记整理工具 organize_notes

   def structure_notes(raw_notes: str) -> str:
       concepts, methods, references = [], [], []
       for line in raw_notes.splitlines():
           normalized = line.strip()
           if not normalized:
               continue
           lowered = normalized.lower()
           if lowered.startswith(("method:", "experiment:", "approach:")):
               methods.append(normalized.split(":", 1)[1].strip())
           elif lowered.startswith(("ref:", "cite:", "doi:")):
               references.append(normalized.split(":", 1)[1].strip())
           else:
               concepts.append(normalized.lstrip("-* "))
       concept_block = format_note_block(concepts)
       method_block = format_note_block(methods)
       reference_block = format_note_block(references)
       return textwrap.dedent(
           f"""\
           ### 核心概念
           - {concept_block}
   
           ### 技术方法 / 实验设计
           - {method_block}
   
           ### 引文 / 参考资料
           - {reference_block}
           """
       ).strip()
   

   @mcp.tool()
   def organize_notes(raw_notes: str) -> str:
       if not raw_notes.strip():
           return "未检测到笔记内容，请输入至少一行文本。"
       return structure_notes(raw_notes)
   

5. 入口函数

   def main() -> None:
       mcp.run(transport="stdio")
   
   if __name__ == "__main__":
       main()
   

6. 客户端 client.py：连接 MCP Server 与大模型

   为了让大模型自动调用上述三个工具，本项目在 src/research_assistant_mcp/client.py 中实现了一个命令行客户端，核心思路如下：

   • 加载环境变量与初始化 LLM 客户端

     from dotenv import load_dotenv
     from openai import OpenAI
     
     load_dotenv()
     
     BASE_URL = os.getenv("BASE_URL")
     MODEL = os.getenv("MODEL")  # 实测可用：deepseek-ai/DeepSeek-V3
     API_KEY = os.getenv("API_KEY")
     
     llm = OpenAI(api_key=API_KEY, base_url=BASE_URL)
     

   • 通过 STDIO 启动并连接本地 MCP Server

     from contextlib import AsyncExitStack
     from mcp import ClientSession, StdioServerParameters, stdio_client
     
     class ResearchAssistantClient:
         def __init__(self, server_script: str) -> None:
             self.server_script = os.path.abspath(server_script)
             self.exit_stack = AsyncExitStack()
             self.session: ClientSession | None = None
             self.llm = OpenAI(api_key=API_KEY, base_url=BASE_URL)
     
         async def connect(self) -> None:
             params = StdioServerParameters(
                 command="python",
                 args=[self.server_script],
                 env=os.environ.copy(),
                 cwd=os.path.dirname(self.server_script),
             )
             read, write = await self.exit_stack.enter_async_context(stdio_client(params))
             self.session = await self.exit_stack.enter_async_context(ClientSession(read, write))
             await self.session.initialize()
     

   • 封装工具调用 _invoke_tool

     async def _invoke_tool(self, name: str, arguments: dict[str, Any]) -> str:
         assert self.session is not None
         result = await self.session.call_tool(name, arguments)
         texts: list[str] = []
         for item in result.content:
             if item.type == "text":
                 texts.append(item.text)
         return "\n\n".join(texts)
     

   • 在对话中启用 Function Calling

     async def chat_with_llm(self) -> None:
         tools = [  # 略，定义 search_papers / build_reading_plan / organize_notes 的 schema
             {...},
         ]
     
         # system 提示与固定问候
         conversation: list[dict[str, Any]] = [
             {
                 "role": "system",
                 "content": "你是一个科研资料云端整理助手，优先使用 MCP 工具完成用户需求。",
             }
         ]
         print("助手：你好，我是“科研资料云端整理助手”。...")
     
         while True:
             user_input = input("\n你：").strip()
             if user_input.lower() in {"q", "quit", "exit"}:
                 break
             conversation.append({"role": "user", "content": user_input})
     
             for _ in range(2):
                 response = self.llm.chat.completions.create(
                     model=MODEL,
                     messages=conversation,
                     tools=tools,
                     tool_choice="auto",
                 )
                 choice = response.choices[0]
     
                 # 若需要调用工具
                 if choice.finish_reason == "tool_calls" and choice.message.tool_calls:
                     conversation.append(
                         {
                             "role": "assistant",
                             "tool_calls": [tc.to_dict() for tc in choice.message.tool_calls],
                         }
                     )
                     for tool_call in choice.message.tool_calls:
                         name = tool_call.function.name
                         args = __import__("json").loads(tool_call.function.arguments or "{}")
                         tool_result = await self._invoke_tool(name, args)
                         conversation.append(
                             {
                                 "role": "tool",
                                 "tool_call_id": tool_call.id,
                                 "name": name,
                                 "content": tool_result,
                             }
                         )
                     continue
     
                 # 返回最终回答
                 assistant_msg = choice.message.content or ""
                 print(f"\n助手：{assistant_msg}")
                 conversation.append({"role": "assistant", "content": assistant_msg})
                 break
     

   如此，用户只需自然语言对话，DeepSeek 模型会在需要时自动调用对应的 MCP 工具，完成论文检索、阅读计划生成与笔记整理等复合任务。

4. 实验步骤（教学流程）

1. 启动 Inspector 并调试 Server（在 E:\research-assistant-mcp 根目录）

   npx @modelcontextprotocol/inspector
   

   • 在 Inspector 界面中选择 Transport Type: STDIO；
   • Command 填写 python，Arguments 填写 src/research_assistant_mcp/server.py，点击连接以拉起本地 Server。
2. 功能调试（Inspector 中）
   • Tools -> List Tools 查看 search_papers, build_reading_plan, organize_notes；
   • 输入示例参数：
     • search_papers: topic=multi-modal llm, max_results=3;
     • build_reading_plan: topic=多模态LLM, available_hours=8, target_papers=4;
     • organize_notes: 粘贴多行笔记，含 method: 与 ref: 前缀。
3. 结果验证
   • 观察 Inspector 输出，确认 Markdown 内容正确；
   • 若检索提示 301，确保 API 使用 HTTPS 并启用 follow_redirects=True。
4. 本地命令行客户端（可选）
   • 使用项目自带的命令行 MCP Client，与 Server 进行交互式测试：

   # 方式一：显式指定 server 脚本路径
   uv run python src/research_assistant_mcp/client.py src/research_assistant_mcp/server.py
   
   # 方式二：使用默认 server 路径（本项目结构未改名时）
   uv run python src/research_assistant_mcp/client.py
   

   • 启动后按照提示输入 papers / plan / notes 指令，并依次填写参数或粘贴笔记内容（以 END 结束），即可在终端中体验论文检索、阅读计划与笔记整理的完整流程。
5. 扩展实践
   • 在 .env 中添加其他 API Key；
   • 借助第 7 章 SSE 教程，将 mcp.run(transport="sse") 部署到云服务器，Inspector 中改用 SSE 连接，支持远程访问；
   • 结合第 6 章客户端示例，将该 MCP Server 接入自定义 MCP Client，实现自动化对话流程。

5. 实验记录模板

结论

本文基于 mcp-lite-dev 项目提出的“科研资料云端整理助手”展示了 MCP Server 在个人效率场景的落地路径。通过统一的工具注册与标准化调试流程，只需实现三个核心工具，即可在 IDE、桌面客户端或云端服务中复用该助手。实验流程以 ch06、ch07 的教学风格整理，使得初学者能够快速完成环境搭建、Server 编写与 Inspector 调试，从而为云计算导论课程的结课论文提供完整、可复现的案例。未来可继续扩展多源检索、向量数据库与团队协同功能，进一步提升智能体在科研领域的价值。

参考文献

1. Datawhale，《MCP 极简开发》，https://github.com/datawhalechina/mcp-lite-dev
2. Model Context Protocol 官方文档：https://modelcontextprotocol.io/
3. arXiv API Help：https://info.arxiv.org/help/api/
4. SiliconFlow 官方网站：https://cloud.siliconflow.cn/
5. Chroma 向量数据库：https://www.trychroma.com/
6. Inspector-MCP中文文档：https://mcp-docs.cn/docs/tools/inspector