PostgreSQL MCP Server

Instructions

Implementation Reference

• src/tools/query.ts:321-371 (handler)

  Main handler for the pg_manage_query tool. Validates input using the schema and executes the appropriate query management operation via executeManageQuery, formats the output.
  export const manageQueryTool: PostgresTool = { name: 'pg_manage_query', description: 'Manage PostgreSQL query analysis and performance - operation="explain" for EXPLAIN plans, operation="get_slow_queries" for slow query analysis, operation="get_stats" for query statistics, operation="reset_stats" for clearing statistics', inputSchema: ManageQueryInputSchema, 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)

  Input schema (Zod) defining parameters for all operations supported by pg_manage_query: explain, get_slow_queries, get_stats, reset_stats.
  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 manageQueryTool (pg_manage_query) in the allTools array, which is passed to the MCP server constructor to make it available.
  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)

  Core dispatcher helper function that routes to specific operation handlers based on input.operation.
  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 for 'explain' operation: executes EXPLAIN (or ANALYZE) on provided query and returns plan with timings.
  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(); } }