DB Explorer MCP

A Model Context Protocol server that lets an AI coding assistant explore, query, and audit a relational database without ever being able to write to it.

Point your MCP client at a database and ask questions in plain language. The client's LLM writes the SQL; this server parses it, refuses anything that is not a single read-only SELECT, executes it with a row cap, and returns structured results. Schema inspection, execution plans, index recommendations, and migration review come along with it.

MCP client LLM  ->  FastMCP tools  ->  safety layer  ->  SQLAlchemy  ->  database
   (writes SQL)       (7 tools)       (rejects writes)   (any dialect)

The server makes no LLM API calls of its own, so there is no API key to configure — reasoning happens in whichever client you connect. Works with SQLite, PostgreSQL, and MySQL through SQLAlchemy.

Why

Giving an assistant raw database credentials means one confused or prompt-injected turn can drop a table. Handing it a read-only replica loses schema context and plan analysis. This server takes the middle path: full introspection and query power, with mutation made structurally impossible at the parser level rather than by asking the model to behave.

Architecture

┌──────────────────────────────────────────────────────────────┐
│  MCP client  (Claude Code / Claude Desktop / Inspector)      │
│  owns the LLM: reads schema, authors SQL, interprets results │
└───────────────────────────┬──────────────────────────────────┘
                            │  MCP  ·  stdio (local)
                            │        ·  streamable HTTP + OAuth 2.0 (remote)
┌───────────────────────────▼──────────────────────────────────┐
│ server.py  —  FastMCP instance + one shared SQLAlchemy engine│
│                                                              │
│   explore_schema   execute_query    explain_query            │
│   validate_schema  suggest_index    migration_context        │
│   validate_migration                                         │
└──────┬───────────────────────┬───────────────────┬───────────┘
       │                       │                   │
       │  read path            │  metadata path    │  review path
       │                       │                   │
┌──────▼────────────┐  ┌───────▼─────────┐  ┌──────▼──────────┐
│ safety.py         │  │ inspector.py    │  │ migration.py    │
│ ── trust boundary │  │ schema_health.py│  │ parses up/down, │
│ sqlparse AST      │  │ index_suggest.py│  │ never executes  │
│ SELECT-only       │  │ explain.py      │  │                 │
│ 1 stmt · no cmnts │  │                 │  │                 │
│ denylist · LIMIT  │  │                 │  │                 │
└──────┬────────────┘  └───────┬─────────┘  └─────────────────┘
       │                       │
       └───────────┬───────────┘
                   │  SQLAlchemy Core (text() + inspect())
┌──────────────────▼───────────────────────────────────────────┐
│  Target database   ·   PostgreSQL  /  MySQL  /  SQLite       │
└──────────────────────────────────────────────────────────────┘

The LLM lives in the client, not the server. Most NL-to-SQL designs put a model call inside the server; this one does not. The client already has a capable model, so the server ships zero LLM dependencies, zero API keys, and zero per-call inference cost — and stays usable from any MCP client, not just Claude.

That split defines the trust boundary: the SQL arriving at safety.py is model-authored and therefore untrusted, so it is parsed rather than pattern-matched, and a rejected query never reaches the driver.

Request lifecycle

A typical execute_query call:

1. Client turns the user's question into SQL, using schema it fetched earlier via explore_schema.
2. FastMCP deserializes the tool call and validates arguments against the tool's type hints.
3. safety.py parses the SQL with sqlparse — one statement, type SELECT, no comments, no blocked keywords. Failure raises before any connection is opened.
4. Row cap applied: if the query has no LIMIT, it is wrapped in SELECT * FROM (…) AS limited_query LIMIT row_limit.
5. SQLAlchemy executes it on a pooled connection and the rows are serialized to plain dicts.
6. Client receives {columns, rows, count} as structured JSON and explains it in natural language.

Errors travel the same path in reverse: a raised ValueError becomes an MCP tool error, which the client surfaces to the user while the server keeps serving.

Module responsibilities

Each tool body delegates to a module-level function that takes an Engine argument, so the whole system is testable against a temporary SQLite database with no MCP client and no network involved.

Transports

Both modes run identical tool code — only MCP_TRANSPORT changes.

Tools

validate_schema reports four codes: missing_primary_key, unindexed_foreign_key, wide_table (50+ columns), and no_indexes.

explore_schema is cheap by default and expensive only on request. With no arguments it
returns table names and column counts — a handful of queries however wide the database
is, and small enough to read before picking a table. Row counts cost a COUNT(*) scan,
so they arrive only with table_name. Narrow a large schema with name_pattern (order
matches any name containing it, order_* is a glob), page with limit/offset (capped
at 1000), and use detail=true to expand a whole page into columns, keys, and indexes.

