analyze_db_health

Instructions

Implementation Reference

• src/postgres_mcp_pro_plus/server.py:610-636 (handler)

  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)

• src/postgres_mcp_pro_plus/database_health/database_health.py:22-31 (schema)

  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"

• src/postgres_mcp_pro_plus/database_health/database_health.py:33-94 (helper)

  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}"