Commit Helper MCP

""" Workflow Tools MCP tools for workflow orchestration and utility functions. """ 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, ValidationError, ConfigurationError, ServiceError, create_validation_error, create_success_response, ) logger = logging.getLogger(__name__) @mcp.tool() @handle_errors(log_errors=True) def get_commit_questions() -> Dict[str, Any]: """ Get interactive questions for commit message generation. Returns: Dict containing the questions and metadata """ questions = service.get_questions() result = { "questions": questions, "count": len(questions), "plugin": service.get_info().get("plugin_name"), } return create_success_response(result) @mcp.tool() @handle_errors(log_errors=True) def health_check() -> Dict[str, Any]: """ Check the health and status of the Commitizen service. Returns: Dict containing service health information """ info = service.get_info() result = { "status": "healthy", "service_info": info, "timestamp": str(__import__("datetime").datetime.now()), } return create_success_response(result) @mcp.tool() @handle_errors(log_errors=True) def refresh_configuration() -> Dict[str, Any]: """ Refresh the Commitizen configuration and reinitialize the service. Returns: Dict containing refresh status and new configuration info """ try: service.refresh_config() except Exception as e: raise ConfigurationError( "Failed to refresh configuration", config_file="commitizen config", cause=e ) info = service.get_info() result = { "status": "refreshed", "new_config": info, "timestamp": str(__import__("datetime").datetime.now()), } return create_success_response(result) @mcp.tool() @handle_errors(log_errors=True) def commit_workflow_step( workflow_data: Dict[str, Any], step: str = "generate" ) -> Dict[str, Any]: """ Multi-step commit workflow with state management. Args: workflow_data: Workflow state data step: Current step ("generate" | "preview" | "approve" | "execute") Returns: Dict with workflow state and next step information """ if step == "generate": # Generate commit message from parameters required_params = ["type", "subject"] for param in required_params: if param not in workflow_data: raise ValidationError( f"Missing required parameter: {param}", validation_type="workflow_parameters", invalid_value=str(workflow_data.keys()), ) # Import message generation function from .message_tools import generate_commit_message message_result = generate_commit_message( type=workflow_data["type"], subject=workflow_data["subject"], body=workflow_data.get("body"), scope=workflow_data.get("scope"), breaking=workflow_data.get("breaking", False), footer=workflow_data.get("footer"), ) if "error" in message_result: return { "error": f"Message generation failed: {message_result['error']}", "step": step, "workflow_data": workflow_data, "next_step": None, } # Update workflow data with generated message updated_workflow = workflow_data.copy() updated_workflow["generated_message"] = message_result["message"] updated_workflow["is_valid"] = message_result["is_valid"] return { "success": True, "step": step, "workflow_data": updated_workflow, "next_step": "preview", "result": message_result, } elif step == "preview": # Show commit preview with git status if "generated_message" not in workflow_data: return { "error": "No generated message found in workflow data", "step": step, "workflow_data": workflow_data, "next_step": None, } # Import git preview function from .git_tools import preview_git_commit # Get repository path from workflow data or use current service repo_path = workflow_data.get("repo_path") if not repo_path and service.git_service: repo_path = str(service.git_service.repo_path) if not repo_path: return { "error": "No repository path available for preview", "step": step, "workflow_data": workflow_data, "next_step": None, } preview_result = preview_git_commit( workflow_data["generated_message"], repo_path ) # Update workflow data with preview updated_workflow = workflow_data.copy() updated_workflow["preview_result"] = preview_result updated_workflow["repo_path"] = repo_path # Store for later steps return { "success": True, "step": step, "workflow_data": updated_workflow, "next_step": "approve", "result": preview_result, } elif step == "approve": # Validate approval and prepare execution if "generated_message" not in workflow_data: return { "error": "No generated message found in workflow data", "step": step, "workflow_data": workflow_data, "next_step": None, } # Check if user has explicitly approved user_approved = workflow_data.get("user_approved", False) if not user_approved: return { "error": "User approval required before execution", "step": step, "workflow_data": workflow_data, "next_step": "execute", "approval_required": True, "message": "Set 'user_approved': true in workflow_data to proceed", } # Import readiness validation function from .git_tools import validate_commit_readiness # Validate readiness - get repo_path from workflow data repo_path = workflow_data.get("repo_path") if not repo_path and service.git_service: repo_path = str(service.git_service.repo_path) if not repo_path: return { "error": "No repository path available for readiness validation", "step": step, "workflow_data": workflow_data, "next_step": None, } readiness = validate_commit_readiness(repo_path) updated_workflow = workflow_data.copy() updated_workflow["readiness_check"] = readiness updated_workflow["ready_for_execution"] = readiness.get( "ready_to_commit", False ) return { "success": True, "step": step, "workflow_data": updated_workflow, "next_step": "execute" if readiness.get("ready_to_commit", False) else None, "result": readiness, } elif step == "execute": # Perform actual commit with all safety checks if "generated_message" not in workflow_data: return { "error": "No generated message found in workflow data", "step": step, "workflow_data": workflow_data, "next_step": None, } if not workflow_data.get("user_approved", False): return { "error": "User approval required for execution", "step": step, "workflow_data": workflow_data, "next_step": None, } # Import git execution function from .git_tools import execute_git_commit # Execute the commit repo_path = workflow_data.get("repo_path") if not repo_path and service.git_service: repo_path = str(service.git_service.repo_path) if not repo_path: return { "error": "No repository path available for commit execution", "step": step, "workflow_data": workflow_data, "next_step": None, } commit_result = execute_git_commit( message=workflow_data["generated_message"], repo_path=repo_path, sign_off=workflow_data.get("sign_off", True), # Default to True for signoff force_execute=True, ) updated_workflow = workflow_data.copy() updated_workflow["commit_result"] = commit_result updated_workflow["completed"] = commit_result.get("success", False) return { "success": commit_result.get("success", False), "step": step, "workflow_data": updated_workflow, "next_step": None, "result": commit_result, "workflow_completed": True, } else: raise ValidationError( f"Unknown workflow step: {step}", validation_type="workflow_step", invalid_value=step, expected_format="One of: generate, preview, approve, execute", )