MCP Postgres Analyst

A Model Context Protocol server that lets AI assistants explore and query a PostgreSQL or SQLite database read-only, behind a SQL allow-list, automatic row limits, timeouts, column masking, rate limits and a JSONL audit log.

Try it live

There is no GitHub Pages site for this repo: an MCP server has no browser UI, and the Codespace gives you the real thing.

Related MCP server: Database Auditor MCP

Why this exists

Every team that connects an assistant to its database ends up with the same question from security: what stops it from running DELETE, reading password hashes, or pulling a whole customer table with phone numbers into a chat log? In payment and e-commerce systems I have worked on, the answer cannot be "the prompt says please". This server puts the controls in code: queries are parsed and validated before they reach the database, the database session itself is read-only, personal data is masked on the way out, and every call is written to an audit log that compliance can read.

Features

• Seven MCP tools: list_schemas, describe_table, sample_rows, run_query, explain_query, find_columns (by name or comment) and run_saved_query, all with JSON schemas, structured output and read-only annotations

• Resources: schema://overview (Markdown data dictionary) and queries://saved / queries://saved/{name}; prompts: analyse_table and write_report

• SQL guard built on sqlglot: one statement only, SELECT/WITH only, no data-modifying CTEs, SELECT INTO, FOR UPDATE, COPY, SET, PRAGMA; blocked functions (pg_sleep, dblink*, lo_*, pg_read_file, set_config, load_extension, ...) and system tables (pg_authid, pg_stat_activity, ...)

• The database only ever receives SQL re-rendered from the validated AST, never the caller's raw text

• Automatic LIMIT injection and clamping (also replaces LIMIT ALL, sub-query limits and FETCH FIRST), plus row and byte caps on the result

• Statement timeout: statement_timeout on PostgreSQL, a progress-handler interrupt on SQLite

• Column masking from a YAML policy (email, phone, last4, hash, redact) that follows aliases, expressions, CTEs and derived tables; blocked columns disappear even from SELECT *; whole-row tricks like row_to_json(c) are rejected

• Per-tool token-bucket rate limits and an append-only JSONL audit log of every call (ok, rejected, timeout, rate limited)

• Both transports: stdio for desktop clients, Streamable HTTP with optional bearer token, DNS-rebinding protection and a /health endpoint

• Deterministic demo database (fictional Bangladeshi online shop: customers, products, orders, bKash/Nagad/card/COD payments) seeded identically into SQLite or PostgreSQL

Architecture

flowchart LR
    subgraph Clients
        CD[Claude Desktop / Cursor / VS Code<br/>stdio]
        HC[Remote MCP client<br/>Streamable HTTP + bearer token]
    end
    CD --> S[MCPServer<br/>tools · resources · prompts]
    HC --> AUTH[BearerTokenMiddleware<br/>/health is public] --> S
    S --> SVC[AnalystService]
    SVC --> RL[ToolRateLimiter<br/>token bucket per tool]
    SVC --> G[SqlGuard<br/>sqlglot AST checks<br/>LIMIT rewrite · lineage]
    G --> DB[(Database adapter)]
    DB --> PG[(PostgreSQL<br/>READ ONLY txn<br/>statement_timeout)]
    DB --> SQ[(SQLite<br/>mode=ro · authorizer<br/>progress interrupt)]
    DB --> M[Masking<br/>name + lineage rules]
    M --> SVC
    SVC --> A[[audit.jsonl]]
    P[policy.yaml] -.-> G
    P -.-> M
    P -.-> RL

Key decisions:

• Parse, validate, re-render. The guard parses with the target dialect and executes ast.sql(), so comment tricks, stacked statements and dollar-quoted payloads cannot smuggle anything past the checks: what was validated is byte-for-byte what runs.

• Defence in depth, not one wall. Even if a query slipped through the guard, PostgreSQL runs it in a READ ONLY transaction under a SELECT-only role (compose creates analyst_ro, and the seed revokes payroll access at the database level too); SQLite opens the file mode=ro with an authorizer that denies anything but reads.

• Masking follows lineage, not just column names. compute_taint walks every projection to a fixed point, so WITH x AS (SELECT email AS contact ...) SELECT upper(contact) AS shout FROM x still masks shout. count(...) is exempt because it reveals a number, not the value.

• One extra row. When the guard writes the LIMIT it asks for limit + 1 rows, so the result can say truncated: row_limit honestly instead of guessing.

• Transport-agnostic core. AnalystService is synchronous and knows nothing about MCP; the server runs it in a worker thread. Tests exercise it directly, through the SDK's in-process client, over real HTTP and over a stdio subprocess.

