MCP Server PostgreSQL — pg driver, parameterized queries, and connection pooling

Driver setup: Pool vs Client

The pg npm package exposes two main classes: Client (a single connection) and Pool (a managed connection pool). For MCP servers, always use Pool. MCP tools are called concurrently — multiple tool calls can arrive before a previous one finishes. A single Client instance handles only one query at a time; a pool distributes concurrent queries across multiple connections.

import { Pool } from 'pg';

// Singleton pool — created once, shared across all tool calls
const pool = new Pool({
  connectionString: process.env.DATABASE_URL,
  // OR explicit fields:
  // host: process.env.PGHOST,
  // port: parseInt(process.env.PGPORT ?? '5432'),
  // database: process.env.PGDATABASE,
  // user: process.env.PGUSER,
  // password: process.env.PGPASSWORD,
  // ssl: { rejectUnauthorized: true },  // required for RDS/Supabase/Neon

  max: 10,           // maximum pool connections
  min: 2,            // keep 2 connections always open
  idleTimeoutMillis: 30000,   // close idle connections after 30s
  connectionTimeoutMillis: 5000,  // fail fast if pool is exhausted
  allowExitOnIdle: false      // keep process alive between tool calls
});

// Surface pool errors to stderr rather than crashing the process
pool.on('error', (err) => {
  console.error('PostgreSQL pool error:', err.message);
});

export { pool };

The DATABASE_URL approach (used by Supabase, Neon, Railway, Render) combines all connection fields into a single environment variable: postgresql://user:pass@host:5432/dbname?sslmode=require. The pg driver parses this automatically. For RDS and Neon you'll need ssl: { rejectUnauthorized: true } added to the Pool options, which DATABASE_URL alone does not enable.

Parameterized query tools: the only safe pattern

SQL injection is the most dangerous failure mode for database-backed MCP tools. The calling agent may pass values from untrusted sources — user input, scraped content, file names. Never interpolate tool parameters into SQL strings. Always use PostgreSQL's parameterized query syntax with $1, $2, … placeholders.

import { z } from 'zod';
import { McpError, ErrorCode } from '@modelcontextprotocol/sdk/types.js';
import { pool } from './db.js';

// ---- query_users ----
server.tool(
  'query_users',
  {
    email: z.string().email().optional(),
    role: z.enum(['admin', 'member', 'viewer']).optional(),
    created_after: z.string().datetime().optional(),
    limit: z.number().int().min(1).max(100).default(20),
    offset: z.number().int().min(0).default(0)
  },
  async ({ email, role, created_after, limit, offset }) => {
    // Build WHERE clause dynamically, but safely
    const conditions: string[] = [];
    const params: unknown[] = [];

    if (email) {
      params.push(email);
      conditions.push(`email = $${params.length}`);
    }
    if (role) {
      params.push(role);
      conditions.push(`role = $${params.length}`);
    }
    if (created_after) {
      params.push(created_after);
      conditions.push(`created_at > $${params.length}`);
    }

    // Limit and offset go through params too — never interpolated
    params.push(limit);
    const limitPlaceholder = `$${params.length}`;
    params.push(offset);
    const offsetPlaceholder = `$${params.length}`;

    const where = conditions.length > 0
      ? `WHERE ${conditions.join(' AND ')}`
      : '';

    const sql = `
      SELECT id, email, role, created_at, updated_at
      FROM users
      ${where}
      ORDER BY created_at DESC
      LIMIT ${limitPlaceholder} OFFSET ${offsetPlaceholder}
    `;

    const result = await pool.query(sql, params);

    return {
      content: [{
        type: 'text',
        text: JSON.stringify({
          rows: result.rows,
          rowCount: result.rowCount
        }, null, 2)
      }]
    };
  }
);

// ---- insert_user ----
server.tool(
  'insert_user',
  {
    email: z.string().email(),
    role: z.enum(['admin', 'member', 'viewer']).default('member'),
    name: z.string().min(1).max(255)
  },
  async ({ email, role, name }) => {
    try {
      const result = await pool.query(
        `INSERT INTO users (email, role, name, created_at, updated_at)
         VALUES ($1, $2, $3, NOW(), NOW())
         RETURNING id, email, role, name, created_at`,
        [email, role, name]
      );
      return {
        content: [{
          type: 'text',
          text: JSON.stringify(result.rows[0], null, 2)
        }]
      };
    } catch (err: unknown) {
      // PostgreSQL unique violation (23505) — email already exists
      if ((err as { code?: string }).code === '23505') {
        throw new McpError(
          ErrorCode.InvalidParams,
          `Email ${email} is already registered`
        );
      }
      throw err;
    }
  }
);

