Build a Safe Read-Only Postgres MCP Server on macOS

mkdir -p readonly-postgres-mcp-server-macos
cd readonly-postgres-mcp-server-macos
touch .gitignore

touch .env.example

# Development-only credentials for the local Docker Postgres (bound to localhost).
# Copy this file to .env (`make setup` does it) and edit for anything beyond local
# dev. .env is gitignored and must never be committed. In production these values
# come from your secrets manager or injected environment, not a file in the repo.

# Superuser password for the Postgres container.
POSTGRES_PASSWORD=postgres

# Password for the SELECT-only role the MCP server logs in as. init.sh creates the
# role with this value; DATABASE_URL below reuses it.
MCP_READONLY_PASSWORD=readonly_pw

# What the MCP server connects with. Reuses MCP_READONLY_PASSWORD above.
DATABASE_URL=postgresql://mcp_readonly:${MCP_READONLY_PASSWORD}@localhost:5544/appdb

mkdir -p db
touch docker-compose.yml db/init.sh

services:
  db:
    image: postgres:17
    container_name: readonly-pg
    environment:
      POSTGRES_USER: postgres
      # Credentials come from .env (copy .env.example). Compose fails with this
      # message if the variable is unset, so a password is never defaulted here.
      POSTGRES_PASSWORD: ${POSTGRES_PASSWORD:?set POSTGRES_PASSWORD in .env (copy .env.example)}
      POSTGRES_DB: appdb
      # Passed through so init.sh can create the read-only role with it.
      MCP_READONLY_PASSWORD: ${MCP_READONLY_PASSWORD:?set MCP_READONLY_PASSWORD in .env}
    ports:
      # Host 5544 avoids colliding with a local Postgres on 5432.
      - "5544:5432"
    volumes:
      # Runs once, on an empty data directory, as the postgres superuser.
      - ./db/init.sh:/docker-entrypoint-initdb.d/init.sh:ro
      - pgdata:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U postgres -d appdb"]
      interval: 2s
      timeout: 3s
      retries: 15

volumes:
  pgdata:

#!/bin/bash
# Runs once, on first container start, inside the Postgres container. Creates the
# schema, seed data, and a SELECT-only role. The role's password comes from the
# MCP_READONLY_PASSWORD environment variable (passed in from .env by compose), so
# no credential is hard-coded in the repo.
set -euo pipefail

psql -v ON_ERROR_STOP=1 \
     --username "$POSTGRES_USER" \
     --dbname "$POSTGRES_DB" \
     --set ro_pw="$MCP_READONLY_PASSWORD" \
     --set dbname="$POSTGRES_DB" <<'EOSQL'
CREATE TABLE customers (
    id         integer PRIMARY KEY,
    name       text        NOT NULL,
    email      text        NOT NULL,
    city       text        NOT NULL
);

CREATE TABLE orders (
    id          integer PRIMARY KEY,
    customer_id integer NOT NULL REFERENCES customers (id),
    total_cents integer NOT NULL,
    status      text        NOT NULL
);

INSERT INTO customers (id, name, email, city) VALUES
    (1, 'Ada Lovelace',      '[email protected]',      'London'),
    (2, 'Alan Turing',       '[email protected]',     'London'),
    (3, 'Grace Hopper',      '[email protected]',    'New York'),
    (4, 'Katherine Johnson', '[email protected]','Hampton');

INSERT INTO orders (id, customer_id, total_cents, status) VALUES
    (1, 1,  1299, 'paid'),
    (2, 1,  4500, 'paid'),
    (3, 2,   999, 'refunded'),
    (4, 3, 15000, 'paid'),
    (5, 3,  2500, 'pending');

-- The MCP server logs in as this role. It can read the two tables and nothing
-- more: no INSERT/UPDATE/DELETE, no DDL, no other schema. The password is
-- injected from the environment via the :'ro_pw' psql variable, not written here.
CREATE ROLE mcp_readonly LOGIN PASSWORD :'ro_pw';
GRANT CONNECT ON DATABASE :"dbname" TO mcp_readonly;
GRANT USAGE ON SCHEMA public TO mcp_readonly;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO mcp_readonly;

-- Any table added to this schema later is readable too, but still only SELECT.
ALTER DEFAULT PRIVILEGES IN SCHEMA public
    GRANT SELECT ON TABLES TO mcp_readonly;
