MCP Database Server: Build a Secure SQL Query Server with FastMCP

By Fadi Updated Jun 14, 2026 12 min read MCP

TL;DR — MCP database server: FastMCP with PostgreSQL, MySQL, SQLite, MSSQL. FastMCP SQL Tools: "FastMCP server for PostgreSQL, MySQL, SQLite. Tools: list_tables, get_table_schema, execute_query, execute_safe_query. Auto-detects from DATABASE_URL." SQLAlchemy MCP: "SQLAlchemy support for 60+ databases. Read-only and read-write modes. Tools: query, list_tables, describe_table, list_schemas, get_table_ddl, explain_query. Prompts: sql_query, explain_schema, optimize_query." DB-MCP: "PostgreSQL, MySQL, SQL Server, SQLite. Read-only by default, query validation, connection pooling. 5 tools." SQL MCP Server: "Safe-by-default SQL validation. Read-only mode, single statement enforcement, forbidden keyword detection, table allowlist, row limiting. Multi-instance runtime." MCP DB Bridge: "MySQL, PostgreSQL, SQLite with granular permissions, multi-DB support, cloud-ready SSL/TLS. Adapter pattern." Learn more with MCP guide, build MCP server, MCP tools, and MCP security.

FastMCP SQL Tools provides the foundation: "A Model Context Protocol (MCP) server built with FastMCP that provides SQL database access with support for PostgreSQL, MySQL, and SQLite. Features: list_tables, get_table_schema, execute_query, execute_safe_query. The server automatically detects the database type from the DATABASE_URL environment variable."

SQLAlchemy MCP extends the scope: "An MCP server that provides SQL database access through SQLAlchemy. Built with FastMCP, it supports any database that SQLAlchemy supports and operates in read-only or read-write mode. Universal database support — SQLite, PostgreSQL, MySQL, Oracle, MSSQL, and 60+ other databases."

MCP Database Server Architecture

