analyze_db_health
Analyze PostgreSQL database health by checking indexes, connections, vacuum status, sequences, replication, buffer cache, and constraints to identify performance issues and maintenance needs.
Instructions
Analyzes database health. Here are the available health checks:
index - checks for invalid, duplicate, and bloated indexes
connection - checks the number of connection and their utilization
vacuum - checks vacuum health for transaction id wraparound
sequence - checks sequences at risk of exceeding their maximum value
replication - checks replication health including lag and slots
buffer - checks for buffer cache hit rates for indexes and tables
constraint - checks for invalid constraints
all - runs all checks You can optionally specify a single health check or a comma-separated list of health checks. The default is 'all' checks.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| health_type | No | Optional. Valid values are: all, buffer, connection, constraint, index, replication, sequence, vacuum. | all |
Implementation Reference
- The MCP tool handler and registration for 'analyze_db_health'. This function executes the tool logic by instantiating DatabaseHealthTool and invoking its health method.@mcp.tool( description="Analyzes database health. Here are the available health checks:\n" "- index - checks for invalid, duplicate, and bloated indexes\n" "- connection - checks the number of connection and their utilization\n" "- vacuum - checks vacuum health for transaction id wraparound\n" "- sequence - checks sequences at risk of exceeding their maximum value\n" "- replication - checks replication health including lag and slots\n" "- buffer - checks for buffer cache hit rates for indexes and tables\n" "- constraint - checks for invalid constraints\n" "- all - runs all checks\n" "You can optionally specify a single health check or a comma-separated list of health checks. The default is 'all' checks." ) async def analyze_db_health( health_type: str = Field( description=f"Optional. Valid values are: {', '.join(sorted([t.value for t in HealthType]))}.", default="all", ), ) -> ResponseType: """Analyze database health for specified components. Args: health_type: Comma-separated list of health check types to perform. Valid values: index, connection, vacuum, sequence, replication, buffer, constraint, all """ health_tool = DatabaseHealthTool(await get_sql_driver()) result = await health_tool.health(health_type=health_type) return format_text_response(result)
- Enum defining the valid health_type parameter values, used for input validation in the tool's Pydantic Field.class HealthType(str, Enum): INDEX = "index" CONNECTION = "connection" VACUUM = "vacuum" SEQUENCE = "sequence" REPLICATION = "replication" BUFFER = "buffer" CONSTRAINT = "constraint" ALL = "all"
- Core helper class implementing the database health checks. The health() method performs the specified checks using specialized calculator classes and formats the results.class DatabaseHealthTool: """Tool for analyzing database health metrics.""" def __init__(self, sql_driver): self.sql_driver = sql_driver async def health(self, health_type: str) -> str: """Run database health checks and return compact, labeled results.""" try: def compact(label: str, text: str) -> str: # Join lines, strip bullets and excess spaces lines = [ln.strip().lstrip("- •") for ln in text.splitlines() if ln.strip()] return f"{label}: " + ("; ".join(lines) if lines else "(none)") try: health_types = {HealthType(x.strip()) for x in health_type.split(",")} except ValueError: return f"Invalid health types: '{health_type}'. Valid: " + ", ".join(sorted([t.value for t in HealthType])) if HealthType.ALL in health_types: health_types = [t.value for t in HealthType if t != HealthType.ALL] out: list[str] = [] if HealthType.INDEX in health_types: index_health = IndexHealthCalc(self.sql_driver) out.append(compact("index.invalid", await index_health.invalid_index_check())) out.append(compact("index.duplicate", await index_health.duplicate_index_check())) out.append(compact("index.bloat", await index_health.index_bloat())) out.append(compact("index.unused", await index_health.unused_indexes())) if HealthType.CONNECTION in health_types: connection_health = ConnectionHealthCalc(self.sql_driver) out.append(compact("connection", await connection_health.connection_health_check())) if HealthType.VACUUM in health_types: vacuum_health = VacuumHealthCalc(self.sql_driver) out.append(compact("vacuum.txid", await vacuum_health.transaction_id_danger_check())) if HealthType.SEQUENCE in health_types: sequence_health = SequenceHealthCalc(self.sql_driver) out.append(compact("sequence", await sequence_health.sequence_danger_check())) if HealthType.REPLICATION in health_types: replication_health = ReplicationCalc(self.sql_driver) out.append(compact("replication", await replication_health.replication_health_check())) if HealthType.BUFFER in health_types: buffer_health = BufferHealthCalc(self.sql_driver) out.append(compact("buffer.index_hit", await buffer_health.index_hit_rate())) out.append(compact("buffer.table_hit", await buffer_health.table_hit_rate())) if HealthType.CONSTRAINT in health_types: constraint_health = ConstraintHealthCalc(self.sql_driver) out.append(compact("constraint", await constraint_health.invalid_constraints_check())) return "\n".join([line for line in out if line]) if out else "No health checks performed" except Exception as e: logger.error(f"Error calculating database health: {e}", exc_info=True) return f"Error calculating database health: {e}"