Demo-nya berakhir dengan bar chart yang rapi dan semua orang manggut. Terus agent-nya lanjut jalan. Dia memutuskan baris paket gratis sudah basi, ngejalanin DELETE, dan dashboard kamu berubah jadi error 500.
Patch yang biasa dipakai adalah satu baris di system prompt: "jangan pernah ubah database." Itu saran yang ditulis dalam bahasa manusia, ditujukan ke model yang menebak token berikutnya. Bukan sistem izin.
Yang di bawah ini sistem izinnya, dari awal sampai akhir. Semua perintah di artikel ini sudah dijalankan di PostgreSQL 18.6 dan mcp 2.3.0 sebelum masuk ke sini.
Yang kamu butuhin
- PostgreSQL 14 ke atas
- Python 3.11 ke atas (MCP Python SDK v2 minta versi ini)
- uv
- Host MCP: Claude Desktop, Claude Code, atau client apa pun yang bisa stdio
Langkah 1: role yang gak bisa nulis
Jalankan sebagai superuser di database aplikasi kamu. Ganti app dengan nama database kamu.
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';
Dua baris GRANT itu pasangan yang sering kelupaan. GRANT SELECT ON ALL TABLES cuma mencakup tabel yang ada sekarang. ALTER DEFAULT PRIVILEGES mencakup tabel yang dibuat setelahnya, dan cuma tabel yang dibuat oleh role yang menjalankannya. Bikin tabel subscriptions bulan depan sebagai user aplikasi, agent bisa baca. Bikin sebagai owner lain, agent gak bisa.
CONNECTION LIMIT 5 ada karena agent yang retry dalam loop bakal buka empat puluh koneksi dan ngabisin pool aplikasi. Lima cukup buat satu agent dan cukup kecil buat kelihatan.
Langkah 2: coba rusakin
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
INSERT kedua itu yang penting. default_transaction_read_only bisa diubah sendiri oleh user, jadi client mana pun bisa mematikannya, termasuk model yang baca pesan error lalu coba lagi. Yang gak bisa dia matikan adalah privilege INSERT yang memang gak pernah dikasih. Default read-only itu sabuk pengaman. Daftar grant itu temboknya.
Dua hal kecil lain dari sesi yang sama. create temp table tetap jalan di dalam transaksi read-only, karena mode read-only PostgreSQL menyasar tulisan permanen dan gak menjanjikan semua byte ke disk diblokir. Dan nextval() gagal, karena role-nya gak punya USAGE di sequence.
Langkah 3: MCP server-nya
uv init pg-ro-mcp
cd pg-ro-mcp
uv add "mcp[cli]" "psycopg[binary]"
Dua import perlu diperhatikan. MCPServer adalah entry point v2, bukan class FastMCP yang lama, dan ToolError ada di mcp.server.mcpserver.exceptions di rilis ini.
# 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
Tiap helper kerjaannya satu. _checked nolak apa pun yang kata pertamanya bukan SELECT atau WITH, lalu buang titik koma di akhir. Ini filter paling murah dan paling gampang dikelabui, makanya gak jalan sendirian. Query yang dibuka dengan komentar SQL juga ditolak; pesan errornya kebaca, jadi model bakal coba lagi tanpa komentar.
_connect menyalakan read_only = True, dan psycopg menerjemahkannya jadi BEGIN READ ONLY. Walaupun filter string-nya lolos sesuatu, transaksinya sendiri yang nolak tulisan itu.
_clip memotong sel teks yang panjang. Satu kolom jsonb bisa nyimpen satu megabyte, dan semua byte itu masuk ke context window model.
Sekarang tool-nya.
@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 memfilter pakai has_table_privilege, jadi agent cuma lihat tabel yang role-nya bisa baca dan gak buang satu call buat tabel yang pasti gagal. NULLIF(c.reltuples, -1) kelihatan sepele tapi berguna: PostgreSQL melaporkan -1 buat tabel yang belum pernah di-analyze, dan model yang baca "est_rows": -1 bakal bilang sesuatu yang yakin tapi salah.
Wrapper LIMIT-nya kerja lebih dari sekadar membatasi baris. Karena ada parameter yang dikirim, psycopg pakai extended query protocol, dan protokol itu nolak beberapa statement dalam satu call. select 1; select 2 gak akan pernah sampai statement keduanya.
set_config('statement_timeout', ..., true) mengatur timeout cuma buat transaksi ini. SET LOCAL statement_timeout = %s yang kelihatan lebih natural gak jalan, karena psycopg mengirim parameter terpisah dan SET bukan statement berparameter. Nilai timeout yang tetap bisa pakai SET tanpa parameter, tapi kamu kehilangan override lewat env.
Langkah 4: panggil tanpa client
SDK-nya punya client in-memory, jadi kamu bisa ngetes semuanya sebelum disambung ke mana-mana.
# 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)
Dengan tabel customers berisi tiga baris, outputnya:
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 ";"
Percobaan INSERT gak pernah nyampe ke database. Percobaan multi-statement nyampe ke PostgreSQL dan mati di parser. Kalau filter _checked kamu hapus lalu INSERT dicoba lagi, hasilnya InsufficientPrivilege: permission denied for table customers, yang berarti lapisan di bawahnya yang bekerja.
Langkah 5: sambungin ke 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"
}
}
}
}
Buat Claude Desktop file-nya claude_desktop_config.json, di macOS ada di ~/Library/Application Support/Claude/. Kalau mau coba tool-nya dulu tanpa host, jalankan uv run mcp dev server.py dan pakai Inspector yang muncul.
Simpan DSN di env host, bukan di repo, dan kasih password yang gak kamu pakai di tempat lain. PostgreSQL 14 ke atas nge-hash dengan scram-sha-256 secara default.
Tiap lapisan sebenarnya nahan apa
Prompt gak nahan apa-apa. Itu konteks, dan konteks kalah sama instruksi yang cukup kuat di user turn.
Role nahan tulisan walaupun client-nya bohong, ngubah GUC sesinya sendiri, atau ngejalanin SQL yang gak kepikiran sama kamu. Privilege dicek server di setiap statement.
Transaksi read-only nahan kecelakaan buat client yang tetap pakai default. Murah, dan dia nangkep kasus di mana enam bulan lagi ada orang nambah satu grant.
Batas baris dan timeout nahan bencana yang membosankan: query yang balikin 4 juta baris, ngisi context window, dan bikin kamu bayar satu dolar buat tahu agent-nya nanya hal yang salah.
Di mana ini masih bocor
Row-level security. Kalau aplikasimu pakai RLS dengan variabel tenant per request, role ini gak nembus apa-apa, tapi dia juga gak nyetel tenant. Pilihannya: satu role per tenant, atau kirim tenant id ke tiap tool dan set pakai set_config di transaksi yang sama.
Kolom yang kamu anggap rahasia. GRANT SELECT ON ALL TABLES mencakup kolom email dan token. Grant per kolom solusinya, dan select * berhenti jalan, yang biasanya justru kamu mau:
GRANT SELECT (id, plan, mrr) ON customers TO agent_ro;
Dengan grant itu saja, select id, plan from customers jalan, sementara select email from customers dan select * from customers balikin permission denied for table customers.
Query analitik. Timeout 5 detik bakal ngebunuh yang jujur juga. Arahkan agent ke read replica, atau kasih MCP server terpisah yang nembak warehouse.
Audit. log_statement = 'all' nulis teks statement ke log Postgres, dan log_line_prefix default gak punya nama user atau application name. Kalau lebih dari satu agent konek, set prefix-nya supaya ada %u dan %a. Parameter yang di-bind juga gak masuk log, jadi kamu lihat query apa yang jalan tapi bukan nilai yang dikirim.
Akses tulis. Server ini read-only dengan sengaja. Suatu hari agent perlu bikin tiket atau ngubah flag, jangan tambah cabang INSERT di tool ini. Bikin server kedua dengan tiga tool tulis yang namanya jelas, allowlist, dan langkah approval manusia.
Setelah ini
Tambah satu kasus gagal di CI buat tiap guardrail yang kamu pedulikan: INSERT-nya, DROP-nya, select 1; select 2-nya, pg_sleep-nya. Jalan dalam satu detik dan bakal gagal dengan suara keras di hari orang ngelonggarin grant.
Lalu lihat apa yang sebenarnya ditanyain agent. pg_stat_statements di replica bakal kasih tahu query mana yang lambat, berulang, atau ngawur, dan daftar itu input prompt engineering yang lebih berguna daripada nebak-nebak.
Agent-nya tetap bakal nulis SQL jelek. Dia cuma gak bisa nulis SQL jelek yang selamat sampai commit.
Referensi
- PostgreSQL: CREATE ROLE, termasuk CONNECTION LIMIT dan setting level role: https://www.postgresql.org/docs/current/sql-createrole.html
- PostgreSQL: SET TRANSACTION, daftar persis apa yang diblokir mode read-only: https://www.postgresql.org/docs/current/sql-set-transaction.html
- PostgreSQL: Privileges, GRANT dan grant per kolom: https://www.postgresql.org/docs/current/ddl-priv.html
- MCP Python SDK: Tools, MCPServer dan ToolAnnotations: https://py.sdk.modelcontextprotocol.io/servers/tools/
- psycopg 3: perbedaan dengan psycopg2, perilaku multi-statement: https://www.psycopg.org/psycopg3/docs/basic/from_pg2.html