Paperlib MCP

Instructions

Implementation Reference

• src/paperlib_mcp/tools/graph_claim_grouping.py:178-241 (handler)

  The @mcp.tool()-decorated handler function that executes the core logic: constructs a SQL query to fetch grouped claims filtered by comm_id or pack_id, aggregates top claims per group, and returns structured output.

  @mcp.tool()
  def export_claim_matrix_grouped_v1(
      comm_id: int | None = None,
      pack_id: int | None = None,
  ) -> dict[str, Any]:
      """导出按分组聚合的结论矩阵。"""
      try:
          where_clauses = []
          params = []
          
          if comm_id:
              where_clauses.append("""
                  EXISTS (
                      SELECT 1 FROM claim_group_members cgm
                      JOIN claims c ON c.claim_id = cgm.claim_id
                      JOIN mentions m ON m.doc_id = c.doc_id
                      JOIN community_members cm ON cm.entity_id = m.entity_id
                      WHERE cm.comm_id = %s AND cgm.group_id = g.group_id
                  )
              """)
              params.append(comm_id)
          elif pack_id:
              where_clauses.append("""
                  EXISTS (
                      SELECT 1 FROM claim_group_members cgm
                      JOIN claims c ON c.claim_id = cgm.claim_id
                      JOIN evidence_pack_items epi ON epi.chunk_id = c.chunk_id
                      WHERE epi.pack_id = %s AND cgm.group_id = g.group_id
                  )
              """)
              params.append(pack_id)
          
          where_sql = " WHERE " + " AND ".join(where_clauses) if where_clauses else ""
          
          sql = f"""
              SELECT 
                  g.group_id, g.group_key, g.sign, g.id_family, g.setting,
                  e.canonical_name as topic_name,
                  (SELECT COUNT(*) FROM claim_group_members cgm WHERE cgm.group_id = g.group_id) as member_count,
                  (
                      SELECT json_agg(json_build_object(
                          'claim_id', c.claim_id,
                          'doc_id', c.doc_id,
                          'claim_text', c.claim_text,
                          'confidence', c.confidence
                      ))
                      FROM claim_group_members cgm
                      JOIN claims c ON c.claim_id = cgm.claim_id
                      WHERE cgm.group_id = g.group_id
                      LIMIT 5
                  ) as top_claims
              FROM claim_groups g
              LEFT JOIN entities e ON e.entity_id = g.topic_entity_id
              {where_sql}
              ORDER BY member_count DESC
          """
          rows = query_all(sql, tuple(params))
          
          return ExportClaimMatrixGroupedOut(groups=rows).model_dump()
          
      except Exception as e:
          return ExportClaimMatrixGroupedOut(
              error=MCPErrorModel(code="SYSTEM_ERROR", message=str(e))
          ).model_dump()

• src/paperlib_mcp/models_graph.py:444-453 (schema)

  Pydantic models defining input (comm_id or pack_id optional) and output (list of group dicts with details like group_id, topic_name, member_count, top_claims; optional error) schemas for the tool.

  class ExportClaimMatrixGroupedIn(BaseModel):
      """export_claim_matrix_grouped_v1 输入"""
      comm_id: Optional[int] = None
      pack_id: Optional[int] = None
  
  
  class ExportClaimMatrixGroupedOut(BaseModel):
      """export_claim_matrix_grouped_v1 输出"""
      groups: list[dict[str, Any]] = Field(default_factory=list)
      error: Optional[MCPErrorModel] = None

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

  Top-level registration call in the main MCP server that invokes the module's register function, thereby registering export_claim_matrix_grouped_v1 via its @mcp.tool() decorator.

  register_graph_claim_grouping_tools(mcp)

