Postgres MCP Server
Secure Architecture for AI Database Access (2026)

We built a Postgres MCP server that lets AI agents query a production database in plain English. The naive approach — give the agent raw SQL access — is a security disaster. This is the architecture we landed on after getting burned, and the one we ship to every client now. For the broader MCP context, see our Model Context Protocol guide.

Why Raw SQL Access Is a Trap

Every team starts by thinking: "We will let the LLM write SQL and just review the queries." This breaks down within the first week of production. The failure modes:

• Prompt injection writes SQL. A user message can influence the query the LLM generates. We have seen cases where a cleverly-worded support ticket caused the LLM to write a query that leaked customer data.
• Expensive queries happen accidentally. An LLM-generated JOIN across five tables without appropriate filters can scan hundreds of gigabytes.
• Destructive queries are one prompt away. Any DELETE, UPDATE, or DROP command the model emits is one role misconfiguration away from executing.
• Debuggability is terrible. When the LLM generates a bad query at 3 AM, you are reading post-hoc logs trying to reconstruct intent.

The Three-Layer Architecture

Our Postgres MCP server has three layers. Each layer addresses a specific class of risk.

Layer 1 — Schema-Aware Tool Generation

On startup, the server introspects Postgres using information_schema and generates one MCP tool per table (or per semantic entity defined in configuration). Each tool has:

• A description derived from the table's Postgres COMMENT.
• Parameters derived from indexed columns — equality filters, range filters, text search.
• A typed return schema derived from the column types.

The LLM never writes SQL. It calls get_orders_by_customer(customer_id: int, limit: int) with typed arguments. The tool description tells the LLM when this query makes sense; the signature constrains what can be passed.

async def build_tools_from_schema(pool: asyncpg.Pool) -> list[Tool]:
    """Introspect Postgres and generate typed MCP tools."""
    tools = []
    async with pool.acquire() as conn:
        tables = await conn.fetch("""
            SELECT
                c.table_schema, c.table_name,
                obj_description(pc.oid) AS table_comment
            FROM information_schema.tables c
            JOIN pg_class pc
              ON pc.relname = c.table_name
              AND pc.relnamespace = (
                SELECT oid FROM pg_namespace WHERE nspname = c.table_schema
              )
            WHERE c.table_schema NOT IN ('pg_catalog','information_schema')
              AND c.table_type = 'BASE TABLE'
        """)
        for t in tables:
            columns = await conn.fetch("""
                SELECT column_name, data_type, is_nullable
                FROM information_schema.columns
                WHERE table_schema = $1 AND table_name = $2
            """, t["table_schema"], t["table_name"])

            tools.append(Tool(
                name=f"query_{t['table_name']}",
                description=(
                    t["table_comment"]
                    or f"Query the {t['table_name']} table with typed filters."
                ),
                input_schema=build_input_schema(columns),
                handler=make_handler(pool, t["table_schema"], t["table_name"]),
            ))
    return tools

Layer 2 — Parameterized Execution

Every tool invocation is a parameterized query. The tool accepts typed filters; the server builds the SQL internally using $1, $2, ... placeholders. There is no code path that interpolates user input into a SQL string. No injection surface.

def make_handler(pool, schema, table):
    async def handler(args: dict) -> list[dict]:
        filters = []
        params = []
        for col, val in args.get("filters", {}).items():
            # whitelist column names against the schema
            if col not in ALLOWED_COLUMNS[table]:
                raise ValueError(f"Column {col} is not queryable")
            params.append(val)
            filters.append(f'"{col}" = ${len(params)}')

        where = f"WHERE {' AND '.join(filters)}" if filters else ""
        limit = min(args.get("limit", 100), 1000)  # hard cap

        sql = f'SELECT * FROM "{schema}"."{table}" {where} LIMIT {limit}'
        async with pool.acquire() as conn:
            rows = await conn.fetch(sql, *params)
            return [dict(r) for r in rows]
    return handler

Layer 3 — Database-Enforced Authorization

The MCP server is not the authorization layer. The database is. Three controls:

