Commit Helper MCP

""" Git Tools MCP tools for git operations and repository management. """ import logging from typing import Dict, List, Any, Optional from .base_server import mcp, service from ..service_facade import CommitzenService from ..errors import ( handle_errors, handle_git_errors, GitOperationError, RepositoryError, ValidationError, create_git_error, create_validation_error, create_success_response, ) logger = logging.getLogger(__name__) @mcp.tool() @handle_errors(log_errors=True) def get_git_implementation_info() -> Dict[str, Any]: """ Get information about the current git implementation and available features. Returns: Dict containing: - git_enabled: Whether git operations are available - implementation: Current git implementation ("GitPython") - enhanced_features: Whether enhanced features are available - features: Dict of available feature flags """ result = service.get_git_implementation_info() return create_success_response(result) @mcp.tool() @handle_errors(log_errors=True) def get_enhanced_git_status(repo_path: str) -> Dict[str, Any]: """ Get enhanced git repository status with detailed information. Uses GitPython features when available for richer information: - Detailed file status (staged, unstaged, untracked) - Recent commit history with statistics - Repository analytics (total commits, branches, tags) - Current branch and HEAD information Args: repo_path: Path to git repository Returns: Dict containing enhanced repository status """ # Initialize service with specific repository try: target_service = CommitzenService(repo_path=repo_path) except Exception as e: raise RepositoryError( f"Failed to initialize service for repository '{repo_path}'", repo_path=repo_path, cause=e, ) if not target_service.git_enabled: raise RepositoryError( "Git operations not available - not in a git repository", repo_path=repo_path, ) status = target_service.get_repository_status() if "error" in status: raise GitOperationError(status["error"], repo_path=repo_path) status["enhanced_features_used"] = True status["implementation"] = target_service.git_implementation return create_success_response( {"repository_status": status, "repository_path": repo_path} ) @mcp.tool() @handle_errors(log_errors=True) def get_git_status(repo_path: str) -> Dict[str, Any]: """ Get current git repository status and staged files. Args: repo_path: Path to git repository Returns: Dict containing: - git_enabled: Whether git operations are available - staged_files: List of staged file paths - staged_count: Number of staged files - repository_path: Path to git repository - repository_status: Additional status information """ # For backward compatibility with tests expecting git_enabled field try: # Initialize service for the specified repository try: target_service = CommitzenService(repo_path=repo_path) except Exception as e: return { "git_enabled": False, "error": f"Failed to initialize service for repository '{repo_path}': {e}", "staged_files": [], "staged_count": 0, "repository_path": repo_path, } if not target_service.git_enabled: return { "git_enabled": False, "error": "Git operations not available - not in a git repository", "staged_files": [], "staged_count": 0, "repository_path": repo_path, } status = target_service.get_repository_status() return { "git_enabled": True, "staged_files": status.get("staged_files", []), "staged_count": status.get("staged_files_count", 0), "repository_path": status.get("repository_path"), "staging_clean": status.get("staging_clean", True), "repository_status": status, "success": True, } except Exception as e: logger.error(f"Failed to get git status: {e}") return { "git_enabled": False, "error": str(e), "staged_files": [], "staged_count": 0, "repository_path": repo_path, "success": False, } @mcp.tool() @handle_errors(log_errors=True) def preview_git_commit( message: str, repo_path: str, stage_all: bool = False, sign_off: bool = True ) -> Dict[str, Any]: """ Preview git commit operation without executing (dry-run mode). Args: message: Commit message to preview repo_path: Path to git repository stage_all: Whether to stage all changes before commit (not implemented yet) sign_off: Whether to add sign-off to commit (default: True) Returns: Dict containing: - message: The commit message - is_valid: Whether message passes validation - files_to_commit: List of files that would be committed - dry_run: Always True for this tool - repository_status: Current repository state """ # For backward compatibility with tests expecting git_enabled field try: # Initialize service for the specified repository try: target_service = CommitzenService(repo_path=repo_path) except Exception as e: return { "git_enabled": False, "error": f"Failed to initialize service for repository '{repo_path}': {e}", "message": message, "dry_run": True, "repository_path": repo_path, } if not target_service.git_enabled: return { "git_enabled": False, "error": "Git operations not available - not in a git repository", "message": message, "dry_run": True, "repository_path": repo_path, } # Get preview from service preview_result = target_service.preview_commit_operation( message, sign_off=sign_off ) if "error" in preview_result: return { "git_enabled": True, "error": preview_result["error"], "message": message, "dry_run": True, "repository_path": repo_path, } git_preview = preview_result.get("git_preview", {}) return { "git_enabled": True, "message": message, "is_valid": preview_result.get("is_valid", False), "files_to_commit": git_preview.get("staged_files", []), "staged_files_count": git_preview.get("staged_files_count", 0), "would_execute": git_preview.get("would_execute", False), "dry_run": True, "repository_status": git_preview, "repository_path": git_preview.get("repository_path"), "success": True, } except Exception as e: logger.error(f"Failed to preview git commit: {e}") return { "git_enabled": False, "error": str(e), "message": message, "dry_run": True, "repository_path": repo_path, "success": False, } @mcp.tool() def execute_git_commit( message: str, repo_path: str, stage_all: bool = False, sign_off: bool = True, force_execute: bool = False, ) -> Dict[str, Any]: """ Execute actual git commit with safety checks and user approval. SAFETY: Requires force_execute=True to perform actual commit. Args: message: Commit message to use repo_path: Path to git repository stage_all: Whether to stage all changes before commit (not implemented yet) sign_off: Whether to add sign-off to commit (default: True) force_execute: Must be True to execute actual commit (safety flag) Returns: Dict containing: - success: Whether commit was successful - message: The commit message used - executed: Whether commit was actually executed - error: Error message (if failed) - dry_run: False if actually executed, True if preview only """ try: # Initialize service for the specified repository try: target_service = CommitzenService(repo_path=repo_path) except Exception as e: return { "git_enabled": False, "error": f"Failed to initialize service for repository '{repo_path}': {e}", "success": False, "executed": False, "message": message, "repository_path": repo_path, } if not target_service.git_enabled: return { "git_enabled": False, "error": "Git operations not available - not in a git repository", "success": False, "executed": False, "message": message, "repository_path": repo_path, } # Execute commit with safety checks result = target_service.execute_commit_operation( message=message, force_execute=force_execute, sign_off=sign_off ) # Add dry_run flag based on execution result["dry_run"] = not result.get("executed", False) result["git_enabled"] = True result["repository_path"] = repo_path return result except Exception as e: logger.error(f"Failed to execute git commit: {e}") return { "git_enabled": False, "error": str(e), "success": False, "executed": False, "message": message, "dry_run": True, "repository_path": repo_path, } @mcp.tool() def generate_and_commit( type: str, subject: str, repo_path: str, body: Optional[str] = None, scope: Optional[str] = None, breaking: Optional[bool] = False, footer: Optional[str] = None, stage_all: bool = True, sign_off: bool = True, preview_only: bool = True, ) -> Dict[str, Any]: """ Generate commit message and optionally execute commit in one step. Combines message generation with git operations for streamlined workflow. Args: type: Commit type (feat, fix, docs, etc.) subject: Commit subject/description repo_path: Path to git repository body: Optional detailed description scope: Optional scope of changes breaking: Whether this is a breaking change footer: Optional footer (e.g., issue references) stage_all: Whether to stage all changes (not implemented yet) sign_off: Whether to add sign-off to commit (default: True) preview_only: If True, only preview (default for safety) Returns: Dict containing: - message: Generated commit message - is_valid: Whether message is valid - git_preview: Preview of git operation (if preview_only=True) - commit_result: Commit execution result (if preview_only=False) """ try: # Import message generation function from .message_tools import generate_commit_message # First generate the commit message message_result = generate_commit_message( type=type, subject=subject, body=body, scope=scope, breaking=breaking, footer=footer, ) if "error" in message_result: return { "error": f"Message generation failed: {message_result['error']}", "message": None, "is_valid": False, "preview_only": preview_only, } generated_message = message_result["message"] is_valid = message_result["is_valid"] if not is_valid: return { "error": "Generated message failed validation", "message": generated_message, "is_valid": False, "preview_only": preview_only, } # Now handle git operations using the provided repo_path if preview_only: # Get git preview using provided repository path git_preview = preview_git_commit(generated_message, repo_path) return { "message": generated_message, "is_valid": is_valid, "git_enabled": git_preview.get("git_enabled", False), "git_preview": git_preview, "preview_only": True, "repository_path": repo_path, } else: # Execute the commit using provided repository path commit_result = execute_git_commit( message=generated_message, repo_path=repo_path, sign_off=sign_off, force_execute=True, # Since user explicitly set preview_only=False ) return { "message": generated_message, "is_valid": is_valid, "git_enabled": commit_result.get("git_enabled", False), "commit_result": commit_result, "preview_only": False, "repository_path": repo_path, } except Exception as e: logger.error(f"Failed to generate and commit: {e}") return { "error": str(e), "message": None, "is_valid": False, "git_enabled": service.git_enabled, "preview_only": preview_only, } @mcp.tool() @handle_errors(log_errors=True) def validate_commit_readiness(repo_path: str) -> Dict[str, Any]: """ Comprehensive validation of repository readiness for commit. Args: repo_path: Path to git repository Returns: Dict containing: - ready_to_commit: Boolean overall readiness - checks: Dict of individual validation checks - recommendations: List of actions to take before commit """ # Initialize service for the specified repository try: target_service = CommitzenService(repo_path=repo_path) except Exception as e: raise RepositoryError( f"Failed to initialize service for repository '{repo_path}'", repo_path=repo_path, cause=e, ) if not target_service.git_enabled: raise RepositoryError( "Git operations not available - not in a git repository", repo_path=repo_path, ) # Get repository status status = target_service.get_repository_status() if "error" in status: raise GitOperationError(status["error"], repo_path=repo_path) # Perform readiness checks checks = { "is_git_repository": status.get("is_git_repository", False), "has_staged_files": not status.get("staging_clean", True), "staged_files_count": status.get("staged_files_count", 0), } # Determine overall readiness ready_to_commit = ( checks["is_git_repository"] and checks["has_staged_files"] and checks["staged_files_count"] > 0 ) # Generate recommendations recommendations = [] if not checks["is_git_repository"]: recommendations.append("Initialize git repository") if not checks["has_staged_files"]: recommendations.append("Stage files for commit using 'git add'") if checks["staged_files_count"] == 0: recommendations.append("Add files to staging area before committing") if ready_to_commit: recommendations.append("Repository is ready for commit") return create_success_response( { "ready_to_commit": ready_to_commit, "git_enabled": True, "checks": checks, "recommendations": recommendations, "repository_status": status, "repository_path": repo_path, } ) @mcp.tool() def stage_files_and_commit( files: List[str], message: str, repo_path: str, sign_off: bool = True, dry_run: bool = True, ) -> Dict[str, Any]: """ Stage specific files and commit with provided message. Useful for selective commits of specific files. Args: files: List of file paths to stage message: Commit message to use repo_path: Path to git repository sign_off: Whether to add sign-off to commit (default: True) dry_run: Preview only (default True for safety) Returns: Dict containing staging and commit results """ try: # Initialize service for the specified repository try: target_service = CommitzenService(repo_path=repo_path) except Exception as e: return { "git_enabled": False, "error": f"Failed to initialize service for repository '{repo_path}': {e}", "success": False, "files": files, "message": message, "dry_run": dry_run, "repository_path": repo_path, } if not target_service.git_enabled: return { "git_enabled": False, "error": "Git operations not available - not in a git repository", "success": False, "files": files, "message": message, "dry_run": dry_run, "repository_path": repo_path, } # Validate message format first is_valid = target_service.validate_message(message) if not is_valid: return { "error": "Invalid commit message format", "success": False, "files": files, "message": message, "is_valid": False, "dry_run": dry_run, "git_enabled": True, "repository_path": repo_path, } if dry_run: # Preview mode - don't actually stage or commit return { "success": True, "files": files, "message": message, "is_valid": is_valid, "dry_run": True, "git_enabled": True, "repository_path": repo_path, "preview": { "would_stage": files, "would_commit": True, "commit_message": message, }, } else: # Actually stage files and commit if not target_service.git_service: return { "error": "Git service not available", "success": False, "files": files, "message": message, "dry_run": dry_run, "git_enabled": True, "repository_path": repo_path, } # Stage the files stage_result = target_service.git_service.add_files( *files, force_execute=True ) if not stage_result.get("success", False): return { "error": f"Failed to stage files: {stage_result.get('error', 'Unknown error')}", "success": False, "files": files, "message": message, "dry_run": dry_run, "git_enabled": True, "repository_path": repo_path, "stage_result": stage_result, } # Now commit the staged files commit_result = target_service.execute_commit_operation( message=message, force_execute=True, sign_off=sign_off ) return { "success": commit_result.get("success", False), "files": files, "message": message, "is_valid": is_valid, "dry_run": False, "git_enabled": True, "repository_path": repo_path, "stage_result": stage_result, "commit_result": commit_result, } except Exception as e: logger.error(f"Failed to stage files and commit: {e}") return { "error": str(e), "success": False, "files": files, "message": message, "dry_run": dry_run, "git_enabled": False, "repository_path": repo_path, }