Paperlib MCP

Instructions

Implementation Reference

• src/paperlib_mcp/tools/graph_extract.py:887-969 (handler)

  The core handler function for the 'graph_health_check' tool. It checks the existence of required GraphRAG tables and indexes in the database, optionally includes row counts, and returns a structured health status using GraphHealthCheckOut model.

  def graph_health_check(include_counts: bool = True) -> dict[str, Any]:
      """检查 GraphRAG 层健康状态
      
      验证 M2 GraphRAG 所需的表和索引是否存在，并返回统计信息。
      
      Args:
          include_counts: 是否包含各表的行数统计，默认 True
          
      Returns:
          健康状态信息，包含：
          - ok: 整体状态是否正常
          - db_ok: 数据库连接状态
          - tables_ok: 必要表是否存在
          - indexes_ok: 必要索引是否存在
          - counts: 各表行数（可选）
      """
      try:
          notes = []
          
          # 检查表存在性
          tables_result = query_all(
              """
              SELECT table_name 
              FROM information_schema.tables
              WHERE table_schema = 'public' 
              AND table_name = ANY(%s)
              """,
              (REQUIRED_TABLES,)
          )
          existing_tables = {r["table_name"] for r in tables_result}
          missing_tables = set(REQUIRED_TABLES) - existing_tables
          tables_ok = len(missing_tables) == 0
          
          if missing_tables:
              notes.append(f"Missing tables: {', '.join(sorted(missing_tables))}")
          
          # 检查索引存在性
          indexes_result = query_all(
              """
              SELECT indexname 
              FROM pg_indexes
              WHERE schemaname = 'public' 
              AND indexname = ANY(%s)
              """,
              (REQUIRED_INDEXES,)
          )
          existing_indexes = {r["indexname"] for r in indexes_result}
          missing_indexes = set(REQUIRED_INDEXES) - existing_indexes
          indexes_ok = len(missing_indexes) == 0
          
          if missing_indexes:
              notes.append(f"Missing indexes: {', '.join(sorted(missing_indexes))}")
          
          # 获取统计信息
          counts = None
          if include_counts and tables_ok:
              counts = {}
              for table in REQUIRED_TABLES:
                  if table in existing_tables:
                      result = query_one(f"SELECT COUNT(*) as count FROM {table}")
                      counts[table] = result["count"] if result else 0
          
          ok = tables_ok and indexes_ok
          
          return GraphHealthCheckOut(
              ok=ok,
              db_ok=True,
              tables_ok=tables_ok,
              indexes_ok=indexes_ok,
              notes=notes,
              counts=counts,
          ).model_dump()
          
      except Exception as e:
          return GraphHealthCheckOut(
              ok=False,
              db_ok=False,
              tables_ok=False,
              indexes_ok=False,
              notes=[str(e)],
              error=MCPErrorModel(code="DB_CONN_ERROR", message=str(e)),
          ).model_dump()

• src/paperlib_mcp/models_graph.py:63-81 (schema)

  Pydantic schema definitions for the input (GraphHealthCheckIn) and output (GraphHealthCheckOut) of the graph_health_check tool.

  # graph_health_check 工具模型
  # ============================================================
  
  
  class GraphHealthCheckIn(BaseModel):
      """graph_health_check 输入"""
      include_counts: bool = True
  
  
  class GraphHealthCheckOut(BaseModel):
      """graph_health_check 输出"""
      ok: bool
      db_ok: bool
      tables_ok: bool
      indexes_ok: bool
      notes: list[str] = Field(default_factory=list)
      counts: Optional[dict[str, int]] = None
      error: Optional[MCPErrorModel] = None

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

  Registration of the graph_extract tools module, which includes the graph_health_check tool, in the main MCP server.

  register_graph_extract_tools(mcp)