How to Build an MCP Server for PostgreSQL (Read-Only, Python SDK v2)

To build an MCP server for PostgreSQL, you write a small Python program that opens a connection pool at startup and exposes two or three tools, such as "list tables" and "run a SELECT." The most important step is not the code. It is the database role the server logs in as: give it read-only access and nothing else.

This guide builds that server with the official Python SDK, version 2. It uses psycopg 3 for the database driver and keeps writes out entirely. Once it works, you can add one narrow write tool if your use case truly needs it.

Checked against: the Python SDK docs and PyPI on 2 November 2026 (mcp 2.3.0, the latest release), psycopg 3.3.6 and psycopg-pool 3.3.3. The server code below ran locally on 2 November 2026 against a PostgreSQL 16.2 test database.

Should you build one at all?

You may not need to. The original reference Postgres server now sits in the archived servers repository, which has been read-only since 29 May 2025. Its README says "NO SECURITY GUARANTEES ARE PROVIDED FOR THESE ARCHIVED SERVERS."

Hosted Postgres vendors ship maintained servers. The Supabase MCP server has a read-only mode and project scoping, and Neon, PlanetScale and others are compared in our best MCP servers for databases. Google's open-source MCP Toolbox for Databases is another option for self-hosted Postgres.

Build your own when you need tight control: a fixed set of tables, queries shaped around your business, or a database none of those servers reach.

Prerequisites

• Python 3.11 or newer. PyPI lists mcp as requiring 3.10+, but the SDK README says 3.11+, so use 3.11 or later.
• A PostgreSQL database you can create roles in. A local copy or a staging replica is best for the first build.
• uv (or pip) to install packages.
• Node.js 22.19 or newer if you want the MCP Inspector CLI.

If you have never written an MCP tool, do lesson 14 of our course first. It covers MCPServer and @mcp.tool() in detail.

Step 1: Create a read-only database role

Run this as a database owner. It creates a login that can read one table and do nothing else:

CREATE ROLE mcp_reader LOGIN PASSWORD 'change-me';
GRANT CONNECT ON DATABASE app TO mcp_reader;
GRANT USAGE ON SCHEMA public TO mcp_reader;
GRANT SELECT ON customers, orders TO mcp_reader;

List tables one by one. Avoid GRANT SELECT ON ALL TABLES unless every table is safe for an AI model to read. Leave out tables with password hashes, payment data or personal notes. Our lesson on least privilege explains why narrow grants beat clever code.

Step 2: Install

uv init pg-mcp --python 3.12
cd pg-mcp
uv add "mcp[cli]" "psycopg[binary,pool]"

The pool extra installs psycopg-pool. The binary extra ships a prebuilt driver, so you do not need PostgreSQL client libraries on the machine.

Step 3: Write the minimal server

Create server.py:

import os
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from dataclasses import dataclass
from typing import Annotated, Any

from psycopg.rows import dict_row
from psycopg_pool import AsyncConnectionPool
from pydantic import Field

from mcp.server import MCPServer
from mcp.server.mcpserver import Context
from mcp.server.mcpserver.exceptions import ToolError

DATABASE_URL = os.environ["DATABASE_URL"]  # a read-only role, never the owner
MAX_ROWS = 200


@dataclass
class AppContext:
    pool: AsyncConnectionPool


@asynccontextmanager
async def lifespan(server: MCPServer) -> AsyncIterator[AppContext]:
    pool = AsyncConnectionPool(
        DATABASE_URL,
        min_size=1,
        max_size=5,
        open=False,
        kwargs={"options": "-c default_transaction_read_only=on -c statement_timeout=5000"},
    )
    await pool.open()
    try:
        yield AppContext(pool=pool)
    finally:
        await pool.close()


mcp = MCPServer("postgres-readonly", lifespan=lifespan)


@mcp.tool()
async def list_tables(ctx: Context[AppContext]) -> list[dict[str, Any]]:
    """List the tables in the public schema with their columns."""
    sql = """
        SELECT table_name, string_agg(column_name || ' ' || data_type, ', '
               ORDER BY ordinal_position) AS columns
        FROM information_schema.columns
        WHERE table_schema = 'public'
        GROUP BY table_name ORDER BY table_name
    """
    async with ctx.request_context.lifespan_context.pool.connection() as conn:
        cur = await conn.cursor(row_factory=dict_row).execute(sql)
        return await cur.fetchall()


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

What matters here:

• The lifespan. The SDK's lifespan guide says the lifespan runs once, before the first request, and whatever it yields is available to every tool as ctx.request_context.lifespan_context. That is where the pool lives.
• open=False then await pool.open(). The psycopg pool docs say opening an async pool in the constructor is deprecated, and recommend explicit opening.
• The options string. Every pooled connection starts with two settings. PostgreSQL's client configuration docs define default_transaction_read_only as the default read-only status of each new transaction, and statement_timeout as the limit after which any statement is aborted. Five seconds stops a runaway query from tying up the database.
• list_tables reads information_schema. The model learns your column names before it writes SQL. That one tool cuts most bad queries.

Step 4: Add a query tool

Add this below list_tables:

@mcp.tool()
async def run_select(
    sql: Annotated[str, Field(description="One SELECT statement. No semicolons.")],
    ctx: Context[AppContext],
) -> list[dict[str, Any]]:
    """Run a read-only SELECT and return up to 200 rows."""
    if not sql.lstrip().lower().startswith(("select", "with")) or ";" in sql:
        raise ToolError("Only a single SELECT (or WITH ... SELECT) statement is allowed.")
    async with ctx.request_context.lifespan_context.pool.connection() as conn:
        try:
            cur = await conn.cursor(row_factory=dict_row).execute(sql)
        except Exception as exc:  # report SQL errors so the model can fix its query
            raise ToolError(f"Query failed: {exc}") from exc
        return await cur.fetchmany(MAX_ROWS)

