MCP-DB Infrastructure

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

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

1. Tool wird aufgerufen (z.B. db_select)
2. Query wird validiert
3. Query wird ausgeführt
4. 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

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)
)

Verzeichnisstruktur

/var/www/mcp-servers/mcp_db/infrastructure/
├── __init__.py
└── db_connection.py    # DatabaseConnection Klasse

Verwandte Kapitel

• Sicherheit - Separate DB-User
• Contracts - Datenstrukturen
• Konfiguration - Environment Variables