Postgres MCP Pro

Instructions

Implementation Reference

• src/postgres_mcp/server.py:482-506 (handler)

  Tool handler for get_top_queries which parses inputs and dispatches to the TopQueriesCalc class.

  @mcp.tool(
      name="get_top_queries",
      description=f"Reports the slowest or most resource-intensive queries using data from the '{PG_STAT_STATEMENTS}' extension.",
  )
  async def get_top_queries(
      sort_by: str = Field(
          description="Ranking criteria: 'total_time' for total execution time or 'mean_time' for mean execution time per call, or 'resources' "
          "for resource-intensive queries",
          default="resources",
      ),
      limit: int = Field(description="Number of queries to return when ranking based on mean_time or total_time", default=10),
  ) -> ResponseType:
      try:
          sql_driver = await get_sql_driver()
          top_queries_tool = TopQueriesCalc(sql_driver=sql_driver)
  
          if sort_by == "resources":
              result = await top_queries_tool.get_top_resource_queries()
              return format_text_response(result)
          elif sort_by == "mean_time" or sort_by == "total_time":
              # Map the sort_by values to what get_top_queries_by_time expects
              result = await top_queries_tool.get_top_queries_by_time(limit=limit, sort_by="mean" if sort_by == "mean_time" else "total")
          else:
              return format_error_response("Invalid sort criteria. Please use 'resources' or 'mean_time' or 'total_time'.")
          return format_text_response(result)

• src/postgres_mcp/top_queries/top_queries_calc.py:38-80 (handler)

  The core logic implementation for retrieving top queries based on time (mean or total).

  async def get_top_queries_by_time(self, limit: int = 10, sort_by: Literal["total", "mean"] = "mean") -> str:
      """Reports the slowest SQL queries based on execution time.
  
      Args:
          limit: Number of slow queries to return
          sort_by: Sort criteria - 'total' for total execution time or
              'mean' for mean execution time per call (default)
  
      Returns:
          A string with the top queries or installation instructions
      """
      try:
          logger.debug(f"Getting top queries by time. limit={limit}, sort_by={sort_by}")
          extension_status = await check_extension(
              self.sql_driver,
              PG_STAT_STATEMENTS,
              include_messages=False,
          )
  
          if not extension_status.is_installed:
              logger.warning(f"Extension {PG_STAT_STATEMENTS} is not installed")
              # Return installation instructions if the extension is not installed
              return install_pg_stat_statements_message
  
          # Check PostgreSQL version to determine column names
          pg_version = await get_postgres_version(self.sql_driver)
          logger.debug(f"PostgreSQL version: {pg_version}")
  
          # Column names changed in PostgreSQL 13
          if pg_version >= 13:
              # PostgreSQL 13 and newer
              total_time_col = "total_exec_time"
              mean_time_col = "mean_exec_time"
          else:
              # PostgreSQL 12 and older
              total_time_col = "total_time"
              mean_time_col = "mean_time"
  
          logger.debug(f"Using time columns: total={total_time_col}, mean={mean_time_col}")
  
          # Determine which column to sort by based on sort_by parameter and version
          order_by_column = total_time_col if sort_by == "total" else mean_time_col