EOSQL

chmod +x db/init.sh

cp .env.example .env   # real values live here; .env is gitignored
docker compose up -d --wait

uv init --name macmcp --no-workspace
rm -f main.py hello.py
uv add fastmcp "psycopg[binary]" python-dotenv
uv add --dev pytest pytest-asyncio

[project]
name = "macmcp"
version = "0.1.0"
description = "A safe read-only Postgres MCP server"
readme = "README.md"
requires-python = ">=3.12"
dependencies = [
    "fastmcp>=3.4.3",
    "psycopg[binary]>=3.2.0",
    "python-dotenv>=1.0.0",
]

[dependency-groups]
dev = [
    "pytest>=9.1.1",
    "pytest-asyncio>=1.4.0",
]

touch config.py

"""Runtime configuration, read once from the environment.

Every security-relevant limit lives here: which role connects, which tables the
structured tool may touch, how long a statement may run, and how many rows a
single call may return. Secrets (the database URL, which carries the role's
password) come from the environment -- loaded from a gitignored .env in local
development -- and are never hard-coded in this file.
"""

import os
from dataclasses import dataclass, field

from dotenv import load_dotenv

# Load .env for local development. Real environment variables (for example, those
# injected by a platform in production) are not overridden, so this is a no-op
# wherever configuration is provided by the environment rather than a file.
load_dotenv()


def _tables_from_env() -> tuple[str, ...]:
    raw = os.environ.get("ALLOWED_TABLES", "customers,orders")
    return tuple(t.strip() for t in raw.split(",") if t.strip())


@dataclass(frozen=True)
class Settings:
    # The connection string carries the read-only role's password, so it comes
    # from the environment (.env in dev), never a literal in source.
    database_url: str = os.environ.get("DATABASE_URL", "")
    allowed_tables: tuple[str, ...] = field(default_factory=_tables_from_env)
    statement_timeout_ms: int = int(os.environ.get("STATEMENT_TIMEOUT_MS", "3000"))
    max_rows: int = int(os.environ.get("MAX_ROWS", "100"))


settings = Settings()

if not settings.database_url:
    raise RuntimeError(
        "DATABASE_URL is not set. Copy .env.example to .env (or run `make setup`), "
        "or set DATABASE_URL in your environment."
    )

touch db.py

"""Read-only, timeout-bounded Postgres connections.

Two libpq startup options are set on every connection:

  * default_transaction_read_only=on -> Postgres refuses any write, so even a
    query that slips past the Python guard cannot mutate data.
  * statement_timeout=<ms> -> the server cancels any statement that runs too
    long, so one broad query cannot stall the database.

Combined with the SELECT-only role, that is three independent layers before a
write or a runaway query can do damage.
"""

import psycopg
from psycopg.rows import dict_row

from config import settings


def connect() -> psycopg.Connection:
    """Open a connection as the read-only role with session guards applied."""
    return psycopg.connect(
        settings.database_url,
        autocommit=True,
        row_factory=dict_row,
        options=(
            f"-c statement_timeout={settings.statement_timeout_ms} "
            "-c default_transaction_read_only=on"
        ),
    )

touch safety.py

"""Input guards for the query tools.

These are the fast, friendly layer: they reject bad input with a clear message
before it reaches Postgres. They are NOT the security boundary -- the read-only
role and read-only transaction are. A guard that only inspects keywords can
always be fooled (a WITH statement can hide a data-modifying CTE), which is
exactly why the database is configured to refuse writes on its own.
"""


class QueryError(ValueError):
    """Raised when a query or identifier fails a safety check."""


def assert_select_only(sql_text: str) -> str:
    """Return a normalized single-statement SELECT, or raise QueryError.

    Rejects empty input, multiple statements, and anything that does not begin
    with SELECT or WITH. This is a usability guard, not the write barrier.
    """
    stripped = sql_text.strip().rstrip(";").strip()
    if not stripped:
        raise QueryError("query is empty")
    if ";" in stripped:
        raise QueryError("only a single statement is allowed")
    first_word = stripped.split(None, 1)[0].lower()
    if first_word not in ("select", "with"):
        raise QueryError("only SELECT (or WITH ... SELECT) queries are allowed")
    return stripped


def validate_identifiers(names: list[str], allowed: set[str], kind: str) -> None:
    """Raise QueryError if any name is not in the allowed set."""
    for name in names:
        if name not in allowed:
            raise QueryError(f"unknown {kind}: {name!r}")

