Postgres MCP Server

Instructions

Implementation Reference

• src/index.ts:718-787 (handler)

  Main handler function for the 'query' tool. Dispatches to QueryAPIClient methods based on the 'action' parameter (execute, transaction, explain, analyze, validate, active, cancel).
  private async handleQuery(args: any) { const { action, sql, parameters, queries, options, pid } = args; switch (action) { case 'execute': ParameterValidator.validateRequired(sql, 'sql'); return { content: [{ type: 'text', text: JSON.stringify(await this.queryClient.executeQuery(sql, parameters, options), null, 2) }] }; case 'transaction': ParameterValidator.validateRequired(queries, 'queries'); return { content: [{ type: 'text', text: JSON.stringify(await this.queryClient.executeTransaction(queries, options?.readOnly), null, 2) }] }; case 'explain': ParameterValidator.validateRequired(sql, 'sql'); return { content: [{ type: 'text', text: JSON.stringify(await this.queryClient.getExecutionPlan(sql, parameters), null, 2) }] }; case 'analyze': ParameterValidator.validateRequired(sql, 'sql'); return { content: [{ type: 'text', text: JSON.stringify(await this.queryClient.analyzeQuery(sql, parameters), null, 2) }] }; case 'validate': ParameterValidator.validateRequired(sql, 'sql'); return { content: [{ type: 'text', text: JSON.stringify(await this.queryClient.validateSyntax(sql), null, 2) }] }; case 'active': return { content: [{ type: 'text', text: JSON.stringify(await this.queryClient.getActiveQueries(), null, 2) }] }; case 'cancel': ParameterValidator.validateRequired(pid, 'pid'); return { content: [{ type: 'text', text: JSON.stringify({ cancelled: await this.queryClient.cancelQuery(pid) }, null, 2) }] }; default: throw new Error(`Unknown query action: ${action}`); } }
• src/index.ts:40-89 (schema)

  JSON Schema defining the input parameters and structure for the 'query' tool.
  { name: 'query', description: 'Execute SQL queries with transaction support, query analysis, and performance monitoring', inputSchema: { type: 'object', properties: { action: { type: 'string', enum: ['execute', 'transaction', 'explain', 'analyze', 'validate', 'cancel', 'active'], description: 'Action: execute (single query), transaction (multiple queries), explain (execution plan), analyze (performance), validate (syntax), cancel (query by PID), active (list active queries)' }, sql: { type: 'string', description: 'SQL query to execute (required for execute, explain, analyze, validate actions)' }, parameters: { type: 'array', items: { type: 'string' }, description: 'Query parameters for parameterized queries' }, queries: { type: 'array', items: { type: 'object', properties: { sql: { type: 'string' }, parameters: { type: 'array', items: { type: 'string' } } }, required: ['sql'] }, description: 'Array of queries for transaction action' }, options: { type: 'object', properties: { timeout: { type: 'integer', description: 'Query timeout in milliseconds' }, limit: { type: 'integer', description: 'Maximum number of rows to return' }, offset: { type: 'integer', description: 'Number of rows to skip' }, readOnly: { type: 'boolean', description: 'Execute as read-only transaction' } }, description: 'Query execution options' }, pid: { type: 'integer', description: 'Process ID of query to cancel (required for cancel action)' } }, required: ['action'] } },
• src/index.ts:634-637 (registration)

  Registration of tool definitions via ListToolsRequestSchema handler, which returns the list including 'query' tool schema.
  this.server.setRequestHandler(ListToolsRequestSchema, async () => ({ tools: toolDefinitions, }));
• src/index.ts:644-645 (registration)

  Dispatch registration in CallToolRequestSchema handler switch statement, routing 'query' calls to handleQuery method.
  return await this.handleQuery(args);
