How to Give Claude & Cursor Safe Database Access Without Crashing Production (Zero-Trust MCP Guard in Python)

How to Give Claude & Cursor Safe Database Access Without Crashing Production (Zero-Trust MCP Guard in Python)

Giving an AI assistant or autonomous agent access to a live database is one of the most powerful workflows for modern software teams: Claude Desktop, Cursor, or Windsurf can write complex analytics queries, optimize slow joins, and inspect schemas in seconds.

However, granting raw database credentials to an LLM is dangerous:

  • Accidental Disasters: One misaligned prompt or prompt injection can lead the agent to execute DROP TABLE, TRUNCATE, or unconstrained UPDATE / DELETE queries.
  • Production Freezes: A query missing a WHERE clause or missing an index on a million-row table can trigger a full sequential table scan, consuming 100% CPU and locking transactions.
  • PII Exposure: Raw customer records (plaintext emails, password hashes, credit cards) get pulled into prompt contexts and third-party LLM server logs.

The modern, standardized solution is to place a Zero-Trust Model Context Protocol (MCP) guard between your agent and the database engine.

In this guide, we will walk through how PgSafe MCP was engineered in Python to provide secure, read-only database exploration, forced LIMIT injection, real-time PII redaction, and dry-run mutation simulation.

The Zero-Trust MCP Architecture

Rather than connecting Claude directly to Postgres or SQLite with an unrestricted superuser connection, the MCP server acts as an intelligent firewall:

[ AI Agent: Claude / Cursor / Antigravity ]
                      │  JSON-RPC 2.0 (stdio)
                      ▼
             [ PgSafe MCP Server ]
                      │
           [ ReadOnlyGuard (AST Analysis) ]
           ├─ Blocks DROP, TRUNCATE, DELETE, ALTER
           └─ Blocks Semicolon Chaining
                      │
           [ CostAnalyzer (LIMIT Guard) ]
           ├─ Forces max 1,000 row limits
           └─ Warns on missing WHERE clauses
                      │
            [ Database Connection ]
            (PostgreSQL / SQLite / MySQL)
                      │
             [ Real-Time PII Masker ]
           ├─ Redacts emails (u***@domain.com)
           ├─ Masks credit cards (****-****-****-1234)
           └─ Redacts password hashes
                      │
                      ▼
[ Clean, Safe JSON-RPC Output to Agent ]

1. Enforcing Read-Only SQL Lexical Validation

A secure read-only guard must inspect every statement before it ever reaches the database cursor, stripping comments and blocking destructive keywords:

import re

BLOCKED_KEYWORDS = [
    r"\bDROP\b", r"\bTRUNCATE\b", r"\bDELETE\b",
    r"\bUPDATE\b", r"\bINSERT\b", r"\bALTER\b",
    r"\bCREATE\b", r"\bGRANT\b", r"\bREVOKE\b"
]

class ReadOnlyGuard:
    @staticmethod
    def validate(sql: str) -> tuple[bool, str]:
        # Disallow multi-statement injection (semicolon chaining)
        statements = [s.strip() for s in sql.split(";") if s.strip()]
        if len(statements) > 1:
            return False, "Semicolon chaining disallowed."

        # Verify query begins with read-only keyword
        first_word = statements[0].split()[0].upper()
        if first_word not in {"SELECT", "EXPLAIN", "WITH", "PRAGMA", "SHOW"}:
            return False, f"Statement '{first_word}' is forbidden."

        for pattern in BLOCKED_KEYWORDS:
            if re.search(pattern, sql, re.IGNORECASE):
                return False, "Destructive SQL operation detected."

        return True, "Query verified safe."

2. Dynamic LIMIT Injection

Runaway SELECT * FROM audit_logs queries can crash your client's memory. PgSafe intercepts every query and caps row limits:

import re

def enforce_limit(sql: str, default_limit: int = 100, max_limit: int = 1000) -> str:
    cleaned = sql.strip().rstrip(";")
    match = re.search(r"\bLIMIT\s+(\d+)\b", cleaned, re.IGNORECASE)
    if match:
        limit_val = int(match.group(1))
        if limit_val > max_limit:
            return re.sub(r"\bLIMIT\s+\d+\b", f"LIMIT {max_limit}", cleaned, flags=re.IGNORECASE)
        return cleaned
    return f"{cleaned} LIMIT {default_limit}"

3. Real-Time PII Redaction Before Agent Ingestion

When the query returns data, sensitive columns (password_hash, ssn, token) and customer PII are redacted on the fly:

import re

EMAIL_REGEX = re.compile(r"\b([A-Za-z0-9._%+-])[A-Za-z0-9._%+-]*@([A-Za-z0-9.-]+\.[A-Z|a-z]{2,})\b")
CARD_REGEX = re.compile(r"\b(?:\d{4}[-\s]?){3}(\d{4})\b")

def redact_pii(text: str) -> str:
    text = CARD_REGEX.sub(r"****-****-****-\1", text)
    text = EMAIL_REGEX.sub(r"\1***@\2", text)
    return text

4. Configuring Claude Desktop & Cursor

Add the server to your claude_desktop_config.json:

{
  "mcpServers": {
    "pgsafe": {
      "command": "python",
      "args": ["-m", "pgsafe.server"],
      "cwd": "/path/to/pgsafe-mcp",
      "env": {
        "DATABASE_URL": "postgresql://user:pass@localhost:5432/my_db"
      }
    }
  }
}

Now you can prompt Claude:

"Inspect the orders table, find the top 5 customers with pending invoices, and suggest indexes to speed up this query."

Enterprise Edition: Dry-Run Mutation Simulation Harness

When agents need to draft data mutations or migrations, the Enterprise Pro edition includes an isolated simulation harness:

  • Executes INSERT, UPDATE, and DELETE queries within an isolated transaction.
  • Returns row diffs, affected count, and execution time.
  • Guarantees auto-ROLLBACK: changes are never committed to your database.
  • Includes automatic index recommendation advisors and Docker Compose setups.

Downloads and Packages:

  • Lite Version (Open Source): Read-only guard and SQLite/Postgres schema inspection.
  • Enterprise Pro Edition ($14.99): PII masking, dry-run mutation simulation, index advisor, Docker Compose, and automated test suite.

👉 Get PgSafe Pro on Gumroad ($14.99)

Story originally reported by Dev.to. View at Dev.to →
← Back to all news