MCP-DB Tools

10 MCP Tools für vollständige Datenbankoperationen.

Übersicht

db_select

Führt SELECT-Abfragen aus.

Signatur

def db_select(
    query: str,
    database: str = "ki_protokoll",
    max_rows: int = 100,
    params: list | None = None
) -> dict

Parameter

Response

{
    "status": "success",
    "data": [{"id": 1, "name": "Test"}, ...],
    "row_count": 10,
    "execution_ms": 45
}

Beispiele

# Einfache Abfrage
db_select("SELECT * FROM mcp_log ORDER BY timestamp DESC LIMIT 10")

# Mit Prepared Statements
db_select(
    query="SELECT * FROM tasks WHERE status = %s",
    database="ki_dev",
    params=["pending"]
)

db_schema

Zeigt Tabellen einer Datenbank mit Metadaten.

Signatur

def db_schema(database: str = "ki_protokoll") -> dict

Response

{
    "tables": [
        {"TABLE_NAME": "mcp_log", "TABLE_ROWS": 1234, "CREATE_TIME": "2025-01-15"},
        ...
    ],
    "database": "ki_protokoll"
}

db_stats

Zeigt Logging-Statistiken aus mcp_log.

Signatur

def db_stats(limit: int = 50) -> dict

Response

{
    "logs": [
        {"id": 123, "timestamp": "2025-01-15 14:30:00", "status": "success", ...},
        ...
    ],
    "count": 50
}

db_databases

Listet alle erlaubten Datenbanken auf.

Signatur

def db_databases() -> dict

Response

{
    "status": "success",
    "databases": [
        {"name": "ki_dev", "table_count": 15},
        {"name": "ki_content", "table_count": 8}
    ]
}

Beispiel

# Erlaubte Datenbanken anzeigen
db_databases()

db_tables

Listet alle Tabellen einer Datenbank auf.

Signatur

def db_tables(
    database: str = "ki_dev",
    include_row_count: bool = False
) -> dict

Parameter

Response

{
    "status": "success",
    "database": "ki_dev",
    "tables": ["tasks", "contracts", "mcp_log", ...],
    "count": 15
}

Beispiel

# Tabellen mit Row-Count
db_tables(database="ki_content", include_row_count=True)

db_describe

Zeigt Tabellenstruktur an (DESCRIBE oder SHOW CREATE TABLE).

Signatur

def db_describe(
    table: str,
    database: str = "ki_dev",
    show_create: bool = False
) -> dict

Parameter

Response (DESCRIBE)

{
    "status": "success",
    "table": "chat_sessions",
    "database": "ki_content",
    "columns": [
        {"Field": "id", "Type": "int(11)", "Null": "NO", "Key": "PRI", "Default": null},
        {"Field": "title", "Type": "varchar(255)", "Null": "YES", ...},
        ...
    ]
}

Response (SHOW CREATE)

{
    "status": "success",
    "table": "chat_sessions",
    "create_statement": "CREATE TABLE `chat_sessions` (...)"
}

Beispiele

# Spalten anzeigen
db_describe(table="tasks", database="ki_dev")

# CREATE Statement anzeigen
db_describe(table="tasks", database="ki_dev", show_create=True)

db_insert

Fügt einen Datensatz ein.

Signatur

def db_insert(
    table: str,
    data: dict,
    database: str = "ki_dev"
) -> dict

Parameter

Response

{
    "status": "success",
    "table": "tasks",
    "inserted_id": 42,
    "execution_ms": 12
}

Beispiel

db_insert(
    table="tasks",
    data={
        "title": "Neue Aufgabe",
        "description": "Beschreibung hier",
        "status": "pending",
        "type": "ai_task"
    },
    database="ki_dev"
)

Sicherheit

• Tabellen- und Spaltennamen werden per Regex validiert: ^[a-zA-Z0-9_]+$
• Werte werden als Prepared Statement Parameter übergeben
• Nur erlaubte Datenbanken (ki_dev, ki_content)

db_update

Aktualisiert Datensätze. WHERE ist Pflicht!

Signatur

def db_update(
    table: str,
    data: dict,
    where: dict,
    database: str = "ki_dev"
) -> dict

Parameter

Response

{
    "status": "success",
    "table": "tasks",
    "affected_rows": 1,
    "execution_ms": 8
}

Beispiel

db_update(
    table="tasks",
    data={"status": "completed"},
    where={"id": 42},
    database="ki_dev"
)

Sicherheit

db_delete

Löscht Datensätze. WHERE ist Pflicht! Default LIMIT: 100

Signatur

def db_delete(
    table: str,
    where: dict,
    database: str = "ki_dev",
    limit: int | None = None
) -> dict

Parameter

Response

{
    "status": "success",
    "table": "mcp_log",
    "deleted_rows": 5,
    "limit_applied": 100,
    "execution_ms": 15
}

Beispiel

# Alte Logs löschen
db_delete(
    table="mcp_log",
    where={"status": "error"},
    database="ki_protokoll",
    limit=50
)

Sicherheit

db_execute

Führt DDL-Statements aus (ALTER, CREATE, DROP, TRUNCATE).

Signatur

def db_execute(
    statement: str,
    database: str = "ki_dev",
    params: list | None = None
) -> dict

Parameter

Erlaubte Statements

• ALTER TABLE - Tabellenstruktur ändern
• CREATE TABLE - Tabelle erstellen
• CREATE INDEX - Index erstellen
• DROP TABLE - Tabelle löschen
• DROP INDEX - Index löschen
• TRUNCATE TABLE - Tabelle leeren

Response

{
    "status": "success",
    "statement_type": "ALTER TABLE",
    "affected_rows": 0,
    "execution_ms": 120
}

Beispiele

# Spalte hinzufügen
db_execute(
    statement="ALTER TABLE chat_sessions ADD COLUMN temperature FLOAT DEFAULT 0.7",
    database="ki_content"
)

# Index erstellen
db_execute(
    statement="CREATE INDEX idx_status ON tasks(status)",
    database="ki_dev"
)

Sicherheit

Fehlerbehandlung

Verwandte Kapitel

• Contracts - Request/Response Types
• Validators - Validierungslogik
• Sicherheit - Schutzmaßnahmen
• Cheat-Sheet - Schnellreferenz