MCP SQLite Server (Nur-Lesen)

Ein produktionsreifer Model Context Protocol-Server, der KI-Agenten sicheren, schreibgeschützten Zugriff auf eine SQLite-Datenbank (shop.db) bietet. Erstellt mit dem offiziellen mcp Python SDK unter Verwendung des stdio-Transports.

Funktionen

• 3 MCP-Tools: list_tables, describe_table, query_database

• Mehrschichtige Schreibschutz-Sicherheit: SQLite-URI im Nur-Lesen-Modus + PRAGMA query_only + SQL-Validator + EXPLAIN-Opcode-Inspektion

• Abfragevalidierung: Lehnt INSERT/UPDATE/DELETE/DROP/ALTER/CREATE/REPLACE/TRUNCATE/ATTACH/DETACH, Mehrfachanweisungen (;), SQL-Kommentare (--, /* */) und modifizierende PRAGMA-Anweisungen ab – ohne Fehlalarme bei String-Literalen

• Paginierung: Standard-Zeilenlimit (100), limit/offset-Parameter, Flag für abgeschnittene Ausgabe

• Nur-Stderr-Protokollierung: Alle Logs/Tracebacks gehen an sys.stderr; stdout ist ausschließlich für JSON-RPC reserviert

• Vollständige Typannotationen: mypy --strict sauber

• TDD: 105 Tests, die Sicherheit, DB-Schicht, MCP-Tools, 8 Benchmark-Abfragen und den Stderr-Schutz abdecken

Related MCP server: sqlite-mcp-server

Schnellstart

Voraussetzungen

• Python 3.10+

• Eine SQLite-Datenbankdatei (Standard: ./shop.db)

Lokale Einrichtung

python -m venv .venv
source .venv/bin/activate
pip install -e ".[dev]"

Konfiguration

Kopieren Sie .env.example und legen Sie den Datenbankpfad fest:

cp .env.example .env
# Edit DATABASE_PATH to point to your SQLite file

Oder setzen Sie die Umgebungsvariable direkt:

export DATABASE_PATH=/abs/path/to/shop.db

Server ausführen

python -m mcp_server.server

Der Server kommuniziert über stdin/stdout mit dem MCP-stdio-Transport. Sie interagieren nicht direkt mit ihm – ein MCP-Client (z. B. Claude Desktop, Ihr KI-Agent) verbindet sich mit ihm.

MCP-Client-Konfigurationen

Standard-Python

Fügen Sie dies zu Ihrer MCP-Client-Konfiguration hinzu (z. B. claude_desktop_config.json von Claude Desktop):

{
  "mcpServers": {
    "sqlite-shop": {
      "command": "python",
      "args": ["-m", "mcp_server.server"],
      "env": {
        "DATABASE_PATH": "/abs/path/to/shop.db"
      }
    }
  }
}

Docker

Zuerst das Image erstellen:

docker build -t mcp-shop:latest .

Dann Ihren MCP-Client konfigurieren:

{
  "mcpServers": {
    "sqlite-shop": {
      "command": "docker",
      "args": [
        "run", "-i", "--rm",
        "-v", "/abs/path/to/shop.db:/app/shop.db",
        "-e", "DATABASE_PATH=/app/shop.db",
        "mcp-shop:latest"
      ]
    }
  }
}

Docker Compose

docker compose up -d

Tools

list_tables

Listet alle Benutzertabellen und -sichten in der Datenbank auf (schließt interne sqlite_*-Tabellen aus).

Parameter: keine

Rückgabe:

{
  "tables": ["customers", "orders", "order_items", "products"],
  "count": 4
}

describe_table

Beschreibt das Schema einer Tabelle: Spalten, Fremdschlüssel, Zeilenanzahl und die CREATE-Anweisung.

Parameter:

• table (Zeichenkette, erforderlich): Name der zu beschreibenden Tabelle.

Rückgabe:

{
  "table": "customers",
  "columns": [
    {"cid": 0, "name": "id", "type": "INTEGER", "notnull": 0, "default": null, "pk": 1},
    {"cid": 1, "name": "first_name", "type": "TEXT", "notnull": 1, "default": null, "pk": 0}
  ],
  "foreign_keys": [],
  "row_count": 150,
  "sql": "CREATE TABLE customers (...)"
}

query_database

Führt eine schreibgeschützte SQL-Abfrage mit Paginierungsunterstützung aus.

Parameter:

• sql (Zeichenkette, erforderlich): Eine einzelne schreibgeschützte SQL-Anweisung (SELECT, WITH, EXPLAIN oder schreibgeschütztes PRAGMA).

• limit (Ganzzahl, optional): Maximale Anzahl zurückzugebender Zeilen. Standard: 100. Maximum: 1000.

• offset (Ganzzahl, optional): Anzahl der zu überspringenden Zeilen. Standard: 0.

Rückgabe:

{
  "columns": ["id", "first_name"],
  "rows": [{"id": 1, "first_name": "Alice"}, {"id": 2, "first_name": "Bob"}],
  "row_count": 2,
  "truncated": false,
  "limit": 100,
  "offset": 0
}

Wenn truncated true ist, sind weitere Zeilen verfügbar – erhöhen Sie offset, um die nächste Seite abzurufen.

