Metadata-Version: 2.4
Name: research-assistant-mcp-huangshuai
Version: 0.1.0
Summary: 科研资料云端整理助手（基于 MCP Server 与 DeepSeek 模型）
Requires-Python: >=3.10
Description-Content-Type: text/markdown
Requires-Dist: httpx>=0.28.1
Requires-Dist: mcp>=1.10.1
Requires-Dist: openai>=1.93.3
Requires-Dist: python-dotenv>=1.1.1
Requires-Dist: textwrap3>=0.9.2

# 科研资料云端整理助手：基于 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`**（示例）
   ```toml
   [build-system]
   requires = ["setuptools>=80.9.0", "wheel>=0.45.1"]
   build-backend = "setuptools.build_meta"

   [project]
   name = "research-assistant-mcp"
   version = "0.1.0"
   description = "科研资料云端整理助手"
   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"
   ]

   [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.10` 安装依赖。

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. **服务初始化**
   ```python
   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：
   ```python
   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：
   ```python
   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 工具：
   ```python
   @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`**
   ```python
   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)),
       ]
   ```
   ```python
   @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`**
   ```python
   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()
   ```
   ```python
   @mcp.tool()
   def organize_notes(raw_notes: str) -> str:
       if not raw_notes.strip():
           return "未检测到笔记内容，请输入至少一行文本。"
       return structure_notes(raw_notes)
   ```

5. **入口函数**
   ```python
   def main() -> None:
       mcp.run(transport="stdio")

   if __name__ == "__main__":
       main()
   ```

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

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

   - **加载环境变量与初始化 LLM 客户端**
     ```python
     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**
     ```python
     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`**
     ```python
     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**
     ```python
     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` 根目录）
   ```bash
   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 进行交互式测试：
   ```bash
   # 方式一：显式指定 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. 实验记录模板

| 步骤 | 预期结果 | 实际输出 | 问题排查 |
|------|----------|----------|----------|
| 环境准备 | `uv --version` 显示版本 | | |
| 启动 server | 终端显示 `ResearchAssistantServer ready` | | |
| Inspector 调试 | `List Tools` 列出 3 个工具 | | |
| 功能测试 | `search_papers` 返回论文列表 | | |
| SSE 部署（可选） | 浏览器访问 `http://0.0.0.0:8000/sse` 返回 200 | | |

## 结论

本文基于 `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
