Enterprise PostgreSQL MCP Server: Secure Natural-Language Data Exploration for Claude Desktop

[ Claude Desktop / Enterprise AI Clients ]
                    │
                    ▼  (Encrypted Server-Sent Events [SSE] + Bearer JWT over TLS 1.3)
[ AWS Application Load Balancer / NGINX Reverse Proxy ]
                    │
                    ▼  (Reverse Proxy Buffering Disabled: X-Accel-Buffering: no)
[ Custom FastMCP Gateway (Containerized Microservice on AWS ECS) ]
                    │
    ┌───────────────┴────────────────┐
    ▼                                ▼
[ Dynamic Schema Whitelister ]    [ SQL AST Guardrail & Parser (sqlglot) ]
(Filters internal tables & PII)   (Blocks DDL/DML mutations & enforces LIMIT)
                    │                                │
                    └────────────────┬───────────────┘
                                     ▼
                  [ PostgreSQL Read-Replica (Pooled asyncpg) ]
                  (Forced 'statement_timeout = 2500ms' & Read-Only Role)
                                     │
                                     ▼
                      [ Dynamic PII Sanitization Layer ]
                      (Regex masking of emails, phones, and IDs)
                                     │
                                     ▼
                [ Formatted JSON-RPC Tool Result Stream ]

"""
Defensive SQL AST Guardrail: AST Validation & Read-Only Enforcement
Inspects generated SQL queries using an Abstract Syntax Tree (AST) visitor
to reject mutation statements, block DDL, enforce LIMIT clauses, and sandbox tables.
"""
import sqlglot
from sqlglot import exp
from typing import Set, Tuple, Optional

class SQLSecurityException(Exception):
    """Raised when an SQL query violates enterprise security constraints."""
    pass

class EnterpriseSQLGuardrail:
    ALLOWED_STATEMENT_TYPES = (exp.Select,)
    DISALLOWED_EXPRESSIONS = (
        exp.Insert, exp.Update, exp.Delete, exp.Drop,
        exp.AlterTable, exp.TruncateTable, exp.Create,
        exp.Command, exp.Transaction
    )
    
    WHITELISTED_TABLES: Set[str] = {
        "v_revenue_quarterly_cohorts",
        "v_merchant_aggregate_volumes",
        "v_churn_retention_metrics",
        "v_customer_subscription_tiers",
        "corporate_general_ledger_summary"
    }

    RESTRICTED_COLUMNS: Set[str] = {
        "password_hash", "ssn_tax_id", "credit_card_token",
        "raw_bank_account_num", "stripe_secret_key"
    }

    @classmethod
    def validate_and_sanitize(cls, raw_sql: str, max_limit: int = 100) -> str:
        """
        Parses SQL into an AST, validates statement types and table access,
        and injects hard LIMIT boundaries.
        """
        try:
            parsed = sqlglot.parse_one(raw_sql, read="postgres")
        except Exception as e:
            raise SQLSecurityException(f"Syntax validation failed: {str(e)}")

        # 1. Reject any non-SELECT statements
        if not isinstance(parsed, cls.ALLOWED_STATEMENT_TYPES):
            raise SQLSecurityException(
                f"Unauthorized SQL statement type: {parsed.key.upper()}. Only read-only SELECT queries are permitted."
            )

        # 2. Check for nested destructive statements inside CTEs or unions
        for node in parsed.walk():
            if isinstance(node, cls.DISALLOWED_EXPRESSIONS):
                raise SQLSecurityException(f"Destructive expression detected: {node.key.upper()}. Execution aborted.")
            
            # 3. Whitelist table verification
            if isinstance(node, exp.Table):
                table_name = node.name.lower()
                if table_name not in cls.WHITELISTED_TABLES:
                    raise SQLSecurityException(
                        f"Access to table '{table_name}' is forbidden. Queries must target approved reporting views."
                    )
            
            # 4. Blacklist column access
            if isinstance(node, exp.Column):
                col_name = node.name.lower()
                if col_name in cls.RESTRICTED_COLUMNS:
                    raise SQLSecurityException(
                        f"Access to restricted column '{col_name}' is blocked by enterprise governance policy."
                    )

        # 5. Enforce hard LIMIT if absent or excessive
        limit_node = parsed.args.get("limit")
        if limit_node is None:
            parsed = parsed.limit(max_limit)
        else:
            try:
                current_limit = int(limit_node.expression.this)
                if current_limit > max_limit or current_limit < 1:
                    parsed = parsed.limit(max_limit)
            except (ValueError, AttributeError):
                parsed = parsed.limit(max_limit)

        return parsed.sql(dialect="postgres")

"""
Serialization Layer with Dynamic PII Masking
Redacts sensitive customer identifiers, emails, telephone numbers,
and IP addresses before tool results enter the LLM context window.
"""
import re
from typing import Any, Dict, List

EMAIL_REGEX = re.compile(r"([a-zA-Z0-9_.+-]+)@([a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+)")
PHONE_REGEX = re.compile(r"(\+?\d{1,3}[-.\s]?)?\(?\d{3}\)?[-.\s]?\d{3}[-.\s]?\d{4}")
IPV4_REGEX = re.compile(r"\b(?:\d{1,3}\.){3}\d{1,3}\b")

