How to Build a Secure Database MCP Server for PostgreSQL and SQLite in Python

Enforcing Read-Only URI Modes Before Exposing SQL to Agents

Give an AI coding agent a raw PostgreSQL connection string with write privileges, and sooner or later it will hallucinate a destructive migration. Application-level regex filters will not save you. They fail on CTEs. You have to lock the connection down at the driver and transaction level.

Install FastMCP and Psycopg 3 to build a read-only database inspector compliant with the Model Context Protocol specification:

pip install fastmcp psycopg[binary]

Save this script as db_mcp.py, which enforces immutable SQLite URI read-only mode at the driver level:

import sqlite3
from fastmcp import FastMCP

mcp = FastMCP("Database-Inspector")

@mcp.tool()
def query_sqlite(db_path: str, sql_query: str) -> list[dict]:
    """Execute a read-only SQL query against a local SQLite database."""
    # Enforce immutable read-only mode at the connection level
    conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True)
    conn.row_factory = sqlite3.Row
    cursor = conn.cursor()
    try:
        cursor.execute(sql_query)
        rows = cursor.fetchmany(100) # Hard limit on returned rows
        return [dict(row) for row in rows]
    finally:
        conn.close()

if __name__ == "__main__":
    mcp.run()

Security Architecture: Agent Database Access Matrix

Never give language models raw write access or administrative credentials. Always enforce safety constraints at the driver and network layers rather than relying on prompt instructions:

Step 1: PostgreSQL Implementation with Connection Pooling

For PostgreSQL, use a connection pool and execute queries inside a read-only transaction block with an enforced statement timeout.

import psycopg
from psycopg.rows import dict_row
from fastmcp import FastMCP

mcp = FastMCP("Postgres-Inspector")

PG_CONN_STRING = "postgresql://readonly_user:secret_pass@localhost:5432/production_replica"

@mcp.tool()
def describe_table_schema(table_name: str) -> list[dict]:
    """Inspect column names, data types, and nullability for a PostgreSQL table."""
    query = """
        SELECT column_name, data_type, is_nullable
        FROM information_schema.columns
        WHERE table_name = %s
        ORDER BY ordinal_position;
    """
    with psycopg.connect(PG_CONN_STRING, row_factory=dict_row) as conn:
        with conn.cursor() as cur:
            cur.execute(query, (table_name,))
            return cur.fetchall()

@mcp.tool()
def execute_safe_select(sql_query: str) -> list[dict]:
    """Execute a SELECT query with strict timeout and read-only enforcement."""
    with psycopg.connect(PG_CONN_STRING, row_factory=dict_row) as conn:
        # Enforce read-only at session level
        conn.read_only = True
        with conn.cursor() as cur:
            # Set 3 second statement timeout to stop runaway queries
            cur.execute("SET statement_timeout = 3000;")
            cur.execute(sql_query)
            return cur.fetchmany(50)

if __name__ == "__main__":
    mcp.run()

Step 2: Database User Provisioning in PostgreSQL

Execute these commands in PostgreSQL (psql) to create a dedicated user with minimal permissions:

-- 1. Create dedicated read-only role
CREATE ROLE ai_reader WITH LOGIN PASSWORD 'strong_random_password';

-- 2. Grant connection rights to target database
GRANT CONNECT ON DATABASE analytics_db TO ai_reader;

-- 3. Grant schema usage and select privileges
GRANT USAGE ON SCHEMA public TO ai_reader;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO ai_reader;

-- 4. Ensure future tables are automatically readable
ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT SELECT ON TABLES TO ai_reader;

-- 5. Revoke write privileges explicitly
REVOKE CREATE ON SCHEMA public FROM ai_reader;

Step 3: Registering the Server in Claude Code, Cursor, and Antigravity

Add the database server configuration to your AI client’s MCP configuration file.

For Claude Desktop / Claude Code (claude_desktop_config.json):

{
  "mcpServers": {
    "postgres-inspector": {
      "command": "python3",
      "args": ["/absolute/path/to/db_mcp.py"],
      "env": {
        "PG_CONN_STRING": "postgresql://ai_reader:[email protected]:5432/analytics_db"
      }
    }
  }
}

For Cursor (.cursor/mcp.json):

{
  "mcpServers": {
    "db-tools": {
      "command": "python3",
      "args": ["/absolute/path/to/db_mcp.py"]
    }
  }
}

The agent will execute describe_table_schema("orders"), analyze the foreign keys, formulate an optimized SQL query, and return structured tabular insights without manual copy-pasting.

Field Notes: 3 Production Bugs That Crash Database MCP Servers

Writing a 30-line MCP wrapper is easy. Keeping it alive during a 40-turn coding session with Claude Code or Cursor is where subtle driver behaviors bite you:

1. TypeError: Object of type Decimal / datetime is not JSON serializable: PostgreSQL NUMERIC and TIMESTAMPTZ columns deserialize into Python Decimal and datetime.datetime objects in psycopg. When FastMCP tries to serialize the tool return dictionary over JSON-RPC, the server crashes unless values are cast. Either pass default=str in your serializer or normalize values ({k: str(v) if isinstance(v, (Decimal, datetime)) else v for k, v in row.items()}) before returning rows.
2. SQLite database is locked When Querying an Active Dev Database: If your local web application writes to an SQLite file in Write-Ahead Logging (WAL) mode while your MCP server opens it with ?mode=ro, SQLite still needs to read the -shm shared-memory index file. Ensure the Linux user running your MCP subprocess has read permissions on the parent directory containing .db-wal and .db-shm, or use ?mode=ro&nolock=1 when inspecting static snapshots.
3. Logging to stdout Breaks the JSON-RPC Pipe: Never put print("Connected to Postgres") or default SQLAlchemy echo logs in an MCP server running over stdio. Any non-JSON character written to sys.stdout corrupts the MCP protocol frame and triggers an immediate Connection closed error in the client (see our MCP stdio debugging guide). Route all diagnostic logging to sys.stderr.