"""
Message Tools
MCP tools for commit message generation and validation.
"""
import logging
from typing import Dict, List, Any, Optional
from .base_server import mcp, service
from ..errors import (
handle_errors,
ValidationError,
create_validation_error,
create_success_response,
)
logger = logging.getLogger(__name__)
@mcp.tool()
@handle_errors(log_errors=True)
def generate_commit_message(
type: str,
subject: str,
body: Optional[str] = None,
scope: Optional[str] = None,
breaking: Optional[bool] = False,
footer: Optional[str] = None,
include_git_preview: bool = False,
) -> Dict[str, Any]:
"""
Generate a commit message with validation using provided parameters.
Args:
type: The commit type (e.g., 'feat', 'fix', 'docs')
subject: The commit subject/description
body: Optional commit body with detailed description
scope: Optional scope of the change
breaking: Whether this is a breaking change
footer: Optional footer (e.g., issue references)
include_git_preview: Whether to include git preview in response
Returns:
Dict containing the generated message and validation status
"""
# Build answers dictionary with all fields (adapter will handle mapping)
answers = {
"type": type,
"prefix": type, # Some plugins expect 'prefix'
"subject": subject,
"body": body or "",
"scope": scope or "",
"breaking": breaking or False,
"is_breaking_change": breaking or False, # Some plugins expect this name
"footer": footer or "",
}
# Generate the message
message = service.generate_message(answers)
# Validate the generated message
is_valid = service.validate_message(message)
if not is_valid:
raise create_validation_error(
"Invalid commit message format",
validation_type="commit_message",
invalid_value=f"{type}: {subject}",
)
result = {
"message": message,
"is_valid": is_valid,
"parameters": {
"type": type,
"subject": subject,
"body": body,
"scope": scope,
"breaking": breaking,
"footer": footer,
},
}
# Add git preview if requested and available
if include_git_preview and service.git_enabled:
try:
# Import preview function from git_tools
from .git_tools import preview_git_commit
# Use the current service's repository path for preview
repo_path = (
str(service.git_service.repo_path) if service.git_service else None
)
if repo_path:
git_preview = preview_git_commit(message, repo_path)
result["git_preview"] = git_preview
else:
result["git_preview_error"] = (
"No repository path available for git preview"
)
except Exception as e:
logger.warning(f"Failed to include git preview: {e}")
result["git_preview_error"] = str(e)
elif include_git_preview and not service.git_enabled:
result["git_preview_error"] = (
"Git operations not available - not in a git repository"
)
return create_success_response(result)
@mcp.tool()
@handle_errors(log_errors=True)
def create_commit_message(answers_dict: Dict[str, Any]) -> Dict[str, Any]:
"""
Generate a commit message from a complete answers dictionary.
Args:
answers_dict: Dictionary containing all answers to commit questions
Returns:
Dict containing the generated message and validation status
"""
# Check for required fields
if "type" not in answers_dict or "subject" not in answers_dict:
raise create_validation_error(
"Missing required fields in answers dictionary",
validation_type="commit_answers",
invalid_value=str(answers_dict.keys()),
)
# Generate the message using the service
message = service.generate_message(answers_dict)
# Validate the generated message
is_valid = service.validate_message(message)
if not is_valid:
raise create_validation_error(
"Generated message failed validation",
validation_type="commit_message",
invalid_value=message,
)
result = {"message": message, "is_valid": is_valid, "answers": answers_dict}
return create_success_response(result)
@mcp.tool()
@handle_errors(log_errors=True)
def validate_commit_message(message: str) -> Dict[str, Any]:
"""
Validate an existing commit message against the current plugin's rules.
Args:
message: The commit message to validate
Returns:
Dict containing validation result and details
"""
is_valid = service.validate_message(message)
# Get additional info for context
info = service.get_info()
pattern = info.get("pattern")
result = {
"is_valid": is_valid,
"message": message,
"pattern": pattern,
"plugin": info.get("plugin_name"),
}
if not is_valid:
# Return error response but don't raise exception
# This allows validation to fail gracefully
return {
"success": False,
"is_valid": False,
"message": message,
"error": "Commit message does not match required pattern",
"pattern": pattern,
"plugin": info.get("plugin_name"),
}
return create_success_response(result)
@mcp.tool()
@handle_errors(log_errors=True)
def get_commit_types() -> Dict[str, Any]:
"""
Get list of available commit types from the current plugin.
Returns:
Dict containing available commit types and their descriptions
"""
commit_types = service.get_commit_types()
result = {
"commit_types": commit_types,
"count": len(commit_types),
"plugin": service.get_info().get("plugin_name"),
}
return create_success_response(result)