Safety model

Known limits, stated plainly: masking hides values in results but a determined caller can still learn about a masked column through filters (WHERE email LIKE 'a%' changes the row count). Treat masking as protection against accidental exposure in transcripts, and keep truly secret columns in blocked_columns or out of the role's grants. Lineage is name-based across the query, so it can over-mask a same-named column from another table; it errs on the safe side.

Tech Stack

Quick Start

Local, SQLite demo (no database server)

git clone https://github.com/AkibUllah/mcp-postgres-analyst.git
cd mcp-postgres-analyst
python3 -m venv .venv && . .venv/bin/activate
pip install -e ".[dev]"
python scripts/seed_demo.py              # -> data/demo.db (300 customers, 2,000 orders)
python examples/client.py                # spawns the server over stdio and calls a few tools
mcp-pg-analyst serve --transport http    # or serve over Streamable HTTP on :8765/mcp

Docker compose, PostgreSQL

cp .env.example .env                     # optional: set MCP_HTTP_TOKEN
docker compose up --build                # postgres + seed job + server on :8765
python examples/client.py --url http://127.0.0.1:8765/mcp --token "$MCP_HTTP_TOKEN"

The server connects as analyst_ro, a login role with default_transaction_read_only = on and SELECT grants only.

Dry-run the guard

$ mcp-pg-analyst check "WITH x AS (SELECT email AS contact FROM customers) SELECT upper(contact) AS shout FROM x LIMIT 5000"
ALLOWED
WITH x AS (
  SELECT
    email AS contact
  FROM customers
)
SELECT
  UPPER(contact) AS shout
FROM x
LIMIT 101
-- tables: customers; limit: 100 (rewritten; one extra row is fetched to detect truncation)
-- masked derived columns: contact (email), shout (email)

$ mcp-pg-analyst check "WITH gone AS (DELETE FROM orders RETURNING *) SELECT count(*) FROM gone"
REJECTED forbidden_statement: DELETE is not allowed

Connecting clients

All stdio clients launch the same command. Use absolute paths; the server reads its settings from env.

Claude Desktop (claude_desktop_config.json) and Cursor (.cursor/mcp.json):

{
  "mcpServers": {
    "shop-analyst": {
      "command": "/home/you/mcp-postgres-analyst/.venv/bin/mcp-pg-analyst",
      "args": ["serve", "--transport", "stdio"],
      "env": {
        "DATABASE_URL": "sqlite:////home/you/mcp-postgres-analyst/data/demo.db",
        "POLICY_PATH": "/home/you/mcp-postgres-analyst/config/policy.yaml",
        "AUDIT_LOG_PATH": "/home/you/mcp-postgres-analyst/logs/audit.jsonl"
      }
    }
  }
}

VS Code (.vscode/mcp.json), stdio or HTTP:

{
  "inputs": [{ "id": "token", "type": "promptString", "description": "MCP_HTTP_TOKEN", "password": true }],
  "servers": {
    "shop-analyst": {
      "type": "stdio",
      "command": "${workspaceFolder}/.venv/bin/mcp-pg-analyst",
      "args": ["serve"],
      "env": { "DATABASE_URL": "sqlite:///${workspaceFolder}/data/demo.db" }
    },
    "shop-analyst-http": {
      "type": "http",
      "url": "http://127.0.0.1:8765/mcp",
      "headers": { "Authorization": "Bearer ${input:token}" }
    }
  }
}

HTTP mode for remote clients: MCP_HOST=0.0.0.0 MCP_HTTP_TOKEN=... mcp-pg-analyst serve --transport http. Cursor takes {"url": "...", "headers": {...}} in place of command; any SDK client can connect as in examples/client.py. For a public hostname, also set MCP_ALLOWED_HOSTS so DNS-rebinding protection accepts it.

MCP Inspector: make run-http in one terminal, npx -y @modelcontextprotocol/inspector in another, then connect with transport "Streamable HTTP" to http://127.0.0.1:8765/mcp.

API

Tools

Failures come back as tool errors with a stable code prefix, e.g. not_select: ..., blocked_function: ..., timeout: ..., rate_limited: ..., unknown_table: ....

Example run_query result on the demo data (delivered revenue by customer city):

{
  "sql": "SELECT c.city, COUNT(*) AS orders, ROUND(SUM(o.total_bdt), 2) AS revenue_bdt FROM orders AS o JOIN customers AS c ON c.id = o.customer_id WHERE o.status = 'delivered' GROUP BY c.city ORDER BY revenue_bdt DESC LIMIT 101",
  "columns": ["city", "orders", "revenue_bdt"],
  "rows": [["Dhaka", 482, 4112160.8], ["Chattogram", 188, 1624927.2], ["Sylhet", 109, 845776.5], ["Khulna", 100, 796680.5]],
  "row_count": 7, "truncated": false, "limit": 100, "limit_rewritten": true, "masked_columns": []
}

