Paperlib MCP

Instructions

Implementation Reference

• src/paperlib_mcp/tools/graph_summarize.py:532-780 (handler)

  Core handler function that executes the tool logic: determines relevant documents from community or topic, queries relations and claims to build paper_matrix (doc metadata + attributes) and claim_matrix, returns structured JSON.

  @mcp.tool()
  def export_evidence_matrix_v1(
      comm_id: int | None = None,
      topic: str | None = None,
      format: str = "json",
      limit_docs: int | None = None,
  ) -> dict[str, Any]:
      """导出证据矩阵
      
      导出 PaperMatrix（论文级）和 ClaimMatrix（结论级）两张表。
      
      Args:
          comm_id: 社区 ID（与 topic 二选一）
          topic: 主题名称或 canonical_key（与 comm_id 二选一）
          format: 输出格式，"json" 或 "csv"
          limit_docs: 限制文档数量
          
      Returns:
          paper_matrix 和 claim_matrix
      """
      try:
          if not comm_id and not topic:
              return ExportEvidenceMatrixOut(
                  error=MCPErrorModel(code="VALIDATION_ERROR", message="Must provide either comm_id or topic"),
              ).model_dump()
          
          # 确定文档集合
          if comm_id:
              # 从社区获取文档
              doc_ids_result = query_all(
                  """
                  SELECT DISTINCT m.doc_id
                  FROM community_members cm
                  JOIN mentions m ON m.entity_id = cm.entity_id
                  WHERE cm.comm_id = %s
                  """,
                  (comm_id,)
              )
          else:
              # 从主题获取文档
              # 先查找 topic entity
              topic_entity = query_one(
                  """
                  SELECT entity_id FROM entities
                  WHERE type = 'Topic' AND (canonical_name ILIKE %s OR canonical_key ILIKE %s)
                  LIMIT 1
                  """,
                  (f"%{topic}%", f"%{topic}%")
              )
              
              if not topic_entity:
                  return ExportEvidenceMatrixOut(
                      error=MCPErrorModel(code="NOT_FOUND", message=f"Topic '{topic}' not found"),
                  ).model_dump()
              
              doc_ids_result = query_all(
                  """
                  SELECT DISTINCT m.doc_id
                  FROM mentions m
                  WHERE m.entity_id = %s
                  """,
                  (topic_entity["entity_id"],)
              )
          
          doc_ids = [r["doc_id"] for r in doc_ids_result]
          
          if not doc_ids:
              return ExportEvidenceMatrixOut(
                  error=MCPErrorModel(code="NOT_FOUND", message="No documents found"),
              ).model_dump()
          
          if limit_docs:
              doc_ids = doc_ids[:limit_docs]
          
          # ===== PaperMatrix =====
          # 获取文档元数据
          docs = query_all(
              """
              SELECT doc_id, title, authors, year, venue, doi
              FROM documents
              WHERE doc_id = ANY(%s)
              """,
              (doc_ids,)
          )
          doc_meta = {d["doc_id"]: d for d in docs}
          
          # 获取每个文档的 topics
          doc_topics = query_all(
              """
              SELECT DISTINCT p.canonical_key AS doc_id, x.canonical_name AS topic
              FROM relations r
              JOIN entities p ON p.entity_id = r.subj_entity_id AND p.type = 'Paper'
              JOIN entities x ON x.entity_id = r.obj_entity_id AND x.type = 'Topic'
              WHERE r.predicate = 'PAPER_HAS_TOPIC' AND p.canonical_key = ANY(%s)
              """,
              (doc_ids,)
          )
          topics_by_doc: dict[str, list[str]] = defaultdict(list)
          for r in doc_topics:
              topics_by_doc[r["doc_id"]].append(r["topic"])
          
          # 获取每个文档的 measures
          doc_measures = query_all(
              """
              SELECT DISTINCT p.canonical_key AS doc_id, x.canonical_name AS measure
              FROM relations r
              JOIN entities p ON p.entity_id = r.subj_entity_id AND p.type = 'Paper'
              JOIN entities x ON x.entity_id = r.obj_entity_id AND x.type = 'MeasureProxy'
              WHERE r.predicate = 'PAPER_USES_MEASURE' AND p.canonical_key = ANY(%s)
              """,
              (doc_ids,)
          )
          measures_by_doc: dict[str, list[str]] = defaultdict(list)
          for r in doc_measures:
              measures_by_doc[r["doc_id"]].append(r["measure"])
          
          # 获取每个文档的 identification strategies
          doc_ids_strat = query_all(
              """
              SELECT DISTINCT p.canonical_key AS doc_id, x.canonical_name AS id_strategy
              FROM relations r
              JOIN entities p ON p.entity_id = r.subj_entity_id AND p.type = 'Paper'
              JOIN entities x ON x.entity_id = r.obj_entity_id AND x.type = 'IdentificationStrategy'
              WHERE r.predicate = 'PAPER_IDENTIFIES_WITH' AND p.canonical_key = ANY(%s)
              """,
              (doc_ids,)
          )
          ids_by_doc: dict[str, list[str]] = defaultdict(list)
          for r in doc_ids_strat:
              ids_by_doc[r["doc_id"]].append(r["id_strategy"])
          
          # 获取每个文档的 methods
          doc_methods = query_all(
              """
              SELECT DISTINCT p.canonical_key AS doc_id, x.canonical_name AS method
              FROM relations r
              JOIN entities p ON p.entity_id = r.subj_entity_id AND p.type = 'Paper'
              JOIN entities x ON x.entity_id = r.obj_entity_id AND x.type = 'Method'
              WHERE r.predicate = 'PAPER_USES_METHOD' AND p.canonical_key = ANY(%s)
              """,
              (doc_ids,)
          )
          methods_by_doc: dict[str, list[str]] = defaultdict(list)
          for r in doc_methods:
              methods_by_doc[r["doc_id"]].append(r["method"])
          
          # 获取每个文档的 settings
          doc_settings = query_all(
              """
              SELECT DISTINCT p.canonical_key AS doc_id, x.canonical_name AS setting
              FROM relations r
              JOIN entities p ON p.entity_id = r.subj_entity_id AND p.type = 'Paper'
              JOIN entities x ON x.entity_id = r.obj_entity_id AND x.type = 'Setting'
              WHERE r.predicate = 'PAPER_IN_SETTING' AND p.canonical_key = ANY(%s)
              """,
              (doc_ids,)
          )
          settings_by_doc: dict[str, list[str]] = defaultdict(list)
          for r in doc_settings:
              settings_by_doc[r["doc_id"]].append(r["setting"])
          
          # 获取每个文档的 top claims
          doc_claims = query_all(
              """
              SELECT doc_id, claim_text, sign, confidence, chunk_id
              FROM claims
              WHERE doc_id = ANY(%s)
              ORDER BY doc_id, confidence DESC
              """,
              (doc_ids,)
          )
          claims_by_doc: dict[str, list[dict]] = defaultdict(list)
          for c in doc_claims:
              if len(claims_by_doc[c["doc_id"]]) < 3:  # Top 3 claims per doc
                  claims_by_doc[c["doc_id"]].append({
                      "claim_text": c["claim_text"],
                      "sign": c["sign"],
                      "chunk_id": c["chunk_id"],
                  })
          
          # 构建 PaperMatrix
          paper_matrix = []
          for doc_id in doc_ids:
              meta = doc_meta.get(doc_id, {})
              paper_matrix.append({
                  "doc_id": doc_id,
                  "title": meta.get("title", ""),
                  "authors": meta.get("authors", ""),
                  "year": meta.get("year"),
                  "venue": meta.get("venue", ""),
                  "topics": topics_by_doc.get(doc_id, []),
                  "measures": measures_by_doc.get(doc_id, []),
                  "identification_strategies": ids_by_doc.get(doc_id, []),
                  "methods": methods_by_doc.get(doc_id, []),
                  "settings": settings_by_doc.get(doc_id, []),
                  "top_claims": claims_by_doc.get(doc_id, []),
              })
          
          # ===== ClaimMatrix =====
          all_claims = query_all(
              """
              SELECT 
                  c.claim_id,
                  c.doc_id,
                  c.chunk_id,
                  c.claim_text,
                  c.sign,
                  c.effect_size_text,
                  c.conditions,
                  c.confidence,
                  ch.page_start,
                  ch.page_end
              FROM claims c
              JOIN chunks ch ON ch.chunk_id = c.chunk_id
              WHERE c.doc_id = ANY(%s)
              ORDER BY c.confidence DESC
              """,
              (doc_ids,)
          )
          
          claim_matrix = []
          for claim in all_claims:
              doc_id = claim["doc_id"]
              claim_matrix.append({
                  "claim_id": claim["claim_id"],
                  "doc_id": doc_id,
                  "chunk_id": claim["chunk_id"],
                  "claim_text": claim["claim_text"],
                  "sign": claim["sign"],
                  "effect_size_text": claim["effect_size_text"],
                  "conditions": claim["conditions"],
                  "confidence": claim["confidence"],
                  "page_start": claim["page_start"],
                  "page_end": claim["page_end"],
                  # 补充 doc 级信息
                  "topics": topics_by_doc.get(doc_id, []),
                  "identification_strategies": ids_by_doc.get(doc_id, []),
              })
          
          return ExportEvidenceMatrixOut(
              paper_matrix=paper_matrix,
              claim_matrix=claim_matrix,
          ).model_dump()
          
      except Exception as e:
          return ExportEvidenceMatrixOut(
              error=MCPErrorModel(code="DB_CONN_ERROR", message=str(e)),
          ).model_dump()

• src/paperlib_mcp/models_graph.py:330-347 (schema)

  Pydantic models defining the input parameters and output structure (paper_matrix and claim_matrix lists) for validation and typing.

  # export_evidence_matrix_v1 工具模型
  # ============================================================
  
  
  class ExportEvidenceMatrixIn(BaseModel):
      """export_evidence_matrix_v1 输入"""
      comm_id: Optional[int] = None
      topic: Optional[str] = None             # topic 名称或 canonical_key
      format: Literal["json", "csv"] = "json"
      limit_docs: Optional[int] = None
  
  
  class ExportEvidenceMatrixOut(BaseModel):
      """export_evidence_matrix_v1 输出"""
      paper_matrix: list[dict[str, Any]] = Field(default_factory=list)
      claim_matrix: list[dict[str, Any]] = Field(default_factory=list)
      error: Optional[MCPErrorModel] = None

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

  Invocation of register_graph_summarize_tools(mcp) in the main MCP server setup, which defines and registers the export_evidence_matrix_v1 tool using @mcp.tool() decorator.

  register_graph_summarize_tools(mcp)