The pattern of building conditions[] and params[] together — incrementing the placeholder index with $${params.length} after each push — is the cleanest way to build dynamic WHERE clauses without string interpolation. Every value the caller provides goes through params, never into the SQL string itself.

Transactions across tool calls

A single tool call that needs atomicity (insert + update + delete as one unit) must run all statements on the same connection. Use pool.connect() to check out a dedicated client, run statements with BEGIN/COMMIT/ROLLBACK, and release the client in a finally block.

server.tool(
  'transfer_credits',
  {
    from_user_id: z.string().uuid(),
    to_user_id: z.string().uuid(),
    amount: z.number().int().positive()
  },
  async ({ from_user_id, to_user_id, amount }) => {
    const client = await pool.connect();
    try {
      await client.query('BEGIN');

      // Check balance first
      const balanceResult = await client.query(
        'SELECT credits FROM users WHERE id = $1 FOR UPDATE',
        [from_user_id]
      );
      if (balanceResult.rows.length === 0) {
        throw new McpError(ErrorCode.InvalidParams, 'Sender not found');
      }
      const currentCredits = balanceResult.rows[0].credits as number;
      if (currentCredits < amount) {
        throw new McpError(
          ErrorCode.InvalidParams,
          `Insufficient credits: have ${currentCredits}, need ${amount}`
        );
      }

      // Debit sender
      await client.query(
        'UPDATE users SET credits = credits - $1, updated_at = NOW() WHERE id = $2',
        [amount, from_user_id]
      );

      // Credit receiver
      await client.query(
        'UPDATE users SET credits = credits + $1, updated_at = NOW() WHERE id = $2',
        [amount, to_user_id]
      );

      // Audit log
      await client.query(
        `INSERT INTO credit_transfers (from_user_id, to_user_id, amount, created_at)
         VALUES ($1, $2, $3, NOW())`,
        [from_user_id, to_user_id, amount]
      );

      await client.query('COMMIT');

      return {
        content: [{ type: 'text', text: JSON.stringify({ transferred: amount }) }]
      };
    } catch (err) {
      await client.query('ROLLBACK');
      throw err;
    } finally {
      client.release();  // always release — even on error
    }
  }
);

The FOR UPDATE lock on the balance row prevents a race condition where two concurrent transfers read the same balance before either commits. Without it, two simultaneous transfers of 100 credits from a 150-credit balance can both succeed — resulting in a -50 balance that violates application logic.

Schema introspection as MCP resources

Exposing the database schema as MCP resources lets calling agents understand what tables and columns exist before constructing queries. This is especially useful for AI agents that generate SQL dynamically.

import {
  ListResourcesRequestSchema,
  ReadResourceRequestSchema
} from '@modelcontextprotocol/sdk/types.js';

server.setRequestHandler(ListResourcesRequestSchema, async () => {
  const result = await pool.query(`
    SELECT
      t.table_name,
      t.table_type,
      pg_stat_user_tables.n_live_tup AS estimated_rows
    FROM information_schema.tables t
    LEFT JOIN pg_stat_user_tables
      ON pg_stat_user_tables.relname = t.table_name
    WHERE t.table_schema = 'public'
      AND t.table_type IN ('BASE TABLE', 'VIEW')
    ORDER BY t.table_name
  `);

  return {
    resources: result.rows.map((row) => ({
      uri: `postgres://public/${row.table_name}`,
      name: row.table_name,
      description: `${row.table_type} — ~${(row.estimated_rows ?? 0).toLocaleString()} rows`,
      mimeType: 'application/json'
    }))
  };
});

server.setRequestHandler(ReadResourceRequestSchema, async ({ params }) => {
  const match = params.uri.match(/^postgres:\/\/public\/([a-zA-Z_][a-zA-Z0-9_]*)$/);
  if (!match) {
    throw new McpError(ErrorCode.InvalidParams, `Unrecognised URI: ${params.uri}`);
  }
  const [, tableName] = match;

  // Fetch column definitions from information_schema
  const columnsResult = await pool.query(
    `SELECT
       column_name,
       data_type,
       is_nullable,
       column_default,
       character_maximum_length
     FROM information_schema.columns
     WHERE table_schema = 'public' AND table_name = $1
     ORDER BY ordinal_position`,
    [tableName]
  );

  if (columnsResult.rows.length === 0) {
    throw new McpError(ErrorCode.InvalidParams, `Table ${tableName} not found`);
  }

  // Also fetch indexes for query planning context
  const indexResult = await pool.query(
    `SELECT indexname, indexdef
     FROM pg_indexes
     WHERE schemaname = 'public' AND tablename = $1`,
    [tableName]
  );

  return {
    contents: [{
      uri: params.uri,
      mimeType: 'application/json',
      text: JSON.stringify({
        table: tableName,
        columns: columnsResult.rows,
        indexes: indexResult.rows
      }, null, 2)
    }]
  };
});

