MCP Server Refactoring Plan

Version: 1.2
Erstellt: 2025-12-28
Aktualisiert: 2025-12-28 (nach Supervision Runde 2)
Status: Final
Task: #507

1. Executive Summary

Ist-Zustand

• 5 MCP-Server: mcp-db, mcp-tasks, mcp-contracts, mcp-docs, mcp-code
• 91 Python-Dateien (ohne venv)
• 9.248 LOC gesamt
• Signifikante Code-Duplikation (~25% redundanter Code)
• Inkonsistente Patterns zwischen Servern

Ziel

Konsolidierung zu einer modularen, wartbaren Architektur unter Einhaltung von:

• DRY (Don't Repeat Yourself)
• SRP (Single Responsibility Principle)
• SOLID Principles
• Clean Architecture
• KISS (Keep It Simple, Stupid)
• YAGNI (You Aren't Gonna Need It)

Erwartete Verbesserungen

• ~30% weniger Code durch Elimination von Duplikaten
• Einheitliche Patterns für alle Server
• Verbesserte Wartbarkeit durch Modularisierung
• Schnellere Feature-Entwicklung durch shared Components

2. Analyse der Code-Duplikation

2.1 Kritische Duplikate (Sofort beheben)

db_connection.py (4 Dateien, ~240 LOC dupliziert)

Problem: 3 von 4 Dateien sind nahezu identisch (mcp-tasks, mcp-contracts, mcp-docs).

Lösung: Zentralisieren in shared/infrastructure/simple_db_connection.py

WICHTIG: Diese Klasse heißt bewusst SimpleDbConnection, nicht DatabaseConnection. Sie bietet kein Pooling und ist nicht für High-Throughput geeignet. Für mcp-db (100+ Queries/Minute) bleibt die eigene Pooling-Implementierung.

# shared/infrastructure/simple_db_connection.py
"""
Einfache DB-Verbindung ohne Pooling.

NICHT geeignet für:
- High-Throughput (> 50 Queries/Minute)
- Connection Pooling Requirements
- mcp-db Server

Für diese Fälle: Eigene Implementierung mit mysql.connector.pooling
"""
from contextlib import contextmanager
from typing import Generator

import pymysql
from pymysql.connections import Connection

from shared.config_base import AppDatabaseConfig, LogDatabaseConfig


class SimpleDbConnection:
    """
    Einfache Datenbankverbindung ohne Pooling.
    
    Für Server mit geringem Query-Volumen (< 50/Minute).
    """
    
    @classmethod
    @contextmanager
    def get_connection(
        cls, 
        config: AppDatabaseConfig,
        database: str | None = None,
        autocommit: bool = False
    ) -> Generator[Connection, None, None]:
        """
        Context Manager für App-DB Connection.
        
        ACHTUNG: Keine weiteren Parameter hinzufügen!
        Bei Bedarf für mehr Logik: Neue Methode erstellen.
        
        Args:
            config: Server-Konfiguration mit DB-Credentials
            database: Optional - überschreibt config.DB_NAME
            autocommit: True für Logging, False für Transaktionen
        """
        conn = None
        db_name = database or getattr(config, 'DB_NAME', None)
        
        try:
            conn = pymysql.connect(
                host=config.DB_HOST,
                port=getattr(config, 'DB_PORT', 3306),
                user=config.DB_USER,
                password=config.DB_PASSWORD,
                database=db_name,
                charset="utf8mb4",
                cursorclass=pymysql.cursors.DictCursor,
                autocommit=autocommit,
            )
            yield conn
            if not autocommit:
                conn.commit()
        except Exception:
            if conn and not autocommit:
                conn.rollback()
            raise
        finally:
            if conn:
                conn.close()
    
    @classmethod
    @contextmanager
    def get_log_connection(
        cls, 
        config: LogDatabaseConfig
    ) -> Generator[Connection, None, None]:
        """Separate Verbindung für Logging (autocommit=True)."""
        conn = None
        try:
            conn = pymysql.connect(
                host=config.LOG_DB_HOST,
                user=config.LOG_DB_USER,
                password=config.LOG_DB_PASSWORD,
                database=config.LOG_DB_NAME,
                charset="utf8mb4",
                cursorclass=pymysql.cursors.DictCursor,
                autocommit=True,
            )
            yield conn
        finally:
            if conn:
                conn.close()

protokoll_logger.py (4 Dateien, ~240 LOC dupliziert)

Lösung: Zentralisieren mit strikter Clean Architecture Trennung

# shared/domain/log_entry.py
"""
Domain Entity für Log-Einträge.

REINE DOMAIN - keine Infrastructure-Abhängigkeiten!
"""
from dataclasses import dataclass
from datetime import datetime
from typing import Optional


@dataclass(frozen=True)
class LogEntry:
    """Standardisierter Log-Eintrag für alle MCP-Server."""
    timestamp: datetime
    client_name: str
    request: str
    status: str
    duration_ms: int
    error_message: Optional[str] = None
    tool_name: Optional[str] = None
    context_id: Optional[int] = None  # task_id, contract_id, etc.

# shared/infrastructure/protokoll_logger.py
"""
Infrastructure Logger für mcp_log Tabelle.

Importiert LogEntry aus Domain - definiert ihn NICHT selbst.
"""
import logging
import sys
from typing import TYPE_CHECKING

from shared.domain.log_entry import LogEntry  # Import aus Domain!
from shared.infrastructure.simple_db_connection import SimpleDbConnection

if TYPE_CHECKING:
    from shared.config_base import LogDatabaseConfig


class ProtokollLogger:
    """Fail-Safe Logger für mcp_log Tabelle."""
    
    def __init__(self, client_name: str, config: "LogDatabaseConfig"):
        self.client_name = client_name
        self.config = config
        self._logger = logging.getLogger(f"mcp.{client_name}")
    
    def log(self, entry: LogEntry) -> None:
        """Schreibt Log-Eintrag. Fehler gehen nur zu stderr."""
        try:
            with SimpleDbConnection.get_log_connection(self.config) as conn:
                with conn.cursor() as cursor:
                    request_str = self._format_request(entry)
                    cursor.execute(
                        """INSERT INTO mcp_log
                           (timestamp, client_name, request, status, duration_ms, error_message)
                           VALUES (%s, %s, %s, %s, %s, %s)""",
                        (
                            entry.timestamp,
                            self.client_name,
                            request_str[:500],
                            entry.status,
                            entry.duration_ms,
                            entry.error_message[:500] if entry.error_message else None,
                        )
                    )
        except Exception as e:
            print(f"CRITICAL: {self.client_name} log failed: {e}", file=sys.stderr)
    
    def _format_request(self, entry: LogEntry) -> str:
        parts = []
        if entry.tool_name:
            parts.append(f"[{entry.tool_name}]")
        if entry.context_id:
            parts.append(f"id={entry.context_id}")
        parts.append(entry.request[:400] if entry.request else "")
        return " ".join(parts)


# === Registry mit Test-Support ===

_logger_instances: dict[str, ProtokollLogger] = {}


def get_logger(client_name: str, config: "LogDatabaseConfig") -> ProtokollLogger:
    """Singleton-Factory für Logger."""
    if client_name not in _logger_instances:
        _logger_instances[client_name] = ProtokollLogger(client_name, config)
    return _logger_instances[client_name]


def clear_logger_registry() -> None:
    """
    Setzt die Logger-Registry zurück.
    
    NUR FÜR TESTS! Ermöglicht isolierte Test-Ausführung.
    """
    _logger_instances.clear()

config.py - Strikte Statische Konfiguration

# shared/config_base.py
"""
Basis-Konfiguration für alle MCP-Server.

WICHTIG: Diese Klasse wird NICHT instanziiert!
Alle Zugriffe erfolgen über Klassenattribute.

Falsch: config = BaseConfig()
Richtig: host = BaseConfig.DB_HOST
"""
import os
from typing import ClassVar, Protocol, runtime_checkable


@runtime_checkable
class AppDatabaseConfig(Protocol):
    """Protocol für App-Datenbank Konfiguration."""
    DB_HOST: str
    DB_PORT: int
    DB_USER: str
    DB_PASSWORD: str
    DB_NAME: str


@runtime_checkable
class LogDatabaseConfig(Protocol):
    """Protocol für Log-Datenbank Konfiguration."""
    LOG_DB_HOST: str
    LOG_DB_NAME: str
    LOG_DB_USER: str
    LOG_DB_PASSWORD: str


class BaseConfig:
    """
    Statische Konfiguration - NICHT INSTANZIIEREN!
    
    Verwendung:
        from shared.config_base import BaseConfig
        host = BaseConfig.DB_HOST  # Korrekt
        
        config = BaseConfig()  # FALSCH!
    """
    
    # App-Datenbank
    DB_HOST: ClassVar[str] = os.getenv("DB_HOST", "localhost")
    DB_PORT: ClassVar[int] = int(os.getenv("DB_PORT", "3306"))
    DB_USER: ClassVar[str] = os.getenv("DB_USER", "root")
    DB_PASSWORD: ClassVar[str] = os.getenv("DB_PASSWORD", "")
    DB_NAME: ClassVar[str] = os.getenv("DB_NAME", "ki_dev")
    
    # Log-Datenbank
    LOG_DB_HOST: ClassVar[str] = os.getenv("LOG_DB_HOST", "localhost")
    LOG_DB_NAME: ClassVar[str] = os.getenv("LOG_DB_NAME", "ki_dev")
    LOG_DB_USER: ClassVar[str] = os.getenv("LOG_DB_USER", "mcp_logger")
    LOG_DB_PASSWORD: ClassVar[str] = os.getenv("LOG_DB_PASSWORD", "")
    
    def __init__(self):
        raise TypeError("BaseConfig ist statisch und darf nicht instanziiert werden")

3. SRP-Verletzungen

3.1 ContractValidator (409 Zeilen) - Pragmatische Teilzerlegung

Empfohlenes Refactoring:

validators/
├── contract_validator.py      # Orchestrierung (< 150 Zeilen)
├── scope_resolver.py          # Pfad-Auflösung + Glob (< 100 Zeilen)
└── rule_evaluator.py          # Domain Service für Regelauswertung

HINWEIS: rule_evaluator.py ist ein Domain Service, kein Repository! Es führt keine CRUD-Operationen durch, sondern Business-Logik. Keine Repository-Nomenklatur für Validatoren verwenden.

Aktuelle Verantwortlichkeiten von rule_evaluator:

1. Regel-Parsing
2. Regel-Auswertung
3. Regel-Aggregation

Status: Als Übergang akzeptabel, da reines In-Memory-Domain-Modul ohne IO.

Grenze: Sobald eine Regel externen Zustand berührt → weitere Zerlegung nötig.

4. Inkonsistenzen

4.1 Datenbank-Bibliotheken - Hybride Strategie

Trade-off explizit dokumentiert:

• pymysql ohne Pooling: ~50-100ms Connection-Overhead pro Request
• Für mcp-tasks/contracts/docs/code: Akzeptabel bei < 50 Queries/Minute
• Für mcp-db: Inakzeptabel, daher eigenes Pooling

4.2 Zwei DB-Abstraktionen (konzedierte Redundanz)

Semantische Abgrenzung:

• SimpleDbConnection = Explizit "einfach", nicht für Hochlast
• mcp-db behält eigene Implementierung mit MySQLConnectionPool

5. Ziel-Architektur

5.1 Verzeichnisstruktur (nach Package-Umbenennung)

/var/www/mcp-servers/
├── shared/                           # Gemeinsame Komponenten
│   ├── __init__.py
│   ├── config_base.py               # Statische BaseConfig + Protocols
│   ├── constants.py                 # ✓ Existiert bereits
│   ├── server_factory.py            # Server Creation Factory
│   ├── domain/
│   │   ├── __init__.py
│   │   └── log_entry.py             # LogEntry Dataclass (REINE DOMAIN)
│   └── infrastructure/
│       ├── __init__.py
│       ├── simple_db_connection.py  # Ohne Pooling, nicht für Hochlast
│       └── protokoll_logger.py      # Importiert LogEntry aus domain/
│
├── mcp_db/                          # Umbenannt von mcp-db!
│   ├── server.py
│   ├── config.py
│   ├── infrastructure/
│   │   └── db_connection.py         # EIGENE Version mit Pooling
│   └── tools/
│
├── mcp_tasks/                       # Umbenannt von mcp-tasks!
│   ├── server.py
│   ├── config.py
│   ├── domain/
│   ├── infrastructure/
│   │   └── task_repository.py
│   └── tools/
│
├── mcp_contracts/                   # Umbenannt!
│   ├── server.py
│   ├── config.py
│   ├── validators/
│   │   ├── contract_validator.py
│   │   ├── scope_resolver.py
│   │   └── rule_evaluator.py        # Domain Service, KEIN Repository!
│   └── tools/
│
├── mcp_docs/                        # Umbenannt!
│   └── ...
│
└── mcp_code/                        # Umbenannt!
    └── ...

5.2 Clean Architecture Schichten

┌─────────────────────────────────────────────────────────┐
│                    SHARED/DOMAIN                         │
│  ┌───────────────────────────────────────────────────┐  │
│  │ log_entry.py (reine Dataclass, keine Deps)       │  │
│  └───────────────────────────────────────────────────┘  │
└─────────────────────────────────────────────────────────┘
                          ▲
                          │ importiert
┌─────────────────────────────────────────────────────────┐
│                 SHARED/INFRASTRUCTURE                    │
│  ┌───────────────────────────────────────────────────┐  │
│  │ simple_db_connection.py, protokoll_logger.py     │  │
│  └───────────────────────────────────────────────────┘  │
└─────────────────────────────────────────────────────────┘
                          ▲
                          │ importiert
┌─────────────────────────────────────────────────────────┐
│                    SERVER-LAYER                          │
│  ┌─────────┐  ┌───────────┐  ┌─────────┐  ┌─────────┐  │
│  │ mcp_db  │  │ mcp_tasks │  │mcp_docs │  │mcp_code │  │
│  └─────────┘  └───────────┘  └─────────┘  └─────────┘  │
└─────────────────────────────────────────────────────────┘

6. Implementierungsplan mit Quality Gates

Phase 0: Package-Umbenennung (VORAUSSETZUNG)

Dauer: 0.5 Tage

Gate:

• [ ] Keine sys.path.insert mehr notwendig
• [ ] pyproject.toml konsistent
• [ ] Alle Server starten und antworten

ACHTUNG: Phase 0 ist nicht optional! Ohne korrekte Package-Namen ist das gesamte Refactoring auf Sand gebaut.

Phase 1: Shared Foundation

Dauer: 2 Tage

Phase 1 Abnahmekriterien:

• [ ] ruff check shared/ = 0 Fehler
• [ ] mypy shared/ = 0 Fehler
• [ ] Domain Coverage: 100%
• [ ] Infrastructure Coverage: >= 90%
• [ ] clear_logger_registry() existiert und funktioniert
• [ ] Kein sys.path.insert in shared/

Phase 2: Server Migration

Phase 2a: mcp_docs (einfachstes)

Dauer: 0.5 Tage

Gate:

• [ ] Nutzt SimpleDbConnection
• [ ] Nutzt ProtokollLogger aus shared/
• [ ] Kein sys.path.insert
• [ ] LOC-Reduktion: >= 50 Zeilen

Phase 2b-2e: Weitere Server

(analog, siehe v1.1)

Universelle Regel für alle Server:

• [ ] Kein sys.path.insert nach Migration
• [ ] Nur relative Imports oder Package-Imports

Phase 3: ContractValidator Refactoring

7. Verbote und Grenzen

7.1 Absolute Verbote nach Phase 1

7.2 Komplexitätsgrenzen

8. Qualitätssicherung

Coverage-Strategie (differenziert)

Test-Isolation

# In jedem Test-Setup:
from shared.infrastructure.protokoll_logger import clear_logger_registry

def setup_function():
    clear_logger_registry()  # Isolierte Tests

9. Metriken

Vorher (Baseline)

Ziel Nachher (mit Phase-Gates)

10. Entscheidungen (Final)

Entscheidung 1: Package-Umbenennung

Entscheidung: Phase 0 (Voraussetzung, nicht optional)

Entscheidung 2: Datenbank-Bibliothek

Entscheidung: Hybrid mit klarer Semantik

• SimpleDbConnection für Standard-Server (kein Pooling)
• Eigene Implementierung für mcp_db (mit Pooling)

Entscheidung 3: Singleton Pattern

Entscheidung: Dict-Registry mit clear_registry() für Tests

Entscheidung 4: Clean Architecture

Entscheidung: Strikte Trennung Domain/Infrastructure

• LogEntry nur in shared/domain/log_entry.py
• Infrastructure importiert aus Domain, nicht umgekehrt

Entscheidung 5: sys.path.insert

Entscheidung: Vollständig verboten nach Phase 0

Anhang A: Korrekturen nach Supervision

Runde 1 (v1.0 → v1.1)

Runde 2 (v1.1 → v1.2)

Dokument Version 1.2 - Final für Implementierung.