• Read-only role. The connection pool uses a Postgres role with SELECT privileges only. If the tool layer has a bug that somehow issues a write, Postgres refuses it.
• Row-level security passthrough. The server sets request.user_id as a session variable. RLS policies on each table enforce per-user data access. The database, not the tool, decides what rows this user can see.
• Statement timeout. SET statement_timeout = 5000 at the start of every session. No query runs longer than 5 seconds. Protects against runaway scans.

async def create_pool(database_url: str):
    return await asyncpg.create_pool(
        database_url,  # connects as role mcp_readonly
        min_size=5,
        max_size=20,
        command_timeout=6,
        setup=lambda conn: conn.execute(
            "SET statement_timeout = 5000"
        ),
    )

async def handle_request(pool, user_id: str, tool_name: str, args: dict):
    async with pool.acquire() as conn:
        # RLS will filter rows based on this session variable
        await conn.execute(
            "SELECT set_config('request.user_id', $1, true)", user_id
        )
        handler = TOOLS[tool_name].handler
        return await handler(args, conn)

Connection Pooling — Do Not Skip This

Our first production deployment ran fine for 4 hours, then everything fell over. Every tool call was opening a new Postgres connection. At ~100 queries/minute, we exhausted the connection limit.

The fix: one asyncpg pool created at server startup, shared across all tool invocations. Pool size tuned for the Postgres connection limit and expected concurrency. We default to min_size=5, max_size=20 for a single MCP server instance. Behind a load balancer, size the pool accordingly.

If you are running managed Postgres (RDS, Supabase, Neon), put PgBouncer between the MCP server and the database. Your connection count will otherwise become the bottleneck.

Tool Descriptions Decide Everything

The generated tool descriptions are the single largest factor in whether the LLM uses your tools correctly. A vague description like "Query the orders table" will be called 10× more than needed. A specific description sets expectations for when and how to call.

Compare:

• Bad: "Query the orders table."
• Good: "Retrieve orders for a specific customer. Use when the user asks about a customer's order history. Filter by customer_id (required) and status (optional). Returns up to 50 recent orders. Do not call repeatedly for the same customer in one session — cache the result."

Generate descriptions from the Postgres COMMENT on each table, and treat those comments as prompt engineering. Invest in writing them well. For a deeper take on this, see the tool description section of our MCP guide.

Schema Change Workflow

Schema changes — new tables, dropped columns, renamed fields — need to flow into the MCP server without stale tool definitions causing failures.

Our approach:

• Tool generation runs on server startup. New deploys pick up schema changes automatically.
• On a long-running server, we re-introspect every 15 minutes and diff. Changes trigger tool re-registration.
• Deprecated columns get a short grace period before removal, logged loudly.
• All schema changes go through migrations that update table COMMENTs where the description logic would change.

Observability — What to Log

You need to reconstruct why the LLM made the decisions it made. Our logging, per tool invocation:

• Tool name, input arguments, user_id, request_id.
• Generated SQL (yes, log the parameterized SQL — not the values, for PII).
• Execution time, row count returned, Postgres query plan for slow queries.
• Any error with full stack trace.
• Link back to the originating LLM trace (LangSmith, Langfuse, OTel span).

Track p99 latency per tool. When a tool's p99 creeps up, it usually means the LLM started using a filter that needs a new index.

When to Skip MCP and Just Build an API

MCP is not always the right answer. If the agent only needs 3–5 specific queries, a normal REST or GraphQL API with function-calling tools is simpler and gives you more control. MCP wins when:

• You want multiple clients (Claude Desktop, internal agents, third-party tools) to use the same capabilities.
• Your schema is large enough that auto-generating tools beats hand-writing them.
• You benefit from the full MCP ecosystem (resources, prompts, sampling).

For a narrow integration owned by one team, a handful of typed function-calling tools backed by an internal API is often the faster path. See our MCP server architecture for enterprise data for when MCP is the right choice.

Frequently Asked Questions

Related Reading

• MCP Server Development Services — Inventiple
• Model Context Protocol: Complete Guide for 2026
• MCP Server Architecture for Enterprise Data
• Agentic AI Guardrails and Hallucination Prevention
• How to Build a Production RAG Pipeline in Python