{
"tool_response": [
{
"type": "text",
"text": "{\n \"success\": true,\n \"doc\": {\n \"id\": 156,\n \"parent_id\": 35,\n \"slug\": \"mcp-server-refactoring-plan\",\n \"path\": \"\/mcp\/mcp-server-refactoring-plan\",\n \"title\": \"MCP Server Refactoring Plan\",\n \"description\": \"Vollständiger Refactoring-Plan für alle 5 MCP-Server nach DRY, SRP, SOLID, Clean Architecture\",\n \"content\": \"# MCP Server Refactoring Plan\\n\\n**Version:** 1.2 \\n**Erstellt:** 2025-12-28 \\n**Aktualisiert:** 2025-12-28 (nach Supervision Runde 2) \\n**Status:** Final \\n**Task:** #507\\n\\n---\\n\\n## 1. Executive Summary\\n\\n### Ist-Zustand\\n- **5 MCP-Server**: mcp-db, mcp-tasks, mcp-contracts, mcp-docs, mcp-code\\n- **91 Python-Dateien** (ohne venv)\\n- **9.248 LOC** gesamt\\n- **Signifikante Code-Duplikation** (~25% redundanter Code)\\n- **Inkonsistente Patterns** zwischen Servern\\n\\n### Ziel\\nKonsolidierung zu einer modularen, wartbaren Architektur unter Einhaltung von:\\n- **DRY** (Don't Repeat Yourself)\\n- **SRP** (Single Responsibility Principle)\\n- **SOLID** Principles\\n- **Clean Architecture**\\n- **KISS** (Keep It Simple, Stupid)\\n- **YAGNI** (You Aren't Gonna Need It)\\n\\n### Erwartete Verbesserungen\\n- **~30% weniger Code** durch Elimination von Duplikaten\\n- **Einheitliche Patterns** für alle Server\\n- **Verbesserte Wartbarkeit** durch Modularisierung\\n- **Schnellere Feature-Entwicklung** durch shared Components\\n\\n---\\n\\n## 2. Analyse der Code-Duplikation\\n\\n### 2.1 Kritische Duplikate (Sofort beheben)\\n\\n#### db_connection.py (4 Dateien, ~240 LOC dupliziert)\\n\\n| Server | Zeilen | Bibliothek | Besonderheit |\\n|--------|--------|------------|--------------|\\n| mcp-db | 59 | mysql.connector + **Pooling** | Dynamisches DB-Switching via `USE` |\\n| mcp-tasks | 62 | pymysql | Festes `DB_NAME` |\\n| mcp-contracts | 60 | pymysql | Festes `DB_NAME` |\\n| mcp-docs | 60 | pymysql | Festes `DB_NAME` |\\n\\n**Problem:** 3 von 4 Dateien sind nahezu identisch (mcp-tasks, mcp-contracts, mcp-docs).\\n\\n**Lösung:** Zentralisieren in `shared\/infrastructure\/simple_db_connection.py`\\n\\n> **WICHTIG:** Diese Klasse heißt bewusst `SimpleDbConnection`, nicht `DatabaseConnection`.\\n> Sie bietet **kein Pooling** und ist **nicht für High-Throughput** geeignet.\\n> Für mcp-db (100+ Queries\/Minute) bleibt die eigene Pooling-Implementierung.\\n\\n```python\\n# shared\/infrastructure\/simple_db_connection.py\\n\\\"\\\"\\\"\\nEinfache DB-Verbindung ohne Pooling.\\n\\nNICHT geeignet für:\\n- High-Throughput (> 50 Queries\/Minute)\\n- Connection Pooling Requirements\\n- mcp-db Server\\n\\nFür diese Fälle: Eigene Implementierung mit mysql.connector.pooling\\n\\\"\\\"\\\"\\nfrom contextlib import contextmanager\\nfrom typing import Generator\\n\\nimport pymysql\\nfrom pymysql.connections import Connection\\n\\nfrom shared.config_base import AppDatabaseConfig, LogDatabaseConfig\\n\\n\\nclass SimpleDbConnection:\\n \\\"\\\"\\\"\\n Einfache Datenbankverbindung ohne Pooling.\\n \\n Für Server mit geringem Query-Volumen (< 50\/Minute).\\n \\\"\\\"\\\"\\n \\n @classmethod\\n @contextmanager\\n def get_connection(\\n cls, \\n config: AppDatabaseConfig,\\n database: str | None = None,\\n autocommit: bool = False\\n ) -> Generator[Connection, None, None]:\\n \\\"\\\"\\\"\\n Context Manager für App-DB Connection.\\n \\n ACHTUNG: Keine weiteren Parameter hinzufügen!\\n Bei Bedarf für mehr Logik: Neue Methode erstellen.\\n \\n Args:\\n config: Server-Konfiguration mit DB-Credentials\\n database: Optional - überschreibt config.DB_NAME\\n autocommit: True für Logging, False für Transaktionen\\n \\\"\\\"\\\"\\n conn = None\\n db_name = database or getattr(config, 'DB_NAME', None)\\n \\n try:\\n conn = pymysql.connect(\\n host=config.DB_HOST,\\n port=getattr(config, 'DB_PORT', 3306),\\n user=config.DB_USER,\\n password=config.DB_PASSWORD,\\n database=db_name,\\n charset=\\\"utf8mb4\\\",\\n cursorclass=pymysql.cursors.DictCursor,\\n autocommit=autocommit,\\n )\\n yield conn\\n if not autocommit:\\n conn.commit()\\n except Exception:\\n if conn and not autocommit:\\n conn.rollback()\\n raise\\n finally:\\n if conn:\\n conn.close()\\n \\n @classmethod\\n @contextmanager\\n def get_log_connection(\\n cls, \\n config: LogDatabaseConfig\\n ) -> Generator[Connection, None, None]:\\n \\\"\\\"\\\"Separate Verbindung für Logging (autocommit=True).\\\"\\\"\\\"\\n conn = None\\n try:\\n conn = pymysql.connect(\\n host=config.LOG_DB_HOST,\\n user=config.LOG_DB_USER,\\n password=config.LOG_DB_PASSWORD,\\n database=config.LOG_DB_NAME,\\n charset=\\\"utf8mb4\\\",\\n cursorclass=pymysql.cursors.DictCursor,\\n autocommit=True,\\n )\\n yield conn\\n finally:\\n if conn:\\n conn.close()\\n```\\n\\n#### protokoll_logger.py (4 Dateien, ~240 LOC dupliziert)\\n\\n**Lösung:** Zentralisieren mit strikter Clean Architecture Trennung\\n\\n```python\\n# shared\/domain\/log_entry.py\\n\\\"\\\"\\\"\\nDomain Entity für Log-Einträge.\\n\\nREINE DOMAIN - keine Infrastructure-Abhängigkeiten!\\n\\\"\\\"\\\"\\nfrom dataclasses import dataclass\\nfrom datetime import datetime\\nfrom typing import Optional\\n\\n\\n@dataclass(frozen=True)\\nclass LogEntry:\\n \\\"\\\"\\\"Standardisierter Log-Eintrag für alle MCP-Server.\\\"\\\"\\\"\\n timestamp: datetime\\n client_name: str\\n request: str\\n status: str\\n duration_ms: int\\n error_message: Optional[str] = None\\n tool_name: Optional[str] = None\\n context_id: Optional[int] = None # task_id, contract_id, etc.\\n```\\n\\n```python\\n# shared\/infrastructure\/protokoll_logger.py\\n\\\"\\\"\\\"\\nInfrastructure Logger für mcp_log Tabelle.\\n\\nImportiert LogEntry aus Domain - definiert ihn NICHT selbst.\\n\\\"\\\"\\\"\\nimport logging\\nimport sys\\nfrom typing import TYPE_CHECKING\\n\\nfrom shared.domain.log_entry import LogEntry # Import aus Domain!\\nfrom shared.infrastructure.simple_db_connection import SimpleDbConnection\\n\\nif TYPE_CHECKING:\\n from shared.config_base import LogDatabaseConfig\\n\\n\\nclass ProtokollLogger:\\n \\\"\\\"\\\"Fail-Safe Logger für mcp_log Tabelle.\\\"\\\"\\\"\\n \\n def __init__(self, client_name: str, config: \\\"LogDatabaseConfig\\\"):\\n self.client_name = client_name\\n self.config = config\\n self._logger = logging.getLogger(f\\\"mcp.{client_name}\\\")\\n \\n def log(self, entry: LogEntry) -> None:\\n \\\"\\\"\\\"Schreibt Log-Eintrag. Fehler gehen nur zu stderr.\\\"\\\"\\\"\\n try:\\n with SimpleDbConnection.get_log_connection(self.config) as conn:\\n with conn.cursor() as cursor:\\n request_str = self._format_request(entry)\\n cursor.execute(\\n \\\"\\\"\\\"INSERT INTO mcp_log\\n (timestamp, client_name, request, status, duration_ms, error_message)\\n VALUES (%s, %s, %s, %s, %s, %s)\\\"\\\"\\\",\\n (\\n entry.timestamp,\\n self.client_name,\\n request_str[:500],\\n entry.status,\\n entry.duration_ms,\\n entry.error_message[:500] if entry.error_message else None,\\n )\\n )\\n except Exception as e:\\n print(f\\\"CRITICAL: {self.client_name} log failed: {e}\\\", file=sys.stderr)\\n \\n def _format_request(self, entry: LogEntry) -> str:\\n parts = []\\n if entry.tool_name:\\n parts.append(f\\\"[{entry.tool_name}]\\\")\\n if entry.context_id:\\n parts.append(f\\\"id={entry.context_id}\\\")\\n parts.append(entry.request[:400] if entry.request else \\\"\\\")\\n return \\\" \\\".join(parts)\\n\\n\\n# === Registry mit Test-Support ===\\n\\n_logger_instances: dict[str, ProtokollLogger] = {}\\n\\n\\ndef get_logger(client_name: str, config: \\\"LogDatabaseConfig\\\") -> ProtokollLogger:\\n \\\"\\\"\\\"Singleton-Factory für Logger.\\\"\\\"\\\"\\n if client_name not in _logger_instances:\\n _logger_instances[client_name] = ProtokollLogger(client_name, config)\\n return _logger_instances[client_name]\\n\\n\\ndef clear_logger_registry() -> None:\\n \\\"\\\"\\\"\\n Setzt die Logger-Registry zurück.\\n \\n NUR FÜR TESTS! Ermöglicht isolierte Test-Ausführung.\\n \\\"\\\"\\\"\\n _logger_instances.clear()\\n```\\n\\n#### config.py - Strikte Statische Konfiguration\\n\\n```python\\n# shared\/config_base.py\\n\\\"\\\"\\\"\\nBasis-Konfiguration für alle MCP-Server.\\n\\nWICHTIG: Diese Klasse wird NICHT instanziiert!\\nAlle Zugriffe erfolgen über Klassenattribute.\\n\\nFalsch: config = BaseConfig()\\nRichtig: host = BaseConfig.DB_HOST\\n\\\"\\\"\\\"\\nimport os\\nfrom typing import ClassVar, Protocol, runtime_checkable\\n\\n\\n@runtime_checkable\\nclass AppDatabaseConfig(Protocol):\\n \\\"\\\"\\\"Protocol für App-Datenbank Konfiguration.\\\"\\\"\\\"\\n DB_HOST: str\\n DB_PORT: int\\n DB_USER: str\\n DB_PASSWORD: str\\n DB_NAME: str\\n\\n\\n@runtime_checkable\\nclass LogDatabaseConfig(Protocol):\\n \\\"\\\"\\\"Protocol für Log-Datenbank Konfiguration.\\\"\\\"\\\"\\n LOG_DB_HOST: str\\n LOG_DB_NAME: str\\n LOG_DB_USER: str\\n LOG_DB_PASSWORD: str\\n\\n\\nclass BaseConfig:\\n \\\"\\\"\\\"\\n Statische Konfiguration - NICHT INSTANZIIEREN!\\n \\n Verwendung:\\n from shared.config_base import BaseConfig\\n host = BaseConfig.DB_HOST # Korrekt\\n \\n config = BaseConfig() # FALSCH!\\n \\\"\\\"\\\"\\n \\n # App-Datenbank\\n DB_HOST: ClassVar[str] = os.getenv(\\\"DB_HOST\\\", \\\"localhost\\\")\\n DB_PORT: ClassVar[int] = int(os.getenv(\\\"DB_PORT\\\", \\\"3306\\\"))\\n DB_USER: ClassVar[str] = os.getenv(\\\"DB_USER\\\", \\\"root\\\")\\n DB_PASSWORD: Cla... [TRUNCATED-a6124601bb869ddb]"
}
]
}