MCP-DB Infrastructure

Connection Pool und Protokoll Logger.

DatabaseConnection

Singleton Connection Pool für Datenbankzugriffe.

"""Database Connection Pool"""
import os
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):
        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."""
        conn = cls.get_pool().get_connection()
        try:
            # Datenbank wechseln (USE statement statt property)
            cursor = conn.cursor()
            cursor.execute(f"USE {database}")

            # Query Timeout setzen (MariaDB verwendet 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 (in Sekunden, nicht ms)

Hinweis: MySQL verwendet max_execution_time in Millisekunden, MariaDB verwendet max_statement_time in Sekunden.

ProtokollLogger

Logging aller Operationen nach ki_protokoll.mcp_log.

"""Logging in ki_protokoll"""
import sys
import mysql.connector.pooling as pooling
from domain.log_contract import LogEntry
from config import Config

class ProtokollLogger:
    """Schreibt in ki_protokoll.mcp_log - SRP: Nur Logging"""

    def __init__(self):
        self._pool = None

    def _get_pool(self):
        if self._pool is None:
            self._pool = pooling.MySQLConnectionPool(
                pool_name="log_pool",
                pool_size=2,
                host=Config.LOG_DB_HOST,
                database=Config.LOG_DB_NAME,
                user=Config.LOG_DB_USER,
                password=Config.LOG_DB_PASSWORD,
                charset="utf8mb4"
            )
        return self._pool

    def log(self, entry: LogEntry) -> None:
        """Schreibt LogEntry in Datenbank."""
        try:
            conn = self._get_pool().get_connection()
            cursor = conn.cursor()
            cursor.execute(
                """INSERT INTO mcp_log
                   (timestamp, client_name, request, status,
                    duration_ms, error_message)
                   VALUES (%s, %s, %s, %s, %s, %s)""",
                (entry.timestamp, entry.client_name, entry.request,
                 entry.status, entry.duration_ms, entry.error_message)
            )
            conn.commit()
            cursor.close()
            conn.close()
        except Exception as e:
            # Fail-Safe: Exception werfen statt silent failure
            error_msg = f"CRITICAL: Failed to write to mcp_log: {str(e)}"
            print(error_msg, file=sys.stderr)
            raise RuntimeError(error_msg) from e

Logger-Pool

Fail-Safe Verhalten

Bei Logging-Fehlern:

• Fehlermeldung nach stderr
• RuntimeError wird geworfen
• Kein silent failure

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;

Nutzung

# DatabaseConnection
with DatabaseConnection.get_connection("ki_protokoll") as conn:
    cursor = conn.cursor(dictionary=True)
    cursor.execute("SELECT * FROM mcp_log LIMIT 10")
    rows = cursor.fetchall()
    cursor.close()

# ProtokollLogger
logger = ProtokollLogger()
logger.log(LogEntry(
    request="SELECT * FROM mcp_log",
    status="success",
    duration_ms=45
))

Verwandte Kapitel

• Sicherheit - Separate DB-User
• Contracts - LogEntry