graph_health_check
Verify GraphRAG layer health by checking required tables and indexes exist, returning status and optional row counts for Paperlib MCP's knowledge graph.
Instructions
检查 GraphRAG 层健康状态
验证 M2 GraphRAG 所需的表和索引是否存在，并返回统计信息。
Args: include_counts: 是否包含各表的行数统计，默认 True
Returns: 健康状态信息，包含： - ok: 整体状态是否正常 - db_ok: 数据库连接状态 - tables_ok: 必要表是否存在 - indexes_ok: 必要索引是否存在 - counts: 各表行数（可选）
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| include_counts | No |
Implementation Reference
- 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()
- 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)