PostgreSQL MCP Server

Instructions

Implementation Reference

• src/tools/query.ts:325-369 (handler)

  The primary handler function for the 'pg_manage_query' tool. Validates input using the schema, dispatches to operation-specific logic via executeManageQuery, handles errors, and returns formatted ToolOutput.

  async execute(params: unknown, getConnectionString: GetConnectionStringFn): Promise<ToolOutput> {
    const validationResult = ManageQueryInputSchema.safeParse(params);
    if (!validationResult.success) {
      return { 
        content: [{ type: 'text', text: `Invalid input: ${validationResult.error.format()}` }], 
        isError: true 
      };
    }
  
    try {
      const result = await executeManageQuery(validationResult.data, getConnectionString);
      
      let message: string;
      switch (validationResult.data.operation) {
        case 'explain':
          message = validationResult.data.analyze 
            ? 'Query execution plan with runtime statistics' 
            : 'Query execution plan';
          break;
        case 'get_slow_queries':
          message = `Top ${validationResult.data.limit || 10} slow queries ordered by ${validationResult.data.orderBy || 'mean_time'}`;
          break;
        case 'get_stats':
          message = `Query statistics ordered by ${validationResult.data.orderBy || 'total_time'}`;
          break;
        case 'reset_stats':
          message = (result as { message: string }).message;
          break;
        default:
          message = 'Query operation completed';
      }
      
      return { 
        content: [
          { type: 'text', text: message }, 
          { type: 'text', text: JSON.stringify(result, null, 2) }
        ] 
      };
    } catch (error) {
      const errorMessage = error instanceof McpError ? error.message : (error instanceof Error ? error.message : String(error));
      return { 
        content: [{ type: 'text', text: `Error in query operation: ${errorMessage}` }], 
        isError: true 
      };
    }

• src/tools/query.ts:48-74 (schema)

  Zod schema defining the input parameters for the pg_manage_query tool, including operation type and specific parameters for explain, slow queries, stats, and reset operations.

  const ManageQueryInputSchema = z.object({
    operation: z.enum(['explain', 'get_slow_queries', 'get_stats', 'reset_stats']).describe(
      'Operation: explain (EXPLAIN/EXPLAIN ANALYZE query), get_slow_queries (find slow queries from pg_stat_statements), get_stats (query statistics with cache hit ratios), reset_stats (reset pg_stat_statements)'
    ),
    connectionString: z.string().optional(),
    
    // EXPLAIN operation parameters
    query: z.string().optional().describe('SQL query to explain (required for explain operation)'),
    analyze: z.boolean().optional().default(false).describe('Use EXPLAIN ANALYZE - actually executes the query (for explain operation)'),
    buffers: z.boolean().optional().default(false).describe('Include buffer usage information (for explain operation)'),
    verbose: z.boolean().optional().default(false).describe('Include verbose output (for explain operation)'),
    costs: z.boolean().optional().default(true).describe('Include cost estimates (for explain operation)'),
    format: z.enum(['text', 'json', 'xml', 'yaml']).optional().default('json').describe('Output format (for explain operation)'),
    
    // GET_SLOW_QUERIES operation parameters
    limit: z.number().optional().default(10).describe('Number of slow queries to return (for get_slow_queries operation)'),
    minDuration: z.number().optional().describe('Minimum average duration in milliseconds (for get_slow_queries operation)'),
    orderBy: z.enum(['mean_time', 'total_time', 'calls', 'cache_hit_ratio']).optional().default('mean_time').describe('Sort order (for get_slow_queries and get_stats operations)'),
    includeNormalized: z.boolean().optional().default(true).describe('Include normalized query text (for get_slow_queries operation)'),
    
    // GET_STATS operation parameters
    minCalls: z.number().optional().describe('Minimum number of calls (for get_stats operation)'),
    queryPattern: z.string().optional().describe('Filter queries containing this pattern (for get_stats operation)'),
    
    // RESET_STATS operation parameters
    queryId: z.string().optional().describe('Specific query ID to reset (for reset_stats operation, resets all if not provided)'),
  });

• src/index.ts:225-257 (registration)

  Registration of the pg_manage_query tool (as manageQueryTool) in the allTools array, which is passed to the PostgreSQLServer constructor to enable the tool in the MCP server.

  const allTools: PostgresTool[] = [
    // Core Analysis & Debugging
    analyzeDatabaseTool,
    debugDatabaseTool,
    
    // Schema & Structure Management (Meta-Tools)
    manageSchemaTools,
    manageFunctionsTool,
    manageTriggersTools,
    manageIndexesTool,
    manageConstraintsTool,
    manageRLSTool,
    
    // User & Security Management
    manageUsersTool,
    
    // Query & Performance Management
    manageQueryTool,
    
    // Data Operations (Enhancement Tools)
    executeQueryTool,
    executeMutationTool,
    executeSqlTool,
    
    // Documentation & Metadata
    manageCommentsTool,
    
    // Data Migration & Monitoring
    exportTableDataTool,
    importTableDataTool,
    copyBetweenDatabasesTool,
    monitorDatabaseTool
  ];

• src/tools/query.ts:299-319 (helper)

  Helper dispatch function that routes the validated input to specific operation handlers (explain, get_slow_queries, get_stats, reset_stats).

  async function executeManageQuery(
    input: ManageQueryInput,
    getConnectionString: GetConnectionStringFn
  ): Promise<ExplainResult | SlowQuery[] | QueryStats[] | { message: string; queryId?: string }> {
    switch (input.operation) {
      case 'explain':
        return executeExplainQuery(input, getConnectionString);
        
      case 'get_slow_queries':
        return executeGetSlowQueries(input, getConnectionString);
        
      case 'get_stats':
        return executeGetQueryStats(input, getConnectionString);
        
      case 'reset_stats':
        return executeResetQueryStats(input, getConnectionString);
        
      default:
        throw new McpError(ErrorCode.InvalidParams, `Unsupported operation: ${input.operation}`);
    }
  }

• src/tools/query.ts:78-142 (helper)

  Helper function for the 'explain' operation: executes EXPLAIN (with options like ANALYZE, BUFFERS, etc.) on the provided query and extracts key plan metrics.

  async function executeExplainQuery(
    input: ManageQueryInput,
    getConnectionString: GetConnectionStringFn
  ): Promise<ExplainResult> {
    if (!input.query) {
      throw new McpError(ErrorCode.InvalidParams, 'query parameter is required for explain operation');
    }
  
    const resolvedConnectionString = getConnectionString(input.connectionString);
    const db = DatabaseConnection.getInstance();
    const { query, analyze, buffers, verbose, costs, format } = input;
    
    try {
      await db.connect(resolvedConnectionString);
      
      // Build EXPLAIN options
      const options = [];
      if (analyze) options.push('ANALYZE');
      if (buffers) options.push('BUFFERS');
      if (verbose) options.push('VERBOSE');
      if (!costs) options.push('COSTS false');
      options.push(`FORMAT ${format?.toUpperCase()}`);
      
      const explainQuery = `EXPLAIN (${options.join(', ')}) ${query}`;
      
      const result = await db.query(explainQuery);
      
      // Extract timing information if available (from EXPLAIN ANALYZE)
      let execution_time: number | undefined;
      let planning_time: number | undefined;
      let total_cost: number | undefined;
      let actual_rows: number | undefined;
      let estimated_rows: number | undefined;
      
      if (format === 'json' && result.length > 0) {
        const plan = result[0]['QUERY PLAN'];
        if (Array.isArray(plan) && plan.length > 0) {
          const planData = plan[0];
          execution_time = planData['Execution Time'];
          planning_time = planData['Planning Time'];
          
          if (planData.Plan) {
            total_cost = planData.Plan['Total Cost'];
            actual_rows = planData.Plan['Actual Rows'];
            estimated_rows = planData.Plan['Plan Rows'];
          }
        }
      }
      
      return {
        query,
        plan: result,
        execution_time,
        planning_time,
        total_cost,
        actual_rows,
        estimated_rows
      };
      
    } catch (error) {
      throw new McpError(ErrorCode.InternalError, `Failed to explain query: ${error instanceof Error ? error.message : String(error)}`);
    } finally {
      await db.disconnect();
    }
  }