• src/paperlib_mcp/tools/graph_claim_grouping.py:35-242 (registration)

  Module-level registration function that defines and registers both build_claim_groups_v1 and export_claim_matrix_grouped_v1 tools using @mcp.tool() decorators.

  def register_graph_claim_grouping_tools(mcp: FastMCP) -> None:
      """注册结论分组工具"""
  
      @mcp.tool()
      def build_claim_groups_v1(
          scope: str = "all",
          max_claims_per_doc: int = 100,  # v1.1: raised from 20 to improve coverage
          dry_run: bool = False,
      ) -> dict[str, Any]:
          """对结论进行分组/聚类。
          
          Args:
              scope: 处理范围，"all", "comm_id:...", "doc_ids:id1,id2"
              max_claims_per_doc: 每个文档最多处理多少条结论
              dry_run: 是否仅预览
          """
          try:
              # 1. 确定范围
              params = []
              where_clauses = []
              
              if scope.startswith("comm_id:"):
                  comm_id = int(scope.split(":", 1)[1])
                  where_clauses.append("""
                      EXISTS (
                          SELECT 1 FROM community_members cm
                          JOIN mentions m ON m.entity_id = cm.entity_id
                          WHERE cm.comm_id = %s AND m.doc_id = c.doc_id
                      )
                  """)
                  params.append(comm_id)
              elif scope.startswith("doc_ids:"):
                  doc_ids = scope.split(":", 1)[1].split(",")
                  where_clauses.append("c.doc_id = ANY(%s)")
                  params.append(doc_ids)
  
              where_sql = " WHERE " + " AND ".join(where_clauses) if where_clauses else ""
              
              # 2. 获取结论及关联信息
              sql = f"""
                  SELECT 
                      c.claim_id, c.doc_id, c.claim_text, c.sign, c.conditions,
                      (SELECT e.canonical_key FROM relations r 
                       JOIN entities e ON e.entity_id = r.obj_entity_id
                       WHERE r.subj_entity_id = (SELECT entity_id FROM entities WHERE type='Paper' AND canonical_key = c.doc_id)
                       AND r.predicate = 'PAPER_HAS_TOPIC' LIMIT 1) as topic_key,
                      (SELECT e.canonical_key FROM relations r 
                       JOIN entities e ON e.entity_id = r.obj_entity_id
                       WHERE r.subj_entity_id = (SELECT entity_id FROM entities WHERE type='Paper' AND canonical_key = c.doc_id)
                       AND r.predicate = 'PAPER_IDENTIFIES_WITH' LIMIT 1) as id_family
                  FROM claims c
                  {where_sql}
                  ORDER BY c.doc_id, c.confidence DESC
              """
              claims = query_all(sql, tuple(params))
              
              if not claims:
                  return BuildClaimGroupsOut(new_groups=0, total_members=0).model_dump()
  
              # 3. 分组逻辑
              groups: dict[str, list[int]] = defaultdict(list)
              group_details: dict[str, dict] = {}
              
              doc_counts: dict[str, int] = defaultdict(int)
              
              for c in claims:
                  if doc_counts[c["doc_id"]] >= max_claims_per_doc:
                      continue
                  doc_counts[c["doc_id"]] += 1
                  
                  # 构造 group_key
                  topic_key = c["topic_key"] or "unknown_topic"
                  sign = c["sign"] or "null"
                  id_family = c["id_family"] or "general"
                  
                  # 尝试从 conditions 或 claim_text 提取 outcome/treatment
                  conditions = c["conditions"] or {}
                  outcome_family = conditions.get("outcome_family") or extract_family(c["claim_text"], "outcome_gen")
                  treatment_family = conditions.get("treatment_family") or extract_family(c["claim_text"], "treatment_gen")
                  setting = conditions.get("setting") or "general"
                  
                  group_key = f"{topic_key}|{outcome_family}|{treatment_family}|{sign}|{id_family}|{setting}"
                  groups[group_key].append(c["claim_id"])
                  
                  if group_key not in group_details:
                      # 查找 topic_entity_id
                      topic_ent = query_one("SELECT entity_id FROM entities WHERE canonical_key = %s", (topic_key,))
                      group_details[group_key] = {
                          "topic_entity_id": topic_ent["entity_id"] if topic_ent else None,
                          "sign": sign,
                          "setting": setting,
                          "id_family": id_family
                      }
  
              if dry_run:
                  return BuildClaimGroupsOut(
                      new_groups=len(groups),
                      total_members=sum(len(v) for v in groups.values())
                  ).model_dump()
  
              # 4. 写入数据库
              total_members = 0
              with get_db() as conn:
                  for key, claim_ids in groups.items():
                      details = group_details[key]
                      try:
                          with conn.cursor() as cur:
                              with conn.transaction():
                                  cur.execute("""
                                      INSERT INTO claim_groups (group_key, topic_entity_id, sign, setting, id_family)
                                      VALUES (%s, %s, %s, %s, %s)
                                      ON CONFLICT (group_key) DO UPDATE SET updated_at = now()
                                      RETURNING group_id
                                  """, (key, details["topic_entity_id"], details["sign"], details["setting"], details["id_family"]))
                                  group_id = cur.fetchone()["group_id"]
                                  
                                  for cid in claim_ids:
                                      cur.execute("""
                                          INSERT INTO claim_group_members (group_id, claim_id)
                                          VALUES (%s, %s)
                                          ON CONFLICT (group_id, claim_id) DO NOTHING
                                      """, (group_id, cid))
                                      total_members += cur.rowcount
                      except Exception as e:
                          print(f"Error processing group {key}: {e}")
  
                  # 计算总数 (在 commit 前查询)
                  with conn.cursor() as cur:
                      cur.execute("SELECT COUNT(*) as count FROM claim_groups")
                      new_groups_total = cur.fetchone()["count"]
  
              return BuildClaimGroupsOut(
                  new_groups=new_groups_total,
                  total_members=total_members
              ).model_dump()
  
          except Exception as e:
              return BuildClaimGroupsOut(
                  new_groups=0,
                  total_members=0,
                  error=MCPErrorModel(code="SYSTEM_ERROR", message=str(e))
              ).model_dump()
  
      @mcp.tool()
      def export_claim_matrix_grouped_v1(
          comm_id: int | None = None,
          pack_id: int | None = None,
      ) -> dict[str, Any]:
          """导出按分组聚合的结论矩阵。"""
          try:
              where_clauses = []
              params = []
              
              if comm_id:
                  where_clauses.append("""
                      EXISTS (
                          SELECT 1 FROM claim_group_members cgm
                          JOIN claims c ON c.claim_id = cgm.claim_id
                          JOIN mentions m ON m.doc_id = c.doc_id
                          JOIN community_members cm ON cm.entity_id = m.entity_id
                          WHERE cm.comm_id = %s AND cgm.group_id = g.group_id
                      )
                  """)
                  params.append(comm_id)
              elif pack_id:
                  where_clauses.append("""
                      EXISTS (
                          SELECT 1 FROM claim_group_members cgm
                          JOIN claims c ON c.claim_id = cgm.claim_id
                          JOIN evidence_pack_items epi ON epi.chunk_id = c.chunk_id
                          WHERE epi.pack_id = %s AND cgm.group_id = g.group_id
                      )
                  """)
                  params.append(pack_id)
              
              where_sql = " WHERE " + " AND ".join(where_clauses) if where_clauses else ""
              
              sql = f"""
                  SELECT 
                      g.group_id, g.group_key, g.sign, g.id_family, g.setting,
                      e.canonical_name as topic_name,
                      (SELECT COUNT(*) FROM claim_group_members cgm WHERE cgm.group_id = g.group_id) as member_count,
                      (
                          SELECT json_agg(json_build_object(
                              'claim_id', c.claim_id,
                              'doc_id', c.doc_id,
                              'claim_text', c.claim_text,
                              'confidence', c.confidence
                          ))
                          FROM claim_group_members cgm
                          JOIN claims c ON c.claim_id = cgm.claim_id
                          WHERE cgm.group_id = g.group_id
                          LIMIT 5
                      ) as top_claims
                  FROM claim_groups g
                  LEFT JOIN entities e ON e.entity_id = g.topic_entity_id
                  {where_sql}
                  ORDER BY member_count DESC
              """
              rows = query_all(sql, tuple(params))
              
              return ExportClaimMatrixGroupedOut(groups=rows).model_dump()
              
          except Exception as e:
              return ExportClaimMatrixGroupedOut(
                  error=MCPErrorModel(code="SYSTEM_ERROR", message=str(e))
              ).model_dump()