Sicherheit

Der Server implementiert eine mehrschichtige Verteidigung, um schreibgeschützten Zugriff zu gewährleisten:

Ebene 1: SQLite-Verbindung (URI im Nur-Lesen-Modus)

Die Datenbank wird mit file:<pfad>?mode=ro geöffnet, was Schreibvorgänge auf der SQLite-Engine-Ebene verhindert. Zusätzlich wird bei jeder Verbindung PRAGMA query_only = ON gesetzt.

Ebene 2: SQL-Abfragevalidator (security.py)

Bevor eine Abfrage SQLite erreicht, durchläuft sie einen mehrstufigen Validator:

1. String-Literal-Entfernung: String-Literale ('...', "...") werden durch Platzhalter ersetzt, damit Schlüsselwörter in Daten (z. B. ein Produkt namens "Deleted Item") keine Fehlalarme auslösen.

2. Kommentarerkennung: SQL-Kommentare (--, /* */) werden abgelehnt, um kommentarbasierte Umgehungen zu verhindern.

3. Ablehnung von Mehrfachanweisungen: Jedes Semikolon (;) wird abgelehnt, wodurch gestapelte Abfragen verhindert werden.

4. Schlüsselwortanalyse: Das erste echte Anweisungsschlüsselwort muss SELECT, WITH, EXPLAIN oder PRAGMA sein. Destruktive Schlüsselwörter (INSERT, UPDATE, DELETE, DROP, ALTER, CREATE, REPLACE, TRUNCATE, ATTACH, DETACH, VACUUM usw.) werden blockiert.

5. PRAGMA-Validierung: Schreibgeschützte PRAGMAs (table_info, database_list usw.) sind erlaubt. Jede PRAGMA mit einer Zuweisung (=) oder in der Blocklist für mutierende PRAGMAs (journal_mode, synchronous, foreign_keys usw.) wird abgelehnt.

Ebene 3: EXPLAIN-Opcode-Inspektion

Als letzte Verteidigung wird die Abfrage über SQLites eigenen Parser mit EXPLAIN <abfrage> ausgeführt. Der resultierende Opcode-Stream wird auf Schreib-Opcodes (OpenWrite, Insert, Delete, Create, Drop usw.) und Schreibtransaktions-Flags untersucht. Wenn solche gefunden werden, wird die Abfrage abgelehnt.

Ebene 4: Bereinigte Fehlermeldungen

Alle an den Client zurückgegebenen Fehler werden bereinigt – Dateisystempfade und interne Details werden entfernt, um Informationslecks zu verhindern.

Tests

Tests verwenden nur temporäre oder In-Memory-Datenbanken – niemals die Produktions-shop.db.

# Run all tests
python -m pytest

# Run with verbose output
python -m pytest -v

# Run a specific test file
python -m pytest tests/test_security.py

Testabdeckung

Statische Analyse

# Type checking
python -m mypy

# Linting
python -m ruff check src/ tests/

Projektstruktur

.
├── .env.example          # Environment variable template
├── Dockerfile            # Docker containerization
├── docker-compose.yml    # Docker Compose config
├── pyproject.toml        # Package config, deps, tool settings
├── README.md             # This file
├── shop.db               # The SQLite database (not included in tests)
├── src/mcp_server/
│   ├── __init__.py
│   ├── config.py         # Configuration (DATABASE_PATH, limits, URI builder)
│   ├── db.py             # Read-only Database class with introspection + query
│   ├── security.py       # SQL validator (multi-layer defense-in-depth)
│   ├── server.py         # MCP server entrypoint (stdio transport)
│   ├── tools.py          # MCP tool definitions and handlers
│   └── py.typed          # PEP 561 marker
└── tests/
    ├── __init__.py
    ├── test_db.py        # Database layer + benchmark tests
    ├── test_security.py  # Query validator tests
    └── test_server.py    # MCP server/tool tests

Benchmark-Aufgaben

Die Tools des Servers ermöglichen einem KI-Agenten die Durchführung dieser analytischen Aufgaben (durch Tests gegen eine kontrollierte Fixture-Datenbank validiert):

1. Tabellenentdeckung: list_tables + describe_table – alle Tabellen auflisten und Schemas beschreiben.

2. Gefilterte Zählung: query_database mit SELECT COUNT(*) FROM customers WHERE country = 'Germany'.

3. Länderaggregation: SELECT country, COUNT(*) ... GROUP BY country ORDER BY ... DESC LIMIT 1.

4. Kunden-LTV: Join customers + orders, SUM(total_amount), nach Gesamtsumme sortieren.

5. Produktleistung: Join order_items + products, nach Menge und Umsatz aggregieren, LIMIT 5.

6. Kategorieaggregation: order_items → products → category durchlaufen, Umsatz aggregieren, LIMIT 3.

7. Datumsfilterung: SUM(total_amount) WHERE substr(order_date,1,4) = '2025'.

8. Bestellaggregation: Join customers + orders, COUNT(o.id), nach Anzahl sortieren.

Konfiguration

Lizenz

Dieses Projekt wird wie besehen zu Demonstrationszwecken bereitgestellt.