flowchart TD subgraph Client["MCP Client (Claude/Cursor)"] Agent["AI Agent
natural language query"] Approval["User Approval
before query execution"] end subgraph Server["MCP Database Server (FastMCP)"] Tools["MCP Tools
list_tables
describe_table
execute_query
explain_query"] Validation["SQL Validation
read-only check
single statement
forbidden keywords
table allowlist"] RowLimit["Row Limiting
DB_MAX_ROWS=100
automatic LIMIT/TOP"] end subgraph Security["Security Layer"] ReadOnly["Read-Only Mode
DB_READ_ONLY=true
SELECT only"] Timeout["Query Timeout
DB_QUERY_TIMEOUT=10s
DB_STATEMENT_TIMEOUT_MS"] Pooling["Connection Pooling
pool_size=5
max_overflow=2
pre-ping enabled"] AllowList["Table Allowlist
DB_ALLOWED_TABLES
comma-separated"] end subgraph Databases["Supported Databases"] PG["PostgreSQL
asyncpg / psycopg2
postgresql://host:port/db"] MySQL["MySQL
aiomysql / PyMySQL
mysql://host:port/db"] SQLite["SQLite
aiosqlite / built-in
sqlite:///path/to/db"] MSSQL["SQL Server
pyodbc
mssql+pyodbc://host:port/db"] Oracle["Oracle
oracledb
oracle+oracledb://host:1521/"] end Agent --> Approval Approval --> Tools Tools --> Validation Validation --> RowLimit RowLimit --> Security ReadOnly --> Validation Timeout --> Validation Pooling --> Databases AllowList --> Validation Security --> PG Security --> MySQL Security --> SQLite Security --> MSSQL Security --> Oracle

Database Support Comparison

Database	Driver	URL Format	Install Extra
PostgreSQL	asyncpg / psycopg2	`postgresql://user:pass@host:port/db`	`[postgresql]`
MySQL	aiomysql / PyMySQL	`mysql://user:pass@host:port/db`	`[mysql]`
SQLite	aiosqlite / built-in	`sqlite:///path/to/db.db`	(none)
MSSQL	pyodbc	`mssql+pyodbc://user:pass@host:port/db`	`[mssql]`
Oracle	oracledb	`oracle+oracledb://user:pass@host:1521/?service_name=svc`	`[oracle]`

MCP Database Tools

Tool	Description	Read-Only	Example
list_tables	List all tables and views	Yes	`await session.call_tool('list_tables')`
describe_table	Columns, PKs, FKs, indexes	Yes	`describe_table('users')`
execute_query	Execute SQL (SELECT in read-only)	Configurable	`execute_query('SELECT * FROM users')`
explain_query	Execution plan for SELECT	Yes	`explain_query('SELECT * FROM users')`
list_schemas	List all database schemas	Yes	`await session.call_tool('list_schemas')`
get_table_ddl	CREATE TABLE DDL	Yes	`get_table_ddl('users')`

Implementation

from dataclasses import dataclass
from typing import Optional
from enum import Enum

class DatabaseType(Enum):
    POSTGRESQL = "postgresql"
    MYSQL = "mysql"
    SQLITE = "sqlite"
    MSSQL = "mssql"
    ORACLE = "oracle"

class AccessMode(Enum):
    READONLY = "readonly"
    READWRITE = "readwrite"

@dataclass
class MCPDatabaseServerGuide:
    """MCP database server implementation guide."""

    def get_install(self) -> str:
        """Installation instructions."""
        return (
            "# Install MCP SDK and database drivers\n"
            "pip install 'mcp>=1.27,<2'\n"
            "\n"
            "# PostgreSQL\n"
            "pip install asyncpg\n"
            "\n"
            "# MySQL\n"
            "pip install aiomysql\n"
            "\n"
            "# SQLite (built-in with Python)\n"
            "pip install aiosqlite\n"
            "\n"
            "# Or use SQLAlchemy for universal support\n"
            "pip install sqlalchemy psycopg2-binary"
        )

    def get_server_code(self) -> str:
        """MCP database server with FastMCP."""
        return (
            "import os\n"
            "import asyncio\n"
            "from mcp.server.fastmcp import FastMCP\n"
            "\n"
            "mcp = FastMCP(\n"
            "    'database-server',\n"
            "    instructions=(\n"
            "        'This server provides database '\n"
            "        'access tools. Use list_tables '\n"
            "        'to discover tables, '\n"
            "        'describe_table for schema, '\n"
            "        'execute_query for SELECT queries.'\n"
            "    ),\n"
            ")\n"
            "\n"
            "DB_URL = os.environ.get('DATABASE_URL')\n"
            "READ_ONLY = os.environ.get(\n"
            "    'DB_READ_ONLY', 'true') == 'true'\n"
            "MAX_ROWS = int(os.environ.get(\n"
            "    'DB_MAX_ROWS', '100'))\n"
            "QUERY_TIMEOUT = int(os.environ.get(\n"
            "    'DB_QUERY_TIMEOUT', '10'))\n"
            "\n"
            "# Database connection pool\n"
            "import asyncpg\n"
            "pool = None\n"
            "\n"
            "async def get_pool():\n"
            "    global pool\n"
            "    if pool is None:\n"
            "        pool = await asyncpg\\\n"
            "            .create_pool(DB_URL,\n"
            "                min_size=2,\n"
            "                max_size=5)\n"
            "    return pool\n"
            "\n"
            "@mcp.tool()\n"
            "async def list_tables() -> list[dict]:\n"
            "    '''List all tables in the database.\n"
            "\n"
            "    Returns table names and row counts.\n"
            "    '''\n"
            "    p = await get_pool()\n"
            "    async with p.acquire() as conn:\n"
            "        rows = await conn.fetch(\n"
            "            '''SELECT tablename \n"
            "            FROM pg_tables \n"
            "            WHERE schemaname = \n"
            "            'public' ORDER BY tablename'''\n"
            "        )\n"
            "    return [{'name': r['tablename']}\n"
            "            for r in rows]\n"
            "\n"
            "@mcp.tool()\n"
            "async def describe_table(\n"
            "    table: str\n"
            ") -> dict:\n"
            "    '''Get schema for a table.\n"
            "\n"
            "    Args:\n"
            "        table: Table name\n"
            "    '''\n"
            "    p = await get_pool()\n"
            "    async with p.acquire() as conn:\n"
            "        cols = await conn.fetch(\n"
            "            '''SELECT column_name,\n"
            "            data_type, is_nullable,\n"
            "            column_default\n"
            "            FROM information_schema.columns\n"
            "            WHERE table_name = $1\n"
            "            ORDER BY ordinal_position''',\n"
            "            table)\n"
            "    return {\n"
            "        'table': table,\n"
            "        'columns': [\n"
            "            {'name': c['column_name'],\n"
            "             'type': c['data_type'],\n"
            "             'nullable': c[\n"
            "                'is_nullable'] == 'YES',\n"
            "             'default': c[\n"
            "                'column_default']}\n"
            "            for c in cols\n"
            "        ]\n"
            "    }\n"
            "\n"
            "@mcp.tool()\n"
            "async def execute_query(\n"
            "    query: str\n"
            ") -> list[dict]:\n"
            "    '''Execute a SQL query.\n"
            "\n"
            "    Args:\n"
            "        query: SQL query (SELECT only\n"
            "        in read-only mode)\n"
            "    '''\n"
            "    # Validate query\n"
            "    query_upper = query.upper().strip()\n"
            "    if READ_ONLY and not (\n"
            "        query_upper.startswith('SELECT')\n"
            "        or query_upper.startswith('WITH')\n"
            "        or query_upper.startswith(\n"
            "            'EXPLAIN')):\n"
            "        return [{'error':\n"
            "            'Read-only mode: '\n"
            "            'only SELECT/WITH/EXPLAIN '\n"
            "            'allowed'}]\n"
            "    \n"
            "    # Check for forbidden keywords\n"
            "    forbidden = ['DROP', 'TRUNCATE',\n"
            "                 'DELETE', 'ALTER']\n"
            "    if READ_ONLY:\n"
            "        for kw in forbidden:\n"
            "            if kw in query_upper:\n"
            "                return [{'error':\n"
            "                    f'Forbidden: {kw}'}]\n"
            "    \n"
            "    # Add row limit if not present\n"
            "    if 'LIMIT' not in query_upper:\n"
            "        query = f'{query.rstrip(\";\")} '\n"
            "                f'LIMIT {MAX_ROWS}'\n"
            "    \n"
            "    p = await get_pool()\n"
            "    async with p.acquire() as conn:\n"
            "        await conn.set_statement_timeout(\n"
            "            QUERY_TIMEOUT * 1000)\n"
            "        rows = await conn.fetch(query)\n"
            "    return [dict(r) for r in rows]\n"
            "\n"
            "@mcp.tool()\n"
            "async def explain_query(\n"
            "    query: str\n"
            ") -> list[dict]:\n"
            "    '''Get execution plan for a query.\n"
            "\n"
            "    Args:\n"
            "        query: SELECT query to explain\n"
            "    '''\n"
            "    p = await get_pool()\n"
            "    async with p.acquire() as conn:\n"
            "        rows = await conn.fetch(\n"
            "            f'EXPLAIN (FORMAT JSON) '\n"
            "            f'{query}')\n"
            "    return [dict(r) for r in rows]\n"
            "\n"
            "if __name__ == '__main__':\n"
            "    mcp.run(transport='stdio')"
        )

    def get_claude_config(self) -> str:
        """Claude Desktop configuration."""
        return (
            '{\n'
            '  "mcpServers": {\n'
            '    "database": {\n'
            '      "command": "/path/to/python",\n'
            '      "args": ["db_server.py"],\n'
            '      "env": {\n'
            '        "DATABASE_URL":\n'
            '          "postgresql://user:pass@localhost/mydb",\n'
            '        "DB_READ_ONLY": "true",\n'
            '        "DB_MAX_ROWS": "100",\n'
            '        "DB_QUERY_TIMEOUT": "10"\n'
            '      }\n'
            '    }\n'
            '  }\n'
            '}'
        )

    def get_security_config(self) -> dict:
        """Security configuration options."""
        return {
            "read_only": {
                "env": "DB_READ_ONLY",
                "default": "true",
                "description": (
                    "Only allow SELECT/WITH/EXPLAIN "
                    "queries. Recommended for production."
                ),
            },
            "max_rows": {
                "env": "DB_MAX_ROWS",
                "default": "100",
                "description": (
                    "Automatic row limiting. "
                    "Prevents large result sets "
                    "from consuming context."
                ),
            },
            "query_timeout": {
                "env": "DB_QUERY_TIMEOUT",
                "default": "10",
                "description": (
                    "Query timeout in seconds. "
                    "Prevents long-running queries."
                ),
            },
            "statement_timeout": {
                "env": "DB_STATEMENT_TIMEOUT_MS",
                "default": "DB_QUERY_TIMEOUT * 1000",
                "description": (
                    "Statement execution timeout "
                    "in milliseconds."
                ),
            },
            "allowed_tables": {
                "env": "DB_ALLOWED_TABLES",
                "default": None,
                "description": (
                    "Comma-separated table allowlist. "
                    "Only listed tables are accessible."
                ),
            },
            "allow_alter": {
                "env": "DB_ALLOW_ALTER",
                "default": "false",
                "description": (
                    "Allow ALTER statements for "
                    "schema evolution."
                ),
            },
            "allow_drop": {
                "env": "DB_ALLOW_DROP",
                "default": "false",
                "description": (
                    "Allow DROP statements. "
                    "Use with extreme caution."
                ),
            },
            "pool_config": {
                "pool_size": 5,
                "max_overflow": 2,
                "pool_timeout": 30,
                "pre_ping": True,
            },
        }

    def get_sql_validation(self) -> dict:
        """SQL validation middleware."""
        return {
            "checks": [
                "Read-only enforcement — only "
                "SELECT/WITH/EXPLAIN in read-only mode",
                "Single statement enforcement — "
                "no multiple statements separated "
                "by semicolons",
                "Forbidden keyword detection — "
                "DROP, TRUNCATE, DELETE, ALTER "
                "blocked in read-only mode",
                "Automatic row limiting — add "
                "LIMIT if not present",
                "Query timeout — statement timeout "
                "to prevent long-running queries",
                "Table allowlist — only allowed "
                "tables accessible",
            ],
            "flow": [
                "Receive SQL query from tool call",
                "Check read-only mode — reject "
                "non-SELECT if read-only",
                "Check for forbidden keywords",
                "Check table allowlist",
                "Enforce single statement",
                "Add row limit if not present",
                "Set statement timeout",
                "Execute query via connection pool",
                "Return results as list of dicts",
            ],
        }

MCP Database Server Checklist

[ ] Use FastMCP to build MCP database server with SQL access tools
[ ] Supported databases: PostgreSQL (asyncpg), MySQL (aiomysql), SQLite (aiosqlite), MSSQL (pyodbc), Oracle (oracledb)
[ ] SQLAlchemy provides universal support for 60+ databases
[ ] Auto-detect database type from DATABASE_URL environment variable
[ ] URL formats: postgresql://, mysql://, sqlite:///path, mssql+pyodbc://, oracle+oracledb://
[ ] Core tools: list_tables, describe_table, execute_query, explain_query
[ ] Additional tools: list_schemas, get_table_ddl
[ ] list_tables: list all tables and views in database
[ ] describe_table: columns, types, constraints, PKs, FKs, indexes
[ ] execute_query: SELECT in read-only mode, any SQL in read-write mode
[ ] explain_query: execution plan for SELECT query optimization
[ ] Read-only mode by default (DB_READ_ONLY=true) — only SELECT/WITH/EXPLAIN
[ ] SQL validation middleware: read-only check, single statement, forbidden keywords
[ ] Forbidden keywords in read-only: DROP, TRUNCATE, DELETE, ALTER
[ ] Granular opt-in for destructive: DB_ALLOW_DROP=true, DB_ALLOW_ALTER=true
[ ] Automatic row limiting: DB_MAX_ROWS=100, add LIMIT if not present
[ ] Query timeout: DB_QUERY_TIMEOUT=10 seconds
[ ] Statement timeout: DB_STATEMENT_TIMEOUT_MS
[ ] Table allowlist: DB_ALLOWED_TABLES (comma-separated)
[ ] Connection pooling: pool_size=5, max_overflow=2, pool_timeout=30s
[ ] Connection pre-ping: detect stale connections
[ ] Multi-instance runtime: expose multiple databases from single MCP server
[ ] Prompts: sql_query (help write SQL), explain_schema (schema overview), optimize_query (EXPLAIN optimization)
[ ] Resources: browse database schema, tables, DDL, sample data
[ ] Configure for Claude Desktop: claude_desktop_config.json with command, args, env
[ ] Set DATABASE_URL in env block of MCP client config
[ ] Set DB_READ_ONLY, DB_MAX_ROWS, DB_QUERY_TIMEOUT in env
[ ] Use read-only database user with SELECT-only grants
[ ] Create dedicated database user for MCP server with minimal privileges
[ ] Connection pooling prevents connection exhaustion
[ ] Row limiting prevents large results from consuming LLM context
[ ] Query timeout prevents long-running queries from blocking
[ ] Table allowlist restricts access to approved tables only
[ ] SQL validation prevents SQL injection and destructive operations
[ ] Single statement enforcement prevents statement stacking attacks
[ ] Log queries for audit (ENABLE_QUERY_LOGS, daily rotation)
[ ] Use SSL/TLS for cloud databases (AWS RDS, Google Cloud SQL, Azure)
[ ] Schema-specific permissions for granular access control
[ ] Multi-DB mode is read-only by default for security
[ ] Transactions: automatic BEGIN/COMMIT/ROLLBACK
[ ] Adapter pattern for extensible database support
[ ] Read MCP developer guide for protocol
[ ] Read build MCP server for server creation
[ ] Read MCP tools for tool development
[ ] Read MCP security for security risks
[ ] Read MCP authentication for auth
[ ] Test: list_tables returns all tables in database
[ ] Test: describe_table returns correct schema with columns and types
[ ] Test: execute_query returns results for SELECT
[ ] Test: read-only mode rejects INSERT/UPDATE/DELETE/DROP
[ ] Test: row limiting adds LIMIT when not present
[ ] Test: query timeout aborts long-running queries
[ ] Test: table allowlist restricts access
[ ] Test: connection pool handles concurrent requests
[ ] Document database URL, tools, security config, and access mode

FAQ

How do you build an MCP database server?

Use FastMCP with database drivers (asyncpg for PostgreSQL, aiomysql for MySQL, aiosqlite for SQLite). Register tools for list_tables, describe_table, execute_query, and explain_query. FastMCP SQL Tools: "MCP server built with FastMCP providing SQL database access for PostgreSQL, MySQL, and SQLite. Tools: list_tables, get_table_schema, execute_query, execute_safe_query. Auto-detects database type from DATABASE_URL env var." SQLAlchemy MCP: "MCP server providing SQL database access through SQLAlchemy. Supports any database SQLAlchemy supports (60+ databases). Read-only and read-write modes. Tools: query, list_tables, describe_table, list_schemas, get_table_ddl, explain_query." DB-MCP: "Python-based MCP server for PostgreSQL, MySQL, SQL Server, SQLite. Read-only mode, query validation, connection pooling. 5 tools: query execution, schema inspection, explain plans." Steps: (1) pip install mcp asyncpg aiosqlite. (2) Create FastMCP server. (3) Register database tools. (4) Set DATABASE_URL env var. (5) Run with stdio transport.

What tools should an MCP database server expose?

Expose list_tables, describe_table, execute_query (read-only SELECT), explain_query, and optionally list_schemas and get_table_ddl. SQLAlchemy MCP: "Tools: query (read-only SELECT/WITH/EXPLAIN), list_tables (all tables and views), describe_table (columns, PKs, FKs, indexes), list_schemas, get_table_ddl (CREATE TABLE DDL), explain_query (execution plan)." FastMCP SQL Tools: "list_tables: list all tables. get_table_schema: columns, types, constraints, indexes. execute_query: any SQL. execute_safe_query: read-only SELECT with safety checks." SQL MCP Server: "list_tables: list accessible tables. describe_table: columns for a table. run_select: validated safe SELECT. run_query: validated query (write allowed when not read-only)." Recommended tools: (1) list_tables — discover available tables. (2) describe_table — schema details (columns, types, constraints, indexes). (3) execute_query or run_select — read-only SELECT queries. (4) explain_query — execution plan for optimization. (5) list_schemas — database schemas. (6) get_table_ddl — CREATE TABLE DDL.

How do you secure an MCP database server?

Use read-only mode by default, SQL validation middleware, single statement enforcement, forbidden keyword detection, table allowlist, row limiting, and connection pooling. SQL MCP Server: "Safe-by-default SQL validation middleware. Read-only mode (DB_READ_ONLY=true) enforced before execution. Single statement enforcement. Forbidden keyword detection. Granular opt-in for destructive statements (DB_ALLOW_DROP=true). Automatic row limiting (LIMIT/TOP). Optional table allowlist (DB_ALLOWED_TABLES). Multi-instance runtime." DB-MCP: "Secure by default: read-only mode, query validation, connection pooling. Read-only: true (recommended). Pool size: 5. Max overflow: 2. Pool timeout: 30s." MCP DB Bridge: "Security first: read-only mode + granular schema permissions. Schema-specific permissions. Multi-DB mode read-only by default. Transactions: automatic BEGIN/COMMIT/ROLLBACK." Security: (1) DB_READ_ONLY=true (default). (2) SQL validation — single statement, no forbidden keywords. (3) Table allowlist — DB_ALLOWED_TABLES. (4) Row limiting — DB_MAX_ROWS=100. (5) Query timeout — DB_QUERY_TIMEOUT=10. (6) Connection pooling with pre-ping. (7) Granular opt-in for DROP/ALTER.

What databases does MCP support for database servers?

MCP database servers support PostgreSQL, MySQL, SQLite, MSSQL, Oracle, and 60+ databases via SQLAlchemy. FastMCP SQL Tools: "PostgreSQL via asyncpg, MySQL via aiomysql, SQLite via aiosqlite. Auto-detects from DATABASE_URL." SQLAlchemy MCP: "Universal database support via SQLAlchemy — SQLite, PostgreSQL, MySQL, Oracle, MSSQL, and 60+ other databases." DB-MCP: "PostgreSQL (psycopg2), MySQL (mysql-connector-python), SQLite (built-in), SQL Server (pyodbc)." URL formats: PostgreSQL: postgresql://user:pass@host:port/db. MySQL: mysql://user:pass@host:port/db. SQLite: sqlite:///path/to/db.db. MSSQL: mssql+pyodbc://user:pass@host:port/db. Oracle: oracle+oracledb://user:pass@host:1521/?service_name=svc.

How do you configure an MCP database server for Claude Desktop?

Add the server to claude_desktop_config.json with command, args, and DATABASE_URL env var. FastMCP SQL Tools: "Add to claude_desktop_config.json: mcpServers with command uvx, args fastmcp-sqltools, env DATABASE_URL." SQLAlchemy MCP: "sqlalchemy-mcp-server --db-url sqlite:///mydb.db for read-only. --mode readwrite for write access. --transport stdio (default) or http." Config: {"mcpServers": {"db-server": {"command": "uvx", "args": ["fastmcp-sqltools"], "env": {"DATABASE_URL": "postgresql://user:pass@localhost/mydb"}}}}. For SQLAlchemy MCP: {"command": "python", "args": ["-m", "sqlalchemy_mcp_server", "--db-url", "postgresql://...", "--mode", "readonly"]}. Restart Claude Desktop after editing.

Want a self-hosted AI company brain that does all of this out of the box?
Book a demo →