┌────────────────────────────────────────────────────────────────────────┐
│                          MCP HOST APPLICATION                          │
│                                                                        │
│   ┌────────────────────┐                   ┌───────────────────────┐   │
│   │  LLM Orchestrator  │ ◄───────────────► │   MCP Client Bridge   │   │
│   │   (LangGraph/Agent)│                   │  (Session Management) │   │
│   └────────────────────┘                   └───────────┬───────────┘   │
└────────────────────────────────────────────────────────┼───────────────┘
                                                         │
                             JSON-RPC 2.0 Protocol       │
                        (Transport: STDIO / HTTP-SSE)    │
                                                         ▼
┌────────────────────────────────────────────────────────────────────────┐
│                        ENTERPRISE MCP SERVER                           │
│                                                                        │
│   ┌────────────────────────────────────────────────────────────────┐   │
│   │                     Zero-Trust Security Gate                   │   │
│   │   [ Capability Tokens ]  [ RBAC Guard ]  [ Rate Limiter ]     │   │
│   └────────────────────────────────┬───────────────────────────────┘   │
│                                    ▼                                   │
│   ┌────────────────────────────────────────────────────────────────┐   │
│   │                     MCP Primitive Handlers                     │   │
│   │                                                                │   │
│   │   • Resources: URI-based context (postgres://schema/tables)    │   │
│   │   • Tools: Executable actions with strict Pydantic schemas    │   │
│   │   • Prompts: Pre-engineered domain prompt templates            │   │
│   └────────────────────────────────┬───────────────────────────────┘   │
│                                    ▼                                   │
│   ┌────────────────────────────────────────────────────────────────┐   │
│   │               Database Connection Pool (PgBouncer)             │   │
│   └────────────────────────────────┬───────────────────────────────┘   │
└────────────────────────────────────┼───────────────────────────────────┘
                                     ▼
                      [ Enterprise PostgreSQL Cluster ]

import os
import re
import json
import logging
from typing import Any
import asyncpg
from mcp.server.fastmcp import FastMCP
from pydantic import BaseModel, Field, field_validator

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("mcp-postgres-server")

# 1. Initialize FastMCP Server
mcp = FastMCP(
    name="Enterprise PostgreSQL Gateway",
    dependencies=["asyncpg", "pydantic>=2.0"]
)

# 2. Database Connection Pool State
DB_POOL: asyncpg.Pool | None = None

async def get_db_pool() -> asyncpg.Pool:
    global DB_POOL
    if DB_POOL is None:
        DB_POOL = await asyncpg.create_pool(
            dsn=os.getenv("DATABASE_URL"),
            min_size=5,
            max_size=20,
            command_timeout=5.0
        )
    return DB_POOL

# 3. Input Validation Schemas
class TableSchemaQuery(BaseModel):
    table_name: str = Field(..., description="The name of the database table to inspect.")

    @field_validator("table_name")
    def validate_table_name(cls, v: str) -> str:
        if not re.match(r"^[a-zA-Z0-9_]+$", v):
            raise ValueError("Invalid table name. Only alphanumeric and underscore allowed.")
        return v

class ParameterizedQuery(BaseModel):
    query: str = Field(..., description="SELECT query with parameter placeholders ($1, $2, etc.).")
    parameters: list[Any] = Field(default_factory=list, description="Positional query parameter values.")

    @field_validator("query")
    def validate_select_only(cls, v: str) -> str:
        clean = v.strip().upper()
        if not clean.startswith("SELECT") and not clean.startswith("WITH"):
            raise ValueError("Only read-only SELECT and WITH statements are permitted.")
        
        # Banned destructive keywords
        forbidden = ["DROP", "DELETE", "UPDATE", "INSERT", "ALTER", "TRUNCATE", "GRANT", "REVOKE"]
        for word in forbidden:
            if re.search(r"\b" + word + r"\b", clean):
                raise ValueError(f"Forbidden SQL keyword detected: {word}")
        return v

# 4. Expose MCP Resources
@mcp.resource("postgres://schema/{table_name}")
async def get_table_schema_resource(table_name: str) -> str:
    """Provides passive context of the table's column definitions and data types."""
    pool = await get_db_pool()
    query = """
        SELECT column_name, data_type, is_nullable
        FROM information_schema.columns
        WHERE table_name = $1
        ORDER BY ordinal_position;
    """
    async with pool.acquire() as conn:
        rows = await conn.fetch(query, table_name)
        if not rows:
            return f"Table '{table_name}' does not exist or has no columns."
        
        schema_info = [dict(row) for row in rows]
        return json.dumps(schema_info, indent=2)

# 5. Expose Secure MCP Tools
@mcp.tool()
async def describe_database_tables() -> str:
    """Lists all accessible tables within the public schema."""
    pool = await get_db_pool()
    query = """
        SELECT table_name 
        FROM information_schema.tables 
        WHERE table_schema = 'public' 
        ORDER BY table_name;
    """
    async with pool.acquire() as conn:
        rows = await conn.fetch(query)
        tables = [row["table_name"] for row in rows]
        return json.dumps({"tables": tables})

@mcp.tool()
async def execute_safe_query(params: ParameterizedQuery) -> str:
    """Executes a sanitized, read-only parameterized query against PostgreSQL."""
    pool = await get_db_pool()
    logger.info(f"Executing query: {params.query} with params: {params.parameters}")
    
    async with pool.acquire() as conn:
        # Enforce transaction timeout and read-only transaction state
        async with conn.transaction(readonly=True):
            await conn.execute("SET LOCAL statement_timeout = '3000ms';")
            
            try:
                rows = await conn.fetch(params.query, *params.parameters)
                results = [dict(row) for row in rows[:100]] # Hard limit 100 rows
                return json.dumps({
                    "row_count": len(results),
                    "truncated": len(rows) > 100,
                    "data": results
                }, default=str)
            except Exception as e:
                logger.error(f"Database query error: {str(e)}")
                return json.dumps({"error": str(e)})

if __name__ == "__main__":
    # Start the server over STDIO transport
    mcp.run(transport="stdio")

┌────────────────────────────────────────────────────────────────────────┐
│               ENTERPRISE INTEGRATION OVERHEAD COMPARISON               │
│                                                                        │
│   [ Traditional Custom APIs ]         [ Model Context Protocol ]       │
│                                                                        │
│     Agent A ──► Custom Shim 1 ──► DB     Agent A ──┐                   │
│     Agent B ──► Custom Shim 2 ──► DB     Agent B ────► [ MCP Server ] ──► DB
│     Agent C ──► Custom Shim 3 ──► DB     Agent C ──┘                   │
│                                                                        │
│   * Result: O(N * M) Connectors        * Result: O(N + M) Open Spec    │
└────────────────────────────────────────────────────────────────────────┘