commit_workflow_step

Instructions

Implementation Reference

• src/commit_helper_mcp/server/workflow_tools.py:90-307 (handler)

  Core handler function implementing the multi-step commit workflow tool: generate commit message, preview changes, approve with validation, and execute git commit.

  @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",
          )

• src/commit_helper_mcp/mcp_server.py:46-51 (registration)

  Explicit import of commit_workflow_step from workflow_tools module, which registers the @mcp.tool()-decorated function for the MCP server.

  from .server.workflow_tools import (
      get_commit_questions,
      health_check,
      refresh_configuration,
      commit_workflow_step,
  )

• src/commit_helper_mcp/mcp_server.py:15-15 (registration)

  Module-level import of workflow_tools, automatically registering all @mcp.tool() functions including commit_workflow_step.

  from .server import workflow_tools