PostgreSQL MCP Server

Instructions

Implementation Reference

• src/tools/data.ts:168-296 (handler)

  Core handler function that constructs and executes SQL for data mutations (INSERT/UPDATE/DELETE/UPSERT) using dynamic SQL building and parameterized queries.
  async function executeMutation( input: ExecuteMutationInput, getConnectionString: GetConnectionStringFn ): Promise<{ operation: string; rowsAffected: number; returning?: unknown[] }> { const resolvedConnectionString = getConnectionString(input.connectionString); const db = DatabaseConnection.getInstance(); const { operation, table, data, where, conflictColumns, returning, schema } = input; try { await db.connect(resolvedConnectionString); const schemaPrefix = (schema && schema !== 'public') ? `"${schema}".` : ''; const tableName = `${schemaPrefix}"${table}"`; switch (operation) { case 'insert': { if (!data || Object.keys(data).length === 0) { throw new McpError(ErrorCode.InvalidParams, 'Data object is required for insert operation'); } const columns = Object.keys(data); const values = Object.values(data); const placeholders = values.map((_, i) => `$${i + 1}`).join(', '); let insertSQL = `INSERT INTO ${tableName} (${columns.map(col => `"${col}"`).join(', ')}) VALUES (${placeholders})`; if (returning) { insertSQL += ` RETURNING ${returning}`; } const result = await db.query(insertSQL, values); return { operation: 'insert', rowsAffected: Array.isArray(result) ? result.length : 1, returning: returning ? result : undefined }; } case 'update': { if (!data || Object.keys(data).length === 0) { throw new McpError(ErrorCode.InvalidParams, 'Data object is required for update operation'); } if (!where) { throw new McpError(ErrorCode.InvalidParams, 'WHERE clause is required for update operation to prevent accidental full table updates'); } const columns = Object.keys(data); const values = Object.values(data); const setClause = columns.map((col, i) => `"${col}" = $${i + 1}`).join(', '); let updateSQL = `UPDATE ${tableName} SET ${setClause} WHERE ${where}`; if (returning) { updateSQL += ` RETURNING ${returning}`; } const result = await db.query(updateSQL, values); return { operation: 'update', rowsAffected: Array.isArray(result) ? result.length : 1, returning: returning ? result : undefined }; } case 'delete': { if (!where) { throw new McpError(ErrorCode.InvalidParams, 'WHERE clause is required for delete operation to prevent accidental full table deletion'); } let deleteSQL = `DELETE FROM ${tableName} WHERE ${where}`; if (returning) { deleteSQL += ` RETURNING ${returning}`; } const result = await db.query(deleteSQL); return { operation: 'delete', rowsAffected: Array.isArray(result) ? result.length : 1, returning: returning ? result : undefined }; } case 'upsert': { if (!data || Object.keys(data).length === 0) { throw new McpError(ErrorCode.InvalidParams, 'Data object is required for upsert operation'); } if (!conflictColumns || conflictColumns.length === 0) { throw new McpError(ErrorCode.InvalidParams, 'Conflict columns are required for upsert operation'); } const columns = Object.keys(data); const values = Object.values(data); const placeholders = values.map((_, i) => `$${i + 1}`).join(', '); const conflictCols = conflictColumns.map(col => `"${col}"`).join(', '); const updateClause = columns .filter(col => !conflictColumns.includes(col)) .map(col => `"${col}" = EXCLUDED."${col}"`) .join(', '); let upsertSQL = `INSERT INTO ${tableName} (${columns.map(col => `"${col}"`).join(', ')}) VALUES (${placeholders}) ON CONFLICT (${conflictCols})`; if (updateClause) { upsertSQL += ` DO UPDATE SET ${updateClause}`; } else { upsertSQL += ' DO NOTHING'; } if (returning) { upsertSQL += ` RETURNING ${returning}`; } const result = await db.query(upsertSQL, values); return { operation: 'upsert', rowsAffected: Array.isArray(result) ? result.length : 1, returning: returning ? result : undefined }; } default: throw new McpError(ErrorCode.InvalidParams, `Unknown operation: ${operation}`); } } catch (error) { throw new McpError(ErrorCode.InternalError, `Failed to execute ${operation}: ${error instanceof Error ? error.message : String(error)}`); } finally { await db.disconnect(); } }
