Paperlib MCP

Instructions

Implementation Reference

• src/paperlib_mcp/tools/writing.py:203-314 (handler)

  The main handler function for the 'build_evidence_pack' tool. It performs hybrid search on literature chunks, saves the results as an evidence pack in the database (evidence_packs and evidence_pack_items tables), fetches document metadata, and returns structured pack information including items with snippets and scores.

  async def build_evidence_pack(
      query: str,
      k: int = 40,
      per_doc_limit: int = 3,
      alpha: float = 0.6,
  ) -> dict[str, Any]:
      """构建证据包
      
      搜索与主题相关的文献片段，并保存为可复用的证据包。
      证据包可用于多次迭代综述写作，避免每次重新检索导致结果漂移。
      
      Args:
          query: 搜索主题/研究问题
          k: 检索数量，默认 40
          per_doc_limit: 每篇文档最多返回的 chunk 数量，默认 3
          alpha: 向量搜索权重，默认 0.6
          
      Returns:
          证据包信息，包含 pack_id 和检索到的条目
      """
      try:
          # 执行搜索
          search_result = await hybrid_search(
              query=query,
              k=k,
              alpha=alpha,
              per_doc_limit=per_doc_limit,
          )
          
          if not search_result.results:
              return {
                  "error": "No relevant literature found",
                  "query": query,
                  "pack_id": None,
              }
          
          # 保存证据包
          params = {
              "k": k,
              "per_doc_limit": per_doc_limit,
              "alpha": alpha,
          }
          
          with get_db() as conn:
              with conn.cursor() as cur:
                  # 创建证据包
                  cur.execute(
                      """
                      INSERT INTO evidence_packs (query, params_json)
                      VALUES (%s, %s)
                      RETURNING pack_id
                      """,
                      (query, json.dumps(params))
                  )
                  pack_result = cur.fetchone()
                  pack_id = pack_result["pack_id"]
                  
                  # 插入条目
                  for rank, result in enumerate(search_result.results):
                      cur.execute(
                          """
                          INSERT INTO evidence_pack_items (pack_id, doc_id, chunk_id, rank)
                          VALUES (%s, %s, %s, %s)
                          """,
                          (pack_id, result.doc_id, result.chunk_id, rank)
                      )
          
          # 获取文档元数据
          doc_ids = list(set(r.doc_id for r in search_result.results))
          doc_metadata = {}
          for doc_id in doc_ids:
              doc = query_one(
                  "SELECT title, authors, year FROM documents WHERE doc_id = %s",
                  (doc_id,)
              )
              if doc:
                  doc_metadata[doc_id] = doc
          
          # 构建返回结果
          items = []
          for result in search_result.results:
              meta = doc_metadata.get(result.doc_id, {})
              items.append({
                  "doc_id": result.doc_id,
                  "chunk_id": result.chunk_id,
                  "page_start": result.page_start,
                  "page_end": result.page_end,
                  "text": result.snippet,
                  "score": result.score_total,
                  "title": meta.get("title"),
                  "authors": meta.get("authors"),
                  "year": meta.get("year"),
              })
          
          return {
              "pack_id": pack_id,
              "query": query,
              "params": params,
              "items": items,
              "stats": {
                  "total_chunks": len(items),
                  "unique_docs": len(doc_ids),
              },
          }
          
      except Exception as e:
          return {
              "error": str(e),
              "query": query,
              "pack_id": None,
          }

• src/paperlib_mcp/server.py:37-37 (registration)

  The call to register_writing_tools(mcp) which registers the build_evidence_pack tool (and other writing tools) to the FastMCP server instance.

  register_writing_tools(mcp)

• src/paperlib_mcp/tools/writing.py:125-132 (schema)

  Pydantic model defining the structure of an EvidencePack, which matches the return type of the build_evidence_pack tool.

  class EvidencePack(BaseModel):
      """证据包"""
      pack_id: int
      query: str
      params: dict[str, Any]
      items: list[EvidencePackItem]
      stats: dict[str, Any]

• src/paperlib_mcp/tools/writing.py:115-123 (schema)

  Pydantic model for individual items in the EvidencePack returned by the tool.

  class EvidencePackItem(BaseModel):
      """证据包条目"""
      doc_id: str
      chunk_id: int
      page_start: int
      page_end: int
      text: str
      score: float

• src/paperlib_mcp/tools/writing.py:134-197 (helper)

  Helper function to retrieve an evidence pack from the database by pack_id, used in related tools but illustrative of the data structure.

  def get_evidence_pack(pack_id: int) -> EvidencePack | None:
      """获取证据包内容
      
      Args:
          pack_id: 证据包 ID
          
      Returns:
          证据包对象，如果不存在返回 None
      """
      # 获取证据包元数据
      pack = query_one(
          """
          SELECT pack_id, query, params_json, created_at::text
          FROM evidence_packs
          WHERE pack_id = %s
          """,
          (pack_id,)
      )
      
      if not pack:
          return None
      
      # 获取证据包条目
      items = query_all(
          """
          SELECT 
              epi.doc_id,
              epi.chunk_id,
              epi.rank,
              c.page_start,
              c.page_end,
              c.text
          FROM evidence_pack_items epi
          JOIN chunks c ON epi.chunk_id = c.chunk_id
          WHERE epi.pack_id = %s
          ORDER BY epi.rank
          """,
          (pack_id,)
      )
      
      # 统计
      unique_docs = len(set(item["doc_id"] for item in items))
      
      return EvidencePack(
          pack_id=pack["pack_id"],
          query=pack["query"],
          params=pack["params_json"] or {},
          items=[
              EvidencePackItem(
                  doc_id=item["doc_id"],
                  chunk_id=item["chunk_id"],
                  page_start=item["page_start"],
                  page_end=item["page_end"],
                  text=item["text"],
                  score=1.0 / (item["rank"] + 1) if item["rank"] is not None else 0.5,  # 基于排名的伪分数
              )
              for item in items
          ],
          stats={
              "total_chunks": len(items),
              "unique_docs": unique_docs,
          }
      )