touch server.py

"""A safe read-only Postgres MCP server.

Tools:
  * list_tables    -> the allow-listed tables this server will expose.
  * describe_table -> column names and types for one allow-listed table.
  * query_table    -> structured SELECT with allow-listed columns, parameterized
                      filters, and a row cap. No raw SQL from the caller.
  * run_query      -> a guarded raw SELECT for real queries, still read-only,
                      timeout-bounded, and row-capped.
"""

import psycopg
from fastmcp import FastMCP
from fastmcp.exceptions import ToolError
from psycopg import sql

import db
from config import settings
from safety import QueryError, assert_select_only, validate_identifiers

mcp = FastMCP("macmcp")


def _columns_of(table: str) -> list[dict]:
    """Return [{name, type}] for an allow-listed table, or raise QueryError."""
    validate_identifiers([table], set(settings.allowed_tables), "table")
    with db.connect() as conn:
        rows = conn.execute(
            "SELECT column_name AS name, data_type AS type "
            "FROM information_schema.columns "
            "WHERE table_schema = 'public' AND table_name = %s "
            "ORDER BY ordinal_position",
            (table,),
        ).fetchall()
    if not rows:
        raise QueryError(f"unknown table: {table!r}")
    return rows


def _cap(rows: list[dict], limit: int) -> dict:
    """Truncate to `limit` rows and report whether more existed.

    Callers over-fetch one extra row (limit + 1) so a full result signals
    truncation instead of silently returning a short answer.
    """
    truncated = len(rows) > limit
    return {"rows": rows[:limit], "truncated": truncated}


@mcp.tool
def list_tables() -> list[str]:
    """List the tables this server is allowed to read."""
    return list(settings.allowed_tables)


@mcp.tool
def describe_table(table: str) -> list[dict]:
    """Return column names and types for one allowed table."""
    try:
        return _columns_of(table)
    except QueryError as exc:
        raise ToolError(str(exc))


@mcp.tool
def query_table(
    table: str,
    columns: list[str] | None = None,
    filters: dict[str, str] | None = None,
    limit: int | None = None,
) -> dict:
    """Select from one allowed table with allow-listed columns and filters.

    columns defaults to all columns. filters is column -> value, combined with
    AND and passed as parameters (never interpolated). limit is capped by
    MAX_ROWS.
    """
    filters = filters or {}
    try:
        real_columns = [c["name"] for c in _columns_of(table)]
        allowed = set(real_columns)
        selected = columns or real_columns
        validate_identifiers(selected, allowed, "column")
        validate_identifiers(list(filters), allowed, "column")
    except QueryError as exc:
        raise ToolError(str(exc))

    row_limit = settings.max_rows if limit is None else min(limit, settings.max_rows)

    query = sql.SQL("SELECT {cols} FROM {tbl}").format(
        cols=sql.SQL(", ").join(sql.Identifier(c) for c in selected),
        tbl=sql.Identifier(table),
    )
    params: list = []
    if filters:
        conditions = [
            sql.SQL("{col} = %s").format(col=sql.Identifier(col)) for col in filters
        ]
        query += sql.SQL(" WHERE ") + sql.SQL(" AND ").join(conditions)
        params.extend(filters.values())
    query += sql.SQL(" LIMIT %s")
    params.append(row_limit + 1)  # one extra row detects truncation

    try:
        with db.connect() as conn:
            rows = conn.execute(query, params).fetchall()
    except psycopg.Error as exc:
        raise ToolError(f"database error: {exc}")
    return _cap(rows, row_limit)


@mcp.tool
def run_query(query: str, params: list | None = None) -> dict:
    """Run a single read-only SELECT and return capped rows.

    Pass values through `params` (with %s placeholders) rather than formatting
    them into the query string.
    """
    try:
        sql_text = assert_select_only(query)
    except QueryError as exc:
        raise ToolError(str(exc))
    try:
        with db.connect() as conn:
            cur = conn.execute(sql_text, params or [])
            rows = cur.fetchmany(settings.max_rows + 1)
    except psycopg.Error as exc:
        raise ToolError(f"database error: {exc}")
    return _cap(rows, settings.max_rows)