• src/tools/data.ts:155-166 (schema)

  Zod schema for validating the input parameters to the pg_execute_mutation tool.
  const ExecuteMutationInputSchema = z.object({ connectionString: z.string().optional().describe('PostgreSQL connection string (optional)'), operation: z.enum(['insert', 'update', 'delete', 'upsert']).describe('Mutation operation: insert (add rows), update (modify rows), delete (remove rows), upsert (insert or update)'), table: z.string().describe('Table name for the operation'), data: z.record(z.unknown()).optional().describe('Data object with column-value pairs (required for insert/update/upsert)'), where: z.string().optional().describe('WHERE clause for update/delete operations (without WHERE keyword)'), conflictColumns: z.array(z.string()).optional().describe('Columns for conflict resolution in upsert (ON CONFLICT)'), returning: z.string().optional().describe('RETURNING clause to get back inserted/updated data'), schema: z.string().optional().default('public').describe('Schema name (defaults to public)') }); type ExecuteMutationInput = z.infer<typeof ExecuteMutationInputSchema>;
• src/tools/data.ts:298-360 (registration)

  Local registration of the pg_execute_mutation tool object, including name, description, input schema, and thin wrapper execute handler that delegates to the core executeMutation function.
  export const executeMutationTool: PostgresTool = { name: 'pg_execute_mutation', description: 'Execute data modification operations (INSERT/UPDATE/DELETE/UPSERT) - operation="insert/update/delete/upsert" with table and data. Examples: operation="insert", table="users", data={"name":"John","email":"john@example.com"}', inputSchema: ExecuteMutationInputSchema, execute: async (args: unknown, getConnectionStringVal: GetConnectionStringFn): Promise<ToolOutput> => { const { connectionString: connStringArg, operation, table, data, where, conflictColumns, returning, schema } = args as { connectionString?: string; operation: 'insert' | 'update' | 'delete' | 'upsert'; table: string; data?: Record<string, unknown>; where?: string; conflictColumns?: string[]; returning?: string; schema?: string; }; const resolvedConnString = getConnectionStringVal(connStringArg); try { // Input validation if (!table?.trim()) { return { content: [{ type: 'text', text: 'Error: table is required' }], isError: true }; } const result = await executeMutation({ connectionString: resolvedConnString, operation, table, data, where, conflictColumns, returning, schema: schema || 'public' } as ExecuteMutationInput, getConnectionStringVal); let responseText = `${operation.toUpperCase()} operation completed successfully. Rows affected: ${result.rowsAffected}`; if (result.returning && result.returning.length > 0) { responseText += `\n\nReturning data:\n${JSON.stringify(result.returning, null, 2)}`; } return { content: [{ type: 'text', text: responseText }] }; } catch (error) { return { content: [{ type: 'text', text: `Error executing ${operation} operation: ${error instanceof Error ? error.message : String(error)}` }], isError: true }; } } };
• src/index.ts:225-257 (registration)

  Global registration: inclusion of executeMutationTool in the allTools array, which is passed to the PostgreSQLServer constructor to register all available tools with 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/data.ts:302-359 (helper)

  MCP tool execution wrapper that handles argument parsing, basic validation, delegation to core handler, and response formatting.
  execute: async (args: unknown, getConnectionStringVal: GetConnectionStringFn): Promise<ToolOutput> => { const { connectionString: connStringArg, operation, table, data, where, conflictColumns, returning, schema } = args as { connectionString?: string; operation: 'insert' | 'update' | 'delete' | 'upsert'; table: string; data?: Record<string, unknown>; where?: string; conflictColumns?: string[]; returning?: string; schema?: string; }; const resolvedConnString = getConnectionStringVal(connStringArg); try { // Input validation if (!table?.trim()) { return { content: [{ type: 'text', text: 'Error: table is required' }], isError: true }; } const result = await executeMutation({ connectionString: resolvedConnString, operation, table, data, where, conflictColumns, returning, schema: schema || 'public' } as ExecuteMutationInput, getConnectionStringVal); let responseText = `${operation.toUpperCase()} operation completed successfully. Rows affected: ${result.rowsAffected}`; if (result.returning && result.returning.length > 0) { responseText += `\n\nReturning data:\n${JSON.stringify(result.returning, null, 2)}`; } return { content: [{ type: 'text', text: responseText }] }; } catch (error) { return { content: [{ type: 'text', text: `Error executing ${operation} operation: ${error instanceof Error ? error.message : String(error)}` }], isError: true }; } }