shop-mcp

Ein lokaler, schreibgeschützter MCP-Server (Model Context Protocol), der es einem KI-Agenten ermöglicht, eine SQLite-E-Commerce-Datenbank (shop.db) zu analysieren – Kunden, Produkte, Bestellungen und Bestellpositionen – über den stdio-Transport. Kein HTTP-Server, kein separater Datenbankprozess: Der Server öffnet shop.db direkt und stellt zwei kleine, allgemein einsetzbare Tools bereit, mit denen ein Agent das Schema erkunden und eigene analytische SQL-Abfragen ausführen kann.

Erstellt mit dem offiziellen Python-MCP-SDK (mcp auf PyPI).

Projektstruktur

mcp-sql/
├── server.py                    # the MCP server (stdio transport)
├── shop.db                      # SQLite database (not modified by this project)
├── requirements.txt
├── .env.example
├── mcp-config.example.json
├── tests/
│   ├── conftest.py
│   ├── test_server.py           # unit tests (call tool functions directly)
│   └── test_stdio_integration.py# protocol-level test (spawns server.py over stdio)
└── README.md

Related MCP server: shop-db MCP Server

Datenbankschema (wie tatsächlich in shop.db vorgefunden)

customers(id PK, first_name, last_name, email UNIQUE, phone, created_at)
products(id PK, name, category, price, stock_quantity, created_at)
orders(id PK, customer_id -> customers.id, order_date, status, total_amount)
order_items(id PK, order_id -> orders.id, product_id -> products.id, quantity, unit_price)

orders.status ist auf folgende Werte beschränkt: new, processing, shipped, completed, cancelled. products.category hat derzeit 5 verschiedene Werte. Fremdschlüssel: orders.customer_id → customers.id, order_items.order_id → orders.id, order_items.product_id → products.id. Der Server leitet all dies zur Abfragezeit aus der Live-Datenbank ab (über sqlite_master / PRAGMA table_info / PRAGMA foreign_key_list) – nichts davon ist fest codiert. Wenn shop.db also durch eine andere Datei mit einem anderen Schema ersetzt wird, spiegelt get_database_schema dies automatisch wider.

Bekannte Datenmerkmale der bereitgestellten shop.db: customers hat keine Spalte country, daher können Fragen wie „Kunden aus Deutschland“ nicht beantwortet werden – das Schema-Tool macht dies erkennbar, und query_database gibt einen klaren Fehler no such column: country zurück, anstatt zu raten. Alle 750 derzeit in der Datenbank vorhandenen Bestellungen sind auf 2026 datiert (keine im Jahr 2025), sodass eine Abfrage nach „Umsatz 2025“ korrekt 0/null zurückgibt, keinen Fehler.

Installation

cd mcp-sql
python3 -m venv .venv
source .venv/bin/activate        # on Windows: .venv\Scripts\activate
pip install -r requirements.txt

Konfiguration

Der Datenbankpfad ist im Quellcode nie fest codiert. Er wird wie folgt aufgelöst:

1. die Umgebungsvariable SHOP_DB_PATH, falls gesetzt;

2. andernfalls shop.db neben server.py.

Kopieren Sie .env.example nach .env und bearbeiten Sie es, wenn Sie den Server auf eine andere Datenbankdatei verweisen möchten (Sie müssen es selbst in Ihre Shell/Agenten-Startumgebung laden, z. B. export $(cat .env | xargs), oder einfach SHOP_DB_PATH direkt setzen):

cp .env.example .env
# edit .env, or simply:
export SHOP_DB_PATH=/absolute/path/to/shop.db

Ausführen

source .venv/bin/activate
python server.py

Der Prozess spricht MCP über stdio und wartet auf einen Client – er wirkt „hängend“ ohne Ausgabe, was erwartet ist: Verbinden Sie einen MCP-Client (einen KI-Agenten oder mcp-inspector, siehe unten), anstatt ihn eigenständig in einem Terminal auszuführen.

Schnelle manuelle Überprüfung mit dem offiziellen MCP Inspector (keine Installation erforderlich):