Safety model

Every execute_query, explain_query, and suggest_index call routes through safety.py before touching the database. A query is rejected unless it satisfies all of:

• Single statement. SELECT 1; DROP TABLE users → Exactly one SQL statement is required

• SELECT only, determined from the parsed statement type rather than a string prefix → Only SELECT queries are allowed. Got: DELETE

• No SQL comments. --, /*, */ are refused outright, closing the classic comment-smuggling route

• No blocked keywords anywhere in the token stream: ALTER, CREATE, DELETE, DROP, EXEC, EXECUTE, GRANT, INSERT, INTO, REVOKE, TRUNCATE, UPDATE

• No locking clause. FOR UPDATE, FOR NO KEY UPDATE, FOR SHARE, FOR KEY SHARE and MySQL's LOCK IN SHARE MODE are refused, because a locking read is not a read: it blocks other transactions from writing those rows. This one is defense in depth rather than a hole being closed — PostgreSQL 16 refuses both forms itself inside a read-only transaction (cannot execute SELECT FOR SHARE in a read-only transaction, verified). What the check adds is a rejection before a connection is opened, an error naming the clause and the fix instead of a generic driver message, and consistency: FOR UPDATE used to be refused only incidentally, because UPDATE is on the denylist for data-modifying CTEs.

  This is matched as a clause, not as a keyword, and the distinction is the point. SHARE alone is a legal column name — sqlparse types the share in SELECT share FROM cap_table as a Keyword — so adding SHARE to the denylist above would reject a real query. A flat set of words is the wrong shape for a rule about multi-word clauses, so safety.py collects the keyword sequence and matches FOR [NO] [KEY] UPDATE|SHARE against it. A FOR belonging to something else (FOR XML, FOR JSON, FOR SYSTEM_TIME) falls through, because its target is not a lock strength.

Every query that passes is wrapped as SELECT * FROM (<your query>) AS limited_query LIMIT <row_limit>, so an unbounded scan cannot flood the client's context. The wrap is unconditional: a LIMIT in your own query narrows the inner result, but row_limit still caps what comes back, so LIMIT 500 with the default row_limit returns 100 rows. row_limit is itself clamped to 1000, so raising it cannot defeat the guard.

Validation is only the first of three layers, because a keyword blocklist cannot see a query that is syntactically fine and still harmful:

• A statement timeout. SELECT pg_sleep(600) passes every check above, so time is bounded independently of syntax: statement_timeout on PostgreSQL, max_execution_time on MySQL, and a progress-handler deadline on SQLite. Configured by QUERY_TIMEOUT_SECONDS, applied in db.py.
• A read-only transaction. Reads run through BEGIN READ ONLY on PostgreSQL, SET SESSION TRANSACTION READ ONLY on MySQL, and PRAGMA query_only on SQLite. The database refuses the write itself, which is a guarantee the blocklist cannot make.
• Privileges. Still the outermost boundary — see .env.example. A SELECT-only user is what stops server-side file reads like pg_read_file() that no keyword check reliably catches.

validate_migration is deliberately the inverse: it rejects SELECT statements, and it never runs either script. You get the parsed statement types back and run the DDL yourself.

Errors

A tool error is only useful to a model if it says what to do next, so every failure carries a code and a corrective hint instead of a bare message:

[table_not_found] Table not found: usrs. Did you mean: users.
Hint: Call explore_schema() to list the tables in this database.

[sql_error] The database rejected the query: no such column: totl.
Hint: Call explore_schema(table_name=...) to confirm the table and column names before retrying.

[query_timeout] The query exceeded the statement timeout of 15s: interrupted.
Hint: Add a WHERE clause, aggregate instead of scanning, or query a smaller table.

Two details worth knowing. Errors are raised as FastMCP ToolError, which is the only error type that survives a server configured with mask_error_details=True — reasonable hardening for an HTTP deployment, and it would otherwise reduce every message above to Error calling tool. And sql_error reports the query you sent, not the row-limit wrapper safety.py builds around it, so the SQL in the message is SQL you can act on.

Internally these stay Python exceptions. errors.py defines ToolInputError, a ValueError subclass whose str() is the plain message, so composition between modules and direct Python use both keep working; conversion happens only at the tool boundary in server.py.

Quickstart

Requires Python 3.11+ and uv.