def main() -> None:
    import os

    if os.environ.get("MCP_TRANSPORT", "stdio").lower() == "http":
        mcp.run(
            transport="http",
            host=os.environ.get("MCP_HOST", "127.0.0.1"),
            port=int(os.environ.get("MCP_PORT", "8000")),
        )
    else:
        mcp.run()  # stdio


if __name__ == "__main__":
    main()

mkdir -p tests
touch tests/__init__.py tests/test_safety.py tests/test_server.py pytest.ini

[pytest]
asyncio_mode = auto

"""Unit tests for the pure guards. No database required."""

import pytest

from safety import QueryError, assert_select_only, validate_identifiers


def test_accepts_plain_select():
    assert assert_select_only("SELECT 1") == "SELECT 1"


def test_accepts_with_and_strips_trailing_semicolon():
    assert assert_select_only("WITH t AS (SELECT 1) SELECT * FROM t;") == (
        "WITH t AS (SELECT 1) SELECT * FROM t"
    )


def test_rejects_empty():
    with pytest.raises(QueryError):
        assert_select_only("   ")


def test_rejects_multiple_statements():
    with pytest.raises(QueryError):
        assert_select_only("SELECT 1; DROP TABLE customers")


@pytest.mark.parametrize("bad", ["INSERT INTO t VALUES (1)", "UPDATE t SET x=1", "DELETE FROM t"])
def test_rejects_non_select(bad):
    with pytest.raises(QueryError):
        assert_select_only(bad)


def test_validate_identifiers_rejects_unknown():
    with pytest.raises(QueryError):
        validate_identifiers(["price"], {"id", "name"}, "column")


def test_validate_identifiers_accepts_known():
    validate_identifiers(["id", "name"], {"id", "name", "city"}, "column")

"""Integration tests against the running Postgres container.

Requires `docker compose up -d --wait` first (or `make up`). If the database is
unreachable, the whole module is skipped with a clear message rather than
failing.
"""

from dataclasses import replace

import psycopg
import pytest
from fastmcp import Client
from fastmcp.exceptions import ToolError

import db
import server


@pytest.fixture(scope="module", autouse=True)
def require_database():
    try:
        with db.connect() as conn:
            conn.execute("SELECT 1")
    except psycopg.Error as exc:
        pytest.skip(f"Postgres not reachable (run `make up`): {exc}")


async def test_list_and_describe():
    async with Client(server.mcp) as client:
        tables = (await client.call_tool("list_tables", {})).data
        assert set(tables) == {"customers", "orders"}
        cols = (await client.call_tool("describe_table", {"table": "customers"})).data
        assert {c["name"] for c in cols} == {"id", "name", "email", "city"}


async def test_query_table_filters_are_parameterized():
    async with Client(server.mcp) as client:
        result = (
            await client.call_tool(
                "query_table",
                {"table": "customers", "columns": ["id", "name"], "filters": {"city": "London"}},
            )
        ).data
        # No ORDER BY in query_table, so compare as a set, not a sequence.
        assert {r["name"] for r in result["rows"]} == {"Ada Lovelace", "Alan Turing"}
        assert result["truncated"] is False


async def test_query_table_limit_is_exact():
    """An explicit limit returns exactly that many rows (plus a truncated flag)."""
    async with Client(server.mcp) as client:
        result = (
            await client.call_tool(
                "query_table", {"table": "orders", "columns": ["id"], "limit": 3}
            )
        ).data
        assert len(result["rows"]) == 3  # 5 orders exist, limit caps to 3
        assert result["truncated"] is True


async def test_query_table_rejects_unknown_table_and_column():
    async with Client(server.mcp) as client:
        with pytest.raises(ToolError):
            await client.call_tool("query_table", {"table": "secrets"})
        with pytest.raises(ToolError):
            await client.call_tool(
                "query_table", {"table": "customers", "columns": ["password"]}
            )


async def test_run_query_reads_and_caps_rows(monkeypatch):
    # Settings is a frozen dataclass, so rebind the module's `settings` to a
    # copy with a smaller cap rather than mutating the instance.
    monkeypatch.setattr(server, "settings", replace(server.settings, max_rows=3))
    async with Client(server.mcp) as client:
        result = (
            await client.call_tool(
                "run_query", {"query": "SELECT id FROM orders ORDER BY id"}
            )
        ).data
        assert [r["id"] for r in result["rows"]] == [1, 2, 3]
        assert result["truncated"] is True  # 5 orders, capped at 3


