Protokoll #15712

{
    "event": "PreToolUse",
    "tool_name": "Write",
    "tool_input": {
        "file_path": "\/var\/www\/mcp-servers\/mcp-contracts\/tools\/contract_tools.py.new",
        "content": "\"\"\"Contract Tools für MCP-Contracts Server - Refactored\"\"\"\nimport sys\nimport time\nimport json\nfrom typing import Optional\n\nsys.path.insert(0, \"\/opt\/mcp-servers\/mcp-contracts\")\n\nfrom domain.contracts import Contract, ContractStatus\nfrom infrastructure.contract_repository import ContractRepository\n\nfrom contract_tools.constants import (\n    DEFAULT_VERSION,\n    DEFAULT_STATUS,\n    DEFAULT_CREATED_BY,\n    DEFAULT_CHANGED_BY,\n    DEFAULT_TRIGGERED_BY,\n    DEFAULT_LIMIT,\n    DEFAULT_VIOLATIONS_LIMIT,\n    DEFAULT_VALIDATIONS_LIMIT,\n    VALID_STATUSES,\n)\nfrom contract_tools.contract_parser import ContractParser\nfrom contract_tools.contract_validator import ContractValidatorService\nfrom contract_tools.contract_reporter import ContractReporter\n\n\ndef register_contract_tools(mcp):\n    \"\"\"Registriert alle Contract-Tools\"\"\"\n\n    repo = ContractRepository()\n    parser = ContractParser()\n    validator_service = ContractValidatorService(repo)\n    reporter = ContractReporter()\n\n    # ==================== contracts_list ====================\n    @mcp.tool()\n    def contracts_list(\n        status: Optional[str] = None,\n        search: Optional[str] = None,\n        compact: bool = True,\n        limit: int = DEFAULT_LIMIT,\n    ) -> dict:\n        \"\"\"\n        Listet alle Contracts aus der Datenbank.\n\n        Args:\n            status: Filter nach Status (draft, active, deprecated)\n            search: Volltextsuche in Name\/Scope\n            compact: True = nur id\/name\/version\/status (Token-sparend)\n            limit: Maximale Anzahl Ergebnisse\n\n        Returns:\n            Liste der Contracts\n        \"\"\"\n        request_data = {\"status\": status, \"search\": search, \"limit\": limit}\n\n        def operation():\n            contracts = repo.find_all(status=status, search=search, limit=limit)\n            total = repo.count(status=status, search=search)\n            return reporter.format_list_response(contracts, total, limit, compact)\n\n        return reporter.execute_with_logging(\"contracts_list\", request_data, operation)\n\n    # ==================== contracts_get ====================\n    @mcp.tool()\n    def contracts_get(\n        id: Optional[int] = None,\n        name: Optional[str] = None,\n        version: Optional[str] = None,\n        include_history: bool = False,\n        include_validations: bool = False,\n    ) -> dict:\n        \"\"\"\n        Holt einen Contract nach ID oder Name.\n\n        Args:\n            id: Contract-ID\n            name: Contract-Name (alternativ zu ID)\n            version: Spezifische Version (optional, sonst neueste aktive)\n            include_history: Änderungshistorie einschließen\n            include_validations: Letzte Validierungen einschließen\n\n        Returns:\n            Contract mit Details\n        \"\"\"\n        request_data = {\"id\": id, \"name\": name, \"version\": version}\n\n        def operation():\n            # Contract laden\n            contract = None\n            if id:\n                contract = repo.find_by_id(id)\n            elif name:\n                contract = repo.find_by_name(name, version)\n            else:\n                return reporter.format_error_response(\"Either id or name required\")\n\n            if not contract:\n                reporter.log_not_found(\"contracts_get\", request_data)\n                return reporter.format_error_response(\"Contract not found\")\n\n            result = {\"contract\": contract.to_dict()}\n\n            if include_history:\n                history = repo.get_history(contract.id)\n                result[\"history\"] = [h.to_dict() for h in history]\n\n            if include_validations:\n                validations = repo.get_validations(contract.id, limit=DEFAULT_VALIDATIONS_LIMIT)\n                result[\"validations\"] = [v.to_dict() for v in validations]\n\n            return reporter.format_success_response(result)\n\n        return reporter.execute_with_logging(\"contracts_get\", request_data, operation)\n\n    # ==================== contracts_create ====================\n    @mcp.tool()\n    def contracts_create(\n        name: str,\n        yaml_content: str,\n        version: str = DEFAULT_VERSION,\n        scope_description: Optional[str] = None,\n        status: str = DEFAULT_STATUS,\n        created_by: str = DEFAULT_CREATED_BY,\n    ) -> dict:\n        \"\"\"\n        Erstellt einen neuen Contract.\n\n        Args:\n            name: Contract-Name (eindeutig pro Version)\n            yaml_content: YAML-Inhalt des Contracts\n            version: Version (default: 1.0)\n            scope_description: Kurze Beschreibung des Scopes\n            status: Status (draft, active, deprecated)\n            created_by: Ersteller\n\n        Returns:\n            Erstellter Contract\n        \"\"\"\n        request_data = {\"name\": name, \"version\": version}\n\n        def operation():\n            # Validiere YAML-Syntax\n            is_valid, error_msg = parser.validate_yaml_syntax(yaml_content)\n            if not is_valid:\n                return reporter.format_error_response(error_msg)\n\n            # Prüfe ob Name+Version bereits existiert\n            existing = repo.find_by_name(name, version)\n            if existing:\n                return reporter.format_error_response(\n                    f\"Contract {name} v{version} already exists\"\n                )\n\n            # Erstelle Contract\n            contract_status = (\n                ContractStatus(status) if status in VALID_STATUSES\n                else ContractStatus.ACTIVE\n            )\n            contract = Contract(\n                name=name,\n                version=version,\n                status=contract_status,\n                yaml_content=yaml_content,\n                scope_description=scope_description,\n                created_by=created_by,\n            )\n\n            contract_id = repo.create(contract)\n            contract.id = contract_id\n\n            return reporter.format_success_response({\n                \"contract\": contract.to_dict(),\n                \"message\": f\"Contract {name} v{version} created with ID {contract_id}\",\n            })\n\n        return reporter.execute_with_logging(\"contracts_create\", request_data, operation)\n\n    # ==================== contracts_update ====================\n    @mcp.tool()\n    def contracts_update(\n        id: int,\n        yaml_content: str,\n        new_version: str,\n        change_description: str,\n        changed_by: str = DEFAULT_CHANGED_BY,\n    ) -> dict:\n        \"\"\"\n        Aktualisiert einen Contract (erstellt neue Version mit Historie).\n\n        Args:\n            id: Contract-ID\n            yaml_content: Neuer YAML-Inhalt\n            new_version: Neue Versionsnummer\n            change_description: Beschreibung der Änderung\n            changed_by: Wer hat geändert\n\n        Returns:\n            Aktualisierter Contract\n        \"\"\"\n        request_data = {\"id\": id, \"new_version\": new_version}\n\n        def operation():\n            # Validiere YAML-Syntax\n            is_valid, error_msg = parser.validate_yaml_syntax(yaml_content)\n            if not is_valid:\n                return reporter.format_error_response(error_msg)\n\n            # Contract existiert?\n            contract = repo.find_by_id(id)\n            if not contract:\n                return reporter.format_error_response(f\"Contract {id} not found\")\n\n            # Neue Version erstellen (mit Historie)\n            repo.create_new_version(\n                contract_id=id,\n                new_yaml=yaml_content,\n                new_version=new_version,\n                change_description=change_description,\n                changed_by=changed_by,\n            )\n\n            # Aktualisierten Contract laden\n            updated = repo.find_by_id(id)\n\n            return reporter.format_success_response({\n                \"contract\": updated.to_dict(),\n                \"message\": f\"Contract updated to v{new_version}\",\n            })\n\n        return reporter.execute_with_logging(\"contracts_update\", request_data, operation)\n\n    # ==================== contracts_deprecate ====================\n    @mcp.tool()\n    def contracts_deprecate(id: int) -> dict:\n        \"\"\"\n        Markiert einen Contract als deprecated.\n\n        Args:\n            id: Contract-ID\n\n        Returns:\n            Bestätigung\n        \"\"\"\n        request_data = {\"id\": id}\n\n        def operation():\n            contract = repo.find_by_id(id)\n            if not contract:\n                return reporter.format_error_response(f\"Contract {id} not found\")\n\n            repo.deprecate(id)\n\n            return reporter.format_success_response({\n                \"message\": f\"Contract {contract.name} v{contract.version} deprecated\",\n            })\n\n        return reporter.execute_with_logging(\"contracts_deprecate\", request_data, operation)\n\n    # ==================== contracts_validate ====================\n    @mcp.tool()\n    def contracts_validate(\n        id: Optional[int] = None,\n        name: Optional[str] = None,\n        path: Optional[str] = None,\n        triggered_by: str = DEFAULT_TRIGGERED_BY,\n    ) -> dict:\n        \"\"\"\n        Führt eine Contract-Validierung aus.\n\n        Args:\n            id: Contract-ID\n            name: Contract-Name (alternativ zu ID)\n            path: Spezifischer Pfad (optional, sonst gesamter Scope)\n            triggered_by: Auslöser (mcp, sync, manual, hook)\n\n        Returns:\n            Validierungsergebnis\n        \"\"\"\n        start = time.time()\n        request_data = {\"id\": id, \"name\": name, \"path\": path}\n\n        def operation():\n            # Contract laden\n            contract = None\n            if id:\n                contract = repo.find_by_id(id)\n            elif name:\n                contract = repo.find_by_name(name)\n            else:\n                return reporter.format_error_response(\"Either id or name required\")\n\n            if not contract:\n                return reporter.format_error_response(\"Contract not found\")\n\n            # Validierung durchführen\n            success, result_data, validation_id = validator_service.validate_contract(\n                contract_id=contract.id,\n      ... [TRUNCATED-3c50bc7b29c2894f]"
    }
}

