feat(literature): add host-side paper tools

CHANGELOG.md

@@ -6,6 +6,10 @@
 > 开发中的用户文案可先写入 `## Unreleased`；该区不会被前端解析，正式发布时再替换为数字版本和日期。
 > 工程口径的完整记录见 `PROGRESS.md` / git log。
 
+## Unreleased
+
+- 文献检索现在可以在服务端直接限定书籍、章节、论文等出版物类型，书籍类查询更准确；平台文献源的访问凭据也不再进入任务沙箱。
+
 ## 0.68.0 — 2026-08-24
 
 - 修复服务更新或多实例切换期间，点击“停止”后对话可能一直停留在“停止中”的问题。

DESIGN.md

@@ -90,7 +90,7 @@ yaml 是手填的,probe 用真实调用对账(basic_chat/parallel_tools/thinking
 ### 3.5 Skill 系统(Anthropic 渐进披露)
 三层加载:Discovery(name+description,几百 token)→ Activation(`load_skill` 完整 SKILL.md)→ Execution(references 按需拉)。写 WHY+WHAT 不写 Step 1/2/3;description 决定触发。
 
-**出版物检索统一入口**(2026-08-25):`literature` 按用户任务而非数据源划分 skill，统一论文、书籍、章节、会议论文、学位论文、报告、标准、专利、预印本及未来出版物的发现、获取、跨源去重与证据核验；`paper_server`、内部材料知识库及未来来源只是可替换后端。出版物类型与来源正交，同一出版物可有多条来源记录；跨源机械层只自动合并 DOI、ISBN+版次、标准号、公开号等精确键或题名+主要责任者+年份完全一致项，疑似重复、不同版次、章节、标准修订版和专利族成员保守保留。题录、摘要、片段、全文四级证据分开标注，下载成功不等于全文已经核验。取舍是不为每个数据库新增 skill，也不把检索编排固化成统一高级工具：skill 负责路由与判断，现有 helper/host-side tool 保持各自信任域，确定性标准化与精确去重下沉到可重建脚本。
+**出版物检索统一入口**(2026-08-25):`literature` 按用户任务而非数据源划分 skill，统一论文、书籍、章节、会议论文、学位论文、报告、标准、专利、预印本及未来出版物的发现、获取、跨源去重与证据核验；`paper_server`、内部材料知识库及未来来源只是可替换后端。出版物类型与来源正交，同一出版物可有多条来源记录；跨源机械层只自动合并 DOI、ISBN+版次、标准号、公开号等精确键或题名+主要责任者+年份完全一致项，疑似重复、不同版次、章节、标准修订版和专利族成员保守保留。题录、摘要、片段、全文四级证据分开标注，下载成功不等于全文已经核验。取舍是不为每个数据库新增 skill，也不把检索编排固化成统一高级工具：skill 负责路由与判断，来源能力保持各自工具边界，确定性标准化与精确去重下沉到可重建脚本。`paper_server` 自 2026-08-26 起使用按配置注册的 host-side `paper_search` / `paper_get` / `paper_fetch`，API Key 不再进入 sandbox；来源原始 `type` 可服务端过滤，统一类型映射仍归出版物模型与合并脚本。
 
 **用户私有 skill**(2026-06-11):registry 收有序来源列表——内置 `ROOT/skills`(只读)+ 用户 `user_root/.skills`(可写)。取舍:① **user wins 同名覆盖**(核心用例是"copy 内置再改",覆盖只作用于本人会话,blast radius 锁死),覆盖显式标注不静默;② **创作走 host-side `save_skill`/`fork_skill`**——fs 工具的 base_dir 跨 backend 够不到 `user_root/.skills`,host 工具一个落点两模式通吃;③ 用户 skill 加载失败收进 `load_errors` 注入 prompt 提示修,不崩整次扫描。
 

PROGRESS.md

@@ -2,7 +2,7 @@
 
 > 配合 `DESIGN.md`。本文件只记 phase 状态、决策偏差、文件量、下一步。每条 1-2 句:做了啥 + 关键判断;细节查 `git log` / `git diff` / `DESIGN §7.9`。
 
-最后更新:2026-08-25(literature 统一出版物检索入口，未发版)
+最后更新:2026-08-26(paper_server 类型过滤与 host-side 凭据隔离，未发版)
 
 ---
 
@@ -20,6 +20,8 @@
 ---
 ## 已完成关键能力
 
+- **08-26 / Unreleased / paper_server 宿主侧受控访问**：新增按配置注册的 `paper_search` / `paper_get` / `paper_fetch` typed tools，`publication_type` 直接映射后端 `type` 过滤且保留开放原始类型；API Key 从 `run_python` 和 Docker sandbox 透传中移除，下载增加同源 URL、100 MiB 上限和原子落盘约束。literature / brief 指引、smoke、RUN、Skill 清单及回归测试同步迁移，线上只读 `type=book` 验证通过，不提升版本号。
+
 - **08-25 / Unreleased / literature 统一出版物检索入口**：删除按数据源拆分的 `research` / `documents` skill，新增 `literature` 统一论文、书籍、章节及未来学术与技术出版物的发现、获取、跨源去重和证据核验；paper_server helper 归入新 skill，内部材料库客户端移到 host-side `tools/` 保持密钥隔离。新增开放出版物模型、来源与证据 references，以及只自动合并精确键、保守标记疑似重复的 `merge_publications.py`；`brief/paper/review/rebuttal/patent/standard/analyze` 和 Skill 清单统一改走新入口，不提升版本号。
 
 - **08-24 / 0.68.0 / Origin 小提琴分布图**：`origin.plot@v2` / adapter 1.3.0 新增 `violin`，每条 `input/y` 系列作为一组原始观测，Worker 复制到受控全 Y 展示表并通过 Origin 2024 原生 `Violin.otpu` 一次性生成分类分布图；默认隐藏冗余图例，保留分类标签并为 X 轴标题预留底部空间。首版不开放逐系列 style、分裂/半小提琴及带箱线变体，避免把未稳定映射的模板属性写入公共契约。固定 QA 新增三组抗压强度分布、中位数 oracle、OPJU 重开与全格式复导；Origin 2024 / originpro 1.1.15 数据、视觉与进程释放门禁通过，Origin/合同/Job/Node 专项 126 项 unittest、Ruff 致命规则、py_compile、diff 检查与独立 adapter 打包通过，未连接或写入数据库。

RUN.md

@@ -31,10 +31,9 @@
   # 离线分析(CIF / POSCAR + SpacegroupAnalyzer + XRDCalculator + CEMENT_PHASES)仍在 sandbox 跑。
   # 申请 https://materialsproject.org/api(免费)
   MP_API_KEY=...
-  # research skill 的 paper_server 文献库:GET 接口要 API Key(paper_server Django admin
-  # 里建 ApiKey,helper 以 ?api_key= 查询参数带上)。research 在 sandbox 里直连 paper_server,
-  # 所以这把 key 会透传进 sandbox / 容器(唯一例外;低价值只读,可随时在 admin 撤销)。
-  # 未设:search/fetch 即报 "paper_server auth failed (HTTP 400, not_authenticated)"。
+  # literature skill 的 paper_server 文献库:GET 接口要 API Key(paper_server Django admin
+  # 里建 ApiKey)。设了后注册 paper_search / paper_get / paper_fetch 三个 host-side tool;
+  # key 不进入 sandbox。未设时本轮不注册这三个工具,literature 会降级使用其他来源。
   PAPER_SERVER_API_KEY=...
   # 可选:覆盖 paper_server 地址(host 与 docker 模式均生效)
   # PAPER_SERVER_URL=http://paper.xxhhcty.xyz:8080
@@ -966,7 +965,7 @@ sudo xfs_quota -x -c "limit -p bhard=10g zcbot_<user_uuid>" /opt
 | DB 级单测把行写进了生产库 | 测试只认 `ZCBOT_TEST_DB_URL`(见「环境」段);别把它指向 `.env` 里的隧道 URL。已发生的:按测试专属 email(`test-*@invalid.local`)过滤清理 usage_events/tasks/scheduled_jobs/users |
 | `ModuleNotFoundError: litellm` | 用了全局 `python`,改 `.venv/Scripts/python.exe ...` |
 | Windows 控制台 emoji 崩 | Python stdout 是 GBK。用 `[OK]` / `[ng]` 等 ASCII 标签(见 memory) |
-| research 报 `paper_server auth failed (HTTP 400, not_authenticated)` | paper_server GET 接口已要求 API Key。`.env` 配 `PAPER_SERVER_API_KEY=`(paper_server Django admin 里建 ApiKey);docker 模式会自动透传进容器,改完重启 web 进程生效 |
+| `paper_search` 等工具未出现或报 paper_server 认证失败 | `.env` 配 `PAPER_SERVER_API_KEY=`(paper_server Django admin 里建 ApiKey)并重启 web 进程；工具在宿主侧调用，key 不进入 sandbox |
 | `db upgrade` 报 `column already exists` | DB 已被改过,`db current` 确认 revision,必要时手 ALTER 或 `db downgrade base` 重来 |
 | Resume 找不到 task | dev SPA 左侧 task 列表看 task_id 是否在;或 `curl /v1/tasks` 拉 |
 | task 删了文件还在 | 现在 `DELETE /v1/tasks/{id}` 是**软删**,本就不动任何磁盘文件(留作语料 + 可恢复);要清磁盘走 `POST /v1/files/delete`。彻底物理删 task(及 messages)留给将来的管理员清理工具;当前如需手动:`psql> DELETE FROM tasks WHERE task_id=...`(messages/usage_events CASCADE) |

SKILL_LIST.md

@@ -1,7 +1,7 @@
 # zcbot Skill 清单
 
 服务对象:中国建筑材料科学研究总院 —— 无机非金属材料 R&D(水泥 / 混凝土 / 玻璃 / 陶瓷 / 耐火 / 新型建材)
-最后更新:2026-08-25(literature 统一所有学术与技术出版物的检索、跨源去重和证据核验)
+最后更新:2026-08-26(literature 的 paper_server 改用宿主侧受控工具并支持出版物类型前置过滤)
 Skill 总数:17
 
 zcbot 的"skill"是一份可加载的工作流脚本(`skills/<name>/SKILL.md` + 配套 templates / scripts / Python helper),模型在识别用户意图后挂载对应 skill,按其内置的阶段化流程产出可交付物。本文档面向**使用方 / 协作方**,按"做什么、什么时候用、什么时候别用、典型产物"组织。
@@ -264,7 +264,7 @@ zcbot 的"skill"是一份可加载的工作流脚本(`skills/<name>/SKILL.md` +
 
 证据统一标记为 `metadata_only` / `abstract_verified` / `snippet_verified` / `fulltext_verified`。下载原件但未实际读取，不算全文已核对；出版物真实存在也不代表它支持当前论断。
 
-**主要能力**:`search` / `get_paper` / `fetch_pdf` / `fetch_xml`，以及 host-side `document_list_kb` / `document_search` / `document_download`；`merge_publications.py` 负责确定性的标准化、精确合并和疑似重复标记。
+**主要能力**:host-side `paper_search` / `paper_get` / `paper_fetch`(支持 `publication_type` 服务端过滤)，以及 `document_list_kb` / `document_search` / `document_download`；`merge_publications.py` 负责确定性的标准化、精确合并和疑似重复标记。
 
 **典型产物**:多类型出版物候选清单、规范化 `publications.json`、可核验摘要或全文、引文与论断证据台账。
 

core/executor_docker.py

@@ -89,20 +89,9 @@ _CONTAINER_ENV = {
     "HOME": "/tmp",
 }
 
-# literature skill 在容器里直连 paper_server,这两个 env 按需透传(host .env → docker exec -e)。
-# PAPER_SERVER_API_KEY 是唯一刻意放进 sandbox 的凭证(低价值只读文献库 key,可随时在
-# paper_server admin 撤销),与「Bocha/ARK 等高价值 key 不入容器」的规矩不冲突;
-# PAPER_SERVER_URL 顺带透传,修掉 host 覆盖 URL 时 docker 模式仍用硬编码默认值的缺口。
-_PASSTHROUGH_ENV = ("PAPER_SERVER_URL", "PAPER_SERVER_API_KEY")
-
-
 def _sandbox_env(extra: Optional[Dict[str, str]] = None) -> Dict[str, str]:
-    """shell / run_python 的容器 env:静态 _CONTAINER_ENV + host 透传项 + 调用点补充。"""
+    """shell / run_python 的容器 env:静态安全环境 + 调用点非敏感补充。"""
     env = dict(_CONTAINER_ENV)
-    for k in _PASSTHROUGH_ENV:
-        v = os.environ.get(k, "").strip()
-        if v:
-            env[k] = v
     if extra:
         env.update(extra)
     return env

core/tool_registry.py

@@ -41,6 +41,7 @@ from tools.materials_project import (
     MaterialsProjectSearchSummaryTool,
 )
 from tools.office_to_pdf import OfficeToPdfTool
+from tools.paper_server import PaperFetchTool, PaperGetTool, PaperSearchTool
 from tools.read_document import ReadDocumentTool
 from tools.register_artifact import RegisterArtifactTool
 from tools.rename_working_dir import RenameWorkingDirTool
@@ -177,6 +178,13 @@ def build_tools(ctx: ToolContext) -> dict[str, Any]:
             MaterialsProjectGetEntriesTool(working_dir=ctx.working_dir_path, **base),
         ]
 
+    def _paper_server() -> list:
+        return [
+            PaperSearchTool(**base),
+            PaperGetTool(**base),
+            PaperFetchTool(working_dir=ctx.working_dir_path, **base),
+        ]
+
     def _external_system_status() -> list:
         return [ExternalSystemListTool(ctx.uid, **base)]
 
@@ -316,6 +324,7 @@ def build_tools(ctx: ToolContext) -> dict[str, Any]:
         # key 绝不进 run_python / 沙箱。
         ("document_search",  _env_set("DOCUMENT_SEARCH_API_KEY"),   _document_search),
         ("materials_project", _env_set("MP_API_KEY"),               _materials_project),
+        ("paper_server",     _env_set("PAPER_SERVER_API_KEY"),      _paper_server),
         (
             "external_system_status",
             lambda: _external_system_status_available(ctx.uid),

scripts/smoke_paper_skill.py

@@ -1,4 +1,4 @@
-"""Smoke: paper_server → zcbot literature skill 三步链路。
+"""Smoke: paper_server → zcbot host-side literature tools 三步链路。
 
 跑法: .venv/Scripts/python.exe scripts/smoke_paper_skill.py
 
@@ -33,7 +33,43 @@ if env_file.exists():
         k, _, v = line.partition("=")
         os.environ.setdefault(k.strip(), v.strip())
 
-from skills.literature.paper import _BASE_URL, fetch_pdf, fetch_xml, get_paper, search
+import json
+
+from tools.paper_server import PaperFetchTool, PaperGetTool, PaperSearchTool
+
+
+_BASE_URL = os.environ.get("PAPER_SERVER_URL", "http://paper.xxhhcty.xyz:8080").rstrip("/")
+
+
+def _json_result(raw: str):
+    if raw.startswith("[Error]"):
+        raise RuntimeError(raw)
+    return json.loads(raw)
+
+
+def search(**kwargs) -> list[dict]:
+    return _json_result(PaperSearchTool().execute(**kwargs))
+
+
+def get_paper(id_or_doi: str) -> dict:
+    return _json_result(PaperGetTool().execute(id_or_doi=id_or_doi))
+
+
+def _fetch(id_or_doi: str, working_dir: str, file_format: str) -> str:
+    raw = PaperFetchTool(working_dir=Path(working_dir)).execute(
+        id_or_doi=id_or_doi, format=file_format
+    )
+    if raw.startswith("[Error]"):
+        raise RuntimeError(raw)
+    return raw.removeprefix("saved:").removesuffix(" (existing)")
+
+
+def fetch_pdf(id_or_doi: str, working_dir: str) -> str:
+    return _fetch(id_or_doi, working_dir, "pdf")
+
+
+def fetch_xml(id_or_doi: str, working_dir: str) -> str:
+    return _fetch(id_or_doi, working_dir, "xml")
 
 
 def _hr(title: str) -> None:
@@ -271,7 +307,7 @@ def step4_fetch_xml() -> None:
 
 def main() -> int:
     print("=" * 60)
-    print("zcbot research skill smoke")
+    print("zcbot literature host-side tools smoke")
     print("=" * 60)
     try:
         step0_trgm_speed()

skills/brief/SKILL.md

@@ -43,16 +43,7 @@ description: 生成科研方向简报(research direction briefing / 重要文献
 
 **先读 `references/journals.md`**。**中文方向先转专业英文术语**(库主语料英文):低碳水泥→low-carbon cement / clinker substitution;SCM→supplementary cementitious materials / fly ash / GGBFS / calcined clay;LC3→limestone calcined clay cement;碳化养护→CO2 curing / carbonation。缩写与全称都试。
 
-**literature / paper_server(逐刊取最新 Elsevier 论文 + DOI)** —— `run_python`:
-
-```python
-from skills.literature.paper import search
-# 逐刊拉最新:publication_name 精确匹配 + 时间窗;list 自带 abstract,看前 200-400 字判切题与分量
-for jname in ["Cement and Concrete Research", "Cement and Concrete Composites",
-              "Construction and Building Materials", "Journal of Cleaner Production"]:
-    papers = search(publication_name=jname, year_gte=2025, limit=50)
-    # 按 publication_date 倒序取最新若干;留重要的(主题居中 + 有实质发现),弃边缘
-```
+**literature / paper_server(逐刊取最新 Elsevier 论文 + DOI)** —— 用宿主侧 `paper_search`。按 `publication_name` 精确匹配并带时间窗；需要限定论文时传 `publication_type="article"`。list 自带 abstract，看前 200–400 字判切题与分量。目标期刊包括 `Cement and Concrete Research`、`Cement and Concrete Composites`、`Construction and Building Materials`、`Journal of Cleaner Production`。
 某刊精确名 0 命中 → 换 `keyword=<方向英文术语>` 再搜,从返回里挑 `publication_name` 命中目标刊的;仍空记"该刊本窗口库内无收录"。
 
 **literature / 内部材料库(取全文,材料类首选)** —— host-side tool `document_search`,中英 query 都行(后端跨语言语义检索);胶凝材料库 `classification_id=1`。取 `md_content` 既做候选也供引文核验抓锚点最顺。
@@ -62,7 +53,7 @@ for jname in ["Cement and Concrete Research", "Cement and Concrete Composites",
 - 汇成证据表 `<task_dir>/evidence.md`:期刊 | 标题 | 第一作者(机构)| 年-月 | 摘要概述 | DOI | 来源(paper_server/materials_library/web)。
 - 跨源去重:按 `literature` 规则合并,同 DOI 一条(内部材料库全文优先作内容证据,paper_server 的 DOI 元数据优先);web 不与论文去重、单列。
 
-> **context 纪律(省时省钱,务必遵守)**:检索结果(尤其全文 abstract)**落进 `evidence.md` / `selected_papers.json` 文件**,**不要在对话里反复 `run_python`/`print` 把整批 abstract 灌进上下文**。工具输出会永久留在 context 并每轮重发——同一批摘要 dump 三次,context 就滚成雪球(实测一次简报因此累计烧 2.5M 输入 token、跑满超时被掐断)。需要看某几篇时按需 `read` 文件片段,看完即弃,别整批重打。
+> **context 纪律(省时省钱,务必遵守)**:检索结果(尤其全文 abstract)**落进 `evidence.md` / `selected_papers.json` 文件**,不要反复查询或把同一批 abstract 灌进上下文。工具输出会永久留在 context 并每轮重发——同一批摘要 dump 三次,context 就滚成雪球(实测一次简报因此累计烧 2.5M 输入 token、跑满超时被掐断)。需要看某几篇时按需 `read` 文件片段,看完即弃,别整批重打。
 
 > **窗口内 0 篇**:如实告知库内该窗口暂无收录(可能该刊本窗口尚未发文),可用 web 补更近的非论文动向,**不脑补文献**。
 

skills/literature/SKILL.md

@@ -25,7 +25,7 @@ description: 检索、获取、合并与核验各类学术和技术出版物，
 ## 来源选择
 
 - 材料主题、性能、配方、工艺、表征或全文语义检索：优先内部材料知识库。
-- DOI、题名、作者、期刊、年份或跨学科发现：优先 `paper_server`。
+- DOI、题名、作者、期刊、年份、出版物类型或跨学科发现：`paper_search` 可用时优先 `paper_server`。
 - 系统调研、综述、重要引用或关键论断：并查可用来源。
 - 用户指定来源：只查指定来源，除非该来源无法满足用户要求并需要说明降级方案。
 - 某一路不可用：继续使用其他来源，并明确实际覆盖范围。

skills/literature/paper.py

@@ -1,218 +0,0 @@
-"""literature skill 的 paper_server 客户端 helper。"""
-from __future__ import annotations
-
-import os
-from pathlib import Path
-from typing import Any, Optional
-
-import httpx
-
-_BASE_URL = os.environ.get("PAPER_SERVER_URL", "http://paper.xxhhcty.xyz:8080").rstrip("/")
-_API = f"{_BASE_URL}/api/resm/paper"
-_TIMEOUT = 30.0
-# paper_server 的 GET 接口要 API Key(走 ?api_key= 查询参数,下载直链也好带)。
-# 这是唯一刻意放进 sandbox 的凭证:run_python env 过滤器对它放行、docker 模式宿主透传;
-# 低价值只读 key,可随时在 paper_server Django admin 撤销。
-_API_KEY = os.environ.get("PAPER_SERVER_API_KEY", "").strip()
-
-
-def _with_key(params: Optional[dict] = None) -> dict:
-    """给请求 params 并上 api_key(未配置时原样返回,由服务端报 401/403)。"""
-    merged = dict(params or {})
-    if _API_KEY:
-        merged["api_key"] = _API_KEY
-    return merged
-
-_LIST_FIELDS = (
-    "id",
-    "doi",
-    "title",
-    "first_author",
-    "first_author_institution",
-    "publication_year",
-    "publication_date",
-    "publication_name",
-    "has_fulltext_pdf",
-    "has_fulltext_xml",
-    "has_abstract",
-    "is_oa",
-    "type",
-    "abstract",
-    "pdf_url",
-    "xml_url",
-)
-
-
-_AUTH_ERR_CODES = {"not_authenticated", "authentication_failed", "permission_denied"}
-
-
-def _raise_with_auth_hint(r: httpx.Response) -> None:
-    """认证类失败转成带指引的报错(比裸 HTTPStatusError 可读),其余走 raise_for_status。
-
-    paper_server 的 custom_exception_hander 把非 401/404 的错误统一压成 400,
-    所以不能只看 status code,要看 body 里的 err_code。
-    """
-    if r.status_code in (400, 401, 403):
-        try:
-            r.read()  # stream 模式下 body 未读,json() 会抛 ResponseNotRead;非 stream 幂等
-            err_code = r.json().get("err_code", "")
-        except Exception:
-            err_code = ""
-        if r.status_code in (401, 403) or err_code in _AUTH_ERR_CODES:
-            raise RuntimeError(
-                f"paper_server auth failed (HTTP {r.status_code}, {err_code or 'no err_code'}): "
-                "PAPER_SERVER_API_KEY not set or invalid"
-            )
-    r.raise_for_status()
-
-
-def _safe_doi(doi: str) -> str:
-    return doi.replace("/", "_")
-
-
-def _is_doi(s: str) -> bool:
-    return "/" in s and s.lstrip().startswith("10.")
-
-
-def _resolve_to_id(id_or_doi: str) -> str:
-    """传 id 直接返,传 doi → 调 list 接口取 id。命中 0 / 多条都抛。"""
-    s = id_or_doi.strip()
-    if not _is_doi(s):
-        return s
-    r = httpx.get(_API + "/", params=_with_key({"doi": s}), timeout=_TIMEOUT)
-    _raise_with_auth_hint(r)
-    data = r.json()
-    results = data.get("results") if isinstance(data, dict) and "results" in data else data
-    if not results:
-        raise ValueError(f"doi 未命中: {s}")
-    if len(results) > 1:
-        raise ValueError(f"doi 命中多条({len(results)}): {s}")
-    return results[0]["id"]
-
-
-def search(
-    keyword: str = "",
-    year: Optional[int] = None,
-    year_gte: Optional[int] = None,
-    year_lte: Optional[int] = None,
-    doi: str = "",
-    first_author: str = "",
-    publication_name: str = "",
-    has_pdf: Optional[bool] = None,
-    is_oa: Optional[bool] = None,
-    limit: int = 10,
-) -> list[dict]:
-    """搜文献,返回精简列表(每条含 abstract 字段,有就是文本,没就是空串)。
-
-    keyword:          paper_server SearchFilter,模糊匹配 title / first_author / first_author_institution
-                      库里主语料是英文,**优先英文关键词**(用户中文输入要先转专业英文术语)
-    year:             精确年份
-    year_gte/year_lte: 年份范围(做"近 N 年文献"用)
-    doi:              精确 DOI(命中 0/1 条)
-    first_author:     精确作者名
-    publication_name: 精确期刊名
-    has_pdf:          True 仅返 PDF 已下好的;False 仅返没 PDF 的;None 都返
-    is_oa:            True 仅返 OA 的;False 仅返非 OA;None 都返
-    limit:            默认 10,上限 50
-    """
-    if limit > 50:
-        limit = 50
-    params: dict[str, Any] = {"page_size": limit}
-    if keyword:
-        params["search"] = keyword
-    if year is not None:
-        params["publication_year"] = year
-    if year_gte is not None:
-        params["publication_year_gte"] = year_gte
-    if year_lte is not None:
-        params["publication_year_lte"] = year_lte
-    if doi:
-        params["doi"] = doi
-    if first_author:
-        params["first_author"] = first_author
-    if publication_name:
-        params["publication_name"] = publication_name
-    if has_pdf is True:
-        params["has_fulltext_pdf"] = "true"
-    elif has_pdf is False:
-        params["has_fulltext_pdf"] = "false"
-    if is_oa is True:
-        params["is_oa"] = "true"
-    elif is_oa is False:
-        params["is_oa"] = "false"
-    r = httpx.get(_API + "/", params=_with_key(params), timeout=_TIMEOUT)
-    _raise_with_auth_hint(r)
-    data = r.json()
-    results = data.get("results") if isinstance(data, dict) and "results" in data else data
-    return [{k: p.get(k) for k in _LIST_FIELDS} for p in results[:limit]]
-
-
-def get_paper(id_or_doi: str) -> dict:
-    """取单条完整 metadata + abstract。
-
-    list 端点已带 abstract,正常工作流不需要调本函数;仅在用户给单个 id/DOI 想拿全字段时用。
-    """
-    pid = _resolve_to_id(id_or_doi)
-    r = httpx.get(f"{_API}/{pid}/", params=_with_key(), timeout=_TIMEOUT)
-    _raise_with_auth_hint(r)
-    return r.json()
-
-
-def _stream_to(url: str, dest: Path) -> None:
-    dest.parent.mkdir(parents=True, exist_ok=True)
-    with httpx.stream("GET", url, params=_with_key(), timeout=60.0) as resp:
-        _raise_with_auth_hint(resp)
-        with open(dest, "wb") as f:
-            for chunk in resp.iter_bytes(chunk_size=64 * 1024):
-                f.write(chunk)
-
-
-def fetch_pdf(id_or_doi: str, working_dir: str) -> str:
-    """下载 PDF 到 <working_dir>/papers/<safe_doi>.pdf,返回相对路径 'papers/<safe_doi>.pdf'。
-
-    走 paper_server media 静态直链(从 list/retrieve 返回的 pdf_url 字段),跟 fetch_xml 同范式。
-    paper.has_fulltext_pdf=False / pdf_url 空(publication_date 缺失时)→ 抛 RuntimeError。
-    已存在跳过下载直接复用。
-    """
-    paper = get_paper(id_or_doi)
-    if not paper.get("has_fulltext_pdf"):
-        reason = paper.get("fail_reason") or "no PDF on server"
-        raise RuntimeError(f"paper has no PDF: id={paper.get('id')} reason={reason}")
-    pdf_url = paper.get("pdf_url") or ""
-    if not pdf_url:
-        raise RuntimeError(
-            f"paper pdf_url unavailable (likely missing publication_date): id={paper.get('id')}"
-        )
-    safe = _safe_doi(paper["doi"])
-    rel = f"papers/{safe}.pdf"
-    dest = Path(working_dir) / rel
-    if dest.exists() and dest.stat().st_size > 0:
-        return rel
-    _stream_to(pdf_url, dest)
-    return rel
-
-
-def fetch_xml(id_or_doi: str, working_dir: str) -> str:
-    """下载 XML 到 <working_dir>/papers/<safe_doi>.xml,返回相对路径 'papers/<safe_doi>.xml'。
-
-    XML 走 paper_server media 静态直链(由 list/retrieve 返回的 xml_url 字段提供);
-    paper_pdf_view 只覆盖 PDF,XML 没对应 API。
-    paper.has_fulltext_xml=False / xml_url 空 → 抛 RuntimeError。
-    已存在跳过下载直接复用。
-    """
-    paper = get_paper(id_or_doi)
-    if not paper.get("has_fulltext_xml"):
-        raise RuntimeError(f"paper has no XML: id={paper.get('id')}")
-    xml_url = paper.get("xml_url") or ""
-    if not xml_url:
-        # publication_date 缺失(unknown 目录)→ paper_server 没暴露这层 media URL
-        raise RuntimeError(
-            f"paper xml_url unavailable (likely missing publication_date): id={paper.get('id')}"
-        )
-    safe = _safe_doi(paper["doi"])
-    rel = f"papers/{safe}.xml"
-    dest = Path(working_dir) / rel
-    if dest.exists() and dest.stat().st_size > 0:
-        return rel
-    _stream_to(xml_url, dest)
-    return rel

skills/literature/references/source-paper-server.md

@@ -4,22 +4,18 @@
 
 ## 调用
 
-通过 `run_python` 使用 helper：
+使用宿主侧 typed tools，凭据不会进入 sandbox：
 
-```python
-from skills.literature.paper import search, get_paper, fetch_pdf, fetch_xml
-```
+- `paper_search(keyword="", publication_type="", year=None, year_gte=None, year_lte=None, doi="", first_author="", publication_name="", has_pdf=None, is_oa=None, limit=10)`：搜索题录并返回摘要和全文可用状态。能确定来源类型时用 `publication_type` 做服务端过滤，例如 `book`、`book-chapter`、`article`。
+- `paper_get(id_or_doi)`：按内部 ID 或 DOI 获取完整记录。
+- `paper_fetch(id_or_doi, format)`：把记录实际提供的 `pdf` 或 `xml` 保存到任务的 `papers/` 目录。
 
-- `search(keyword="", year=None, year_gte=None, year_lte=None, doi="", first_author="", publication_name="", has_pdf=None, is_oa=None, limit=10)`：搜索题录并返回摘要和全文可用状态。
-- `get_paper(id_or_doi)`：按内部 ID 或 DOI 获取完整记录。
-- `fetch_xml(id_or_doi, working_dir)` / `fetch_pdf(id_or_doi, working_dir)`：把可用全文保存到任务的 `papers/` 目录。
-
-helper 自动使用 `PAPER_SERVER_API_KEY` 和可选的 `PAPER_SERVER_URL`。只通过 helper 访问，由它处理认证、URL 和下载路径。
+工具由平台自动使用 `PAPER_SERVER_API_KEY` 和可选的 `PAPER_SERVER_URL`；未配置时本轮不会注册这些工具。只通过这些工具访问，由它们处理认证、URL、响应裁剪和下载路径。
 
 ## 查询与全文
 
 - 将中文概念转换为常用专业英文术语，并按需补中文、缩写、全称或同义词。
-- 用 DOI、题名、作者、出版物、年份和来源返回的 `type` 筛选候选。
+- `publication_type` 使用 paper_server/OpenAlex 的来源原始值；例如统一类型 `book_chapter` 在这里对应 `book-chapter`。能前置确定类型时服务端过滤，跨类型发现或类型不确定时再按来源返回的 `type` 筛选候选。
 - 相关性初筛使用题名和摘要；精确数据、章节、图表或页码必须读取正文。
 - 结构化段落和参考文献优先 XML；版式、页码、公式和图表优先 PDF。只获取记录实际提供的格式。
 

skills/paper/references/citation_verify.md

@@ -18,7 +18,7 @@
 
 每条引文先确认"这篇文献真实存在":
 
-1. 用 `literature` 检索：材料类优先内部材料库语义检索，精确 DOI 使用 paper_server 的 `search()` / `get_paper(doi)`
+1. 用 `literature` 检索：材料类优先内部材料库语义检索，精确 DOI 使用 paper_server 的 `paper_search(doi=...)` / `paper_get(id_or_doi=...)`
 2. 命中 → 记下真实 DOI / 作者 / 年份 / 期刊 / 卷期页;**以库里返回为准**,不沿用记忆里的字段
 3. 两个库都查不到 → 标 `[未核实]`,**不得编造条目**;告诉用户"这条找不到来源,请提供 PDF/DOI 或删去该论断"
 
@@ -34,7 +34,7 @@
 
 最容易翻车的一层:文献存在,但**并不支撑你写的那句话**。逐条做:
 
-1. 抓回该文献的 `md_content`(内部材料库正文片段)/ `fetch_xml` / `fetch_pdf`(paper_server)
+1. 抓回该文献的 `md_content`(内部材料库正文片段)，或用 paper_server 的 `paper_fetch(format="xml"|"pdf")` 获取原件
 2. 在原文里定位与论断相关的**锚点证据**:一句 ≤25 词的原文引语 + 出现的段落/小节位置
 3. 判定支撑度三档:
    - **support**:原文明确支撑该论断 → 通过

tests/test_paper_server_tools.py

@@ -0,0 +1,203 @@
+import os
+import tempfile
+import unittest
+from contextlib import ExitStack
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import patch
+from uuid import uuid4
+
+from tools.paper_server import PaperFetchTool, PaperSearchTool, _media_url
+from tools.run_python import RunPythonTool
+
+
+class PaperServerToolTests(unittest.TestCase):
+    def test_search_forwards_publication_type_to_server_filter(self):
+        captured = {}
+
+        def fake_get_json(url, *, api_key, params=None):
+            captured.update(params or {})
+            self.assertEqual(api_key, "secret")
+            return {"results": [{"id": "p1", "type": "book", "title": "Book"}]}
+
+        with (
+            patch(
+                "tools.paper_server._config",
+                return_value=(
+                    "https://paper.test",
+                    "https://paper.test/api/resm/paper",
+                    "secret",
+                ),
+            ),
+            patch("tools.paper_server._get_json", side_effect=fake_get_json),
+        ):
+            result = PaperSearchTool().execute(publication_type="book", limit=20)
+
+        self.assertEqual(captured["type"], "book")
+        self.assertEqual(captured["page_size"], 20)
+        self.assertIn('"type": "book"', result)
+        self.assertNotIn("secret", result)
+
+    def test_search_accepts_open_raw_type_without_closed_enum(self):
+        with (
+            patch(
+                "tools.paper_server._config",
+                return_value=(
+                    "https://paper.test",
+                    "https://paper.test/api/resm/paper",
+                    "secret",
+                ),
+            ),
+            patch("tools.paper_server._get_json", return_value=[]) as request,
+        ):
+            result = PaperSearchTool().execute(publication_type="future-type")
+        self.assertEqual(request.call_args.kwargs["params"]["type"], "future-type")
+        self.assertEqual(result, "[]")
+
+    def test_search_rejects_unsafe_raw_type(self):
+        with patch("tools.paper_server._get_json") as request:
+            result = PaperSearchTool().execute(publication_type="book&api_key=leak")
+        request.assert_not_called()
+        self.assertTrue(result.startswith("[Error]"))
+
+    def test_media_download_url_must_match_configured_origin(self):
+        self.assertEqual(
+            _media_url("/media/a.pdf", "https://paper.test"),
+            "https://paper.test/media/a.pdf",
+        )
+        with self.assertRaisesRegex(RuntimeError, "out-of-origin"):
+            _media_url("https://attacker.test/a.pdf", "https://paper.test")
+
+    def test_fetch_schema_limits_format(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            tool = PaperFetchTool(working_dir=Path(tmp))
+            self.assertEqual(
+                tool.parameters["properties"]["format"]["enum"], ["pdf", "xml"]
+            )
+
+    def test_fetch_downloads_atomically_and_reuses_existing_file(self):
+        class FakeStreamResponse:
+            status_code = 200
+
+            def __enter__(self):
+                return self
+
+            def __exit__(self, *_args):
+                return False
+
+            def iter_bytes(self, chunk_size=0):
+                self.test_chunk_size = chunk_size
+                yield b"%PDF-test"
+
+        paper = {
+            "id": "p1",
+            "doi": "10.1/example",
+            "has_fulltext_pdf": True,
+            "pdf_url": "/media/example.pdf",
+        }
+        with (
+            tempfile.TemporaryDirectory() as tmp,
+            patch("tools.paper_server._get_paper", return_value=paper),
+            patch(
+                "tools.paper_server._config",
+                return_value=(
+                    "https://paper.test",
+                    "https://paper.test/api/resm/paper",
+                    "secret",
+                ),
+            ),
+            patch(
+                "tools.paper_server.httpx.stream", return_value=FakeStreamResponse()
+            ) as stream,
+        ):
+            tool = PaperFetchTool(working_dir=Path(tmp))
+            first = tool.execute(id_or_doi="p1", format="pdf")
+            second = tool.execute(id_or_doi="p1", format="pdf")
+            destination = Path(tmp) / "papers" / "10.1_example.pdf"
+            saved_bytes = destination.read_bytes()
+
+        self.assertEqual(saved_bytes, b"%PDF-test")
+        self.assertTrue(first.startswith("saved:"))
+        self.assertTrue(second.endswith("(existing)"))
+        self.assertEqual(stream.call_count, 1)
+
+    def test_run_python_no_longer_passes_paper_server_key(self):
+        with patch.dict(os.environ, {"PAPER_SERVER_API_KEY": "secret"}, clear=False):
+            env = RunPythonTool()._filtered_env()
+        self.assertNotIn("PAPER_SERVER_API_KEY", env)
+
+    def test_registry_exposes_host_tools_only_when_platform_key_exists(self):
+        from core.tool_registry import ToolContext, build_tools
+
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            common = dict(
+                tool_base=root,
+                ur_path=root,
+                working_dir_path=root,
+                task_id=uuid4(),
+                uid=uuid4(),
+                cfg={},
+                caps=SimpleNamespace(enable_run_python=False),
+                skills=SimpleNamespace(skills={}),
+                cancel_check=None,
+                scheduled_run=False,
+                deferred_actions=SimpleNamespace(),
+                ark_cfg=None,
+                img_provider="",
+                img_key="",
+                img_cfg=None,
+                img_provider_cfg=None,
+                video_variant="",
+                office_to_pdf_available=False,
+            )
+            gates = dict(
+                DOCUMENT_SEARCH_API_KEY="",
+                MP_API_KEY="",
+                PAPER_SERVER_API_KEY="",
+            )
+
+            def enter_common_patches(stack):
+                stack.enter_context(
+                    patch(
+                        "core.tool_registry._external_system_status_available",
+                        return_value=False,
+                    )
+                )
+                stack.enter_context(
+                    patch(
+                        "core.tool_registry._external_systems_available",
+                        return_value=False,
+                    )
+                )
+                stack.enter_context(
+                    patch("core.tool_registry.smtp_configured", return_value=False)
+                )
+                stack.enter_context(
+                    patch(
+                        "core.tool_registry.wechat_push_available", return_value=False
+                    )
+                )
+                stack.enter_context(
+                    patch("core.tool_registry.lfasr_configured", return_value=False)
+                )
+                stack.enter_context(
+                    patch("core.tool_registry.BochaConfig.load", return_value=None)
+                )
+
+            with ExitStack() as stack:
+                stack.enter_context(patch.dict(os.environ, gates, clear=False))
+                enter_common_patches(stack)
+                without_key = build_tools(ToolContext(**common))
+            gates["PAPER_SERVER_API_KEY"] = "secret"
+            with ExitStack() as stack:
+                stack.enter_context(patch.dict(os.environ, gates, clear=False))
+                enter_common_patches(stack)
+                with_key = build_tools(ToolContext(**common))
+
+        self.assertNotIn("paper_search", without_key)
+        self.assertTrue({"paper_search", "paper_get", "paper_fetch"}.issubset(with_key))
+
+
+if __name__ == "__main__":
+    unittest.main()

tools/paper_server.py

@@ -0,0 +1,362 @@
+"""Host-side paper_server tools.
+
+PAPER_SERVER_API_KEY stays in the host control plane.  The model receives only
+typed business arguments and trimmed JSON / task-local file paths.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import re
+from pathlib import Path
+from typing import Any, Optional
+from urllib.parse import urljoin, urlparse
+
+import httpx
+
+from .base import Tool
+
+_DEFAULT_BASE_URL = "http://paper.xxhhcty.xyz:8080"
+_TIMEOUT = 30.0
+_DOWNLOAD_TIMEOUT = 60.0
+_MAX_DOWNLOAD_BYTES = 100 * 1024 * 1024
+_TYPE_RE = re.compile(r"^[A-Za-z0-9_-]{1,64}$")
+_AUTH_ERR_CODES = {"not_authenticated", "authentication_failed", "permission_denied"}
+_LIST_FIELDS = (
+    "id",
+    "doi",
+    "title",
+    "first_author",
+    "first_author_institution",
+    "publication_year",
+    "publication_date",
+    "publication_name",
+    "has_fulltext_pdf",
+    "has_fulltext_xml",
+    "has_abstract",
+    "is_oa",
+    "type",
+    "abstract",
+    "pdf_url",
+    "xml_url",
+)
+
+
+def _config() -> tuple[str, str, str]:
+    base_url = os.environ.get("PAPER_SERVER_URL", _DEFAULT_BASE_URL).strip().rstrip("/")
+    api_key = os.environ.get("PAPER_SERVER_API_KEY", "").strip()
+    if not api_key:
+        raise RuntimeError("PAPER_SERVER_API_KEY env 未设置,无法查询 paper_server")
+    return base_url, f"{base_url}/api/resm/paper", api_key
+
+
+def _params(api_key: str, values: Optional[dict[str, Any]] = None) -> dict[str, Any]:
+    result = dict(values or {})
+    result["api_key"] = api_key
+    return result
+
+
+def _raise_response_error(response: httpx.Response) -> None:
+    err_code = ""
+    if response.status_code in (400, 401, 403):
+        try:
+            response.read()
+            body = response.json()
+            if isinstance(body, dict):
+                err_code = str(body.get("err_code") or "")
+        except Exception:
+            pass
+    if response.status_code in (401, 403) or err_code in _AUTH_ERR_CODES:
+        raise RuntimeError(
+            f"paper_server auth failed (HTTP {response.status_code}, "
+            f"{err_code or 'no err_code'}):请管理员检查平台 PAPER_SERVER_API_KEY"
+        )
+    if response.status_code >= 400:
+        # 不调用 raise_for_status():HTTPStatusError 会携带含 api_key 的完整请求 URL。
+        raise RuntimeError(f"paper_server request failed (HTTP {response.status_code})")
+
+
+def _get_json(
+    url: str, *, api_key: str, params: Optional[dict[str, Any]] = None
+) -> Any:
+    try:
+        response = httpx.get(
+            url,
+            params=_params(api_key, params),
+            timeout=_TIMEOUT,
+        )
+    except httpx.RequestError as exc:
+        raise RuntimeError(
+            f"paper_server connection failed: {type(exc).__name__}"
+        ) from None
+    _raise_response_error(response)
+    try:
+        return response.json()
+    except ValueError:
+        raise RuntimeError("paper_server returned invalid JSON") from None
+
+
+def _results(data: Any) -> list[dict[str, Any]]:
+    values = (
+        data.get("results") if isinstance(data, dict) and "results" in data else data
+    )
+    if not isinstance(values, list):
+        raise RuntimeError("paper_server returned an unexpected result shape")
+    return [item for item in values if isinstance(item, dict)]
+
+
+def _is_doi(value: str) -> bool:
+    return "/" in value and value.lstrip().startswith("10.")
+
+
+def _resolve_to_id(id_or_doi: str, api_url: str, api_key: str) -> str:
+    value = str(id_or_doi or "").strip()
+    if not value:
+        raise ValueError("id_or_doi 不能为空")
+    if not _is_doi(value):
+        return value
+    matches = _results(_get_json(api_url + "/", api_key=api_key, params={"doi": value}))
+    if not matches:
+        raise ValueError(f"doi 未命中:{value}")
+    if len(matches) > 1:
+        raise ValueError(f"doi 命中多条({len(matches)}):{value}")
+    paper_id = str(matches[0].get("id") or "").strip()
+    if not paper_id:
+        raise RuntimeError("paper_server DOI 查询结果缺少 id")
+    return paper_id
+
+
+def _get_paper(id_or_doi: str) -> dict[str, Any]:
+    _base_url, api_url, api_key = _config()
+    paper_id = _resolve_to_id(id_or_doi, api_url, api_key)
+    data = _get_json(f"{api_url}/{paper_id}/", api_key=api_key)
+    if not isinstance(data, dict):
+        raise RuntimeError("paper_server returned an unexpected paper shape")
+    return data
+
+
+def _safe_stem(value: str) -> str:
+    stem = re.sub(r"[^A-Za-z0-9._-]+", "_", value.strip()).strip("._")
+    return stem[:180] or "paper"
+
+
+def _media_url(raw_url: str, base_url: str) -> str:
+    url = urljoin(base_url + "/", raw_url)
+    parsed = urlparse(url)
+    expected = urlparse(base_url)
+    if parsed.scheme not in ("http", "https") or (
+        parsed.scheme.lower(),
+        parsed.netloc.lower(),
+    ) != (expected.scheme.lower(), expected.netloc.lower()):
+        raise RuntimeError("paper_server returned an out-of-origin media URL")
+    return url
+
+
+class PaperSearchTool(Tool):
+    name = "paper_search"
+    description = (
+        "Search the platform paper_server metadata collection. "
+        "Use publication_type for server-side source-type filtering, e.g. 'book', "
+        "'book-chapter' or 'article'. publication_type uses paper_server/OpenAlex raw "
+        "values; normalized literature type book_chapter is represented here as raw "
+        "'book-chapter'. Prefer English keywords."
+    )
+    parameters = {
+        "type": "object",
+        "properties": {
+            "keyword": {
+                "type": "string",
+                "description": "Fuzzy title/author/institution query.",
+            },
+            "publication_type": {
+                "type": "string",
+                "description": "Exact raw source type, e.g. book or book-chapter.",
+            },
+            "year": {"type": "integer", "description": "Exact publication year."},
+            "year_gte": {"type": "integer", "description": "Minimum publication year."},
+            "year_lte": {"type": "integer", "description": "Maximum publication year."},
+            "doi": {"type": "string", "description": "Exact DOI."},
+            "first_author": {
+                "type": "string",
+                "description": "Exact first-author name.",
+            },
+            "publication_name": {
+                "type": "string",
+                "description": "Exact journal/container name.",
+            },
+            "has_pdf": {
+                "type": "boolean",
+                "description": "Filter by PDF availability.",
+            },
+            "is_oa": {
+                "type": "boolean",
+                "description": "Filter by open-access status.",
+            },
+            "limit": {
+                "type": "integer",
+                "default": 10,
+                "description": "Maximum records, 1-50.",
+            },
+        },
+    }
+
+    def execute(
+        self,
+        keyword: str = "",
+        publication_type: str = "",
+        year: Optional[int] = None,
+        year_gte: Optional[int] = None,
+        year_lte: Optional[int] = None,
+        doi: str = "",
+        first_author: str = "",
+        publication_name: str = "",
+        has_pdf: Optional[bool] = None,
+        is_oa: Optional[bool] = None,
+        limit: int = 10,
+    ) -> str:
+        raw_type = str(publication_type or "").strip()
+        if raw_type and not _TYPE_RE.fullmatch(raw_type):
+            return (
+                "[Error] publication_type 只能包含字母、数字、下划线或连字符,"
+                "最长 64 字符"
+            )
+        limit = min(max(int(limit), 1), 50)
+        params: dict[str, Any] = {"page_size": limit}
+        optional = {
+            "search": str(keyword or "").strip(),
+            "type": raw_type,
+            "doi": str(doi or "").strip(),
+            "first_author": str(first_author or "").strip(),
+            "publication_name": str(publication_name or "").strip(),
+        }
+        params.update({key: value for key, value in optional.items() if value})
+        if year is not None:
+            params["publication_year"] = int(year)
+        if year_gte is not None:
+            params["publication_year_gte"] = int(year_gte)
+        if year_lte is not None:
+            params["publication_year_lte"] = int(year_lte)
+        if has_pdf is not None:
+            params["has_fulltext_pdf"] = "true" if has_pdf else "false"
+        if is_oa is not None:
+            params["is_oa"] = "true" if is_oa else "false"
+        try:
+            _base_url, api_url, api_key = _config()
+            papers = _results(_get_json(api_url + "/", api_key=api_key, params=params))
+        except Exception as exc:
+            return f"[Error] paper_search failed:{type(exc).__name__}:{exc}"
+        trimmed = [
+            {key: paper.get(key) for key in _LIST_FIELDS} for paper in papers[:limit]
+        ]
+        return json.dumps(trimmed, ensure_ascii=False, indent=2)
+
+
+class PaperGetTool(Tool):
+    name = "paper_get"
+    description = (
+        "Get one complete paper_server metadata record by internal id or exact DOI."
+    )
+    parameters = {
+        "type": "object",
+        "properties": {
+            "id_or_doi": {"type": "string", "description": "paper_server id or DOI."},
+        },
+        "required": ["id_or_doi"],
+    }
+
+    def execute(self, id_or_doi: str) -> str:
+        try:
+            paper = _get_paper(id_or_doi)
+        except Exception as exc:
+            return f"[Error] paper_get failed:{type(exc).__name__}:{exc}"
+        return json.dumps(paper, ensure_ascii=False, indent=2)
+
+
+class PaperFetchTool(Tool):
+    name = "paper_fetch"
+    description = (
+        "Download an available PDF or XML from paper_server into the current task's "
+        "papers/ directory. Use the format actually reported by paper_search/paper_get."
+    )
+    parameters = {
+        "type": "object",
+        "properties": {
+            "id_or_doi": {"type": "string", "description": "paper_server id or DOI."},
+            "format": {
+                "type": "string",
+                "enum": ["pdf", "xml"],
+                "description": "File format to download.",
+            },
+        },
+        "required": ["id_or_doi", "format"],
+    }
+
+    def __init__(
+        self,
+        *,
+        working_dir: Path,
+        base_dir: Optional[Path] = None,
+        user_root: Optional[Path] = None,
+    ) -> None:
+        super().__init__(base_dir=base_dir, user_root=user_root)
+        self.working_dir = Path(working_dir)
+
+    def execute(self, id_or_doi: str, format: str) -> str:  # noqa: A002 - JSON tool contract
+        file_format = str(format or "").lower().strip()
+        if file_format not in {"pdf", "xml"}:
+            return "[Error] format 必须是 pdf 或 xml"
+        try:
+            paper = _get_paper(id_or_doi)
+            available_key = f"has_fulltext_{file_format}"
+            if not paper.get(available_key):
+                reason = (
+                    paper.get("fail_reason") or f"no {file_format.upper()} on server"
+                )
+                raise RuntimeError(f"paper has no {file_format.upper()}:{reason}")
+            raw_url = str(paper.get(f"{file_format}_url") or "").strip()
+            if not raw_url:
+                raise RuntimeError(f"paper {file_format}_url unavailable")
+            base_url, _api_url, api_key = _config()
+            media_url = _media_url(raw_url, base_url)
+            identity = str(paper.get("doi") or paper.get("id") or id_or_doi)
+            destination = (
+                self.working_dir / "papers" / f"{_safe_stem(identity)}.{file_format}"
+            )
+            if destination.exists() and destination.stat().st_size > 0:
+                return f"saved:{self._display(destination)} (existing)"
+            destination.parent.mkdir(parents=True, exist_ok=True)
+            partial = destination.with_suffix(destination.suffix + ".part")
+            total = 0
+            try:
+                with httpx.stream(
+                    "GET",
+                    media_url,
+                    params=_params(api_key),
+                    timeout=_DOWNLOAD_TIMEOUT,
+                ) as response:
+                    _raise_response_error(response)
+                    with partial.open("wb") as handle:
+                        for chunk in response.iter_bytes(chunk_size=64 * 1024):
+                            total += len(chunk)
+                            if total > _MAX_DOWNLOAD_BYTES:
+                                message = (
+                                    "paper_server file exceeds "
+                                    f"{_MAX_DOWNLOAD_BYTES} bytes"
+                                )
+                                raise RuntimeError(message)
+                            handle.write(chunk)
+                partial.replace(destination)
+            except httpx.RequestError as exc:
+                raise RuntimeError(
+                    f"paper_server download connection failed:{type(exc).__name__}"
+                ) from None
+            finally:
+                if partial.exists():
+                    try:
+                        partial.unlink()
+                    except OSError:
+                        pass
+        except Exception as exc:
+            return f"[Error] paper_fetch failed:{type(exc).__name__}:{exc}"
+        return f"saved:{self._display(destination)}"

tools/run_python.py

@@ -22,9 +22,7 @@ from .base import Tool
 from .output import compact_tool_output, format_timeout_result
 
 _SENSITIVE_PATTERNS = ("API_KEY", "TOKEN", "SECRET", "PASSWORD", "PRIVATE_KEY")
-# 刻意放行的例外:literature skill 在 sandbox 里直连 paper_server,这把只读文献库 key
-# 的消费方就是沙盒代码本身;低价值、可随时在 paper_server admin 撤销。
-_ENV_ALLOWLIST = frozenset({"PAPER_SERVER_API_KEY"})
+_ENV_ALLOWLIST: frozenset[str] = frozenset()
 
 
 class RunPythonTool(Tool):