npx @modelcontextprotocol/inspector --cli .venv/bin/python server.py --method tools/list

Mit einem KI-Agenten verbinden

Die meisten MCP-kompatiblen Clients (Claude Desktop, Claude Code usw.) lesen einen JSON-Konfigurationsblock wie mcp-config.example.json:

{
  "mcpServers": {
    "shop-mcp": {
      "command": "/absolute/path/to/mcp-sql/.venv/bin/python",
      "args": ["/absolute/path/to/mcp-sql/server.py"],
      "env": {
        "SHOP_DB_PATH": "/absolute/path/to/mcp-sql/shop.db"
      }
    }
  }
}

Hinweise:

• Verwenden Sie den absoluten Pfad zum Python-Interpreter der venv (wie oben), damit das Paket mcp gefunden wird, ohne die venv manuell zu aktivieren; ein nacktes python3 funktioniert ebenfalls, wenn mcp in der Umgebung installiert ist, auf die es aufgelöst wird.

• SHOP_DB_PATH ist optional – weglassen, um die mitgelieferte shop.db zu verwenden.

• Absolute Pfade gehören in diese Konfigurationsdatei, die von demjenigen bereitgestellt wird, der den Server verbindet – niemals in server.py selbst.

• Die client-spezifische Platzierung dieses Blocks variiert (z. B. verwendet Claude Desktop claude_desktop_config.json mit derselben mcpServers-Form; andere Clients möchten möglicherweise nur das innere Objekt {"command": ..., "args": ..., "env": ...}). Überprüfen Sie die Dokumentation Ihres Clients, wo die Datei liegt.

Testen

source .venv/bin/activate
python -m pytest tests/ -v

Dies führt 48 Tests aus, darunter:

• Schema-Erkennung (Tabellen, Spalten, PK/FK, Beziehungen, Zeilenanzahlen);

• SELECT, JOIN, WHERE, GROUP BY, ORDER BY, Aggregatfunktionen (COUNT/SUM/AVG/MIN/MAX), Unterabfragen, ein sicheres WITH ... SELECT-CTE und Datumsfilterung (strftime);

• Begrenzung der Zeilenanzahl und Offset-basierte Paginierung;

• verständliche Fehlerbehandlung für ungültiges SQL, unbekannte Tabellen/Spalten, eine leere Abfrage und eine fehlende Datenbankdatei;

• Schreibschutz-Sicherheit: Jeder in der Aufgabenstellung aufgeführte Anweisungstyp (DELETE, UPDATE, DROP, CREATE, INSERT, plus ALTER, REPLACE, TRUNCATE, ATTACH, DETACH, VACUUM, REINDEX, ein destruktives PRAGMA, ein gestapeltes SELECT 1; DROP TABLE ... und ein als CTE getarntes WITH x AS (...) DELETE ...) wird abgelehnt, und die Zeilenanzahlen sowie der SHA-256-Hash der Datenbankdatei werden danach als unverändert bestätigt;

• tests/test_stdio_integration.py startet server.py als echten Subprozess und steuert ihn über das tatsächliche MCP-Client-SDK über stdio (initialize → list_tools → call_tool), anstatt Python-Funktionen direkt aufzurufen – dies ist derselbe Pfad, den ein echter Agent verwendet.

MCP-Tools

get_database_schema()

Keine Parameter. Rufen Sie dies zuerst auf, wenn Sie die genauen Tabellen-/Spaltennamen nicht bereits kennen – raten Sie nicht. Gibt pro Tabelle zurück: row_count, columns (Name, SQLite-Typ, not_null, default_value, is_primary_key), primary_key, foreign_keys (Spalte, referenzierte Tabelle/Spalte, ON DELETE/ON UPDATE) sowie einige sample_rows, damit der Agent echte Datumsformate, Statuswerte, Preisgrößen usw. sehen kann. Eine Liste relationships auf oberster Ebene gibt Zeichenfolgen der Form table.column -> other_table.column an, die aus den Live-Fremdschlüsseln abgeleitet werden.

query_database(sql, limit=100, offset=0)

