# Registrierung in Claude Code
claude mcp add mcp-db \
--transport stdio \
-- /var/www/mcp-servers/mcp_db/venv/bin/python \
/var/www/mcp-servers/mcp_db/server.py
# Status prüfen
claude mcp list
Verfügbare Tools (10)
Lese-Operationen
Tool
Beschreibung
Parameter
db_select
SELECT-Abfrage ausführen
query, database, max_rows, params
db_schema
Tabellen mit Metadaten anzeigen
database
db_stats
Logging-Statistiken aus mcp_log
limit
db_databases
Erlaubte Datenbanken auflisten
-
db_tables
Tabellen einer DB auflisten
database, include_row_count
db_describe
Tabellenstruktur anzeigen
table, database, show_create
Schreib-Operationen
Tool
Beschreibung
Parameter
db_insert
Datensatz einfügen
table, data, database
db_update
Datensätze aktualisieren
table, data, where, database
db_delete
Datensätze löschen
table, where, database, limit
DDL-Operationen
Tool
Beschreibung
Parameter
db_execute
DDL ausführen (ALTER, CREATE, DROP, TRUNCATE)
statement, database, params
Datenbank-Struktur
Datenbank
Zweck
Beispiel-Tabellen
ki_dev
Development/Infrastructure
tasks, contracts, dokumentation, mcp_log
ki_content
Content/User-facing
chat_sessions, chat_messages, content, personas
Sicherheitsübersicht
Maßnahme
Status
Betrifft
Prepared Statements
Implementiert
Alle Tools
Identifier-Validierung
Regex ^[a-zA-Z0-9_]+$
INSERT, UPDATE, DELETE
Mandatory WHERE
Erzwungen
UPDATE, DELETE
Default LIMIT 100
Aktiv
DELETE
DDL Statement-Validierung
ExecuteValidator
db_execute
Database Allowlist
ki_dev, ki_content
Alle Tools
Keyword Blocklist (15)
Aktiv
db_select
Query Timeout
30 Sekunden
db_select
Row Limit
Max 100
db_select
Logging
Jede Operation in mcp_log
Alle Tools
Blocking-Hook
PreToolUse für Bash
mysql/mariadb CLI
Changelog
Version
Datum
Änderungen
2.0.0
2025-12-20
+7 neue Tools: db_execute, db_describe, db_databases, db_tables, db_insert, db_update, db_delete
# Einfache Abfrage
db_select("SELECT * FROM mcp_log ORDER BY timestamp DESC LIMIT 10", database="ki_dev")
# Mit Prepared Statements
db_select(
query="SELECT * FROM tasks WHERE status = %s",
database="ki_dev",
params=["pending"]
)
# Datenbank für Queries (mcp_readonly)
DB_HOST=localhost
DB_USER=mcp_readonly
DB_PASSWORD=SECURE_PASSWORD_HERE
# Datenbank für Logging (mcp_logger)
LOG_DB_HOST=localhost
LOG_DB_NAME=ki_protokoll
LOG_DB_USER=mcp_logger
LOG_DB_PASSWORD=DIFFERENT_SECURE_PASSWORD_HERE
5. Datenbank vorbereiten
mcp_log Tabelle
CREATE TABLE IF NOT EXISTS mcp_log (
id INT AUTO_INCREMENT PRIMARY KEY,
timestamp DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
client_name VARCHAR(50) NOT NULL DEFAULT 'mcp-db',
request TEXT,
status ENUM('success', 'error', 'denied') NOT NULL,
duration_ms INT DEFAULT 0,
error_message TEXT,
INDEX idx_timestamp (timestamp),
INDEX idx_status (status)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
DB-User erstellen
-- mcp_readonly
CREATE USER IF NOT EXISTS 'mcp_readonly'@'localhost'
IDENTIFIED BY 'SECURE_PASSWORD_HERE';
GRANT SELECT ON ki_protokoll.mcp_log TO 'mcp_readonly'@'localhost';
GRANT SELECT ON ki_protokoll.ki_eintraege TO 'mcp_readonly'@'localhost';
GRANT SELECT ON ki_protokoll.ki_kategorien TO 'mcp_readonly'@'localhost';
GRANT SELECT ON ki_protokoll.ki_tags TO 'mcp_readonly'@'localhost';
GRANT SELECT ON ki_protokoll.ki_settings TO 'mcp_readonly'@'localhost';
GRANT SELECT ON information_schema.TABLES TO 'mcp_readonly'@'localhost';
GRANT SELECT ON ki_system.* TO 'mcp_readonly'@'localhost';
-- mcp_logger
CREATE USER IF NOT EXISTS 'mcp_logger'@'localhost'
IDENTIFIED BY 'DIFFERENT_SECURE_PASSWORD_HERE';
GRANT INSERT ON ki_protokoll.mcp_log TO 'mcp_logger'@'localhost';
FLUSH PRIVILEGES;
6. Server testen
# Manuell starten
cd /opt/mcp-servers/mcp-db
source venv/bin/activate
python server.py
# Bei Erfolg: Server wartet auf stdio Input
# Strg+C zum Beenden
Registrierung nur für ein bestimmtes Projekt (z.B. /var/www):
cd /var/www
claude mcp add mcp-db --transport stdio \
-- /opt/mcp-servers/mcp-db/venv/bin/python \
/opt/mcp-servers/mcp-db/server.py
Hinweis: Im Projekt-Scope wird die Konfiguration in /root/.claude.json unter dem entsprechenden Projekt gespeichert, nicht in .claude/settings.json im Projektverzeichnis.
8. Registrierung prüfen
# Liste aller MCP Server
claude mcp list
# Details zu mcp-db
claude mcp get mcp-db
# In Claude Code Session
/mcp
9. Tools testen
# In Claude Code:
> db_select("SELECT * FROM mcp_log LIMIT 5")
> db_schema()
> db_stats(limit=10)
Troubleshooting
Server startet nicht
# Log prüfen
tail -f /tmp/mcp_db_server.log
# Dependencies prüfen
pip list | grep mcp
# Python Version
python --version # Muss 3.10+ sein
Verbindungsfehler
# DB-Verbindung mit MCP-DB testen (empfohlen)
db_select("SELECT 1")
# Alternativ: Direkter Test (nur für Debugging)
mysql -u mcp_readonly -p -e "SELECT 1"
# .env prüfen
cat .env | grep -v PASSWORD
Permission Denied
# Berechtigungen prüfen
ls -la .env # Muss 600 sein
# User prüfen
mysql -u mcp_readonly -p -e "SHOW GRANTS"
Systemd Service (Optional)
MCP-DB läuft normalerweise on-demand (von Claude Code gestartet). Für Monitoring oder Debugging kann ein Systemd Service verwendet werden.
Service installiert
Der Service ist bereits unter /etc/systemd/system/mcp-db.service installiert und aktiviert.
Verwendung
# Status prüfen (zeigt on-demand oder systemd)
/opt/scripts/mcp-db-status.sh
# Service manuell starten (nicht empfohlen)
systemctl start mcp-db
# Service stoppen
systemctl stop mcp-db
# Logs ansehen
journalctl -u mcp-db -f
Wichtig
Empfohlen: On-demand Betrieb (Claude startet MCP automatisch)
Service nur für: Debugging, permanente Logs, externe Monitoring
Service läuft nicht standardmäßig, nur nach systemctl start mcp-db
Updates
# Dependencies aktualisieren
cd /opt/mcp-servers/mcp-db
source venv/bin/activate
pip install --upgrade mcp[cli] mysql-connector-python
# Nach Updates: Service neu laden (falls aktiv)
systemctl daemon-reload
systemctl restart mcp-db # nur falls Service läuft
Deinstallation
# Aus Claude Code entfernen
claude mcp remove mcp-db
# Verzeichnis löschen (optional)
rm -rf /opt/mcp-servers/mcp-db
# DB-User löschen (optional)
DROP USER 'mcp_readonly'@'localhost';
DROP USER 'mcp_logger'@'localhost';
SQL-Validierung nach dem Single Responsibility Principle.
QueryValidator
Validiert SELECT-Queries für das db_select Tool.
"""SRP: Separate Validierungslogik"""
from typing import Tuple
import re
from config import Config
class QueryValidator:
"""Validiert SQL Queries - SRP: Nur Validierung"""
@staticmethod
def validate_query(query: str, database: str, max_rows: int) -> Tuple[bool, str]:
"""
Validiert eine Query gegen alle Sicherheitsregeln.
Returns:
(is_valid, error_message)
"""
# Basis-Validierung
if not query or len(query) < 1:
return False, "Query must not be empty"
if len(query) > Config.MAX_QUERY_LENGTH:
return False, f"Query must be max {Config.MAX_QUERY_LENGTH} chars"
# Nur SELECT erlaubt
query_upper = query.strip().upper()
if not query_upper.startswith("SELECT"):
return False, "Only SELECT queries allowed"
# Dangerous Keyword Blocklist
for keyword in Config.BLOCKED_KEYWORDS:
pattern = r'\b' + re.escape(keyword) + r'\b'
if re.search(pattern, query_upper):
return False, f"Blocked keyword detected: {keyword}"
# Database Allowlist
if database not in Config.ALLOWED_DATABASES:
return False, f"Database '{database}' not allowed"
# Max Rows prüfen
if max_rows < 1 or max_rows > Config.MAX_ROWS:
return False, f"max_rows must be 1-{Config.MAX_ROWS}"
# Table Allowlist
from_tables = QueryValidator._extract_table_names(query_upper)
for table in from_tables:
if "INFORMATION_SCHEMA" in table:
continue
table_clean = table.split('.')[-1]
if table_clean not in [t.upper() for t in Config.ALLOWED_TABLES]:
return False, f"Table '{table}' not allowed"
return True, ""
Validierungsschritte (QueryValidator)
#
Prüfung
Fehler bei
1
Query nicht leer
Leere Query
2
Query-Länge
> 2000 Zeichen
3
SELECT-Only
Nicht mit SELECT beginnend
4
Keyword-Blocklist
Blockiertes Keyword gefunden
5
Database-Allowlist
Datenbank nicht erlaubt
6
Max-Rows
< 1 oder > 100
7
Table-Allowlist
Tabelle nicht erlaubt
ExecuteValidator
Validiert DDL-Statements für das db_execute Tool.
"""Execute Validator - Validierung für DDL/DML Statements."""
import re
from config import Config
class ExecuteValidator:
"""Validiert DDL/DML Statements gegen Sicherheitsregeln."""
# Erlaubte Statement-Typen
ALLOWED_STATEMENTS: list[str] = ["ALTER", "CREATE", "DROP", "TRUNCATE", "SET"]
# Erlaubte SET-Variablen (Whitelist)
ALLOWED_SET_VARS: list[str] = ["FOREIGN_KEY_CHECKS"]
# Verbotene Patterns (kritisch!)
FORBIDDEN_PATTERNS: list[str] = [
r"\bDROP\s+DATABASE\b",
r"\bDROP\s+SCHEMA\b",
r"\bCREATE\s+DATABASE\b",
r"\bCREATE\s+SCHEMA\b",
r"\bGRANT\b",
r"\bREVOKE\b",
r"\bSHUTDOWN\b",
]
# Verbotene Datenbanken (System-DBs)
FORBIDDEN_DATABASES: list[str] = [
"information_schema",
"mysql",
"performance_schema",
"sys",
]
@classmethod
def validate_statement(cls, statement: str, database: str) -> tuple[bool, str]:
"""
Validiert ein DDL/DML Statement.
Returns:
Tuple (is_valid, error_message)
"""
if not statement or len(statement.strip()) < 5:
return False, "Statement must not be empty"
if len(statement) > Config.MAX_QUERY_LENGTH:
return False, f"Statement exceeds max length"
statement_upper = statement.strip().upper()
# Prüfe ob Statement mit erlaubtem Keyword beginnt
statement_type = cls._get_statement_type(statement_upper)
if statement_type is None:
return False, f"Statement type not allowed. Allowed: {', '.join(cls.ALLOWED_STATEMENTS)}"
# Bei SET: nur erlaubte Variablen
if statement_type == "SET":
allowed = any(var in statement_upper for var in cls.ALLOWED_SET_VARS)
if not allowed:
return False, f"SET variable not allowed. Allowed: {', '.join(cls.ALLOWED_SET_VARS)}"
# Prüfe verbotene Patterns
for pattern in cls.FORBIDDEN_PATTERNS:
if re.search(pattern, statement_upper):
return False, f"Forbidden pattern detected"
# Prüfe Datenbank
if database not in Config.ALLOWED_DATABASES:
return False, f"Database '{database}' not allowed"
# Prüfe ob Statement gegen System-DB zielt
for forbidden_db in cls.FORBIDDEN_DATABASES:
if re.search(rf"\b{forbidden_db}\b", statement_upper, re.IGNORECASE):
return False, f"Operations on '{forbidden_db}' are forbidden"
return True, ""
Validierungsschritte (ExecuteValidator)
#
Prüfung
Fehler bei
1
Statement nicht leer
Leeres Statement
2
Statement-Länge
> 2000 Zeichen
3
Statement-Typ
Nicht ALTER/CREATE/DROP/TRUNCATE/SET
4
SET-Variablen
Nicht in ALLOWED_SET_VARS
5
Forbidden Patterns
DROP DATABASE, GRANT, etc.
6
Database-Allowlist
Datenbank nicht erlaubt
7
System-DB-Schutz
mysql, information_schema, etc.
Erlaubte DDL-Statements
Typ
Beispiel
Status
ALTER TABLE
ALTER TABLE tasks ADD COLUMN priority INT
Erlaubt
CREATE TABLE
CREATE TABLE temp_data (...)
Erlaubt
CREATE INDEX
CREATE INDEX idx_name ON table(col)
Erlaubt
DROP TABLE
DROP TABLE temp_data
Erlaubt
DROP INDEX
DROP INDEX idx_name ON table
Erlaubt
TRUNCATE
TRUNCATE TABLE temp_data
Erlaubt
SET
SET FOREIGN_KEY_CHECKS = 0
Erlaubt
Verbotene Patterns
Pattern
Beschreibung
DROP DATABASE
Datenbank-Löschung verboten
DROP SCHEMA
Schema-Löschung verboten
CREATE DATABASE
Datenbank-Erstellung verboten
CREATE SCHEMA
Schema-Erstellung verboten
GRANT
Rechte-Vergabe verboten
REVOKE
Rechte-Entzug verboten
SHUTDOWN
Server-Shutdown verboten
Keyword-Erkennung
Keywords werden mit Word Boundaries erkannt:
# Regex-Pattern für Keyword "DROP"
pattern = r'\bDROP\b'
# Blockiert:
"SELECT * FROM t; DROP TABLE t" # DROP erkannt
# Nicht blockiert (kein vollständiges Wort):
"SELECT dropdown FROM t" # DROP nicht erkannt
Testfälle
# QueryValidator - Muss DENIED returnen:
"SELECT * FROM mcp_log; DROP TABLE mcp_log" # DROP
"SELECT LOAD_FILE('/etc/passwd')" # LOAD_FILE
"SELECT * FROM users" # users nicht erlaubt
"SELECT SLEEP(10)" # SLEEP
# ExecuteValidator - Muss DENIED returnen:
"DROP DATABASE ki_dev" # Forbidden Pattern
"GRANT ALL ON ki_dev.* TO user" # GRANT verboten
"ALTER TABLE mysql.user ADD COLUMN x INT" # System-DB
# ExecuteValidator - Muss SUCCESS returnen:
"ALTER TABLE tasks ADD COLUMN priority INT" # Erlaubt
"CREATE INDEX idx_status ON tasks(status)" # Erlaubt
"SET FOREIGN_KEY_CHECKS = 0" # Erlaubte Variable
Database Connection Pool für sichere Datenbankzugriffe.
DatabaseConnection
Singleton Connection Pool für Datenbankzugriffe.
"""Database Connection Pool"""
from contextlib import contextmanager
import mysql.connector
from mysql.connector import pooling
from config import Config
class DatabaseConnection:
"""Singleton Connection Pool - SRP: Nur Verbindungsmanagement"""
_pool = None
@classmethod
def get_pool(cls):
"""Erstellt oder gibt existierenden Connection Pool zurück"""
if cls._pool is None:
cls._pool = pooling.MySQLConnectionPool(
pool_name="mcp_pool",
pool_size=5,
host=Config.DB_HOST,
user=Config.DB_USER,
password=Config.DB_PASSWORD,
charset="utf8mb4",
connection_timeout=10,
autocommit=True,
)
return cls._pool
@classmethod
@contextmanager
def get_connection(cls, database: str):
"""
Context Manager für DB Connection.
Args:
database: Datenbankname (bereits validiert)
"""
conn = cls.get_pool().get_connection()
try:
# Datenbank wechseln (USE statement)
cursor = conn.cursor()
cursor.execute(f"USE {database}")
# Query Timeout setzen (MariaDB: max_statement_time in Sekunden)
try:
cursor.execute(
f"SET SESSION max_statement_time = {Config.QUERY_TIMEOUT_SEC}"
)
except Exception:
pass # Falls Variable nicht unterstützt wird
cursor.close()
yield conn
finally:
conn.close()
Pool-Konfiguration
Parameter
Wert
Beschreibung
pool_size
5
Maximale Verbindungen
connection_timeout
10
Verbindungs-Timeout in Sekunden
autocommit
True
Automatisches Commit
charset
utf8mb4
Unicode-Support
Query Timeout
Bei jeder Verbindung wird max_statement_time gesetzt (MariaDB-spezifisch):
SET SESSION max_statement_time = 30 # 30 Sekunden
Hinweis: MySQL verwendet max_execution_time in Millisekunden, MariaDB verwendet max_statement_time in Sekunden.
Logging
Alle MCP-DB Operationen werden direkt in ki_dev.mcp_log protokolliert. Das Logging erfolgt inline in den Tool-Implementierungen, nicht über eine separate Logger-Klasse.
Logging-Ablauf
Tool wird aufgerufen (z.B. db_select)
Query wird validiert
Query wird ausgeführt
Ergebnis wird in mcp_log geschrieben
mcp_log Tabelle
CREATE TABLE IF NOT EXISTS mcp_log (
id INT AUTO_INCREMENT PRIMARY KEY,
timestamp DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
client_name VARCHAR(50) NOT NULL DEFAULT 'mcp-db',
request TEXT,
status ENUM('success', 'error', 'denied') NOT NULL,
duration_ms INT DEFAULT 0,
error_message TEXT,
INDEX idx_timestamp (timestamp),
INDEX idx_status (status)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
Log-Felder
Feld
Typ
Beschreibung
timestamp
DATETIME
Zeitstempel der Operation
client_name
VARCHAR(50)
Client-Identifikation (mcp-db)
request
TEXT
Query-Text (gekürzt auf 200 Zeichen)
status
ENUM
success, error, denied
duration_ms
INT
Ausführungsdauer in Millisekunden
error_message
TEXT
Fehlerdetails (bei Fehlern)
Nutzung
# DatabaseConnection mit Context Manager
with DatabaseConnection.get_connection("ki_dev") as conn:
cursor = conn.cursor(dictionary=True)
cursor.execute("SELECT * FROM mcp_log LIMIT 10")
rows = cursor.fetchall()
cursor.close()
# Log-Eintrag schreiben (inline in Tool)
cursor.execute(
"""INSERT INTO mcp_log
(client_name, request, status, duration_ms, error_message)
VALUES (%s, %s, %s, %s, %s)""",
("mcp-db", query[:200], "success", duration_ms, None)
)
Vollständiger Funktionstest aller MCP-DB Features.
Testübersicht
Kategorie
Tests
Status
db_select/basic
6
100%
db_select/params
3
100%
db_select/db_allow
4
100%
db_select/table_allow
8
100%
db_select/blocklist
17
100%
db_select/limits
8
100%
db_schema
3
100%
db_stats
5
100%
Testmatrix: db_select
Basis-Funktionalität
Test
Query
Erwartung
Status
Simple SELECT
SELECT 1 as test
success
OK
SELECT with columns
SELECT id, status FROM mcp_log
success
OK
SELECT with WHERE
SELECT * FROM mcp_log WHERE status = 'success'
success
OK
SELECT with ORDER BY
SELECT * FROM mcp_log ORDER BY id DESC
success
OK
SELECT with COUNT
SELECT COUNT(*) as total FROM mcp_log
success
OK
SELECT with GROUP BY
SELECT status, COUNT(*) FROM mcp_log GROUP BY status
success
OK
Prepared Statements
Test
Query
Params
Status
Single param
SELECT * FROM mcp_log WHERE status = %s
["success"]
OK
Multiple params
SELECT * FROM mcp_log WHERE status = %s AND id > %s
["success", 0]
OK
LIKE param
SELECT * FROM mcp_log WHERE request LIKE %s
["%SELECT%"]
OK
Database Allowlist
Database
Erwartung
Status
ki_dev
allowed
OK
ki_content
allowed
OK
mysql
denied
OK
information_schema
denied
OK
Table Allowlist
Tabelle
Database
Erwartung
Status
mcp_log
ki_dev
allowed
OK
tasks
ki_dev
allowed
OK
documents
ki_content
allowed
OK
chunks
ki_content
allowed
OK
chat_sessions
ki_content
allowed
OK
users
-
denied
OK
secrets
-
denied
OK
mysql.user
-
denied
OK
Keyword Blocklist (15 Keywords)
Kategorie
Keyword
Test-Query
Status
DML
DROP
SELECT ...; DROP TABLE x
BLOCKED
DML
DELETE
SELECT ...; DELETE FROM x
BLOCKED
DML
INSERT
SELECT ...; INSERT INTO x
BLOCKED
DML
UPDATE
SELECT ...; UPDATE x SET
BLOCKED
DDL
TRUNCATE
SELECT ...; TRUNCATE TABLE
BLOCKED
DDL
ALTER
SELECT ...; ALTER TABLE
BLOCKED
DDL
CREATE
SELECT ...; CREATE TABLE
BLOCKED
DDL
RENAME
SELECT ...; RENAME TABLE
BLOCKED
DCL
GRANT
SELECT ...; GRANT ALL
BLOCKED
DCL
REVOKE
SELECT ...; REVOKE ALL
BLOCKED
File
LOAD_FILE
SELECT LOAD_FILE('/etc/passwd')
BLOCKED
File
INTO OUTFILE
SELECT ... INTO OUTFILE
BLOCKED
File
INTO DUMPFILE
SELECT ... INTO DUMPFILE
BLOCKED
DoS
BENCHMARK
SELECT BENCHMARK(...)
BLOCKED
DoS
SLEEP
SELECT SLEEP(10)
BLOCKED
Word Boundary Tests
Test
Query
Erwartung
Status
dropdown != DROP
SELECT 'dropdown' as word
success
OK
updated_at != UPDATE
SELECT 'updated_at' as col
success
OK
Limits
Test
Parameter
Erwartung
Status
max_rows=1
max_rows=1
success
OK
max_rows=50
max_rows=50
success
OK
max_rows=100
max_rows=100
success (max)
OK
max_rows=0
max_rows=0
denied
OK
max_rows=101
max_rows=101
denied (>100)
OK
Empty query
query=""
denied
OK
Non-SELECT
query="SHOW TABLES"
denied
OK
Query > 2000 chars
query=long_string
denied
OK
Testmatrix: db_schema
Test
Database
Erwartung
Status
ki_dev schema
ki_dev
tables[]
OK
ki_content schema
ki_content
tables[]
OK
mysql denied
mysql
error
OK
Testmatrix: db_stats
Test
Parameter
Erwartung
Status
Default limit
limit=50
logs[]
OK
Custom limit
limit=5
logs[] (5)
OK
Max limit
limit=100
logs[] (max 100)
OK
Clamped min
limit=0
logs[] (clamped to 1)
OK
Clamped max
limit=200
logs[] (clamped to 100)
OK
Konfiguration
Parameter
Wert
Erlaubte Datenbanken
ki_dev, ki_content
Erlaubte Tabellen (ki_dev)
23
Erlaubte Tabellen (ki_content)
35
Blockierte Keywords
15
Max Query Length
2000 Zeichen
Max Rows
100
Query Timeout
30 Sekunden
Test ausführen
# Comprehensive Test Suite
cd /var/www/mcp-servers/mcp_db
source venv/bin/activate
python comprehensive_test.py
# Erwartete Ausgabe:
# STATUS: ALLE TESTS BESTANDEN