def mask_email(match: re.Match) -> str:
    user, domain = match.group(1), match.group(2)
    masked_user = user[0] + "***" if len(user) > 1 else "*"
    return f"{masked_user}@{domain}"

def mask_string(val: str) -> str:
    val = EMAIL_REGEX.sub(mask_email, val)
    val = PHONE_REGEX.sub("[PHONE_REDACTED]", val)
    val = IPV4_REGEX.sub("[IP_MASKED]", val)
    return val

def sanitize_record(record: Dict[str, Any]) -> Dict[str, Any]:
    """Recursively mask PII in dictionary payloads."""
    sanitized = {}
    for key, value in record.items():
        if isinstance(value, str):
            sanitized[key] = mask_string(value)
        elif isinstance(value, dict):
            sanitized[key] = sanitize_record(value)
        elif isinstance(value, list):
            sanitized[key] = [
                sanitize_record(item) if isinstance(item, dict) else (mask_string(item) if isinstance(item, str) else item)
                for item in value
            ]
        else:
            sanitized[key] = value
    return sanitized

"""
Production FastMCP Enterprise PostgreSQL Analytics Server
Features SSE transport, asyncpg pooling, SQL AST security validation,
statement timeouts (2500ms), and automated PII redaction.
"""
import os
import sys
import logging
from typing import Dict, Any, List
import asyncpg
from mcp.server.fastmcp import FastMCP
from ast_guardrail import EnterpriseSQLGuardrail, SQLSecurityException
from pii_sanitizer import sanitize_record

logging.basicConfig(stream=sys.stderr, level=logging.INFO)
logger = logging.getLogger("mcp_postgres_gateway")

mcp = FastMCP(
    name="enterprise-fintech-analytics",
    instructions="Production MCP server exposing read-only financial views to Claude Desktop."
)

class DatabasePool:
    _pool: asyncpg.Pool = None

    @classmethod
    async def get_pool(cls) -> asyncpg.Pool:
        if cls._pool is None:
            dsn = os.environ.get("DATABASE_READ_REPLICA_URL")
            cls._pool = await asyncpg.create_pool(
                dsn=dsn,
                min_size=4,
                max_size=20,
                command_timeout=3.0,
                server_settings={
                    "application_name": "mcp_claude_analytics",
                    "statement_timeout": "2500ms",  # Hard 2.5s timeout
                    "default_transaction_read_only": "on"  # Engine-level read-only
                }
            )
            logger.info("Connected to PostgreSQL read-replica pool successfully.")
        return cls._pool

@mcp.tool()
async def execute_analytical_query(query: str) -> Dict[str, Any]:
    """
    Execute a read-only SQL query against approved enterprise financial reporting views.
    
    All queries are inspected by an AST parser to ensure no mutation occurs.
    Tables available:
      - v_revenue_quarterly_cohorts (signup_quarter, arr, nrr, customer_tier)
      - v_merchant_aggregate_volumes (merchant_id, volume_usd, chargeback_rate)
      - v_churn_retention_metrics (fiscal_year, churn_percentage, expansion_arr)
    """
    # 1. AST Validation & Forced Limit Injection
    try:
        safe_sql = EnterpriseSQLGuardrail.validate_and_sanitize(query, max_limit=100)
    except SQLSecurityException as sec_err:
        logger.warning(f"AST Guardrail Rejected Query: {sec_err}")
        return {
            "status": "error",
            "error_code": "SQL_GOVERNANCE_VIOLATION",
            "message": str(sec_err),
            "remedy": "Ensure your query is a read-only SELECT targeting approved 'v_*' reporting views."
        }

    # 2. Pooled Query Execution
    try:
        pool = await DatabasePool.get_pool()
        async with pool.acquire() as conn:
            records = await conn.fetch(safe_sql)
            raw_dicts = [dict(r) for r in records]
            
            # 3. Dynamic PII Masking
            sanitized_data = [sanitize_record(row) for row in raw_dicts]

            return {
                "status": "success",
                "row_count": len(sanitized_data),
                "data": sanitized_data
            }

    except asyncpg.QueryCanceledError:
        return {
            "status": "error",
            "error_code": "STATEMENT_TIMEOUT",
            "message": "Query exceeded the 2500ms execution timeout.",
            "remedy": "Apply tighter date ranges or filter by specific merchant/customer tiers."
        }
    except Exception as exc:
        logger.error(f"Database error during execution: {exc}")
        return {"status": "error", "message": f"Query execution failed: {str(exc)}"}

if __name__ == "__main__":
    mcp.run(transport="sse", host="0.0.0.0", port=8000)

{
  "mcpServers": {
    "fintech-postgres-analytics": {
      "url": "https://mcp-analytics.internal.fintech.company/sse",
      "headers": {
        "Authorization": "Bearer prd_sec_jwt_auth_token_8891"
      }
    }
  }
}