Führt eine einzelne schreibgeschützte SQL-Anweisung aus (SELECT oder WITH ... SELECT) und gibt {columns, rows, row_count, limit, offset, truncated, total_matching_rows} zurück. Unterstützt JOIN, WHERE, GROUP BY, ORDER BY, Aggregatfunktionen, Unterabfragen und CTEs. limit wird auf 1..500 begrenzt (Standard 100); verwenden Sie offset, um durch größere Ergebnisse zu blättern. total_matching_rows und truncated teilen dem Aufrufer mit, ob die aktuelle Seite das gesamte Ergebnis ist oder ob weitere Daten abgerufen werden müssen. Fehler (falsche Syntax, unbekannte Tabelle/Spalte oder ein abgelehnter Schreibversuch) werden als kurze, spezifische Meldung ausgegeben – niemals als roher Python-Traceback.

Sicherheit: Wie der Schreibschutz durchgesetzt wird

Die Aufgabenstellung verlangt ausdrücklich, sich nicht auf eine einzelne Regex-/Schlüsselwortprüfung zu verlassen. Daher schichtet dieser Server vier unabhängige Verteidigungslinien – verifiziert in tests/test_server.py:

1. Schreibgeschützter Dateihandle auf Betriebssystemebene. Die SQLite-Datei wird mit der URI file:<path>?mode=ro geöffnet. SQLite selbst verweigert dann jeden Schreibzugriff (OperationalError: attempt to write a readonly database), unabhängig davon, welches SQL ausgeführt wird – dies gilt selbst dann, wenn jede der folgenden Prüfungen einen Fehler aufweist.

2. PRAGMA query_only = ON wird auf jeder Verbindung als zweite, unabhängige SQLite-Ebene gegen Schreibzugriffe gesetzt.

3. Ein sqlite3-Autorisierungs-Callback (Connection.set_authorizer) erlaubt nur die Aktionen SELECT / READ / FUNCTION / RECURSIVE auf SQLite-Engine-Ebene und verweigert alles andere – INSERT, UPDATE, DELETE, DROP, ALTER, CREATE, REPLACE, TRUNCATE, ATTACH, DETACH, VACUUM, REINDEX, PRAGMA, Transaktionen usw. Dies läuft auf der geparsten Anweisung, sodass es auch den klassischen CTE-Bypass WITH x AS (SELECT 1) DELETE FROM ... abfängt, den eine naive Textprüfung „muss mit SELECT beginnen“ übersehen würde.

4. Anweisungsform-Prüfungen in server.py: Der übermittelte Text muss mit SELECT/WITH beginnen (schnelle, verständliche Ablehnung vor dem Zugriff auf SQLite), und jede Abfrage wird als SELECT * FROM (<query>) LIMIT :limit OFFSET :offset ausgeführt – für das Parsen ist genau eine Anweisung erforderlich, sodass ein gestapeltes SELECT 1; DROP TABLE customers zu einem einfachen SQL-Syntaxfehler wird, anstatt zwei Anweisungen auszuführen.

Da Ebene 1 (mode=ro) von SQLite/Betriebssystem unabhängig von der eigenen Logik des Servers durchgesetzt wird, kann shop.db über diesen Server nicht verändert werden, selbst wenn in den Ebenen 2–4 ein Fehler existierte.

Bekannte Einschränkungen

• customers hat in der bereitgestellten shop.db keine Spalte country/Standort, sodass Fragen wie „Kunden aus Deutschland“ aus diesen Daten nicht beantwortet werden können – das Schema-Tool macht dies sichtbar, anstatt dass der Server eine Spalte erfindet.

• Alle Bestellungen in den bereitgestellten Daten sind auf 2026 datiert; eine Umsatzabfrage für 2025 gibt korrekt 0 zurück, keinen Fehler.

• total_matching_rows in query_database wird mit einem zweiten COUNT(*) berechnet, das dieselbe Abfrage umschließt; bei sehr teuren Abfragen verdoppelt dies den Aufwand ungefähr. Angesichts der Größe dieser Datenbank (Hunderte bis wenige Tausend Zeilen pro Tabelle) ist dies kein praktisches Problem.