The string check is a courtesy, not security. It gives the model a fast, readable "no" for obvious mistakes. The real protection is the read-only role from Step 1 plus the read-only transaction setting.

That distinction matters. PostgreSQL's SET TRANSACTION page says a read-only transaction blocks INSERT, UPDATE, DELETE, CREATE, DROP and similar commands. It also says the defaults "can be overridden per transaction." A session setting is a second layer. Only the role's grants are a hard limit.

Raising ToolError sends the database's message back as a tool result with isError set. The model reads "column does not exist" and tries again. Our MCP tool errors page explains the two error types.

A safer alternative: named queries

Free-form SQL suits analysts exploring data. For everyday staff use, consider fixed tools instead:

@mcp.tool()
async def orders_for_customer(customer_id: int, ctx: Context[AppContext]) -> list[dict[str, Any]]:
    """List one customer's orders, newest first."""
    async with ctx.request_context.lifespan_context.pool.connection() as conn:
        cur = await conn.cursor(row_factory=dict_row).execute(
            "SELECT id, total, status, created_at FROM orders "
            "WHERE customer_id = %s ORDER BY created_at DESC LIMIT 50",
            (customer_id,),
        )
        return await cur.fetchall()

The value goes in as a parameter, never pasted into the string. The psycopg parameter guide is blunt: "Never use % or + to merge values into queries."

Step 5: Test it

First test in memory with the SDK's Client, which needs no subprocess. Create check.py:

import anyio
from mcp import Client
from server import mcp


async def main() -> None:
    async with Client(mcp) as c:
        print((await c.call_tool("list_tables", {})).structured_content)
        r = await c.call_tool("run_select", {
            "sql": "SELECT city, count(*) AS n FROM customers GROUP BY city ORDER BY city"})
        print(r.structured_content)
        r = await c.call_tool("run_select", {
            "sql": "WITH x AS (DELETE FROM customers RETURNING *) SELECT * FROM x"})
        print(r.is_error, r.content[0].text)

anyio.run(main)

Run it with DATABASE_URL pointing at the mcp_reader role. On 2 November 2026, against a three-row test table, the output was:

{'result': [{'table_name': 'customers', 'columns': 'id integer, name text, city text'}]}
{'result': [{'city': 'Austin', 'n': 2}, {'city': 'Boston', 'n': 1}]}
True Error executing tool run_select: Query failed: cannot execute SELECT in a read-only transaction

Look at the last line. The query started with WITH, so it passed the string check. It still contained a DELETE. PostgreSQL refused it because the transaction was read-only, and the role had no DELETE right anyway. No rows were lost.

Then run the MCP Inspector CLI for an end-to-end check over stdio:

npx @modelcontextprotocol/inspector --cli \
  -e "DATABASE_URL=postgresql://mcp_reader:change-me@localhost/app" \
  uv --directory /ABSOLUTE/PATH/pg-mcp run server.py \
  --method tools/call --tool-name list_tables --format json

Pass the connection string with -e. The Inspector launches the server as a child process, so set the variable explicitly rather than relying on your shell. Its README also warns that it stores stdio env values in the OS keychain, or in ~/.mcp-inspector/secrets.json on machines without one, so use a test password here. Inspector 2.13.0 needs Node.js 22.19 or newer. Our Inspector lesson covers the flags, exit codes and protocol eras.

Step 6: Connect it to a host

Hosts start your server with their own working directory and a near-empty environment. Pass the connection string as an environment variable and use absolute paths. A Claude Desktop entry looks like this:

{
  "mcpServers": {
    "postgres-readonly": {
      "command": "/ABSOLUTE/PATH/TO/uv",
      "args": ["run", "--directory", "/ABSOLUTE/PATH/pg-mcp", "server.py"],
      "env": { "DATABASE_URL": "postgresql://mcp_reader:change-me@localhost/app" }
    }
  }
}

The SDK's real-host guide shows the same pattern for Claude Code, Cursor and VS Code. Our free MCP config generator writes the JSON for each client.

Deploy notes

A local stdio server is fine for one analyst on one laptop. For a team, run it as a remote server over Streamable HTTP with mcp.run(transport="streamable-http"), put it behind sign-in, and keep it on the same private network as the database.

• Never expose the database port to reach the server. The MCP server should be the only thing the AI side can touch.
• Use a replica for heavy analytics so long queries cannot slow production.
• Log every query with the user and time. Our page on excessive permissions explains why an audit trail matters.
• Add OAuth before anyone outside your network can reach it. See adding OAuth to a remote MCP server.

Pitfalls

1. Old imports. Tutorials written for SDK v1 use from mcp.server.fastmcp import FastMCP. The SDK's What's New page says the old import path "is gone rather than deprecated." Use from mcp.server import MCPServer.
2. Logging to stdout. On stdio, stdout carries the protocol. Use the logging module, which writes to stderr.
3. Connecting as the owner. If the connection string uses the database owner, a read-only transaction setting is the only thing between the model and DROP TABLE. Use a dedicated role.
4. Huge results. Cap rows. A 50,000-row result fills the model's context and costs tokens without helping it answer.
5. Prompt injection through data. Text stored in your tables can contain instructions. Treat query results as untrusted. See MCP prompt injection.

Next step

For a guided path from first tool to deployment, start the free Learn MCP course. If you want a Postgres server scoped to your schema, with OAuth and audit logs, eSEOspace's MCP development team can plan and build it with you. For a lighter local database, see building an MCP server for SQLite.

Frequently asked questions