The read-only Postgres MCP server built its safety on the database server: a SELECT-only role, a read-only transaction, a statement timeout, all enforced by a process on the other side of a socket. SQLite has none of that machinery. There is no server, no accounts, no GRANT. The database is a file, the engine runs inside your process, and any connection that can open the file read-write can do anything to it.

So the hardening moves into the connection itself. SQLite ships engine hooks that, layered together, give the same guarantee (a write cannot succeed even when the Python above it is wrong), plus one thing the Postgres version never had: an engine-level veto on reading tables you did not allow-list. In this tutorial you will build a FastMCP server that exposes a SQLite database to an MCP client as strictly read-only, with these layers:

  • the connection opens the file with mode=ro, so the engine refuses every write on that handle;
  • PRAGMA query_only=ON re-states the same refusal at the session level, so the guarantee survives even if the open mode is changed later;
  • an authorizer callback, a per-statement engine hook with no Postgres equivalent, that default-denies every operation except reading allow-listed tables, so a write is rejected at prepare time and a SELECT against an unlisted table never runs at all;
  • a query deadline via a progress handler, interrupting any statement that runs too long, since SQLite has no statement_timeout;
  • a SELECT-only guard, an identifier allow-list, parameterized values, and a row cap in Python, same as the Postgres version;
  • OS file permissions (chmod 444) as the outermost layer, because with an embedded database the filesystem is the perimeter.

No Docker this time: the whole stack is a uv project and a file on disk, and pytest proves each layer holds on its own.

What you will build

  • A seed script that creates a deterministic db/app.db (including a secrets table the server must never read) and marks the file read-only at the OS level.
  • A config.py that reads the database path, table allow-list, query deadline, and row cap from the environment.
  • A db.py that opens read-only connections with query_only, the authorizer, and the deadline installed.
  • A safety.py with the query guard, identifier validation, and identifier quoting, unit-tested with no database.
  • A server.py exposing four tools: list_tables, describe_table, query_table (allow-list driven), and run_query (guarded raw SELECT).
  • A pytest suite that proves each layer independently refuses writes, the authorizer blocks unlisted reads, the deadline interrupts a runaway query, and rows are capped.

Prerequisites

  • macOS 13+ with Homebrew (brew.sh).
  • uv 0.5+ — brew install uv; verify uv --version.
  • Xcode Command Line Tools (xcode-select --install) for make.
  • Basic SQL familiarity. SQLite itself needs no install: Python’s standard library bundles the sqlite3 module, and uv manages Python. Check the engine version any time with uv run python -c "import sqlite3; print(sqlite3.sqlite_version)".

Everything runs through uv and make.

Step 1: Scaffold and lock down hygiene

Create the workspace and a .gitignore first, before any other file, so the database file and runtime artifacts can never be committed.

Create the files

mkdir -p readonly-sqlite-mcp-server-macos
cd readonly-sqlite-mcp-server-macos
touch .gitignore

Add the code: .gitignore

