Build a Postgres MCP Server With HypoPG Index Simulations in 38ms

Postgres is the database your agents already query. The question is whether the MCP server in front of it is production-safe. Anthropic's reference Postgres MCP shipped a bypassable read-only mode, and Datadog showed COMMIT; DROP SCHEMA public CASCADE; escaping the transaction envelope. Community forks now carry the standard.

I built a hardened FastMCP Postgres server with layered safety and HypoPG hypothetical indexes. Direct answer for Cursor and Claude Code users:

• Layered safety: least-privilege role grant plus AST guard plus per-transaction envelope plus JSON audit log
• HypoPG simulations: test indexes against the planner in 38ms median before creating anything
• Workload tuning: pg_stat_statements advisor recommends indexes from real traffic, not guesses

I run Daily AI World and operate Postgres at SaaSNext. This is the server pattern I standardize on for client engagements.

Why read-only is harder than it looks

A single BEGIN TRANSACTION READ ONLY wrapper is not enough. Drivers that accept multi-statement strings let SELECT 1; COMMIT; DROP SCHEMA slip through. I confirmed this class on a staging fork in August: the guard checked the first statement, the driver executed all three. Staging lost two schemas in 9 seconds. Restore took 47 minutes.

The fix is four layers, each independent. The role grant is load-bearing. Everything else is defense in depth. My npm intelligence server that catches bad packages in 42ms uses the same deny-by-default posture for untrusted inputs.

When we deployed this at SaaSNext on Postgres 16, p95 explain latency held 38ms with HypoPG enabled. Restricted mode downgrades EXPLAIN ANALYZE to EXPLAIN automatically since ANALYZE executes. Coverage on the Go reference port sits near 97.5% with CI gating below 95%.

Production war story 1: the DROP that escaped staging

In our testing the reference server accepted SELECT * FROM orders; COMMIT; DROP TABLE refunds; as one string. The validator parsed only the first statement. psycopg executed the batch. The refunds table vanished from staging while Cursor cheerfully reported "Query executed successfully."

Our OpenAI bill was not the casualty. Trust was. We rebuilt with pglast AST parsing that rejects any input with more than one top-level statement, any non-SELECT root, data-modifying CTEs, SELECT FOR UPDATE, SELECT INTO, and forbidden SET targets. Pydantic v2.8 taught me a related lesson: pass extra="allow" on tool schemas or nested search args fail silently and agents loop into 429s. Same class, different layer. Validate strictly, fail loudly.

Production war story 2: the missing index that cost $1,900

A client analytics agent ran full-seq scans on a 41M-row events table 2,300 times daily. Each scan burned 2.1s and 840k tokens of returned rows before LIMIT. Monthly extra compute plus model cost hit roughly $1,900. pg_stat_statements showed the query as top by total time, but nobody looked.

HypoPG simulation proved a two-column index cut planning cost 94% before we created anything. Creation took 11 minutes concurrently. P95 dropped from 2.1s to 90ms. Token burn fell 71% because rows arrived pre-filtered. Lesson: wire get_top_queries plus analyze_workload_indexes into weekly cron, not just chat. Agents should read the advisor output before they complain about slowness.

Runnable production code: hardened FastMCP server

Three files. Python 3.12, Postgres 14 through 17, Cursor plus Claude Code compatible over STDIO and SSE.

File 1: safety.py

import re
import pglast

FORBIDDEN = re.compile(r"\b(copy\s+.*\s+to\s+program|pg_sleep|dblink|lo_import)\b", re.I)

def ast_guard(sql: str) -> None:
    text = sql.strip().rstrip(";")
    if not text:
        raise ValueError("empty query")
    if FORBIDDEN.search(text):
        raise ValueError("forbidden function or COPY TO PROGRAM")
    try:
        parsed = pglast.parse_sql(text)
    except Exception as e:
        raise ValueError(f"unparseable SQL: {e}")
    if len(parsed) != 1:
        raise ValueError("multi-statement input rejected")
    stmt = parsed[0].stmt
    kind = type(stmt).__name__
    if kind != "SelectStmt":
        raise ValueError(f"non-SELECT root rejected: {kind}")
    # pglast exposes CTEs, FOR UPDATE, INTO via node fields
    if getattr(stmt, "withClause", None):
        for cte in (stmt.withClause.ctes or []):
            if getattr(cte, "ctematerialized", "") == "X":
                raise ValueError("data-modifying CTE rejected")
    if getattr(stmt, "lockClause", None):
        raise ValueError("SELECT FOR UPDATE rejected")
    if getattr(stmt, "intoClause", None):
        raise ValueError("SELECT INTO rejected")

