Postgres MCP Pro Plus

Instructions

Implementation Reference

• src/postgres_mcp_pro_plus/server.py:562-581 (handler)

  Primary MCP tool handler for 'analyze_workload_indexes', including @mcp.tool registration decorator and Pydantic input schema. Orchestrates analysis by choosing DTA or LLM tuner and calling TextPresentation.analyze_workload.
  @mcp.tool(description="Analyze frequently executed queries in the database and recommend optimal indexes") @validate_call async def analyze_workload_indexes( max_index_size_mb: int = Field(description="Max index size in MB", default=10000), method: Literal["dta", "llm"] = Field(description="Method to use for analysis", default="dta"), ) -> ResponseType: """Analyze frequently executed queries in the database and recommend optimal indexes.""" try: sql_driver = await get_sql_driver() if method == "dta": index_tuning = DatabaseTuningAdvisor(sql_driver) else: index_tuning = LLMOptimizerTool(sql_driver) dta_tool = TextPresentation(sql_driver, index_tuning) result = await dta_tool.analyze_workload(max_index_size_mb=max_index_size_mb) return format_text_response(result) except Exception as e: logger.error(f"Error analyzing workload: {e}") return format_error_response(str(e))
• src/postgres_mcp_pro_plus/index/presentation.py:34-54 (helper)

  Helper in TextPresentation class: delegates to index_tuning.analyze_workload with workload parameters and formats the IndexTuningResult into structured text response.
  async def analyze_workload(self, max_index_size_mb=10000): """ Analyze SQL workload and recommend indexes. This method analyzes queries from database query history, examining frequently executed and costly queries to recommend the most beneficial indexes. Args: max_index_size_mb: Maximum total size for recommended indexes in MB Returns: String with recommendations or error """ result = await self._execute_analysis( min_calls=50, min_avg_time_ms=5.0, limit=100, max_index_size_mb=max_index_size_mb, ) return self._format_as_text(result)
• src/postgres_mcp_pro_plus/index/index_opt_base.py:182-295 (helper)

  Core helper in IndexTuningBase: implements workload analysis - fetches top queries from pg_stat_statements, parses SQL, uses HypoPG for hypothetical index cost evaluation, greedily selects best indexes within size budget.
  async def analyze_workload( self, workload: list[dict[str, Any]] | None = None, sql_file: str | None = None, query_list: list[str] | None = None, min_calls: int = 50, min_avg_time_ms: float = 5.0, limit: int = MAX_NUM_INDEX_TUNING_QUERIES, max_index_size_mb: int = -1, ) -> IndexTuningResult: """ Analyze query workload and recommend indexes. This method can analyze workload from three different sources (in order of priority): 1. Explicit workload passed as a parameter 2. Direct list of SQL queries passed as query_list 3. SQL file with queries 4. Query statistics from pg_stat_statements Args: workload: Optional explicit workload data sql_file: Optional path to a file containing SQL queries query_list: Optional list of SQL query strings to analyze min_calls: Minimum number of calls for a query to be considered (for pg_stat_statements) min_avg_time_ms: Minimum average execution time in ms (for pg_stat_statements) limit: Maximum number of queries to analyze (for pg_stat_statements) max_index_size_mb: Maximum total size of recommended indexes in MB Returns: IndexTuningResult with analysis results """ session_id = str(int(time.time())) self._analysis_start_time = time.time() self._dta_traces = [] # Reset traces at start of analysis # Clear the cache at the beginning of each analysis self._size_estimate_cache = {} if max_index_size_mb > 0: self.budget_mb = max_index_size_mb session = IndexTuningResult( session_id=session_id, budget_mb=max_index_size_mb, ) try: # Run pre-checks precheck_result = await self._run_prechecks(session) if precheck_result: return precheck_result # First try to use explicit workload if provided if workload: logger.debug(f"Using explicit workload with {len(workload)} queries") session.workload_source = "args" session.workload = workload # Then try direct query list if provided elif query_list: logger.debug(f"Using provided query list with {len(query_list)} queries") session.workload_source = "query_list" session.workload = [] for i, query in enumerate(query_list): # Create a synthetic workload entry for each query session.workload.append( { "query": query, "queryid": f"direct-{i}", } ) # Then try SQL file if provided elif sql_file: logger.debug(f"Reading queries from file: {sql_file}") session.workload_source = "sql_file" session.workload = self._get_workload_from_file(sql_file) # Finally fall back to query stats else: logger.debug("Using query statistics from the database") session.workload_source = "query_store" session.workload = await self._get_query_stats(min_calls, min_avg_time_ms, limit) if not session.workload: logger.warning("No workload to analyze") return session session.workload = await self._validate_and_parse_workload(session.workload) query_weights = self._covert_workload_to_query_weights(session.workload) if query_weights is None or len(query_weights) == 0: self.dta_trace("No query provided") session.recommendations = [] else: # Gather queries as strings workload_queries = [q for q, _, _ in query_weights] self.dta_trace(f"Workload queries ({len(workload_queries)}): {pp_list(workload_queries)}") # Generate and evaluate index recommendations recommendations: tuple[set[IndexRecommendation], float] = await self._generate_recommendations(query_weights) session.recommendations = await self._format_recommendations(query_weights, recommendations) # Reset HypoPG only once at the end await self.sql_driver.execute_query("SELECT hypopg_reset();") except Exception as e: logger.error(f"Error in workload analysis: {e}", exc_info=True) session.error = f"Error in workload analysis: {e}" session.dta_traces = self._dta_traces return session