The demo ends with a clean bar chart and everyone nods. Then the agent keeps going. It decides the free-tier rows are stale, runs a DELETE, and your dashboard turns into a 500.
The usual patch is a line in the system prompt: "never modify the database." That is a suggestion written in English, aimed at a model that guesses the next token. It is not a permission system.
What follows is the permission system, end to end. Every command below ran against PostgreSQL 18.6 and mcp 2.3.0 before it landed in this post.
What you need
- PostgreSQL 14 or newer
- Python 3.11 or newer (the MCP Python SDK v2 requires it)
- uv
- An MCP host: Claude Desktop, Claude Code, or any client that speaks stdio
Step 1: a role that cannot write
Run this as a superuser against your app database. Replace app with your database name.
CREATE ROLE agent_ro LOGIN PASSWORD 'change-me' CONNECTION LIMIT 5;
GRANT CONNECT ON DATABASE app TO agent_ro;
GRANT USAGE ON SCHEMA public TO agent_ro;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO agent_ro;
ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT SELECT ON TABLES TO agent_ro;
ALTER ROLE agent_ro SET default_transaction_read_only = on;
ALTER ROLE agent_ro SET statement_timeout = '5s';
ALTER ROLE agent_ro SET application_name = 'mcp-sql-ro';
ALTER ROLE agent_ro SET log_statement = 'all';
The two GRANT lines are the pair people forget. GRANT SELECT ON ALL TABLES covers what exists now. ALTER DEFAULT PRIVILEGES covers tables created later, and only tables created by the role that ran it. Create next month's subscriptions table as your app user and the agent can read it. Create it as a different owner and it cannot.
CONNECTION LIMIT 5 exists because an agent retrying in a loop will open forty connections and starve the app pool. Five is enough for one agent and small enough that you notice.
Step 2: try to break it
PGPASSWORD=change-me psql -h 127.0.0.1 -U agent_ro -d app
app=> select current_user, current_setting('default_transaction_read_only');
current_user | current_setting
--------------+-----------------
agent_ro | on
app=> insert into customers(email, plan, mrr) values ('tester','pro',1);
ERROR: cannot execute INSERT in a read-only transaction
app=> set default_transaction_read_only = off;
SET
app=> insert into customers(email, plan, mrr) values ('tester','pro',1);
ERROR: permission denied for table customers
That second insert is the one that matters. default_transaction_read_only is a user-settable setting, so any client can turn it off, including a model that read the error message and tried again. What it cannot turn off is the missing INSERT privilege. The read-only default is a seatbelt. The grant list is the wall.
Two smaller things the same session shows. create temp table still works inside a read-only transaction, because PostgreSQL's read-only mode targets permanent writes and does not claim to stop every byte that reaches disk. And nextval() fails, because the role has no USAGE on the sequence.
Step 3: the MCP server
uv init pg-ro-mcp
cd pg-ro-mcp
uv add "mcp[cli]" "psycopg[binary]"
Two imports are worth flagging. MCPServer is the v2 entry point, not the older FastMCP class, and ToolError lives at mcp.server.mcpserver.exceptions in this release.
# server.py
import json
import os
from typing import Annotated
import psycopg
from mcp.server import MCPServer
from mcp.server.mcpserver.exceptions import ToolError
from mcp.types import ToolAnnotations
from psycopg.rows import dict_row
from pydantic import Field
DSN = os.environ["PG_DSN"]
MAX_ROWS = 200
MAX_CELL = 2000
TIMEOUT_MS = int(os.environ.get("SQL_TIMEOUT_MS", "8000"))
mcp = MCPServer(
"postgres-ro",
instructions=(
"Read-only access to the app database. Call list_tables first, then "
"describe_table, then run_query with a single SELECT or WITH statement."
),
)
def _checked(sql: str) -> str:
sql = sql.strip()
first = sql.split(maxsplit=1)[0].lower() if sql else ""
if first not in {"select", "with"}:
raise ToolError("Only SELECT and WITH statements are allowed.")
return sql.rstrip().rstrip(";")
def _connect() -> psycopg.Connection:
conn = psycopg.connect(
DSN, autocommit=True, row_factory=dict_row, connect_timeout=5
)
conn.read_only = True
return conn
def _clip(value: object) -> object:
if isinstance(value, str) and len(value) > MAX_CELL:
return value[:MAX_CELL] + "... [truncated]"
return value
Each helper does one job. _checked rejects anything whose first word is not SELECT or WITH, then drops the trailing semicolon. It is the cheapest filter and the easiest to fool, which is why it does not work alone. A query that opens with a SQL comment gets rejected too; the error is readable, so the model retries without the comment.
_connect sets read_only = True, and psycopg turns that into BEGIN READ ONLY. Even when the string filter misses something, the transaction itself refuses the write.
_clip truncates long text cells. A single jsonb column can hold a megabyte, and every byte of it lands in the model's context window.
Now the tools.
@mcp.tool(annotations=ToolAnnotations(read_only_hint=True, open_world_hint=False))
def list_tables() -> str:
"""List tables and views the agent can read, with estimated row counts."""
with _connect() as conn, conn.transaction():
cur = conn.cursor()
cur.execute("SELECT set_config('statement_timeout', %s, true)", (str(TIMEOUT_MS),))
cur.execute(
"""
SELECT c.relname AS name,
CASE c.relkind
WHEN 'r' THEN 'table'
WHEN 'v' THEN 'view'
ELSE 'matview'
END AS kind,
NULLIF(c.reltuples, -1)::bigint AS est_rows
FROM pg_class c
JOIN pg_namespace n ON n.oid = c.relnamespace
WHERE n.nspname = 'public'
AND c.relkind IN ('r', 'v', 'm')
AND has_table_privilege(c.oid, 'SELECT')
ORDER BY 1
"""
)
return json.dumps(cur.fetchall(), default=str)
@mcp.tool(annotations=ToolAnnotations(read_only_hint=True, open_world_hint=False))
def describe_table(
table: Annotated[str, Field(description="Table name in the public schema.")],
) -> str:
"""Show columns, types and defaults for one table."""
with _connect() as conn, conn.transaction():
cur = conn.cursor()
cur.execute("SELECT set_config('statement_timeout', %s, true)", (str(TIMEOUT_MS),))
cur.execute(
"""
SELECT column_name, data_type, is_nullable, column_default
FROM information_schema.columns
WHERE table_schema = 'public' AND table_name = %s
ORDER BY ordinal_position
""",
(table,),
)
rows = cur.fetchall()
if not rows:
raise ToolError(f"No readable table named {table!r} in schema public.")
return json.dumps(rows, default=str)
@mcp.tool(annotations=ToolAnnotations(read_only_hint=True, open_world_hint=False))
def run_query(
sql: Annotated[str, Field(description="One SELECT or WITH query. No semicolon needed.")],
max_rows: Annotated[int, Field(ge=1, le=MAX_ROWS, description="Row cap.")] = 50,
) -> str:
"""Run one read-only SQL query and return rows as JSON."""
wrapped = f"SELECT * FROM ({_checked(sql)}) AS agent_query LIMIT %s"
try:
with _connect() as conn, conn.transaction():
cur = conn.cursor()
cur.execute(
"SELECT set_config('statement_timeout', %s, true)", (str(TIMEOUT_MS),)
)
cur.execute(wrapped, (max_rows,))
rows = cur.fetchall()
columns = [d.name for d in cur.description]
except psycopg.errors.QueryCanceled:
raise ToolError(
f"Query canceled after {TIMEOUT_MS} ms. Filter on an indexed column or aggregate less."
)
except psycopg.Error as exc:
raise ToolError(f"{type(exc).__name__}: {exc}".strip())
payload = {
"columns": columns,
"row_count": len(rows),
"rows": [{k: _clip(v) for k, v in row.items()} for row in rows],
}
return json.dumps(payload, default=str, ensure_ascii=False)
if __name__ == "__main__":
mcp.run(transport="stdio")
list_tables filters on has_table_privilege, so the agent sees exactly the tables its role can read and does not waste a call on one that will fail. The NULLIF(c.reltuples, -1) is cosmetic but worth it: PostgreSQL reports -1 for a table that has never been analyzed, and a model reading "est_rows": -1 will say something confident and wrong about it.
The LIMIT wrapper is doing more than capping rows. Because a parameter is passed, psycopg uses the extended query protocol, and that protocol refuses multiple statements in one call. select 1; select 2 never gets a second statement through.
set_config('statement_timeout', ..., true) sets the timeout for this transaction only. The obvious SET LOCAL statement_timeout = %s does not work, because psycopg sends parameters separately and SET is not a parameterized statement. Simpler timeout values can be set with SET and no parameter, but then you lose the env override.
Step 4: call it without a client
The SDK ships an in-memory client, so you can test the whole thing before wiring it into anything.
# test_server.py
import anyio
from mcp import Client
from server import mcp
CASES = [
("select plan, count(*) as n, sum(mrr) from customers group by plan", False),
("insert into customers (email, plan, mrr) values ('tester', 'pro', 1)", True),
("select 1; select 2", True),
]
async def main() -> None:
async with Client(mcp, raise_exceptions=True) as client:
for sql, should_fail in CASES:
result = await client.call_tool("run_query", {"sql": sql})
print(result.is_error == should_fail, result.content[0].text[:120])
anyio.run(main)
Against a customers table with three rows, that prints:
True {"columns": ["plan", "n", "mrr"], "row_count": 3, "rows": [{"plan": ...
True Error executing tool run_query: Only SELECT and WITH statements are allowed.
True Error executing tool run_query: SyntaxError: syntax error at or near ";"
The write attempt never reached the database. The multi-statement attempt reached PostgreSQL and died in the parser. If you drop the _checked filter and retry the INSERT, you get InsufficientPrivilege: permission denied for table customers instead, which is the layer below doing its job.
Step 5: plug it into your host
{
"mcpServers": {
"postgres-ro": {
"command": "uv",
"args": ["--directory", "/srv/pg-ro-mcp", "run", "server.py"],
"env": {
"PG_DSN": "host=127.0.0.1 dbname=app user=agent_ro password=change-me",
"SQL_TIMEOUT_MS": "8000"
}
}
}
}
For Claude Desktop the file is claude_desktop_config.json, on macOS under ~/Library/Application Support/Claude/. To poke at the tools by hand first, run uv run mcp dev server.py and use the Inspector it prints.
Keep the DSN in the host's env block, not in the repo, and give the role a password you do not reuse. Postgres 14 and newer hash it with scram-sha-256 by default.
What each layer actually stops
A prompt stops nothing. It is context, and context loses to a strong enough instruction in the user turn.
The role stops writes even if the client lies, sets its own session GUCs, or runs SQL you never imagined. Privileges are checked by the server on every statement.
The read-only transaction stops accidents for clients that keep the default. It is cheap and it catches the case where someone grants an extra privilege six months from now.
The row cap and timeout stop the boring disaster: a query that returns 4 million rows, fills the context window, and costs you a dollar to discover the agent asked for the wrong thing.
Where this still breaks
Row-level security. If your app relies on RLS with a per-request tenant variable, this role bypasses none of it, but it also sets no tenant. Either grant a role per tenant, or pass the tenant id into every tool and set it with set_config inside the same transaction.
Columns you consider secret. GRANT SELECT ON ALL TABLES includes the email and token columns. Column grants are the fix, and select * stops working, which is usually what you want:
GRANT SELECT (id, plan, mrr) ON customers TO agent_ro;
With only that grant, select id, plan from customers works and both select email from customers and select * from customers return permission denied for table customers.
Analytics queries. A 5 second statement timeout will kill the honest ones too. Point the agent at a read replica, or give it a separate MCP server that hits your warehouse.
Audit. log_statement = 'all' writes the statement text to the Postgres log, and the default log_line_prefix has no user or application name in it. If more than one agent connects, set the prefix to include %u and %a. Bound parameters are not in the log either, so you see which query ran but not the values it was given.
Write access. This server is read-only on purpose. The day the agent needs to create a ticket or flip a flag, do not add an INSERT branch to this tool. Build a second server with three named write tools, an allowlist, and a human approval step.
What to do next
Add a failing case to CI for every guardrail you care about: the INSERT, the DROP, the select 1; select 2, the pg_sleep. They run in a second and they fail loudly the day someone loosens a grant.
Then look at what the agent actually asks for. pg_stat_statements on the replica will tell you which of its queries are slow, repeated, or nonsense, and that list is a better prompt-engineering input than any amount of guessing.
The agent will still write bad SQL. It just cannot write bad SQL that survives a commit.
References
- PostgreSQL: CREATE ROLE, including CONNECTION LIMIT and role-level settings: https://www.postgresql.org/docs/current/sql-createrole.html
- PostgreSQL: SET TRANSACTION, the exact list of what read-only blocks: https://www.postgresql.org/docs/current/sql-set-transaction.html
- PostgreSQL: Privileges, GRANT and column-level grants: https://www.postgresql.org/docs/current/ddl-priv.html
- MCP Python SDK: Tools, MCPServer and ToolAnnotations: https://py.sdk.modelcontextprotocol.io/servers/tools/
- psycopg 3: differences from psycopg2, multi-statement behavior: https://www.psycopg.org/psycopg3/docs/basic/from_pg2.html