The pg_stat_user_tables.n_live_tup value is an estimate maintained by autovacuum — it's fast (no scan required) but can be stale by millions of rows on heavily-written tables. It's appropriate for resource descriptions (helping agents decide which tables are large vs small) but not for precise counts. Use SELECT COUNT(*) when accuracy matters.

Health endpoint: /health/postgres

A robust PostgreSQL health endpoint checks connectivity, query latency, and pool availability — not just whether the process is running.

import express from 'express';

const app = express();

app.get('/health/postgres', async (_req, res) => {
  const start = Date.now();
  try {
    // Lightweight connectivity + query latency check
    await pool.query('SELECT 1');
    const queryMs = Date.now() - start;

    // Pool stats — exposed directly on the Pool instance
    const poolStats = {
      totalCount: pool.totalCount,       // connections currently open
      idleCount: pool.idleCount,         // connections waiting for work
      waitingCount: pool.waitingCount    // requests waiting for a free connection
    };

    // waitingCount > 0 means callers are queuing — approaching exhaustion
    const healthy = poolStats.waitingCount === 0;

    res.status(healthy ? 200 : 503).json({
      status: healthy ? 'ok' : 'degraded',
      query_ms: queryMs,
      pool: poolStats
    });
  } catch (err) {
    res.status(503).json({
      status: 'error',
      error: (err as Error).message,
      elapsed_ms: Date.now() - start
    });
  }
});

app.listen(3001);

Wire AliveMCP to this endpoint with a 30-second check interval. A 503 indicates a connection failure. A 200 with pool.waitingCount > 0 means the pool is alive but under pressure — all connections are busy and queries are queuing. This is the early warning before timeouts begin.

PostgreSQL error codes in MCP tool handlers

PostgreSQL error codes are returned in the code field of thrown errors. Map them to McpError so callers receive actionable error messages.

import { McpError, ErrorCode } from '@modelcontextprotocol/sdk/types.js';

function mapPgError(err: unknown): never {
  const pgErr = err as { code?: string; constraint?: string; detail?: string };

  switch (pgErr.code) {
    case '23505': // unique_violation
      throw new McpError(
        ErrorCode.InvalidParams,
        `Duplicate value violates constraint: ${pgErr.constraint ?? 'unique'}`
      );
    case '23503': // foreign_key_violation
      throw new McpError(
        ErrorCode.InvalidParams,
        `Foreign key violation: ${pgErr.detail ?? pgErr.constraint}`
      );
    case '23514': // check_violation
      throw new McpError(
        ErrorCode.InvalidParams,
        `Check constraint failed: ${pgErr.constraint}`
      );
    case '42P01': // undefined_table
      throw new McpError(
        ErrorCode.InvalidParams,
        'Table does not exist'
      );
    case '42703': // undefined_column
      throw new McpError(
        ErrorCode.InvalidParams,
        'Column does not exist'
      );
    case '08006': // connection_failure
    case '08001': // sqlclient_unable_to_establish_sqlconnection
      throw new McpError(
        ErrorCode.InternalError,
        'Database connection failed — please retry'
      );
    default:
      throw err;
  }
}

Wrap each tool handler's database calls with try { … } catch (err) { mapPgError(err); }. The most important codes to handle are 23505 (unique violation — usually from caller providing a duplicate email or ID) and 23503 (foreign key violation — from referencing a non-existent parent row). Both should return ErrorCode.InvalidParams with a message that explains which constraint failed.

Frequently asked questions

Further reading

• MCP Server Postgres — pg driver basics and query patterns
• MCP Server MySQL — mysql2 driver, connection pooling, and query tools
• MCP Server Drizzle ORM — type-safe queries for MCP tools
• MCP Server Connection Pooling — pool sizing and exhaustion detection
• MCP Server Error Handling — mapping database errors to McpError codes