# Python
__pycache__/
*.py[cod]
.venv/
.uv/
.pytest_cache/
.ruff_cache/
.mypy_cache/
# The database and its journals -- generated by `make seed`, never committed
db/*.db
db/*.db-wal
db/*.db-shm
db/*.db-journal
# Runtime artifacts
logs/
*.log
# OS / editor noise
.DS_Store

Detailed breakdown

  • The database is generated, not committed. make seed recreates db/app.db deterministically, so the file is a build artifact. Ignoring db/*.db plus the -wal/-shm/-journal sidecar patterns keeps every form the engine might leave on disk out of git.
  • There is no .env entry because this project has no credentials at all. SQLite has no accounts; access is file access. That is a real difference from the Postgres article, where the connection string carried a password and keeping it out of source took a whole step. Here the outer boundary is the file’s permission bits, handled by the seed script.
  • The rest is standard uv/macOS hygiene.

Step 2: Initialize the uv project

Create the uv project and add FastMCP. There is no database driver to install: sqlite3 ships with Python.

Create the files

uv init --name sqlmcp --no-workspace
rm -f main.py hello.py
uv add fastmcp
uv add --dev pytest pytest-asyncio

Add the code: pyproject.toml

[project]
name = "sqlmcp"
version = "0.1.0"
description = "A safe read-only SQLite MCP server"
readme = "README.md"
requires-python = ">=3.12"
dependencies = [
    "fastmcp>=3.4.4",
]

[dependency-groups]
dev = [
    "pytest>=9.1.1",
    "pytest-asyncio>=1.4.0",
]

Detailed breakdown

  • fastmcp is the MCP server framework; its in-memory client also drives the tests and the demo script.
  • requires-python = ">=3.12" matters here: the code uses sqlite3.connect(..., autocommit=True), a parameter added in Python 3.12. uv provisions a matching interpreter automatically.
  • No driver dependency. The sqlite3 module is the standard library binding to the bundled SQLite engine, and it exposes everything this server needs: URI filenames, the authorizer hook, and the progress handler.

Step 3: Seed a deterministic database

The seed script builds the entire database from scratch on every run: two tables the server will expose, one secrets table it must never read, fixed rows so every count and sum in this article is reproducible. It finishes by marking the file read-only at the OS level.

Create the files

mkdir -p db scripts
touch scripts/seed.py

Add the code: scripts/seed.py

"""Create the tutorial database deterministically, then mark it read-only.

Deletes any existing database file first, so reseeding always yields the same
bytes. The `secrets` table exists to prove a point: it is right there in the
file, but the server's authorizer will refuse to read it.
"""

import os
import sqlite3
from pathlib import Path

from config import settings

SCHEMA = """
CREATE TABLE customers (
    id    INTEGER PRIMARY KEY,
    name  TEXT NOT NULL,
    email TEXT NOT NULL,
    city  TEXT NOT NULL
);

CREATE TABLE orders (
    id          INTEGER PRIMARY KEY,
    customer_id INTEGER NOT NULL REFERENCES customers (id),
    total_cents INTEGER NOT NULL,
    status      TEXT NOT NULL
);

-- Deliberately NOT in the allow-list. The engine-level authorizer refuses to
-- read it, which the tests prove. The value is a placeholder, not a real key.
CREATE TABLE secrets (
    id      INTEGER PRIMARY KEY,
    api_key TEXT NOT NULL
);

INSERT INTO customers (id, name, email, city) VALUES
    (1, 'Ada Lovelace',      '[email protected]',       'London'),
    (2, 'Alan Turing',       '[email protected]',      'London'),
    (3, 'Grace Hopper',      '[email protected]',     'New York'),
    (4, 'Katherine Johnson', '[email protected]', 'Hampton');

INSERT INTO orders (id, customer_id, total_cents, status) VALUES
    (1, 1,  1299, 'paid'),
    (2, 1,  4500, 'paid'),
    (3, 2,   999, 'refunded'),
    (4, 3, 15000, 'paid'),
    (5, 3,  2500, 'pending');

INSERT INTO secrets (id, api_key) VALUES
    (1, 'demo-placeholder-not-a-real-key');
"""


def main() -> None:
    path = Path(settings.db_path)
    path.parent.mkdir(parents=True, exist_ok=True)
    path.unlink(missing_ok=True)

    conn = sqlite3.connect(path)
    conn.executescript(SCHEMA)
    conn.commit()
    counts = {
        table: conn.execute(f"SELECT count(*) FROM {table}").fetchone()[0]
        for table in ("customers", "orders", "secrets")
    }
    conn.close()

    # The outermost layer: the file itself is read-only at the OS level.
    os.chmod(path, 0o444)
    summary = ", ".join(f"{table}={n}" for table, n in counts.items())
    print(f"seeded {path} ({summary}); file mode is now 0444")


if __name__ == "__main__":
    main()

Detailed breakdown

  • Reseeding is destructive on purpose. The script unlinks the old file and rebuilds it, so make seed always produces the same database and stale state can never leak into a test run. Unlinking a 0444 file works because delete permission comes from the directory, not the file.
  • The seed data is deterministic (fixed ids and values), so every number in this article is checkable: London has two customers, and the paid orders total 1299 + 4500 + 15000 = 20799 cents.
  • The secrets table is bait. It lives in the same file as the allowed tables, exactly like a real production database where sensitive tables sit next to boring ones. In Postgres you would withhold GRANT SELECT; here Step 5’s authorizer refuses the read inside the engine, and a test proves it.
  • os.chmod(path, 0o444) is the layer no SQL can bypass: even code that opens the file without mode=ro gets a filesystem-level refusal when it tries to flush a write. It is the embedded-database analogue of the network boundary a client/server database gets for free.

Step 4: Centralize configuration

Read every setting from the environment so the same code serves a different database file, allow-list, or limit without edits. Unlike the Postgres version there is no secret to manage: nothing here is more sensitive than a file path.

Create the file

touch config.py

Add the code: config.py

"""Runtime configuration, read once from the environment.

Every security-relevant limit lives here: which file is opened, which tables
the tools may touch, how long a query may run, and how many rows a single call
may return. There are no credentials -- SQLite access is file access, so the
sensitive settings are the allow-list and the limits, not a password.
"""

import os
from dataclasses import dataclass, field


def _tables_from_env() -> tuple[str, ...]:
    raw = os.environ.get("ALLOWED_TABLES", "customers,orders")
    return tuple(t.strip() for t in raw.split(",") if t.strip())


@dataclass(frozen=True)
class Settings:
    db_path: str = os.environ.get("SQLITE_DB_PATH", "db/app.db")
    allowed_tables: tuple[str, ...] = field(default_factory=_tables_from_env)
    query_timeout_ms: int = int(os.environ.get("QUERY_TIMEOUT_MS", "2000"))
    max_rows: int = int(os.environ.get("MAX_ROWS", "100"))


settings = Settings()

Detailed breakdown

  • db_path defaults to db/app.db, the file make seed creates, and is resolved relative to the project root. The Makefile and the client configuration in the final step both run the server from that directory, so the relative path holds; set SQLITE_DB_PATH absolute if you run it from anywhere else.
  • allowed_tables drives two layers. The structured query_table tool checks names against it in Python, and db.connect compiles it into the engine authorizer, so the same list is enforced twice at different depths. secrets is deliberately absent.
  • query_timeout_ms and max_rows are the runaway-query guards, integers read from the environment so an operator can tighten them without touching code.
  • frozen=True makes the settings immutable after load; the tests that vary a limit rebind a dataclasses.replace copy rather than mutating.

Step 5: Read-only, engine-guarded connections

All database access goes through one helper, and this is where three of the layers live. The connection opens the file mode=ro, turns on query_only, snapshots the stored tables, installs the authorizer, and arms a deadline. Order matters: query_only and the snapshot run before the authorizer is installed, because the authorizer would deny both the PRAGMA statement and the read of sqlite_master.

One SQLite subtlety shapes the authorizer’s read rule. SQLITE_READ fires not only for stored tables but also for names that exist only inside a statement: a recursive CTE’s name, or the pragma_table_info table-valued function. Those names cannot be allow-listed in advance, so the rule is built from a snapshot of what is actually stored in the file: reads of allow-listed tables pass, reads of every other stored table (and the sqlite_* catalogs) are denied, and names that are not stored objects pass. A CTE that happens to shadow a hidden table’s name is still denied — the rule fails closed.

Create the file

touch db.py

Add the code: db.py

"""Read-only, engine-guarded SQLite connections.

Four independent layers are applied to every connection:

  1. mode=ro         -> the engine refuses writes on this handle.
  2. query_only=ON   -> the session refuses writes even if the mode changes.
  3. authorizer      -> default-deny per-statement hook: only SELECT machinery
                        and reads of allow-listed tables are permitted, checked
                        at prepare time inside the engine.
  4. progress handler -> a wall-clock deadline; SQLite has no statement
                        timeout, so a callback interrupts long statements.

The authorizer is the layer with no Postgres equivalent: it also refuses to
READ any table outside the allow-list, so `SELECT * FROM secrets` fails inside
the engine no matter what SQL reaches it.
"""

import sqlite3
import time
from pathlib import Path

from config import settings

# Operations any legitimate SELECT needs. Everything else is denied.
_ALWAYS_ALLOWED = frozenset({
    sqlite3.SQLITE_SELECT,     # the statement itself
    sqlite3.SQLITE_FUNCTION,   # count(), sum(), lower(), ...
    sqlite3.SQLITE_RECURSIVE,  # WITH RECURSIVE common table expressions
})

def _make_authorizer(allowed_tables: frozenset[str], stored_tables: frozenset[str]):
    """Build a default-deny authorizer closed over the table allow-list.

    SQLITE_READ fires for more than stored tables: recursive CTE names and
    pragma table-valued functions surface here too, and those names cannot be
    known in advance. So the read rule is: allow-listed stored tables pass,
    every OTHER stored table (and the sqlite_* catalogs) is denied, and
    ephemeral names -- which are not stored objects -- pass. A CTE that shadows
    a hidden table's name is still denied, which fails closed.
    """
    hidden_tables = stored_tables - allowed_tables

    def authorize(action, arg1, arg2, db_name, source):
        if action in _ALWAYS_ALLOWED:
            return sqlite3.SQLITE_OK
        if action == sqlite3.SQLITE_READ:
            # arg1 is the table-ish name, arg2 the column.
            name = arg1 or ""
            if name in allowed_tables:
                return sqlite3.SQLITE_OK
            if name in hidden_tables or name.startswith("sqlite_"):
                return sqlite3.SQLITE_DENY
            return sqlite3.SQLITE_OK  # ephemeral: CTEs, pragma functions
        if action == sqlite3.SQLITE_PRAGMA:
            # Only table_info, and only for allow-listed tables, so even the
            # schema of a hidden table stays hidden.
            if arg1 == "table_info" and arg2 in allowed_tables:
                return sqlite3.SQLITE_OK
            return sqlite3.SQLITE_DENY
        return sqlite3.SQLITE_DENY  # INSERT, UPDATE, DELETE, DDL, ATTACH, ...

    return authorize


def _dict_row(cursor: sqlite3.Cursor, row: tuple) -> dict:
    """Return each row as {column: value}, ready for JSON."""
    return {desc[0]: value for desc, value in zip(cursor.description, row)}


def connect() -> sqlite3.Connection:
    """Open a fully guarded read-only connection to the configured database."""
    path = Path(settings.db_path)
    if not path.exists():
        raise RuntimeError(
            f"database not found at {settings.db_path}; run `make seed` first"
        )

    conn = sqlite3.connect(f"file:{path}?mode=ro", uri=True, autocommit=True)
    conn.row_factory = _dict_row

    # Layer 2 -- must run before the authorizer is installed, because the
    # authorizer denies PRAGMA statements other than table_info.
    conn.execute("PRAGMA query_only = ON")

    # Snapshot the stored tables and views so the authorizer can tell a real
    # (deniable) table apart from an ephemeral name like a CTE. Must also run
    # before the authorizer is installed: it reads sqlite_master.
    stored_tables = frozenset(
        row["name"]
        for row in conn.execute(
            "SELECT name FROM sqlite_master WHERE type IN ('table', 'view')"
        )
    )

    # Layer 4 -- the deadline. The handler runs every 1000 VM instructions;
    # returning a truthy value interrupts the statement with
    # sqlite3.OperationalError: interrupted.
    deadline = time.monotonic() + settings.query_timeout_ms / 1000
    conn.set_progress_handler(
        lambda: 1 if time.monotonic() > deadline else 0, 1000
    )

    # Layer 3 -- the engine-level default-deny hook.
    conn.set_authorizer(
        _make_authorizer(frozenset(settings.allowed_tables), stored_tables)
    )
    return conn

Detailed breakdown

  • mode=ro in the URI filename opens the handle read-only inside the engine. Any write attempt fails with attempt to write a readonly database, before the filesystem is even consulted. uri=True tells sqlite3.connect to parse the file: string instead of treating it as a literal filename.
  • PRAGMA query_only = ON looks redundant next to mode=ro, and that is the point: it is a second, independent statement of the same policy. If a future edit swaps the URI for a plain path “just to debug something,” query_only still refuses every write on the session.
  • The authorizer is a default-deny allow-list, which is the only safe shape for this hook. It runs inside sqlite3_prepare, so a denied operation fails before the statement ever executes. SELECT machinery, functions, and recursive CTEs are allowed; SQLITE_READ follows the snapshot rule above, so hidden stored tables are unreadable (even count(*) over one is denied) while CTE names pass. The table_info pragma is allowed only for allow-listed tables, because describe_table needs it and a hidden table’s schema should stay as hidden as its rows. Everything else falls through to SQLITE_DENY: every write action, every DDL action, ATTACH (which could target a writable second file), other pragmas. A denied statement raises sqlite3.DatabaseError with a not authorized or prohibited message.
  • The progress handler is the timeout. SQLite has no server to cancel a statement, so the embedding process must do it: every 1000 virtual-machine instructions the callback compares time.monotonic() to a deadline fixed at connect time, and a truthy return interrupts the statement. Each tool call opens a fresh connection, so the deadline spans exactly one call.
  • autocommit=True keeps the connection out of implicit transactions, which a read-only server has no use for. It also means no BEGIN is ever issued — convenient, because the authorizer would deny the transaction opcode.
  • _dict_row returns plain dicts that serialize straight to JSON for the MCP client, matching the dict_row factory the Postgres version used.

Step 6: The query guard and identifier validation

Three helpers, all pure functions with no database access, so they unit-test instantly. assert_select_only gives raw SQL a fast, clear rejection when it is obviously not a single read query. validate_identifiers checks names against the allow-list. quote_identifier double-quotes a validated name for composition into SQL.

Create the file

touch safety.py

Add the code: safety.py

"""Input guards for the query tools.

These are the fast, friendly layer: they reject bad input with a clear message
before it reaches the engine. They are NOT the security boundary -- the
read-only open mode, query_only, and the authorizer are. A guard that only
inspects keywords can always be fooled, which is exactly why the engine is
configured to refuse writes and unlisted reads on its own.
"""


class QueryError(ValueError):
    """Raised when a query or identifier fails a safety check."""


def assert_select_only(sql_text: str) -> str:
    """Return a normalized single-statement SELECT, or raise QueryError.

    Rejects empty input, multiple statements, and anything that does not begin
    with SELECT or WITH. This is a usability guard, not the write barrier.
    """
    stripped = sql_text.strip().rstrip(";").strip()
    if not stripped:
        raise QueryError("query is empty")
    if ";" in stripped:
        raise QueryError("only a single statement is allowed")
    first_word = stripped.split(None, 1)[0].lower()
    if first_word not in ("select", "with"):
        raise QueryError("only SELECT (or WITH ... SELECT) queries are allowed")
    return stripped


def validate_identifiers(names: list[str], allowed: set[str], kind: str) -> None:
    """Raise QueryError if any name is not in the allowed set."""
    for name in names:
        if name not in allowed:
            raise QueryError(f"unknown {kind}: {name!r}")


def quote_identifier(name: str) -> str:
    """Double-quote an identifier for SQL composition.

    Callers must validate the name against an allow-list first; quoting is
    belt-and-braces, not a substitute for validation.
    """
    return '"' + name.replace('"', '""') + '"'

Detailed breakdown

  • assert_select_only trims one trailing semicolon, then rejects any remaining one. That blocks the classic ...; DROP TABLE ... multi-statement injection with a readable error. It also requires the first keyword to be SELECT or WITH.
  • It is deliberately honest about its limits. A statement can begin with WITH and still not be a read. In the Postgres article the backstop was the role and the read-only transaction; here it is mode=ro, query_only, and the authorizer, all of which reject the write regardless of what the string looks like. The guard exists to return a clean error fast, not to be trusted as the wall.
  • quote_identifier follows the SQL standard rule (double the internal double quotes, wrap in double quotes). Since every name is allow-list checked before it is quoted, this is defense in depth against a malformed entry in the allow-list itself, not a sanitizer for arbitrary input.

Step 7: The server and its tools

Four tools, same shape as the Postgres version. list_tables and describe_table let a client discover what it may query. query_table is the structured path: allow-listed table and columns, parameterized filter values, capped rows. run_query accepts real SQL through the guard, and underneath it the engine still enforces read-only and the table allow-list.

Create the file

touch server.py

Add the code: server.py

"""A safe read-only SQLite MCP server.

Tools:
  * list_tables    -> the allow-listed tables this server will expose.
  * describe_table -> column names and types for one allow-listed table.
  * query_table    -> structured SELECT with allow-listed columns,
                      parameterized filters, and a row cap. No raw SQL.
  * run_query      -> a guarded raw SELECT, still read-only, deadline-bounded,
                      allow-list-read-checked by the engine, and row-capped.
"""

import sqlite3

from fastmcp import FastMCP
from fastmcp.exceptions import ToolError

import db
from config import settings
from safety import (
    QueryError,
    assert_select_only,
    quote_identifier,
    validate_identifiers,
)

mcp = FastMCP("sqlmcp")


def _columns_of(table: str) -> list[dict]:
    """Return [{name, type}] for an allow-listed table, or raise QueryError."""
    validate_identifiers([table], set(settings.allowed_tables), "table")
    with db.connect() as conn:
        rows = conn.execute(
            "SELECT name, type FROM pragma_table_info(?)", (table,)
        ).fetchall()
    if not rows:
        raise QueryError(f"unknown table: {table!r}")
    return rows


def _cap(rows: list[dict], limit: int) -> dict:
    """Truncate to `limit` rows and report whether more existed.

    Callers over-fetch one extra row (limit + 1) so a full result signals
    truncation instead of silently returning a short answer.
    """
    truncated = len(rows) > limit
    return {"rows": rows[:limit], "truncated": truncated}


@mcp.tool
def list_tables() -> list[str]:
    """List the tables this server is allowed to read."""
    return list(settings.allowed_tables)


@mcp.tool
def describe_table(table: str) -> list[dict]:
    """Return column names and types for one allowed table."""
    try:
        return _columns_of(table)
    except QueryError as exc:
        raise ToolError(str(exc))


@mcp.tool
def query_table(
    table: str,
    columns: list[str] | None = None,
    filters: dict[str, str] | None = None,
    limit: int | None = None,
) -> dict:
    """Select from one allowed table with allow-listed columns and filters.

    columns defaults to all columns. filters is column -> value, combined with
    AND and passed as parameters (never interpolated). limit is capped by
    MAX_ROWS.
    """
    filters = filters or {}
    try:
        real_columns = [c["name"] for c in _columns_of(table)]
        allowed = set(real_columns)
        selected = columns or real_columns
        validate_identifiers(selected, allowed, "column")
        validate_identifiers(list(filters), allowed, "column")
    except QueryError as exc:
        raise ToolError(str(exc))

    row_limit = settings.max_rows if limit is None else min(limit, settings.max_rows)

    col_sql = ", ".join(quote_identifier(c) for c in selected)
    query = f"SELECT {col_sql} FROM {quote_identifier(table)}"
    params: list = []
    if filters:
        conditions = [f"{quote_identifier(col)} = ?" for col in filters]
        query += " WHERE " + " AND ".join(conditions)
        params.extend(filters.values())
    query += " LIMIT ?"
    params.append(row_limit + 1)  # one extra row detects truncation

    try:
        with db.connect() as conn:
            rows = conn.execute(query, params).fetchall()
    except sqlite3.Error as exc:
        raise ToolError(f"database error: {exc}")
    return _cap(rows, row_limit)


@mcp.tool
def run_query(query: str, params: list | None = None) -> dict:
    """Run a single read-only SELECT and return capped rows.

    Pass values through `params` (with ? placeholders) rather than formatting
    them into the query string.
    """
    try:
        sql_text = assert_select_only(query)
    except QueryError as exc:
        raise ToolError(str(exc))
    try:
        with db.connect() as conn:
            cur = conn.execute(sql_text, params or [])
            rows = cur.fetchmany(settings.max_rows + 1)
    except sqlite3.Error as exc:
        raise ToolError(f"database error: {exc}")
    return _cap(rows, settings.max_rows)


def main() -> None:
    import os

    if os.environ.get("MCP_TRANSPORT", "stdio").lower() == "http":
        mcp.run(
            transport="http",
            host=os.environ.get("MCP_HOST", "127.0.0.1"),
            port=int(os.environ.get("MCP_PORT", "8000")),
        )
    else:
        mcp.run()  # stdio


if __name__ == "__main__":
    main()

Detailed breakdown

  • _columns_of does double duty. It validates the table against the allow-list and reads the real columns from pragma_table_info(?), SQLite’s table-valued pragma function, with the table name bound as a parameter. query_table reuses the result as its column allow-list, so no hand-written column list can drift out of date. This is also why the authorizer permits the table_info pragma and nothing else.
  • query_table composes SQL only from validated, quoted identifiers. Every name passes validate_identifiers first and quote_identifier second; every filter value is a ? placeholder bound as a parameter. The pattern is identical to the Postgres version with psycopg.sql swapped for explicit quoting, because the standard library has no composition helper.
  • Truncation is detected by over-fetching one row. Each tool fetches one more row than its effective limit (row_limit + 1 for query_table, max_rows + 1 for run_query), and _cap trims to that limit and sets truncated, so the client learns the result was clipped.
  • run_query leans on the engine. The Python guard rejects the obvious non-reads, but the interesting property is what happens to a well-formed SELECT * FROM secrets: the guard passes it, and the authorizer kills it at prepare time. In the Postgres version the equivalent request would have succeeded unless you remembered to withhold the grant — the allow-list here is enforced against raw SQL too.
  • Errors surface as ToolError. FastMCP returns that to the client as a typed tool error rather than a stack trace, whether the cause was a bad identifier, a denied read, an interrupted query, or a write attempt.

Step 8: Prove every layer holds

Three test files. test_safety.py unit-tests the pure guards. test_layers.py is the SQLite-specific one: it builds scratch databases and proves each engine layer refuses a write on its own, with the other layers absent. test_server.py drives the tools through FastMCP’s in-memory client against the seeded database.

Create the files

mkdir -p tests
touch tests/__init__.py tests/test_safety.py tests/test_layers.py tests/test_server.py pytest.ini

Add the code: pytest.ini

[pytest]
asyncio_mode = auto

Add the code: tests/test_safety.py

"""Unit tests for the pure guards. No database required."""

import pytest

from safety import (
    QueryError,
    assert_select_only,
    quote_identifier,
    validate_identifiers,
)


def test_accepts_plain_select():
    assert assert_select_only("SELECT 1") == "SELECT 1"


def test_accepts_with_and_strips_trailing_semicolon():
    assert assert_select_only("WITH t AS (SELECT 1) SELECT * FROM t;") == (
        "WITH t AS (SELECT 1) SELECT * FROM t"
    )


def test_rejects_empty():
    with pytest.raises(QueryError):
        assert_select_only("   ")


def test_rejects_multiple_statements():
    with pytest.raises(QueryError):
        assert_select_only("SELECT 1; DROP TABLE customers")


@pytest.mark.parametrize(
    "bad", ["INSERT INTO t VALUES (1)", "UPDATE t SET x=1", "DELETE FROM t"]
)
def test_rejects_non_select(bad):
    with pytest.raises(QueryError):
        assert_select_only(bad)


def test_validate_identifiers_rejects_unknown():
    with pytest.raises(QueryError):
        validate_identifiers(["price"], {"id", "name"}, "column")


def test_validate_identifiers_accepts_known():
    validate_identifiers(["id", "name"], {"id", "name", "city"}, "column")


def test_quote_identifier_doubles_quotes():
    assert quote_identifier("plain") == '"plain"'
    assert quote_identifier('we"ird') == '"we""ird"'

Add the code: tests/test_layers.py

"""Each engine layer refuses a write ON ITS OWN.

Every test builds a scratch database and enables exactly one layer, so a
failure pinpoints which guarantee broke. This is the heart of the article:
defense in depth only counts if each layer holds independently.
"""

import sqlite3

import pytest

import db


def _scratch(tmp_path):
    """A writable scratch database with one visible and one hidden table."""
    path = tmp_path / "scratch.db"
    conn = sqlite3.connect(path)
    conn.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, v TEXT)")
    conn.execute("INSERT INTO t VALUES (1, 'a')")
    conn.execute("CREATE TABLE hidden (k TEXT)")
    conn.execute("INSERT INTO hidden VALUES ('x')")
    conn.commit()
    conn.close()
    return path


def test_mode_ro_alone_refuses_write(tmp_path):
    """Layer 1: the read-only open mode, with no pragma and no authorizer."""
    conn = sqlite3.connect(f"file:{_scratch(tmp_path)}?mode=ro", uri=True)
    with pytest.raises(sqlite3.OperationalError, match="readonly"):
        conn.execute("INSERT INTO t VALUES (2, 'b')")
    conn.close()


def test_query_only_alone_refuses_write(tmp_path):
    """Layer 2: query_only on a deliberately read-WRITE connection."""
    conn = sqlite3.connect(_scratch(tmp_path))
    conn.execute("PRAGMA query_only = ON")
    with pytest.raises(sqlite3.OperationalError, match="readonly"):
        conn.execute("INSERT INTO t VALUES (2, 'b')")
    conn.close()


def _authorizer():
    """The project authorizer, allow-listing t over the scratch schema."""
    return db._make_authorizer(frozenset({"t"}), frozenset({"t", "hidden"}))


def test_authorizer_alone_refuses_write(tmp_path):
    """Layer 3: the authorizer on a read-write connection, at prepare time."""
    conn = sqlite3.connect(_scratch(tmp_path))
    conn.set_authorizer(_authorizer())
    with pytest.raises(sqlite3.DatabaseError):
        conn.execute("INSERT INTO t VALUES (2, 'b')")
    # Reads of the allow-listed table still work.
    assert conn.execute("SELECT v FROM t").fetchone() == ("a",)
    conn.close()


def test_authorizer_blocks_unlisted_table_read(tmp_path):
    """Layer 3, read side: a table outside the allow-list cannot be read."""
    conn = sqlite3.connect(_scratch(tmp_path))
    conn.set_authorizer(_authorizer())
    with pytest.raises(sqlite3.DatabaseError):
        conn.execute("SELECT * FROM hidden")
    # Even counting its rows is denied.
    with pytest.raises(sqlite3.DatabaseError):
        conn.execute("SELECT count(*) FROM hidden")
    conn.close()

Add the code: tests/test_server.py

"""Integration tests against the seeded database.

Requires `make seed` first. If the database file is missing, the whole module
is skipped with a clear message rather than failing.
"""

from dataclasses import replace

import pytest
import sqlite3
from fastmcp import Client
from fastmcp.exceptions import ToolError

import db
import server


@pytest.fixture(scope="module", autouse=True)
def require_database():
    try:
        with db.connect() as conn:
            conn.execute("SELECT count(*) FROM customers")
    except (RuntimeError, sqlite3.Error) as exc:
        pytest.skip(f"database not ready (run `make seed`): {exc}")


async def test_list_and_describe():
    async with Client(server.mcp) as client:
        tables = (await client.call_tool("list_tables", {})).data
        assert set(tables) == {"customers", "orders"}
        cols = (await client.call_tool("describe_table", {"table": "customers"})).data
        assert {c["name"] for c in cols} == {"id", "name", "email", "city"}


async def test_query_table_filters_are_parameterized():
    async with Client(server.mcp) as client:
        result = (
            await client.call_tool(
                "query_table",
                {"table": "customers", "columns": ["id", "name"],
                 "filters": {"city": "London"}},
            )
        ).data
        # No ORDER BY in query_table, so compare as a set, not a sequence.
        assert {r["name"] for r in result["rows"]} == {"Ada Lovelace", "Alan Turing"}
        assert result["truncated"] is False


async def test_query_table_limit_is_exact():
    """An explicit limit returns exactly that many rows plus a truncated flag."""
    async with Client(server.mcp) as client:
        result = (
            await client.call_tool(
                "query_table", {"table": "orders", "columns": ["id"], "limit": 3}
            )
        ).data
        assert len(result["rows"]) == 3  # 5 orders exist, limit caps to 3
        assert result["truncated"] is True


async def test_query_table_rejects_unknown_table_and_column():
    async with Client(server.mcp) as client:
        with pytest.raises(ToolError):
            await client.call_tool("query_table", {"table": "secrets"})
        with pytest.raises(ToolError):
            await client.call_tool(
                "query_table", {"table": "customers", "columns": ["password"]}
            )


async def test_run_query_reads_and_caps_rows(monkeypatch):
    # Settings is a frozen dataclass, so rebind the module's `settings` to a
    # copy with a smaller cap rather than mutating the instance.
    monkeypatch.setattr(server, "settings", replace(server.settings, max_rows=3))
    async with Client(server.mcp) as client:
        result = (
            await client.call_tool(
                "run_query", {"query": "SELECT id FROM orders ORDER BY id"}
            )
        ).data
        assert [r["id"] for r in result["rows"]] == [1, 2, 3]
        assert result["truncated"] is True  # 5 orders, capped at 3


async def test_run_query_parameters_prevent_injection():
    async with Client(server.mcp) as client:
        injection = "London'; DROP TABLE customers; --"
        result = (
            await client.call_tool(
                "run_query",
                {"query": "SELECT count(*) AS n FROM customers WHERE city = ?",
                 "params": [injection]},
            )
        ).data
        assert result["rows"][0]["n"] == 0  # no such city; nothing dropped


async def test_multi_statement_is_rejected():
    async with Client(server.mcp) as client:
        with pytest.raises(ToolError):
            await client.call_tool(
                "run_query", {"query": "SELECT 1; INSERT INTO customers VALUES (9)"}
            )


async def test_secrets_table_is_unreadable():
    """The Python guard passes this SELECT; the engine authorizer kills it."""
    async with Client(server.mcp) as client:
        with pytest.raises(ToolError):
            await client.call_tool("run_query", {"query": "SELECT * FROM secrets"})


async def test_schema_of_hidden_tables_stays_hidden():
    """Neither the catalog nor the table_info pragma leaks unlisted schema."""
    async with Client(server.mcp) as client:
        with pytest.raises(ToolError):
            await client.call_tool("run_query", {"query": "SELECT * FROM sqlite_master"})
        with pytest.raises(ToolError):
            await client.call_tool(
                "run_query",
                {"query": "SELECT name FROM pragma_table_info('secrets')"},
            )


async def test_with_cte_select_is_allowed():
    """A WITH ... SELECT over allow-listed tables passes the authorizer."""
    async with Client(server.mcp) as client:
        result = (
            await client.call_tool(
                "run_query",
                {"query": "WITH t AS (SELECT count(*) AS n FROM orders) "
                          "SELECT n FROM t"},
            )
        ).data
        assert result["rows"][0]["n"] == 5


async def test_deadline_interrupts_runaway_query(monkeypatch):
    """An unbounded recursive CTE is interrupted by the progress handler."""
    monkeypatch.setattr(db, "settings", replace(db.settings, query_timeout_ms=200))
    async with Client(server.mcp) as client:
        with pytest.raises(ToolError, match="interrupted"):
            await client.call_tool(
                "run_query",
                {"query": "WITH RECURSIVE c(x) AS (SELECT 1 UNION ALL "
                          "SELECT x + 1 FROM c) SELECT count(*) FROM c"},
            )


async def test_direct_write_is_refused_by_engine():
    """Even bypassing every Python guard, the connection refuses a write."""
    with db.connect() as conn:
        with pytest.raises(sqlite3.Error):
            conn.execute(
                "INSERT INTO customers (id, name, email, city) "
                "VALUES (99, 'x', 'x@x', 'x')"
            )

Detailed breakdown

  • test_layers.py is the article’s thesis as code. Each test enables one layer on a scratch database and shows the write (or unlisted read) still fails. mode=ro and query_only fail at execution with attempt to write a readonly database; the authorizer fails earlier, at prepare, with not authorized — which is why its test asserts sqlite3.DatabaseError rather than matching the readonly message.
  • test_secrets_table_is_unreadable is the capability Postgres grants could only imitate. The statement is a syntactically perfect SELECT, the Python guard passes it, and the read still dies inside the engine because secrets is not in the allow-list.
  • test_schema_of_hidden_tables_stays_hidden closes the metadata side channel. sqlite_master would reveal every hidden table’s DDL and pragma_table_info('secrets') its columns; the authorizer denies both, so a hidden table leaks neither rows nor shape.
  • The two CTE tests pin down the authorizer’s trickiest behavior. A non-recursive CTE read never reaches the hook, but a recursive CTE’s name does arrive as a SQLITE_READ — which is exactly why the read rule is built from the stored-table snapshot instead of denying every unknown name. If the rule were written too aggressively, these are the tests that would catch it.
  • test_deadline_interrupts_runaway_query lowers the deadline to 200 ms and runs an unbounded recursive CTE. The progress handler interrupts it and the error message contains interrupted, which the test asserts so a denial for the wrong reason (say, an authorizer bug) cannot masquerade as a pass.
  • The limit-tweaking tests rebind, never mutate. Settings is frozen, so they monkeypatch.setattr the module’s settings name to a dataclasses.replace copy — server.settings for the row cap, db.settings for the deadline.

Run everything:

make test

The target reseeds the database first, then runs the suite; the deadline test takes about a fifth of a second by design.

Step 9: A demo script

A short script that drives the tools the way a client would, including one request the server must refuse.

Create the file

touch scripts/demo.py

Add the code: scripts/demo.py

"""Drive the tools in-memory against the seeded database and print results."""

import asyncio
import json

from fastmcp import Client
from fastmcp.exceptions import ToolError

import server


async def main() -> None:
    async with Client(server.mcp) as client:
        tables = (await client.call_tool("list_tables", {})).data
        print("tables:", tables)

        london = (
            await client.call_tool(
                "query_table",
                {"table": "customers", "columns": ["name", "city"],
                 "filters": {"city": "London"}},
            )
        ).data
        print("London customers:", json.dumps(london["rows"]))

        revenue = (
            await client.call_tool(
                "run_query",
                {"query": "SELECT sum(total_cents) AS paid_cents "
                          "FROM orders WHERE status = ?", "params": ["paid"]},
            )
        ).data
        print("paid revenue (cents):", revenue["rows"][0]["paid_cents"])

        try:
            await client.call_tool("run_query", {"query": "SELECT * FROM secrets"})
        except ToolError as exc:
            print("secrets table:", f"refused ({exc})")


if __name__ == "__main__":
    asyncio.run(main())

Detailed breakdown

  • The script uses the same in-memory Client as the tests, so it exercises the real tools and the real database without an external MCP host.
  • The revenue query binds status = ? and sums total_cents for paid orders: 1299 + 4500 + 15000 = 20799 cents with the seed data.
  • The last call is supposed to fail. Printing the refusal makes the demo show the security property, not just the happy path: the engine denies the read (access to secrets.id is prohibited) and the server surfaces a clean tool error. FastMCP also logs an Error calling tool 'run_query' line for it, which is expected noise, not a bug.

Run it:

make demo

Step 10: The Makefile

Wrap seeding, serving, testing, and the demo. Plain make prints help.

Create the file

touch Makefile

Add the code: Makefile

.DEFAULT_GOAL := help

.PHONY: help seed reset serve test demo clean

help:  ## Show this help screen
	@grep -E '^[a-zA-Z_-]+:.*?## .*$$' $(MAKEFILE_LIST) \
		| awk 'BEGIN {FS = ":.*?## "}; {printf "  \033[36m%-8s\033[0m %s\n", $$1, $$2}'

seed:  ## Create and seed db/app.db (idempotent; ends chmod 444)
	uv run python -m scripts.seed

reset:  ## Delete the database and its journals, then reseed
	rm -f db/app.db db/app.db-wal db/app.db-shm db/app.db-journal
	uv run python -m scripts.seed

serve: seed  ## Run the MCP server over HTTP in the foreground (Ctrl-C to stop)
	MCP_TRANSPORT=http uv run python server.py

test: seed  ## Reseed the database, then run the test suite
	uv run pytest -v

demo: seed  ## Run the demo script against the database
	uv run python -m scripts.demo

clean:  ## Remove Python caches (keeps the database)
	rm -rf .pytest_cache __pycache__ tests/__pycache__ scripts/__pycache__

Detailed breakdown

  • seed is safe to run any time: the script deletes and rebuilds the file deterministically, so serve, test, and demo depend on it and always run against known data. There is no up/down pair because there is no server process to manage — the “database lifecycle” is one file.
  • reset also clears journal sidecars, covering the case where an experiment outside the tutorial left a -wal or -journal file behind.
  • Scripts run as modules (python -m scripts.seed) so the project root is on sys.path and import config / import server resolve. Running python scripts/seed.py directly would put scripts/ on the path instead and break those imports.
  • make with no arguments prints the help screen, per the house rule; the awk pattern reads the ## comments off each target.

Try it against a real client

Point any MCP client at the server over stdio. For Claude Desktop, add this to claude_desktop_config.json (adjust the path):

{
  "mcpServers": {
    "sqlite-readonly": {
      "command": "uv",
      "args": ["run", "python", "server.py"],
      "cwd": "/absolute/path/to/readonly-sqlite-mcp-server-macos"
    }
  }
}

Run make seed once before the client launches the server. Ask the client to “list the tables,” “describe the orders table,” or “show total paid revenue,” and watch it use list_tables, describe_table, and run_query in turn. Then ask it to “read the secrets table” and watch the refusal come back from the engine, not from a prompt.

Recap

You built a SQLite MCP server that is read-only by construction, with the enforcement moved into the places an embedded database actually has: the open mode, the session pragma, the authorizer, a deadline, and the file’s own permission bits. The authorizer is the piece worth remembering — a default-deny hook checked at prepare time that refuses writes and reads of anything you did not allow-list, which is a stronger property against raw SQL than the Postgres version’s grants. Any single layer failing still leaves the others standing.

Next improvements

  • Per-table column masking — extend the authorizer’s SQLITE_READ check to deny by (table, column), hiding customers.email while leaving the rest of the table readable. Returning SQLITE_IGNORE instead of SQLITE_DENY makes the column read as NULL rather than failing the query.
  • immutable=1 in the URI for databases that truly never change while served, which lets SQLite skip locking entirely. Only safe when no writer exists; mode=ro is the honest default.
  • An audit log of every statement and its caller, written to a separate writable database or a log file, so read access is reviewable.
  • Snapshot serving — copy the database bytes at startup (sqlite3.Connection.serialize/deserialize) and serve from memory, so even OS-level file changes cannot affect a running server.