warehouse-mcp

An MCP server that lets AI agents query a SQL database without being able to change it, stall it, or leak personal data from it.

When a team puts an agent in front of its real data, the first question is "what stops it from running DELETE, or dumping every customer's email?". This server answers that with layered controls you can explain to a security reviewer, plus an audit log they can read.

agent ──MCP──▶ list_tables / describe_table / run_query
                         │
          ┌──────────────▼──────────────┐
          │ 1. SQL guard (parsed AST)   │  one SELECT only; no writes, locks, INTO,
          │                             │  side-effect functions, system catalogs
          │ 2. Row cap                  │  wraps the query: LIMIT max_rows + 1
          ├─────────────────────────────┤
          │ 3. Read-only transaction    │  enforced by the database itself
          │ 4. Statement timeout        │  stopped by the database itself
          ├─────────────────────────────┤
          │ 5. PII masking              │  by column, through aliases, and by value
          │ 6. Audit log (JSONL)        │  every call; never the result data
          └─────────────────────────────┘

Why each layer exists

Rejections come back as tool errors that say why (for example Query rejected: DELETE is not allowed in a read-only query.), so the agent can fix its query instead of retrying the same one.

Related MCP server: Governor SQL MCP

Quickstart

pip install -e ".[postgres]"          # drop [postgres] for SQLite only
python examples/seed_demo.py          # demo e-commerce DB with fake PII
WAREHOUSE_URL=sqlite:///examples/demo.db warehouse-mcp

Connect it to an MCP client

Claude Desktop, Cursor, VS Code and other MCP clients use the same shape:

{
  "mcpServers": {
    "warehouse": {
      "command": "warehouse-mcp",
      "env": {
        "WAREHOUSE_URL": "postgresql+psycopg://readonly_agent@db.internal:5432/analytics",
        "WAREHOUSE_ALLOWED_SCHEMAS": "analytics,reporting",
        "WAREHOUSE_MAX_ROWS": "200"
      }
    }
  }
}

Example

Agent: Which product categories sold the most units, delivered orders only?

// run_query
{"sql": "SELECT p.category, SUM(o.quantity) AS units FROM orders o JOIN products p ON p.id = o.product_id WHERE o.status = 'delivered' GROUP BY p.category ORDER BY units DESC"}
// → {"columns": ["category","units"], "rows": [...], "row_count": 6, "truncated": false, "masked_columns": [], "elapsed_ms": 4}

// run_query
{"sql": "SELECT id, email AS contact, region FROM customers LIMIT 3"}
// → {"columns": ["id","contact","region"], "rows": [[1,"***","EU"], ...], "masked_columns": ["contact"]}

// run_query
{"sql": "WITH d AS (DELETE FROM orders RETURNING *) SELECT * FROM d"}
// → error: Query rejected: DELETE is not allowed in a read-only query.

Configuration

In production, also connect as a role with only SELECT grants. This server is defense in depth, not a replacement for database permissions.

Tests

pip install -e ".[dev]"
pytest -q

There are 45 tests: guard attack cases, PII tracing through aliases, value-level masking, end-to-end calls through a real MCP client, and audit-log contents. Read-only and timeout behaviour are tested against the database with the guard bypassed. CI runs everything against a real PostgreSQL 16 on Python 3.10, 3.11 and 3.12.

Limitations

• PII detection is name- and pattern-based. It won't recognise a column called notes that holds home addresses, so add those columns with WAREHOUSE_PII_COLUMNS.

• Masking follows aliases and expressions but not every path. For example, string_agg over a PII column is masked, while a join that reconstructs PII from non-PII columns isn't detected.

• Snowflake, BigQuery and other SQLAlchemy dialects will connect, but only Postgres and SQLite have read-only and timeout enforcement at the database level today.

Roadmap

• Snowflake and BigQuery read-only session enforcement

• Per-table row-level filters (for example region = 'EU' for an EU-only agent)

• Query cost estimate from EXPLAIN before execution, rejecting full scans over a threshold

• OpenTelemetry spans per tool call

• Streamable HTTP transport with OAuth for shared deployments