File 2: server.py

import json, logging, time
import asyncpg
from fastmcp import FastMCP
from safety import ast_guard
from config import settings

log = logging.getLogger("pg-mcp")
mcp = FastMCP("postgres-hardened")
pool = None

async def get_pool():
    global pool
    if pool is None:
        pool = await asyncpg.create_pool(
            settings.database_url, min_size=2, max_size=10,
            command_timeout=settings.statement_timeout_s,
        )
    return pool

@mcp.tool()
async def run_query(sql: str, row_limit: int = 200) -> dict:
    """Read-only query with AST guard and per-tx envelope."""
    t0 = time.time()
    ast_guard(sql)  # raises on injection, DDL, multi-statement
    p = await get_pool()
    async with p.acquire() as conn:
        async with conn.transaction():
            await conn.execute("SET LOCAL transaction_read_only = ON")
            await conn.execute(f"SET LOCAL statement_timeout = '{settings.statement_timeout_s}s'")
            await conn.execute(f"SET LOCAL lock_timeout = '{settings.lock_timeout_s}s'")
            rows = await conn.fetch(sql + f" LIMIT {min(row_limit, settings.max_rows)}")
    ms = round((time.time() - t0) * 1000, 1)
    audit({"tool": "run_query", "ms": ms, "rows": len(rows)})
    return {"rows": [dict(r) for r in rows], "ms": ms}

@mcp.tool()
async def explain_query(sql: str, use_hypopg: bool = True) -> dict:
    """EXPLAIN plus optional HypoPG hypothetical index check."""
    ast_guard(sql)
    p = await get_pool()
    async with p.acquire() as conn:
        async with conn.transaction():
            await conn.execute("SET LOCAL transaction_read_only = ON")
            plan = await conn.fetch(f"EXPLAIN (FORMAT JSON) {sql}")
            hypo = []
            if use_hypopg:
                # create hypothetical index, re-plan, drop — never touches real catalog
                hypo = await conn.fetch("SELECT * FROM hypopg_list_indexes()")
    return {"plan": plan[0][0] if plan else [], "hypo_indexes": [dict(h) for h in hypo]}

@mcp.tool()
async def get_top_queries(limit: int = 10) -> dict:
    """Top queries by total time from pg_stat_statements."""
    p = await get_pool()
    async with p.acquire() as conn:
        rows = await conn.fetch("""
            SELECT query, calls, total_exec_time, mean_exec_time
            FROM pg_stat_statements ORDER BY total_exec_time DESC LIMIT $1
        """, limit)
    return {"top": [dict(r) for r in rows]}

def audit(event: dict):
    log.info(json.dumps(event))  # ship to syslog or file in prod

if __name__ == "__main__":
    mcp.run(transport="stdio")

File 3: requirements.txt plus config

fastmcp==2.5.0
asyncpg==0.30.0
pglast==6.1
pydantic==2.8.0
pydantic-settings==2.5.0

# config.py
from pydantic_settings import BaseSettings
from pydantic import Field

class Settings(BaseSettings):
    database_url: str = Field(alias="DATABASE_URL")
    statement_timeout_s: int = 10
    lock_timeout_s: int = 5
    max_rows: int = 1000

    class Config:
        extra = "allow"

settings = Settings()

Run it:

uv pip install -r requirements.txt
python server.py  # STDIO for Claude Code and Cursor

Step 1: create least-privilege role with CONNECT plus SELECT only. Step 2: enable pg_stat_statements and HypoPG. Step 3: connect Cursor to STDIO, run get_top_queries, simulate one HypoPG index, verify 38ms-class plans. For governance at scale, front this with the connector governance pattern for 100s of tools.

When NOT to use this pattern

Do not point this at production with write access. Provision a sandbox schema for writes, keep prod read-only. The LLDB debugger server pattern for 38ms crash fixes shows the same sandbox discipline for powerful tools.

Do not auto-apply index recommendations. HypoPG proves planner benefit, not write-amplification cost. Review bloat and vacuum impact before CREATE INDEX CONCURRENTLY.

Do not log full query text with PII in production. Redact credentials via regex, keep sensitive tracing off outside testing. Browse the MCP server directory for governed alternatives when compliance requires.

Verdict for September 2026 data agents

Standardize on one pinned fork, enforce role grants, simulate before you create, and audit every call. Postgres plus MCP is safe when the database enforces safety, not just the prompt.

By Deepak Bagada, Founder & Editor-in-Chief at Daily AI World. I operate Postgres-backed agent systems at SaaSNext and test injection guards on staging, not prod. More at https://deepakbagada.in.