async def test_run_query_parameters_prevent_injection():
    async with Client(server.mcp) as client:
        injection = "London'; DROP TABLE customers; --"
        result = (
            await client.call_tool(
                "run_query",
                {"query": "SELECT count(*) AS n FROM customers WHERE city = %s",
                 "params": [injection]},
            )
        ).data
        assert result["rows"][0]["n"] == 0  # no such city; nothing dropped


async def test_write_is_rejected():
    async with Client(server.mcp) as client:
        with pytest.raises(ToolError):
            await client.call_tool(
                "run_query", {"query": "SELECT 1; INSERT INTO customers VALUES (9)"}
            )


async def test_direct_write_is_refused_by_postgres():
    """Even bypassing the guard, the read-only role/transaction refuses a write."""
    with db.connect() as conn:
        with pytest.raises(psycopg.Error):
            conn.execute("INSERT INTO customers (id, name, email, city) "
                         "VALUES (99, 'x', 'x@x', 'x')")


async def test_statement_timeout_cancels_slow_query(monkeypatch):
    """A deliberately slow query is cancelled by statement_timeout."""
    monkeypatch.setattr(db, "settings", replace(db.settings, statement_timeout_ms=500))
    async with Client(server.mcp) as client:
        with pytest.raises(ToolError):
            await client.call_tool("run_query", {"query": "SELECT pg_sleep(3)"})

make up
make test

mkdir -p scripts
touch scripts/demo.py

"""Drive the tools in-memory against the running database and print results."""

import asyncio
import json

from fastmcp import Client

import server


async def main() -> None:
    async with Client(server.mcp) as client:
        tables = (await client.call_tool("list_tables", {})).data
        print("tables:", tables)

        london = (
            await client.call_tool(
                "query_table",
                {"table": "customers", "columns": ["name", "city"],
                 "filters": {"city": "London"}},
            )
        ).data
        print("London customers:", json.dumps(london["rows"]))

        revenue = (
            await client.call_tool(
                "run_query",
                {"query": "SELECT sum(total_cents) AS paid_cents "
                          "FROM orders WHERE status = %s", "params": ["paid"]},
            )
        ).data
        print("paid revenue (cents):", revenue["rows"][0]["paid_cents"])


if __name__ == "__main__":
    asyncio.run(main())

make demo

touch Makefile

.DEFAULT_GOAL := help

COMPOSE := docker compose
PSQL_SUPER := $(COMPOSE) exec -T db psql -U postgres -d appdb

.PHONY: help setup up down reset psql serve test demo logs clean

help:  ## Show this help screen
	@grep -E '^[a-zA-Z_-]+:.*?## .*$$' $(MAKEFILE_LIST) \
		| awk 'BEGIN {FS = ":.*?## "}; {printf "  \033[36m%-8s\033[0m %s\n", $$1, $$2}'

# Create .env from the committed template on first use. Real credentials live in
# .env (gitignored); .env.example is the checked-in placeholder.
.env:
	@cp .env.example .env
	@echo "Created .env from .env.example (dev defaults; edit for anything beyond local)."

setup: .env  ## Create .env from .env.example if it is missing
	@echo ".env is ready."

up: .env  ## Start Postgres and wait until it is ready
	$(COMPOSE) up -d --wait

down:  ## Stop Postgres (keeps the data volume)
	$(COMPOSE) down

reset:  ## Stop Postgres and delete the data volume (re-runs init.sh next up)
	$(COMPOSE) down -v

psql:  ## Open a psql shell as the superuser
	$(PSQL_SUPER)

serve: .env  ## Run the MCP server over HTTP in the foreground (Ctrl-C to stop)
	MCP_TRANSPORT=http uv run python server.py

test: up  ## Bring the database up, then run the test suite
	uv run pytest -v

demo: up  ## Run the demo script against the database
	uv run python -m scripts.demo

logs:  ## Tail the Postgres container logs
	$(COMPOSE) logs -f db

clean:  ## Remove Python caches (keeps the database volume and .env)
	rm -rf .pytest_cache __pycache__ tests/__pycache__

{
  "mcpServers": {
    "postgres-readonly": {
      "command": "uv",
      "args": ["run", "python", "server.py"],
      "cwd": "/absolute/path/to/readonly-postgres-mcp-server-macos"
    }
  }
}