and with personal data involved:

{
  "sql": "SELECT full_name, email, phone FROM customers WHERE city = 'Sylhet' ORDER BY id LIMIT 4",
  "rows": [["Rafiq Uddin", "r***@outlook.com", "+**********654"], ["Nusrat Rahman", "n***@gmail.com", "+**********103"]],
  "truncated": true, "truncated_reason": "row_limit", "masked_columns": ["email", "phone"]
}

Resources and prompts

Audit log

One JSON object per line in AUDIT_LOG_PATH; mcp-pg-analyst audit -n 20 prints the tail.

{"ts": "2026-10-24T18:07:11.722+00:00", "tool": "sample_rows", "status": "ok", "duration_ms": 2.15, "args": {"table": "customers", "n": 3}, "sql": "SELECT * FROM \"main\".\"customers\" LIMIT 4", "rows": 3, "masked": ["email", "phone", "nid", "address_line"]}
{"ts": "2026-10-24T18:07:11.743+00:00", "tool": "run_query", "status": "rejected", "duration_ms": 0.61, "args": {"sql": "DELETE FROM orders", "limit": null}, "code": "not_select", "reason": "only SELECT / WITH queries are allowed, got DELETE"}

Configuration

The policy file is documented inline in config/policy.yaml.

Project Structure

mcp-postgres-analyst/
├── src/pg_analyst/
│   ├── server.py        # MCPServer: tools, resources, prompts, /health
│   ├── http.py          # Streamable HTTP app, bearer auth, DNS-rebinding settings
│   ├── service.py       # rate limit -> guard -> execute -> mask -> audit
│   ├── guard.py         # sqlglot validation, LIMIT rewrite, lineage for masking
│   ├── masking.py       # email / phone / last4 / hash / redact
│   ├── policy.py        # YAML policy models
│   ├── db.py            # SQLite and PostgreSQL adapters (read-only, timeouts, introspection)
│   ├── ratelimit.py     # token buckets per tool
│   ├── audit.py         # JSONL audit log
│   ├── demo.py          # deterministic demo shop data
│   └── cli.py           # serve | check | seed | audit
├── config/policy.yaml
├── scripts/seed_demo.py
├── examples/client.py   # SDK client over stdio or HTTP
├── docker/initdb/       # read-only role for PostgreSQL
├── tests/               # guard, masking, policy, service, MCP protocol, transports, PostgreSQL
├── Dockerfile · docker-compose.yml · render.yaml · .devcontainer/
└── Makefile · pyproject.toml

Testing

make test        # 164 tests, no database server needed (~9 s)
make test-pg     # 4 more against PostgreSQL: docker compose up -d postgres first
make lint

Coverage by area:

• Guard (100+ cases across both dialects): stacked statements, DML/DDL, data-modifying CTEs, SELECT INTO, FOR UPDATE, COPY/SET/PRAGMA/ATTACH, comment and dollar-quote smuggling, blocked functions in every position (schema-qualified, quoted, upper-case, nested in CTEs and sub-queries), system and policy tables, blocked columns, whole-row references, LIMIT injection/clamping/LIMIT ALL/FETCH FIRST/unions/offsets, lineage through aliases and CTEs

• Masking and policy: every strategy, glob rules, taint-based masking, rate-limit parsing, token-bucket refill with a fake clock

• Service: masking of direct and derived columns, SELECT * hiding, row and byte caps, timeout cancellation, audit entries for ok/rejected/rate-limited calls, introspection, saved queries

• Protocol: tools, schemas and annotations, structured content, tool errors, resources, templates and prompts through the SDK's in-process Client

• Transports: Streamable HTTP through uvicorn on a free port (health, 401s, authenticated round trip) and stdio through a spawned subprocess

• PostgreSQL (CI service container): catalog comments, FK introspection, saved queries, statement_timeout, read-only transaction rejects DDL

Roadmap

• Per-session rate limits and audit attribution in HTTP mode (session id / OAuth subject instead of one bucket per tool)

• OAuth 2.1 resource-server mode using the SDK's TokenVerifier, so HTTP deployments can drop the shared token

• Optional result sampling (TABLESAMPLE) for sample_rows on very large tables

• MySQL adapter (sqlglot already speaks the dialect; the adapter needs read-only session handling)

• Policy hot-reload on file change

License

MIT, see LICENSE.