Skip to main content
Glama

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
NameRequiredDescriptionDefault
include_countsNo

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
  • Registration of the graph_extract tools module, which includes the graph_health_check tool, in the main MCP server.
    register_graph_extract_tools(mcp)

Latest Blog Posts

MCP directory API

We provide all the information about MCP servers via our MCP API.

curl -X GET 'https://glama.ai/api/mcp/v1/servers/h-lu/paperlib-mcp'

If you have feedback or need assistance with the MCP directory API, please join our Discord server