analyze_db_health
Check and optimize PostgreSQL database performance with health assessments for indexes, connections, vacuum, sequences, replication, buffer cache, and constraints. Specify individual checks or run all by default.
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
- src/postgres_mcp/server.py:453-480 (handler)MCP tool handler and registration for 'analyze_db_health'. Delegates to DatabaseHealthTool for the actual health checks.@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 valid health_type values used in the tool schema.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 analysis logic, orchestrating various sub-checks.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 for the specified components. Args: health_type: Comma-separated list of health check types to perform Valid values: index, connection, vacuum, sequence, replication, buffer, constraint, all Returns: A string with the health check results """ try: result = "" try: health_types = {HealthType(x.strip()) for x in health_type.split(",")} except ValueError: return ( f"Invalid health types provided: '{health_type}'. " + f"Valid values are: {', '.join(sorted([t.value for t in HealthType]))}. " + "Please try again with a comma-separated list of valid health types." ) if HealthType.ALL in health_types: health_types = [t.value for t in HealthType if t != HealthType.ALL] if HealthType.INDEX in health_types: index_health = IndexHealthCalc(self.sql_driver) result += "Invalid index check: " + await index_health.invalid_index_check() + "\n" result += "Duplicate index check: " + await index_health.duplicate_index_check() + "\n" result += "Index bloat: " + await index_health.index_bloat() + "\n" result += "Unused index check: " + await index_health.unused_indexes() + "\n" if HealthType.CONNECTION in health_types: connection_health = ConnectionHealthCalc(self.sql_driver) result += "Connection health: " + await connection_health.connection_health_check() + "\n" if HealthType.VACUUM in health_types: vacuum_health = VacuumHealthCalc(self.sql_driver) result += "Vacuum health: " + await vacuum_health.transaction_id_danger_check() + "\n" if HealthType.SEQUENCE in health_types: sequence_health = SequenceHealthCalc(self.sql_driver) result += "Sequence health: " + await sequence_health.sequence_danger_check() + "\n" if HealthType.REPLICATION in health_types: replication_health = ReplicationCalc(self.sql_driver) result += "Replication health: " + await replication_health.replication_health_check() + "\n" if HealthType.BUFFER in health_types: buffer_health = BufferHealthCalc(self.sql_driver) result += "Buffer health for indexes: " + await buffer_health.index_hit_rate() + "\n" result += "Buffer health for tables: " + await buffer_health.table_hit_rate() + "\n" if HealthType.CONSTRAINT in health_types: constraint_health = ConstraintHealthCalc(self.sql_driver) result += "Constraint health: " + await constraint_health.invalid_constraints_check() + "\n" return result if result else "No health checks were performed." except Exception as e: logger.error(f"Error calculating database health: {e}", exc_info=True) return f"Error calculating database health: {e}"