• src/api/domains/query-api.ts:27-284 (helper)

  QueryAPIClient class providing the core query execution, transaction, analysis, and monitoring functionality used by the 'query' tool handler.
  export class QueryAPIClient { constructor(private dbManager: DatabaseConnectionManager) {} /** * Execute a SQL query */ async executeQuery( sql: string, parameters?: any[], options: QueryOptions = {} ): Promise<QueryExecutionResult> { const startTime = Date.now(); // Validate SQL const validatedSql = ParameterValidator.validateSql(sql); // Add LIMIT if specified and not already present let finalSql = validatedSql; if (options.limit && !finalSql.toUpperCase().includes('LIMIT')) { finalSql += ` LIMIT ${ParameterValidator.validateLimit(options.limit)}`; } // Add OFFSET if specified if (options.offset && !finalSql.toUpperCase().includes('OFFSET')) { finalSql += ` OFFSET ${ParameterValidator.validateOffset(options.offset)}`; } // Add EXPLAIN if requested if (options.explain) { const explainPrefix = options.analyze ? 'EXPLAIN (ANALYZE, BUFFERS)' : 'EXPLAIN'; finalSql = `${explainPrefix} ${finalSql}`; } try { const result = await this.dbManager.query(finalSql, parameters, { timeout: options.timeout, readOnly: this.isReadOnlyQuery(validatedSql) }); const duration = Date.now() - startTime; return { rows: result.rows, rowCount: result.rowCount || 0, fields: result.fields.map((field: any) => ({ name: field.name, dataTypeID: field.dataTypeID, dataTypeSize: field.dataTypeSize, dataTypeModifier: field.dataTypeModifier, format: field.format })), command: result.command, duration }; } catch (error) { logger.error('Query execution failed', { sql: sql.substring(0, 100), parameters, error: error instanceof Error ? error.message : error }); throw error; } } /** * Execute multiple queries in a transaction */ async executeTransaction( queries: Array<{ sql: string; parameters?: any[] }>, readOnly: boolean = false ): Promise<QueryExecutionResult[]> { const transactionId = await this.dbManager.beginTransaction(readOnly); try { const results: QueryExecutionResult[] = []; for (const query of queries) { const startTime = Date.now(); const validatedSql = ParameterValidator.validateSql(query.sql); const result = await this.dbManager.queryInTransaction( transactionId, validatedSql, query.parameters ); const duration = Date.now() - startTime; results.push({ rows: result.rows, rowCount: result.rowCount || 0, fields: result.fields.map((field: any) => ({ name: field.name, dataTypeID: field.dataTypeID, dataTypeSize: field.dataTypeSize, dataTypeModifier: field.dataTypeModifier, format: field.format })), command: result.command, duration }); } await this.dbManager.commitTransaction(transactionId); return results; } catch (error) { await this.dbManager.rollbackTransaction(transactionId); throw error; } } /** * Get query execution plan */ async getExecutionPlan( sql: string, parameters?: any[], analyze: boolean = false ): Promise<any[]> { const validatedSql = ParameterValidator.validateSql(sql); const explainSql = analyze ? `EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) ${validatedSql}` : `EXPLAIN (FORMAT JSON) ${validatedSql}`; const result = await this.dbManager.query(explainSql, parameters, { readOnly: true }); return result.rows[0]['QUERY PLAN']; } /** * Analyze query performance */ async analyzeQuery(sql: string, parameters?: any[]): Promise<{ executionPlan: any[]; statistics: { planningTime: number; executionTime: number; totalCost: number; rows: number; }; }> { const plan = await this.getExecutionPlan(sql, parameters, true); return { executionPlan: plan, statistics: { planningTime: plan[0]['Planning Time'] || 0, executionTime: plan[0]['Execution Time'] || 0, totalCost: plan[0]['Total Cost'] || 0, rows: plan[0]['Actual Rows'] || 0 } }; } /** * Validate SQL syntax without execution */ async validateSyntax(sql: string): Promise<{ valid: boolean; error?: string }> { try { // Use EXPLAIN to validate syntax without execution const validatedSql = ParameterValidator.validateSql(sql); await this.dbManager.query(`EXPLAIN ${validatedSql}`, [], { readOnly: true }); return { valid: true }; } catch (error) { return { valid: false, error: error instanceof Error ? error.message : 'Unknown syntax error' }; } } /** * Get active queries */ async getActiveQueries(): Promise<any[]> { const sql = ` SELECT pid, now() - pg_stat_activity.query_start AS duration, query, state, client_addr, application_name FROM pg_stat_activity WHERE state = 'active' AND query NOT ILIKE '%pg_stat_activity%' ORDER BY duration DESC `; const result = await this.dbManager.query(sql, [], { readOnly: true }); return result.rows; } /** * Cancel a query by PID */ async cancelQuery(pid: number): Promise<boolean> { const validatedPid = ParameterValidator.validateNumber(pid, 'pid', 1); const result = await this.dbManager.query( 'SELECT pg_cancel_backend($1) as cancelled', [validatedPid] ); return result.rows[0]?.cancelled || false; } /** * Get query statistics */ async getQueryStatistics( schemaName?: string, limit: number = 50 ): Promise<any[]> { const validatedLimit = ParameterValidator.validateLimit(limit); let sql = ` SELECT schemaname, tablename, seq_scan, seq_tup_read, idx_scan, idx_tup_fetch, n_tup_ins, n_tup_upd, n_tup_del FROM pg_stat_user_tables `; const params: any[] = []; if (schemaName) { const validatedSchema = ParameterValidator.validateSchemaName(schemaName); sql += ` WHERE schemaname = $1`; params.push(validatedSchema); } sql += ` ORDER BY seq_scan + idx_scan DESC LIMIT $${params.length + 1}`; params.push(validatedLimit); const result = await this.dbManager.query(sql, params, { readOnly: true }); return result.rows; } /** * Check if query is read-only */ private isReadOnlyQuery(sql: string): boolean { const writeOperations = [ 'INSERT', 'UPDATE', 'DELETE', 'CREATE', 'DROP', 'ALTER', 'TRUNCATE', 'GRANT', 'REVOKE', 'COPY' ]; const normalizedSql = sql.trim().toUpperCase(); return !writeOperations.some(op => normalizedSql.startsWith(op)); } }