uv sync
uv run python tests/seed_test_db.py   # creates sample.db
uv run pytest                         # 151 tests, no external database needed
uv run server.py                      # stdio transport

If uv is not on PATH, prefix with py -m (py -m uv sync).

The default database is sqlite:///sample.db. Point at your own with DATABASE_URL:

$env:DATABASE_URL = "postgresql+psycopg2://user:password@localhost:5432/example"
$env:DATABASE_URL = "mysql+pymysql://user:password@localhost:3306/example"
$env:DATABASE_URL = "sqlite:///C:/data/example.db"

Percent-encode special characters in passwords (@ → %40, # → %23, / → %2F).

Connect a client

Claude Code — local

claude mcp add db-explorer --env DATABASE_URL="postgresql+psycopg2://user:pass@localhost:5432/example" -- uv --directory "C:/path/to/DB-Explorer-MCP" run server.py

Then run /mcp in a session to confirm the 7 tools are listed. Add -s user to make it available in every project.

Claude Code — remote

claude mcp add --transport http db-explorer https://your-deployment.fastmcp.app/mcp

Run /mcp → Authenticate for the OAuth flow; tokens are cached and refreshed automatically.

Claude Desktop

Local, in claude_desktop_config.json:

{
  "mcpServers": {
    "db-explorer": {
      "command": "uv",
      "args": ["--directory", "C:/path/to/DB-Explorer-MCP", "run", "server.py"],
      "env": { "DATABASE_URL": "postgresql+psycopg2://user:pass@localhost:5432/example" }
    }
  }
}

To reach a remote deployment without a custom connector, proxy it over stdio:

{
  "mcpServers": {
    "db-explorer": {
      "command": "npx",
      "args": ["-y", "mcp-remote", "https://your-deployment.fastmcp.app/mcp"]
    }
  }
}

VS Code

.vscode/mcp.json is checked in and starts the server over stdio — no extra setup for anyone who clones the repo.

MCP Inspector

npx @modelcontextprotocol/inspector

Use transport Streamable HTTP with your /mcp URL, or stdio with uv run server.py. The Inspector shows raw tool responses and unparaphrased errors, which makes it the fastest way to tell a server problem from a client problem.

Python

import asyncio
from fastmcp import Client

async def main():
    async with Client("https://your-deployment.fastmcp.app/mcp", auth="oauth") as client:
        print([tool.name for tool in await client.list_tools()])
        print(await client.call_tool("explore_schema", {}))

asyncio.run(main())

Try it

Once connected, prompts like these work directly:

• "What tables exist, and which ones are missing primary keys?"
• "Show me 5 rows from orders with the highest total."
• "Why is this query slow? SELECT * FROM orders WHERE customer_id = 42"
• "Which foreign keys in this database lack indexes? Give me the CREATE INDEX statements."
• "Draft a migration adding a status column to orders, then validate the up and down scripts."

To watch the guardrails work, ask it to run DELETE FROM users. The call fails with Unsafe query blocked: Only SELECT queries are allowed. Got: DELETE and the database is untouched.

Configuration

The sqlite fallback exists for local development only. config.py raises RuntimeError: DATABASE_URL must be set when serving over HTTP rather than silently serving an empty local file from a deployment — a failure mode that otherwise surfaces much later as a confusing unable to open database file.

.env is loaded at startup by config.py, so copying .env.example to .env works as that file instructs. The file next to config.py is read first and a .env in the working directory second, because an MCP client launches this server with a working directory you do not control. Real environment variables always win over both, so a host's secret store overrides the file without editing it. .env stays gitignored.

Serve over HTTP

$env:MCP_TRANSPORT = "streamable-http"
$env:MCP_HOST = "0.0.0.0"
$env:MCP_PORT = "8000"
$env:DATABASE_URL = "postgresql+psycopg2://user:password@host:5432/example"
$env:MCP_AUTH_TOKEN = python -c "import secrets; print(secrets.token_urlsafe(32))"
uv run server.py

An HTTP endpoint publishes SELECT on the configured database to anyone who can reach the port, so the server fails to start without MCP_AUTH_TOKEN rather than coming up unprotected. Clients send it as Authorization: Bearer <token>; it is compared in constant time in auth.py. Set MCP_ALLOW_UNAUTHENTICATED=true to override on a genuinely trusted network — the server then warns on stderr at every startup.

For real user identity rather than one shared secret, swap SharedSecretVerifier for one of FastMCP's OAuth providers. See DEPLOYMENT.md for FastMCP Cloud / Prefect Horizon deployment, where OAuth 2.0 with dynamic client registration and PKCE is handled by the platform.

Hosted Supabase note: direct connections (db.<ref>.supabase.co) are IPv6-only, which fails from IPv4-only containers with an empty-looking psycopg2.OperationalError. Use the pooler host from the dashboard's Connect panel, and note that the username becomes postgres.<project-ref>.

Tests

uv run pytest

151 tests covering the safety layer, value serialization, read-only enforcement and timeouts, HTTP authentication, inspector, explain, index suggestions, schema health, migration validation, error reporting, and the tool wrappers. Each uses a temporary SQLite database, so the suite needs no credentials and no running server.

SQLite cannot produce the types that break a real driver -- it has no NUMERIC and returns str/int for nearly everything -- so tests/test_serialization.py exercises Decimal, datetime, UUID, and binary values directly rather than through a query. A PostgreSQL and MySQL test path is the next gap worth closing.

Project layout

server.py         FastMCP instance, engine, and the 7 tool definitions
safety.py         query validation and row-limited execution
db.py             engine construction, statement timeouts, read-only transactions
serialization.py  driver values to JSON-safe primitives
auth.py           bearer-token verification for HTTP transports
inspector.py      schema reflection (columns, PK, FKs, indexes, samples)
explain.py        dialect-aware EXPLAIN
index_suggest.py  index recommendations from plans or FK metadata
schema_health.py  objective schema issue reporting
migration.py      migration context and non-executing script validation
errors.py         coded errors with hints, and driver-error classification
config.py         environment configuration with fail-fast checks
tests/            pytest suite over temporary SQLite databases

Design notes and limits

• Migrations are never executed. The server returns schema context and validates scripts; you run the DDL. That keeps the connection read-only in practice, not just by policy.

• Query-mode suggest_index is tuned to SQLite plan output, which exposes a detail column containing SCAN. On PostgreSQL and MySQL the plan is still returned in full, but automatic recommendations will usually be empty — use table_name mode there, which works from foreign-key metadata on every dialect.

• The keyword denylist matches whole tokens, not substrings, so a keyword that merely contains a blocked word is unaffected: GROUPING SETS and SELECT grant_date FROM permissions both pass, where a naive "SET" in sql check would reject the first and "GRANT" in sql the second.

• Each denylist entry has to earn its place. INTO does: SELECT * INTO archive FROM users has statement type SELECT but creates a table, so only the keyword scan catches it. SET did not, and was removed — every statement that changes session state (SET ROLE, SET search_path, even SET x = (SELECT 1)) parses as type UNKNOWN and is refused by the type check, while UPDATE ... SET inside a data-modifying CTE is caught by UPDATE. All it added was rejecting SELECT set FROM config, since sqlparse types a bare set as a keyword rather than a column name.

• Four of the twelve entries are load-bearing — INSERT, UPDATE, DELETE and INTO are reachable in a statement whose type is SELECT, the first three through Postgres data-modifying CTEs. The rest are redundant, because a CTE accepts only INSERT, UPDATE, DELETE and MERGE, never DDL: no legal SELECT-typed statement can contain DROP. They stay as a second line if sqlparse type detection ever regresses.

• A few read-only statements are rejected for lack of a statement type. sqlparse reports UNKNOWN for a parenthesized (SELECT 1), for VALUES (1) and for TABLE users, and the type check refuses anything that is not SELECT. All are harmless; none is currently accepted.

  This is left as a rejection rather than fixed by unwrapping, because the incidence is near zero — a caller writes the plain SELECT — and widening what the parser accepts to serve a query nobody sends is a poor trade against the risk. What was fixed is the explanation: these reasons and a genuine write both contain Only SELECT, so they used to collapse to the same hint, and a caller that sent a read was told the server is read-only and pointed at validate_migration. Got: UNKNOWN now carries its own hint naming the shapes that cause it. A rejection the caller can recover from in one turn is an acceptable cost; a rejection that misdiagnoses itself is not.

• The row cap is a context guard, not a performance guard. A heavy aggregate still runs in full on the database before its output is limited. QUERY_TIMEOUT_SECONDS is what bounds the cost of that work.

• Binary columns are summarised, not returned. Values up to 256 bytes arrive hex-encoded, which suits BINARY(16) UUIDs and digests; anything larger is reported as a size only. Inlining a multi-megabyte blob would consume the context window it was sent to.

• Wide NUMERIC values arrive as strings. A decimal that fits a float is a JSON number so it sorts and compares correctly; one that does not keeps its exact digits rather than being silently rounded.