Protokoll #4176

{
    "event": "PreToolUse",
    "tool_name": "Read",
    "tool_input": {
        "file_path": "\/opt\/mcp-servers\/mcp-contracts\/tools\/contract_tools.py"
    }
}

{
    "tool_response": {
        "type": "text",
        "file": {
            "filePath": "\/opt\/mcp-servers\/mcp-contracts\/tools\/contract_tools.py",
            "content": "\"\"\"Contract Tools für MCP-Contracts Server\"\"\"\nimport sys\nimport time\nimport json\nfrom typing import Optional\nfrom datetime import datetime\n\nimport yaml\n\nsys.path.insert(0, \"\/opt\/mcp-servers\/mcp-contracts\")\n\nfrom domain.contracts import (\n    LogEntry, Contract, ContractValidation, ContractStatus\n)\nfrom infrastructure.protokoll_logger import get_logger\nfrom infrastructure.contract_repository import ContractRepository\nfrom validators.contract_validator import run_contract_validation\n\n\ndef register_contract_tools(mcp):\n    \"\"\"Registriert alle Contract-Tools\"\"\"\n\n    logger = get_logger()\n    repo = ContractRepository()\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 = 20,\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        start = time.time()\n        request_str = json.dumps({\"status\": status, \"search\": search, \"limit\": limit})\n\n        try:\n            contracts = repo.find_all(status=status, search=search, limit=limit)\n            total = repo.count(status=status, search=search)\n\n            duration = int((time.time() - start) * 1000)\n            logger.log(LogEntry(\n                tool_name=\"contracts_list\",\n                request=request_str,\n                status=\"success\",\n                duration_ms=duration,\n            ))\n\n            return {\n                \"success\": True,\n                \"contracts\": [c.to_dict_compact() if compact else c.to_dict() for c in contracts],\n                \"total\": total,\n                \"limit\": limit,\n            }\n\n        except Exception as e:\n            duration = int((time.time() - start) * 1000)\n            logger.log(LogEntry(\n                tool_name=\"contracts_list\",\n                request=request_str,\n                status=\"error\",\n                duration_ms=duration,\n                error_message=str(e)[:200],\n            ))\n            return {\"success\": False, \"error\": str(e)}\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        start = time.time()\n        request_str = json.dumps({\"id\": id, \"name\": name, \"version\": version})\n\n        try:\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 {\"success\": False, \"error\": \"Either id or name required\"}\n\n            if not contract:\n                logger.log(LogEntry(\n                    tool_name=\"contracts_get\",\n                    request=request_str,\n                    status=\"denied\",\n                    error_message=\"Contract not found\",\n                ))\n                return {\"success\": False, \"error\": \"Contract not found\"}\n\n            result = {\n                \"success\": True,\n                \"contract\": contract.to_dict(),\n            }\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=5)\n                result[\"validations\"] = [v.to_dict() for v in validations]\n\n            duration = int((time.time() - start) * 1000)\n            logger.log(LogEntry(\n                tool_name=\"contracts_get\",\n                request=request_str,\n                status=\"success\",\n                duration_ms=duration,\n            ))\n\n            return result\n\n        except Exception as e:\n            duration = int((time.time() - start) * 1000)\n            logger.log(LogEntry(\n                tool_name=\"contracts_get\",\n                request=request_str,\n                status=\"error\",\n                duration_ms=duration,\n                error_message=str(e)[:200],\n            ))\n            return {\"success\": False, \"error\": str(e)}\n\n    # ==================== contracts_create ====================\n    @mcp.tool()\n    def contracts_create(\n        name: str,\n        yaml_content: str,\n        version: str = \"1.0\",\n        scope_description: Optional[str] = None,\n        status: str = \"active\",\n        created_by: str = \"mcp-contracts\",\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        start = time.time()\n        request_str = json.dumps({\"name\": name, \"version\": version})\n\n        try:\n            # Validiere YAML-Syntax\n            try:\n                yaml.safe_load(yaml_content)\n            except yaml.YAMLError as e:\n                return {\"success\": False, \"error\": f\"Invalid YAML: {e}\"}\n\n            # Prüfe ob Name+Version bereits existiert\n            existing = repo.find_by_name(name, version)\n            if existing:\n                return {\"success\": False, \"error\": f\"Contract {name} v{version} already exists\"}\n\n            contract = Contract(\n                name=name,\n                version=version,\n                status=ContractStatus(status) if status in [\"draft\", \"active\", \"deprecated\"] else ContractStatus.ACTIVE,\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            duration = int((time.time() - start) * 1000)\n            logger.log(LogEntry(\n                tool_name=\"contracts_create\",\n                request=request_str,\n                status=\"success\",\n                duration_ms=duration,\n            ))\n\n            return {\n                \"success\": True,\n                \"contract\": contract.to_dict(),\n                \"message\": f\"Contract {name} v{version} created with ID {contract_id}\",\n            }\n\n        except Exception as e:\n            duration = int((time.time() - start) * 1000)\n            logger.log(LogEntry(\n                tool_name=\"contracts_create\",\n                request=request_str,\n                status=\"error\",\n                duration_ms=duration,\n                error_message=str(e)[:200],\n            ))\n            return {\"success\": False, \"error\": str(e)}\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 = \"mcp-contracts\",\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        start = time.time()\n        request_str = json.dumps({\"id\": id, \"new_version\": new_version})\n\n        try:\n            # Validiere YAML-Syntax\n            try:\n                yaml.safe_load(yaml_content)\n            except yaml.YAMLError as e:\n                return {\"success\": False, \"error\": f\"Invalid YAML: {e}\"}\n\n            # Contract existiert?\n            contract = repo.find_by_id(id)\n            if not contract:\n                return {\"success\": False, \"error\": 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            duration = int((time.time() - start) * 1000)\n            logger.log(LogEntry(\n                tool_name=\"contracts_update\",\n                request=request_str,\n                status=\"success\",\n                duration_ms=duration,\n            ))\n\n            return {\n                \"success\": True,\n                \"contract\": updated.to_dict(),\n                \"message\": f\"Contract updated to v{new_version}\",\n            }\n\n        except Exception as e:\n            duration = int((time.time() - start) * 1000)\n            logger.log(LogEntry(\n                tool_name=\"contracts_update\",\n                request=request_str,\n                status=\"error\",\n                duration_ms=duration,\n                error_message=str(e)[:200],\n            ))\n            return {\"success\": False, \"error\": str(e)}\n\n    # =... [TRUNCATED-f75062997101459b]",
            "numLines": 587,
            "startLine": 1,
            "totalLines": 587
        }
    }
}