contract_tools.py

Code Hygiene Score: 61

Issues 1

Zeile Typ Beschreibung
- coupling Klasse hat 19 Dependencies (max: 15)

Dependencies 19

Funktionen 1

Code

"""Contract Tools für MCP-Contracts Server - Refactored"""
import sys
import time
import json
from typing import Optional

sys.path.insert(0, "/var/www/mcp-servers/mcp_contracts")

from domain.contracts import Contract, ContractStatus
from infrastructure.contract_repository import ContractRepository

from tools.contract_tools_components.constants import (
    DEFAULT_VERSION,
    DEFAULT_STATUS,
    DEFAULT_CREATED_BY,
    DEFAULT_CHANGED_BY,
    DEFAULT_TRIGGERED_BY,
    DEFAULT_LIMIT,
    DEFAULT_VIOLATIONS_LIMIT,
    DEFAULT_VALIDATIONS_LIMIT,
    VALID_STATUSES,
)
from tools.contract_tools_components.contract_parser import ContractParser
from tools.contract_tools_components.contract_validator import ContractValidatorService
from tools.contract_tools_components.contract_reporter import ContractReporter


def register_contract_tools(mcp):
    """Registriert alle Contract-Tools"""

    repo = ContractRepository()
    parser = ContractParser()
    validator_service = ContractValidatorService(repo)
    reporter = ContractReporter()

    # ==================== contracts_list ====================
    @mcp.tool()
    def contracts_list(
        status: Optional[str] = None,
        search: Optional[str] = None,
        compact: bool = True,
        limit: int = DEFAULT_LIMIT,
    ) -> dict:
        """
        Listet alle Contracts aus der Datenbank.

        Args:
            status: Filter nach Status (draft, active, deprecated)
            search: Volltextsuche in Name/Scope
            compact: True = nur id/name/version/status (Token-sparend)
            limit: Maximale Anzahl Ergebnisse

        Returns:
            Liste der Contracts
        """
        request_data = {"status": status, "search": search, "limit": limit}

        def operation():
            contracts = repo.find_all(status=status, search=search, limit=limit)
            total = repo.count(status=status, search=search)
            return reporter.format_list_response(contracts, total, limit, compact)

        return reporter.execute_with_logging("contracts_list", request_data, operation)

    # ==================== contracts_get ====================
    @mcp.tool()
    def contracts_get(
        id: Optional[int] = None,
        name: Optional[str] = None,
        version: Optional[str] = None,
        include_history: bool = False,
        include_validations: bool = False,
    ) -> dict:
        """
        Holt einen Contract nach ID oder Name.

        Args:
            id: Contract-ID
            name: Contract-Name (alternativ zu ID)
            version: Spezifische Version (optional, sonst neueste aktive)
            include_history: Änderungshistorie einschließen
            include_validations: Letzte Validierungen einschließen

        Returns:
            Contract mit Details
        """
        request_data = {"id": id, "name": name, "version": version}

        def operation():
            # Contract laden
            contract = None
            if id:
                contract = repo.find_by_id(id)
            elif name:
                contract = repo.find_by_name(name, version)
            else:
                return reporter.format_error_response("Either id or name required")

            if not contract:
                reporter.log_not_found("contracts_get", request_data)
                return reporter.format_error_response("Contract not found")

            result = {"contract": contract.to_dict()}

            if include_history:
                history = repo.get_history(contract.id)
                result["history"] = [h.to_dict() for h in history]

            if include_validations:
                validations = repo.get_validations(contract.id, limit=DEFAULT_VALIDATIONS_LIMIT)
                result["validations"] = [v.to_dict() for v in validations]

            return reporter.format_success_response(result)

        return reporter.execute_with_logging("contracts_get", request_data, operation)

    # ==================== contracts_create ====================
    @mcp.tool()
    def contracts_create(
        name: str,
        yaml_content: str,
        version: str = DEFAULT_VERSION,
        scope_description: Optional[str] = None,
        status: str = DEFAULT_STATUS,
        created_by: str = DEFAULT_CREATED_BY,
    ) -> dict:
        """
        Erstellt einen neuen Contract.

        Args:
            name: Contract-Name (eindeutig pro Version)
            yaml_content: YAML-Inhalt des Contracts
            version: Version (default: 1.0)
            scope_description: Kurze Beschreibung des Scopes
            status: Status (draft, active, deprecated)
            created_by: Ersteller

        Returns:
            Erstellter Contract
        """
        request_data = {"name": name, "version": version}

        def operation():
            # Validiere YAML-Syntax
            is_valid, error_msg = parser.validate_yaml_syntax(yaml_content)
            if not is_valid:
                return reporter.format_error_response(error_msg)

            # Prüfe ob Name+Version bereits existiert
            existing = repo.find_by_name(name, version)
            if existing:
                return reporter.format_error_response(
                    f"Contract {name} v{version} already exists"
                )

            # Erstelle Contract
            contract_status = (
                ContractStatus(status) if status in VALID_STATUSES
                else ContractStatus.ACTIVE
            )
            contract = Contract(
                name=name,
                version=version,
                status=contract_status,
                yaml_content=yaml_content,
                scope_description=scope_description,
                created_by=created_by,
            )

            contract_id = repo.create(contract)
            contract.id = contract_id

            return reporter.format_success_response({
                "contract": contract.to_dict(),
                "message": f"Contract {name} v{version} created with ID {contract_id}",
            })

        return reporter.execute_with_logging("contracts_create", request_data, operation)

    # ==================== contracts_update ====================
    @mcp.tool()
    def contracts_update(
        id: int,
        yaml_content: str,
        new_version: str,
        change_description: str,
        changed_by: str = DEFAULT_CHANGED_BY,
    ) -> dict:
        """
        Aktualisiert einen Contract (erstellt neue Version mit Historie).

        Args:
            id: Contract-ID
            yaml_content: Neuer YAML-Inhalt
            new_version: Neue Versionsnummer
            change_description: Beschreibung der Änderung
            changed_by: Wer hat geändert

        Returns:
            Aktualisierter Contract
        """
        request_data = {"id": id, "new_version": new_version}

        def operation():
            # Validiere YAML-Syntax
            is_valid, error_msg = parser.validate_yaml_syntax(yaml_content)
            if not is_valid:
                return reporter.format_error_response(error_msg)

            # Contract existiert?
            contract = repo.find_by_id(id)
            if not contract:
                return reporter.format_error_response(f"Contract {id} not found")

            # Neue Version erstellen (mit Historie)
            repo.create_new_version(
                contract_id=id,
                new_yaml=yaml_content,
                new_version=new_version,
                change_description=change_description,
                changed_by=changed_by,
            )

            # Aktualisierten Contract laden
            updated = repo.find_by_id(id)

            return reporter.format_success_response({
                "contract": updated.to_dict(),
                "message": f"Contract updated to v{new_version}",
            })

        return reporter.execute_with_logging("contracts_update", request_data, operation)

    # ==================== contracts_deprecate ====================
    @mcp.tool()
    def contracts_deprecate(id: int) -> dict:
        """
        Markiert einen Contract als deprecated.

        Args:
            id: Contract-ID

        Returns:
            Bestätigung
        """
        request_data = {"id": id}

        def operation():
            contract = repo.find_by_id(id)
            if not contract:
                return reporter.format_error_response(f"Contract {id} not found")

            repo.deprecate(id)

            return reporter.format_success_response({
                "message": f"Contract {contract.name} v{contract.version} deprecated",
            })

        return reporter.execute_with_logging("contracts_deprecate", request_data, operation)

    # ==================== contracts_validate ====================
    @mcp.tool()
    def contracts_validate(
        id: Optional[int] = None,
        name: Optional[str] = None,
        path: Optional[str] = None,
        triggered_by: str = DEFAULT_TRIGGERED_BY,
    ) -> dict:
        """
        Führt eine Contract-Validierung aus.

        Args:
            id: Contract-ID
            name: Contract-Name (alternativ zu ID)
            path: Spezifischer Pfad (optional, sonst gesamter Scope)
            triggered_by: Auslöser (mcp, sync, manual, hook)

        Returns:
            Validierungsergebnis
        """
        start = time.time()
        request_data = {"id": id, "name": name, "path": path}

        def operation():
            # Contract laden
            contract = None
            if id:
                contract = repo.find_by_id(id)
            elif name:
                contract = repo.find_by_name(name)
            else:
                return reporter.format_error_response("Either id or name required")

            if not contract:
                return reporter.format_error_response("Contract not found")

            # Validierung durchführen
            success, result_data, validation_id = validator_service.validate_contract(
                contract_id=contract.id,
                yaml_content=contract.yaml_content,
                path=path,
                triggered_by=triggered_by,
                start_time=start
            )

            if not success:
                return reporter.format_error_response(result_data.get("error", "Unknown error"))

            return reporter.format_success_response({
                "contract": contract.name,
                "version": contract.version,
                "validation_id": validation_id,
                "result": result_data,
            })

        return reporter.execute_with_logging("contracts_validate", request_data, operation)

    # ==================== contracts_history ====================
    @mcp.tool()
    def contracts_history(id: int) -> dict:
        """
        Holt die Änderungshistorie eines Contracts.

        Args:
            id: Contract-ID

        Returns:
            Versions-Historie mit Änderungsbeschreibungen
        """
        request_data = {"id": id}

        def operation():
            contract = repo.find_by_id(id)
            if not contract:
                return reporter.format_error_response(f"Contract {id} not found")

            versions = repo.get_versions(id)
            history = repo.get_history(id)

            return reporter.format_success_response({
                "contract": contract.name,
                "current_version": contract.version,
                "versions": versions,
                "history": [h.to_dict() for h in history],
            })

        return reporter.execute_with_logging("contracts_history", request_data, operation)

    # ==================== contracts_violations ====================
    @mcp.tool()
    def contracts_violations(
        id: Optional[int] = None,
        name: Optional[str] = None,
        limit: int = DEFAULT_VIOLATIONS_LIMIT,
    ) -> dict:
        """
        Holt die letzten Validierungen/Violations eines Contracts.

        Args:
            id: Contract-ID
            name: Contract-Name (alternativ)
            limit: Maximale Anzahl (default: 10)

        Returns:
            Liste der letzten Validierungen
        """
        request_data = {"id": id, "name": name, "limit": limit}

        def operation():
            contract = None
            if id:
                contract = repo.find_by_id(id)
            elif name:
                contract = repo.find_by_name(name)
            else:
                return reporter.format_error_response("Either id or name required")

            if not contract:
                return reporter.format_error_response("Contract not found")

            validations = validator_service.get_latest_validations(contract.id, limit=limit)

            return reporter.format_success_response({
                "contract": contract.name,
                "version": contract.version,
                "validations": validations,
            })

        return reporter.execute_with_logging("contracts_violations", request_data, operation)

    # ==================== contracts_statistics ====================
    @mcp.tool()
    def contracts_statistics() -> dict:
        """
        Holt Statistiken über alle Contracts und Validierungen.

        Returns:
            Statistiken: Contracts nach Status, Validierungen, etc.
        """
        request_data = {}

        def operation():
            stats = repo.get_statistics()
            return reporter.format_success_response({"statistics": stats})

        return reporter.execute_with_logging("contracts_statistics", request_data, operation)
← Übersicht