{
    "tool_response": {
        "type": "create",
        "filePath": "\/var\/www\/mcp-servers\/mcp-contracts\/tools\/contract_tools.py.new",
        "content": "\"\"\"Contract Tools für MCP-Contracts Server - Refactored\"\"\"\nimport sys\nimport time\nimport json\nfrom typing import Optional\n\nsys.path.insert(0, \"\/opt\/mcp-servers\/mcp-contracts\")\n\nfrom domain.contracts import Contract, ContractStatus\nfrom infrastructure.contract_repository import ContractRepository\n\nfrom contract_tools.constants import (\n    DEFAULT_VERSION,\n    DEFAULT_STATUS,\n    DEFAULT_CREATED_BY,\n    DEFAULT_CHANGED_BY,\n    DEFAULT_TRIGGERED_BY,\n    DEFAULT_LIMIT,\n    DEFAULT_VIOLATIONS_LIMIT,\n    DEFAULT_VALIDATIONS_LIMIT,\n    VALID_STATUSES,\n)\nfrom contract_tools.contract_parser import ContractParser\nfrom contract_tools.contract_validator import ContractValidatorService\nfrom contract_tools.contract_reporter import ContractReporter\n\n\ndef register_contract_tools(mcp):\n    \"\"\"Registriert alle Contract-Tools\"\"\"\n\n    repo = ContractRepository()\n    parser = ContractParser()\n    validator_service = ContractValidatorService(repo)\n    reporter = ContractReporter()\n\n    # ==================== contracts_list ====================\n    @mcp.tool()\n    def contracts_list(\n        status: Optional[str] = None,\n        search: Optional[str] = None,\n        compact: bool = True,\n        limit: int = DEFAULT_LIMIT,\n    ) -> dict:\n        \"\"\"\n        Listet alle Contracts aus der Datenbank.\n\n        Args:\n            status: Filter nach Status (draft, active, deprecated)\n            search: Volltextsuche in Name\/Scope\n            compact: True = nur id\/name\/version\/status (Token-sparend)\n            limit: Maximale Anzahl Ergebnisse\n\n        Returns:\n            Liste der Contracts\n        \"\"\"\n        request_data = {\"status\": status, \"search\": search, \"limit\": limit}\n\n        def operation():\n            contracts = repo.find_all(status=status, search=search, limit=limit)\n            total = repo.count(status=status, search=search)\n            return reporter.format_list_response(contracts, total, limit, compact)\n\n        return reporter.execute_with_logging(\"contracts_list\", request_data, operation)\n\n    # ==================== contracts_get ====================\n    @mcp.tool()\n    def contracts_get(\n        id: Optional[int] = None,\n        name: Optional[str] = None,\n        version: Optional[str] = None,\n        include_history: bool = False,\n        include_validations: bool = False,\n    ) -> dict:\n        \"\"\"\n        Holt einen Contract nach ID oder Name.\n\n        Args:\n            id: Contract-ID\n            name: Contract-Name (alternativ zu ID)\n            version: Spezifische Version (optional, sonst neueste aktive)\n            include_history: Änderungshistorie einschließen\n            include_validations: Letzte Validierungen einschließen\n\n        Returns:\n            Contract mit Details\n        \"\"\"\n        request_data = {\"id\": id, \"name\": name, \"version\": version}\n\n        def operation():\n            # Contract laden\n            contract = None\n            if id:\n                contract = repo.find_by_id(id)\n            elif name:\n                contract = repo.find_by_name(name, version)\n            else:\n                return reporter.format_error_response(\"Either id or name required\")\n\n            if not contract:\n                reporter.log_not_found(\"contracts_get\", request_data)\n                return reporter.format_error_response(\"Contract not found\")\n\n            result = {\"contract\": contract.to_dict()}\n\n            if include_history:\n                history = repo.get_history(contract.id)\n                result[\"history\"] = [h.to_dict() for h in history]\n\n            if include_validations:\n                validations = repo.get_validations(contract.id, limit=DEFAULT_VALIDATIONS_LIMIT)\n                result[\"validations\"] = [v.to_dict() for v in validations]\n\n            return reporter.format_success_response(result)\n\n        return reporter.execute_with_logging(\"contracts_get\", request_data, operation)\n\n    # ==================== contracts_create ====================\n    @mcp.tool()\n    def contracts_create(\n        name: str,\n        yaml_content: str,\n        version: str = DEFAULT_VERSION,\n        scope_description: Optional[str] = None,\n        status: str = DEFAULT_STATUS,\n        created_by: str = DEFAULT_CREATED_BY,\n    ) -> dict:\n        \"\"\"\n        Erstellt einen neuen Contract.\n\n        Args:\n            name: Contract-Name (eindeutig pro Version)\n            yaml_content: YAML-Inhalt des Contracts\n            version: Version (default: 1.0)\n            scope_description: Kurze Beschreibung des Scopes\n            status: Status (draft, active, deprecated)\n            created_by: Ersteller\n\n        Returns:\n            Erstellter Contract\n        \"\"\"\n        request_data = {\"name\": name, \"version\": version}\n\n        def operation():\n            # Validiere YAML-Syntax\n            is_valid, error_msg = parser.validate_yaml_syntax(yaml_content)\n            if not is_valid:\n                return reporter.format_error_response(error_msg)\n\n            # Prüfe ob Name+Version bereits existiert\n            existing = repo.find_by_name(name, version)\n            if existing:\n                return reporter.format_error_response(\n                    f\"Contract {name} v{version} already exists\"\n                )\n\n            # Erstelle Contract\n            contract_status = (\n                ContractStatus(status) if status in VALID_STATUSES\n                else ContractStatus.ACTIVE\n            )\n            contract = Contract(\n                name=name,\n                version=version,\n                status=contract_status,\n                yaml_content=yaml_content,\n                scope_description=scope_description,\n                created_by=created_by,\n            )\n\n            contract_id = repo.create(contract)\n            contract.id = contract_id\n\n            return reporter.format_success_response({\n                \"contract\": contract.to_dict(),\n                \"message\": f\"Contract {name} v{version} created with ID {contract_id}\",\n            })\n\n        return reporter.execute_with_logging(\"contracts_create\", request_data, operation)\n\n    # ==================== contracts_update ====================\n    @mcp.tool()\n    def contracts_update(\n        id: int,\n        yaml_content: str,\n        new_version: str,\n        change_description: str,\n        changed_by: str = DEFAULT_CHANGED_BY,\n    ) -> dict:\n        \"\"\"\n        Aktualisiert einen Contract (erstellt neue Version mit Historie).\n\n        Args:\n            id: Contract-ID\n            yaml_content: Neuer YAML-Inhalt\n            new_version: Neue Versionsnummer\n            change_description: Beschreibung der Änderung\n            changed_by: Wer hat geändert\n\n        Returns:\n            Aktualisierter Contract\n        \"\"\"\n        request_data = {\"id\": id, \"new_version\": new_version}\n\n        def operation():\n            # Validiere YAML-Syntax\n            is_valid, error_msg = parser.validate_yaml_syntax(yaml_content)\n            if not is_valid:\n                return reporter.format_error_response(error_msg)\n\n            # Contract existiert?\n            contract = repo.find_by_id(id)\n            if not contract:\n                return reporter.format_error_response(f\"Contract {id} not found\")\n\n            # Neue Version erstellen (mit Historie)\n            repo.create_new_version(\n                contract_id=id,\n                new_yaml=yaml_content,\n                new_version=new_version,\n                change_description=change_description,\n                changed_by=changed_by,\n            )\n\n            # Aktualisierten Contract laden\n            updated = repo.find_by_id(id)\n\n            return reporter.format_success_response({\n                \"contract\": updated.to_dict(),\n                \"message\": f\"Contract updated to v{new_version}\",\n            })\n\n        return reporter.execute_with_logging(\"contracts_update\", request_data, operation)\n\n    # ==================== contracts_deprecate ====================\n    @mcp.tool()\n    def contracts_deprecate(id: int) -> dict:\n        \"\"\"\n        Markiert einen Contract als deprecated.\n\n        Args:\n            id: Contract-ID\n\n        Returns:\n            Bestätigung\n        \"\"\"\n        request_data = {\"id\": id}\n\n        def operation():\n            contract = repo.find_by_id(id)\n            if not contract:\n                return reporter.format_error_response(f\"Contract {id} not found\")\n\n            repo.deprecate(id)\n\n            return reporter.format_success_response({\n                \"message\": f\"Contract {contract.name} v{contract.version} deprecated\",\n            })\n\n        return reporter.execute_with_logging(\"contracts_deprecate\", request_data, operation)\n\n    # ==================== contracts_validate ====================\n    @mcp.tool()\n    def contracts_validate(\n        id: Optional[int] = None,\n        name: Optional[str] = None,\n        path: Optional[str] = None,\n        triggered_by: str = DEFAULT_TRIGGERED_BY,\n    ) -> dict:\n        \"\"\"\n        Führt eine Contract-Validierung aus.\n\n        Args:\n            id: Contract-ID\n            name: Contract-Name (alternativ zu ID)\n            path: Spezifischer Pfad (optional, sonst gesamter Scope)\n            triggered_by: Auslöser (mcp, sync, manual, hook)\n\n        Returns:\n            Validierungsergebnis\n        \"\"\"\n        start = time.time()\n        request_data = {\"id\": id, \"name\": name, \"path\": path}\n\n        def operation():\n            # Contract laden\n            contract = None\n            if id:\n                contract = repo.find_by_id(id)\n            elif name:\n                contract = repo.find_by_name(name)\n            else:\n                return reporter.format_error_response(\"Either id or name required\")\n\n            if not contract:\n                return reporter.format_error_response(\"Contract not found\")\n\n            # Validierung durchführen\n            success, result_data, validation_id = validator_service.validate_contract(\n                contract_id=contract.id,\n      ... [TRUNCATED-3c50bc7b29c2894f]",
        "structuredPatch": [],
        "originalFile": null
    }
}