# Task 7: Enhanced MCP Tools with GitPython Features (Minutes 105-135)
## Objective
Create enhanced MCP tools that leverage GitPython's advanced capabilities while maintaining backward compatibility with existing tools.
## Background
With GitPython service implemented in Task 6, this task focuses on creating enhanced MCP tools that provide richer functionality when GitPython is available, while gracefully falling back to basic functionality when only commitizen.git is available.
## Implementation Steps
### Step 1: Enhanced Repository Analysis Tools
**File**: `src/commitizen_mcp_connector/commitizen_server.py` (additions)
```python
@mcp.tool()
def analyze_repository_health(repo_path: str) -> Dict[str, Any]:
"""
Comprehensive repository health analysis using GitPython features.
Provides detailed analysis including:
- Repository statistics and metrics
- Commit frequency analysis
- Branch and tag information
- File change patterns
- Author contribution analysis
Args:
repo_path: Path to git repository
Returns:
Dict containing comprehensive repository analysis
"""
try:
from .git_compatibility import GitCompatibilityService
git_service = GitCompatibilityService(repo_path)
if not git_service.enhanced_features_available:
return {
"success": False,
"error": "Enhanced features not available - GitPython required",
"fallback_available": True,
"basic_status": git_service.get_repository_status()
}
# Get enhanced repository analysis
status = git_service.get_repository_status()
# Analyze commit patterns (last 30 commits)
recent_commits = git_service.implementation.get_commits(max_count=30)
# Calculate commit frequency
commit_dates = [commit["committed_date"][:10] for commit in recent_commits]
unique_dates = list(set(commit_dates))
commits_per_day = len(recent_commits) / max(len(unique_dates), 1)
# Analyze authors
authors = {}
for commit in recent_commits:
author = commit["author_name"]
if author not in authors:
authors[author] = {"commits": 0, "insertions": 0, "deletions": 0}
authors[author]["commits"] += 1
authors[author]["insertions"] += commit["stats"]["insertions"]
authors[author]["deletions"] += commit["stats"]["deletions"]
# Calculate repository health score
health_score = min(100, max(0,
(status["repository_stats"]["total_commits"] / 10) * 20 + # Commit history
(len(authors) / 3) * 20 + # Author diversity
(commits_per_day * 10) * 20 + # Activity level
(40 if status["staged_files_count"] == 0 else 20) # Clean state
))
return {
"success": True,
"repository_path": repo_path,
"implementation": git_service.implementation_type,
"health_analysis": {
"overall_score": round(health_score, 1),
"commit_frequency": {
"commits_per_day": round(commits_per_day, 2),
"total_commits": status["repository_stats"]["total_commits"],
"recent_activity": len(recent_commits)
},
"author_analysis": {
"total_authors": len(authors),
"top_contributors": sorted(
authors.items(),
key=lambda x: x[1]["commits"],
reverse=True
)[:5]
},
"repository_structure": {
"total_branches": status["repository_stats"]["total_branches"],
"total_tags": status["repository_stats"]["total_tags"],
"current_branch": status["current_branch"]
},
"working_directory": {
"staged_files": status["staged_files_count"],
"unstaged_files": status["unstaged_files_count"],
"untracked_files": status["untracked_files_count"],
"is_clean": status["staging_clean"] and
status["unstaged_files_count"] == 0 and
status["untracked_files_count"] == 0
}
}
}
except Exception as e:
logger.error(f"Failed to analyze repository health: {e}")
return {
"success": False,
"error": str(e),
"repository_path": repo_path
}
@mcp.tool()
def get_detailed_diff_analysis(
repo_path: str,
compare_with: str = "HEAD",
include_content: bool = False
) -> Dict[str, Any]:
"""
Get detailed diff analysis between working directory and specified commit.
Uses GitPython to provide:
- File-by-file change analysis
- Line-level insertion/deletion counts
- Change type classification (added, modified, deleted, renamed)
- Binary file detection
- Optional diff content inclusion
Args:
repo_path: Path to git repository
compare_with: Commit/branch to compare with (default: HEAD)
include_content: Whether to include actual diff content
Returns:
Dict containing detailed diff analysis
"""
try:
from .git_compatibility import GitCompatibilityService
git_service = GitCompatibilityService(repo_path)
if not git_service.enhanced_features_available:
return {
"success": False,
"error": "Enhanced diff analysis requires GitPython",
"fallback_available": False
}
# Get staged and unstaged diffs
staged_diff = git_service.implementation.repo.index.diff(compare_with)
unstaged_diff = git_service.implementation.repo.index.diff(None)
def analyze_diff_items(diff_items, diff_type):
analysis = []
total_insertions = 0
total_deletions = 0
for item in diff_items:
try:
# Get diff statistics
if item.diff:
diff_text = item.diff.decode('utf-8', errors='ignore')
insertions = diff_text.count('\n+') - diff_text.count('\n+++')
deletions = diff_text.count('\n-') - diff_text.count('\n---')
else:
insertions = deletions = 0
file_analysis = {
"file": item.a_path or item.b_path,
"change_type": item.change_type,
"insertions": max(0, insertions),
"deletions": max(0, deletions),
"is_binary": item.diff == b'',
"old_file": item.a_path,
"new_file": item.b_path,
"diff_type": diff_type
}
if include_content and not file_analysis["is_binary"]:
file_analysis["diff_content"] = diff_text
analysis.append(file_analysis)
total_insertions += max(0, insertions)
total_deletions += max(0, deletions)
except Exception as e:
logger.warning(f"Could not analyze diff for {item.a_path}: {e}")
analysis.append({
"file": item.a_path or item.b_path,
"change_type": item.change_type,
"error": str(e),
"diff_type": diff_type
})
return analysis, total_insertions, total_deletions
staged_analysis, staged_insertions, staged_deletions = analyze_diff_items(staged_diff, "staged")
unstaged_analysis, unstaged_insertions, unstaged_deletions = analyze_diff_items(unstaged_diff, "unstaged")
return {
"success": True,
"repository_path": repo_path,
"compare_with": compare_with,
"implementation": git_service.implementation_type,
"diff_analysis": {
"staged_changes": {
"files": staged_analysis,
"file_count": len(staged_analysis),
"total_insertions": staged_insertions,
"total_deletions": staged_deletions,
"total_changes": staged_insertions + staged_deletions
},
"unstaged_changes": {
"files": unstaged_analysis,
"file_count": len(unstaged_analysis),
"total_insertions": unstaged_insertions,
"total_deletions": unstaged_deletions,
"total_changes": unstaged_insertions + unstaged_deletions
},
"summary": {
"total_files_changed": len(staged_analysis) + len(unstaged_analysis),
"total_insertions": staged_insertions + unstaged_insertions,
"total_deletions": staged_deletions + unstaged_deletions,
"has_staged_changes": len(staged_analysis) > 0,
"has_unstaged_changes": len(unstaged_analysis) > 0
}
}
}
except Exception as e:
logger.error(f"Failed to get detailed diff analysis: {e}")
return {
"success": False,
"error": str(e),
"repository_path": repo_path
}
@mcp.tool()
def get_branch_analysis(repo_path: str) -> Dict[str, Any]:
"""
Get comprehensive branch analysis using GitPython.
Provides information about:
- All local and remote branches
- Branch relationships and merging status
- Commit counts per branch
- Last activity per branch
Args:
repo_path: Path to git repository
Returns:
Dict containing branch analysis
"""
try:
from .git_compatibility import GitCompatibilityService
git_service = GitCompatibilityService(repo_path)
if not git_service.enhanced_features_available:
return {
"success": False,
"error": "Branch analysis requires GitPython",
"fallback_available": False
}
repo = git_service.implementation.repo
# Analyze local branches
local_branches = []
for branch in repo.branches:
try:
last_commit = branch.commit
commit_count = sum(1 for _ in repo.iter_commits(branch))
local_branches.append({
"name": branch.name,
"is_current": branch == repo.active_branch,
"last_commit": {
"sha": last_commit.hexsha[:8],
"message": last_commit.summary,
"author": last_commit.author.name,
"date": last_commit.committed_datetime.isoformat()
},
"commit_count": commit_count
})
except Exception as e:
logger.warning(f"Could not analyze branch {branch.name}: {e}")
# Analyze remote branches
remote_branches = []
try:
for remote in repo.remotes:
for ref in remote.refs:
if ref.name.endswith('/HEAD'):
continue
try:
last_commit = ref.commit
remote_branches.append({
"name": ref.name,
"remote": remote.name,
"last_commit": {
"sha": last_commit.hexsha[:8],
"message": last_commit.summary,
"author": last_commit.author.name,
"date": last_commit.committed_datetime.isoformat()
}
})
except Exception as e:
logger.warning(f"Could not analyze remote branch {ref.name}: {e}")
except Exception as e:
logger.warning(f"Could not analyze remote branches: {e}")
return {
"success": True,
"repository_path": repo_path,
"implementation": git_service.implementation_type,
"branch_analysis": {
"current_branch": repo.active_branch.name if repo.active_branch else "HEAD (detached)",
"local_branches": {
"branches": local_branches,
"count": len(local_branches)
},
"remote_branches": {
"branches": remote_branches,
"count": len(remote_branches)
},
"summary": {
"total_branches": len(local_branches) + len(remote_branches),
"local_count": len(local_branches),
"remote_count": len(remote_branches)
}
}
}
except Exception as e:
logger.error(f"Failed to get branch analysis: {e}")
return {
"success": False,
"error": str(e),
"repository_path": repo_path
}
```
### Step 2: Enhanced Workflow Tools
**File**: `src/commitizen_mcp_connector/commitizen_server.py` (continued)
```python
@mcp.tool()
def smart_commit_suggestion(
repo_path: str,
analyze_changes: bool = True,
suggest_type: bool = True,
suggest_scope: bool = True
) -> Dict[str, Any]:
"""
Intelligent commit message suggestions based on repository changes.
Uses GitPython to analyze staged changes and suggest:
- Appropriate commit type based on file patterns
- Scope based on affected directories/modules
- Subject line based on change patterns
Args:
repo_path: Path to git repository
analyze_changes: Whether to analyze file changes for suggestions
suggest_type: Whether to suggest commit type
suggest_scope: Whether to suggest scope
Returns:
Dict containing intelligent commit suggestions
"""
try:
from .git_compatibility import GitCompatibilityService
git_service = GitCompatibilityService(repo_path)
# Get current staged files
status = git_service.get_repository_status()
if status["staging_clean"]:
return {
"success": False,
"error": "No staged changes to analyze",
"suggestions": None
}
suggestions = {
"types": [],
"scopes": [],
"subjects": [],
"analysis": {}
}
if analyze_changes and git_service.enhanced_features_available:
# Analyze file patterns for intelligent suggestions
staged_files = status["staged_files"]
# Analyze file types and patterns
file_patterns = {
"docs": [".md", ".rst", ".txt", "README", "CHANGELOG"],
"tests": ["test_", "_test", ".test.", "/tests/", "/test/"],
"config": [".json", ".yaml", ".yml", ".toml", ".ini", ".cfg"],
"frontend": [".js", ".ts", ".jsx", ".tsx", ".vue", ".css", ".scss"],
"backend": [".py", ".java", ".go", ".rs", ".cpp", ".c"],
"build": ["Makefile", "setup.py", "pyproject.toml", "package.json", "Dockerfile"],
"ci": [".github/", ".gitlab-ci", "Jenkinsfile", ".travis"]
}
detected_patterns = set()
affected_dirs = set()
for file_path in staged_files:
# Detect file patterns
for pattern_type, patterns in file_patterns.items():
if any(pattern in file_path for pattern in patterns):
detected_patterns.add(pattern_type)
# Extract directory for scope suggestion
if "/" in file_path:
dir_parts = file_path.split("/")
if len(dir_parts) > 1:
affected_dirs.add(dir_parts[0])
# Suggest commit types based on patterns
if suggest_type:
if "docs" in detected_patterns:
suggestions["types"].append({"type": "docs", "confidence": 0.9, "reason": "Documentation files modified"})
if "tests" in detected_patterns:
suggestions["types"].append({"type": "test", "confidence": 0.8, "reason": "Test files modified"})
if "config" in detected_patterns:
suggestions["types"].append({"type": "chore", "confidence": 0.7, "reason": "Configuration files modified"})
if "build" in detected_patterns or "ci" in detected_patterns:
suggestions["types"].append({"type": "ci", "confidence": 0.8, "reason": "Build/CI files modified"})
if "frontend" in detected_patterns or "backend" in detected_patterns:
suggestions["types"].append({"type": "feat", "confidence": 0.6, "reason": "Code files modified (likely feature)"})
suggestions["types"].append({"type": "fix", "confidence": 0.5, "reason": "Code files modified (possibly fix)"})
# Suggest scopes based on affected directories
if suggest_scope:
for dir_name in affected_dirs:
suggestions["scopes"].append({
"scope": dir_name,
"confidence": 0.7,
"reason": f"Files in {dir_name}/ directory modified"
})
# Generate subject suggestions
if len(staged_files) == 1:
file_name = staged_files[0].split("/")[-1]
suggestions["subjects"].append({
"subject": f"update {file_name}",
"confidence": 0.6,
"reason": "Single file modification"
})
elif "docs" in detected_patterns:
suggestions["subjects"].append({
"subject": "update documentation",
"confidence": 0.8,
"reason": "Documentation files modified"
})
elif "tests" in detected_patterns:
suggestions["subjects"].append({
"subject": "add/update tests",
"confidence": 0.7,
"reason": "Test files modified"
})
suggestions["analysis"] = {
"staged_files_count": len(staged_files),
"detected_patterns": list(detected_patterns),
"affected_directories": list(affected_dirs),
"file_types": list(set(f.split(".")[-1] for f in staged_files if "." in f))
}
# Sort suggestions by confidence
suggestions["types"].sort(key=lambda x: x["confidence"], reverse=True)
suggestions["scopes"].sort(key=lambda x: x["confidence"], reverse=True)
suggestions["subjects"].sort(key=lambda x: x["confidence"], reverse=True)
return {
"success": True,
"repository_path": repo_path,
"implementation": git_service.implementation_type,
"enhanced_features_used": git_service.enhanced_features_available,
"suggestions": suggestions,
"staged_files": status["staged_files"]
}
except Exception as e:
logger.error(f"Failed to generate smart commit suggestions: {e}")
return {
"success": False,
"error": str(e),
"repository_path": repo_path
}
@mcp.tool()
def batch_commit_analysis(
repo_path: str,
file_groups: List[Dict[str, Any]],
generate_messages: bool = True
) -> Dict[str, Any]:
"""
Analyze multiple file groups for batch commit operations.
Helps organize staged changes into logical commit groups with
appropriate commit messages for each group.
Args:
repo_path: Path to git repository
file_groups: List of file groups with metadata
generate_messages: Whether to generate commit messages for each group
Returns:
Dict containing batch commit analysis and suggestions
"""
try:
from .git_compatibility import GitCompatibilityService
git_service = GitCompatibilityService(repo_path)
batch_analysis = []
for i, group in enumerate(file_groups):
group_files = group.get("files", [])
group_description = group.get("description", f"Group {i+1}")
if not group_files:
continue
# Analyze this group of files
group_analysis = {
"group_id": i,
"description": group_description,
"files": group_files,
"file_count": len(group_files)
}
if generate_messages:
# Generate smart suggestions for this group
# (This would use the smart_commit_suggestion logic for the specific files)
group_analysis["suggested_messages"] = [
{
"type": "feat",
"subject": f"implement {group_description}",
"confidence": 0.7
}
]
batch_analysis.append(group_analysis)
return {
"success": True,
"repository_path": repo_path,
"implementation": git_service.implementation_type,
"batch_analysis": batch_analysis,
"total_groups": len(batch_analysis),
"total_files": sum(group["file_count"] for group in batch_analysis)
}
except Exception as e:
logger.error(f"Failed to analyze batch commits: {e}")
return {
"success": False,
"error": str(e),
"repository_path": repo_path
}
```
### Step 3: Enhanced Existing Tools
**File**: `src/commitizen_mcp_connector/commitizen_server.py` (modifications)
```python
# Enhance existing tools to use GitPython features when available
@mcp.tool()
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,
repo_path: Optional[str] = None,
enhanced_analysis: bool = False
) -> Dict[str, Any]:
"""
Enhanced commit message generation with optional GitPython features.
Args:
type: Commit type (feat, fix, docs, etc.)
subject: Commit subject/description
body: Optional commit body
scope: Optional scope
breaking: Whether this is a breaking change
footer: Optional footer
include_git_preview: Include git repository preview
repo_path: Optional repository path for enhanced features
enhanced_analysis: Use GitPython for enhanced analysis
Returns:
Dict containing generated message and optional enhanced information
"""
try:
# Generate message using existing logic
result = service.generate_message({
"type": type,
"subject": subject,
"body": body,
"scope": scope,
"breaking": breaking,
"footer": footer
})
# Add enhanced features if requested and available
if (include_git_preview or enhanced_analysis) and repo_path:
try:
from .git_compatibility import GitCompatibilityService
git_service = GitCompatibilityService(repo_path)
if include_git_preview:
preview = git_service.preview_commit(result["message"])
result["git_preview"] = preview
result["git_preview"]["implementation"] = git_service.implementation_type
if enhanced_analysis and git_service.enhanced_features_available:
# Add smart suggestions for improvement
suggestions = smart_commit_suggestion(repo_path)
if suggestions["success"]:
result["enhancement_suggestions"] = suggestions["suggestions"]
result["enhanced_features_used"] = git_service.enhanced_features_available
except Exception as e:
logger.warning(f"Enhanced features failed: {e}")
result["enhanced_features_error"] = str(e)
return result
except Exception as e:
logger.error(f"Failed to generate commit message: {e}")
return {
"error": str(e),
"success": False
}
@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 = False,
preview_only: bool = True,
enhanced_preview: bool = True
) -> Dict[str, Any]:
"""
Enhanced generate and commit with GitPython features.
Args:
type: Commit type
subject: Commit subject
repo_path: Repository path
body: Optional body
scope: Optional scope
breaking: Breaking change flag
footer: Optional footer
stage_all: Stage all changes
sign_off: Add sign-off
preview_only: Preview only (safety)
enhanced_preview: Use enhanced preview features
Returns:
Dict containing generation and commit results with enhancements
"""
try:
# Generate message
message_result = generate_commit_message(
type=type,
subject=subject,
body=body,
scope=scope,
breaking=breaking,
footer=footer,
repo_path=repo_path,
enhanced_analysis=enhanced_preview
)
if not message_result.get("success", True):
return message_result
message = message_result["message"]
# Initialize git service
from .git_compatibility import GitCompatibilityService
git_service = GitCompatibilityService(repo_path)
if preview_only:
# Enhanced preview
if enhanced_preview and git_service.enhanced_features_available:
preview = preview_commit_enhanced(message, repo_path)
return {
"success": True,
"message": message,
"message_generation": message_result,
"enhanced_preview": preview,
"preview_only": True,
"implementation": git_service.implementation_type
}
else:
# Basic preview
preview = git_service.preview_commit(message)
return {
"success": True,
"message": message,
"message_generation": message_result,
"preview": preview,
"preview_only": True,
"implementation": git_service.implementation_type
}
else:
# Execute commit
commit_result = git_service.execute_commit(
message=message,
force_execute=True,
sign_off=sign_off
)
return {
"success": commit_result.get("success", False),
"message": message,
"message_generation": message_result,
"commit_result": commit_result,
"preview_only": False,
"implementation": git_service.implementation_type
}
except Exception as e:
logger.error(f"Failed to generate and commit: {e}")
return {
"success": False,
"error": str(e),
"repository_path": repo_path
}
```
## Testing Strategy
### Step 4: Enhanced Tool Testing
**File**: `tests/test_enhanced_mcp_tools.py`
```python
"""
Tests for enhanced MCP tools with GitPython features.
"""
import pytest
from unittest.mock import Mock, patch, MagicMock
class TestEnhancedRepositoryTools:
"""Test enhanced repository analysis tools."""
def test_analyze_repository_health_with_gitpython(self):
"""Test repository health analysis with GitPython."""
pytest.skip("Implementation pending")
def test_analyze_repository_health_fallback(self):
"""Test repository health analysis fallback to basic features."""
pytest.skip("Implementation pending")
def test_detailed_diff_analysis(self):
"""Test detailed diff analysis with GitPython."""
pytest.skip("Implementation pending")
def test_branch_analysis(self):
"""Test comprehensive branch analysis."""
pytest.skip("Implementation pending")
class TestSmartCommitSuggestions:
"""Test intelligent commit suggestion features."""
def test_smart_commit_suggestion_docs(self):
"""Test smart suggestions for documentation changes."""
pytest.skip("Implementation pending")
def test_smart_commit_suggestion_tests(self):
"""Test smart suggestions for test changes."""
pytest.skip("Implementation pending")
def test_smart_commit_suggestion_mixed(self):
"""Test smart suggestions for mixed file types."""
pytest.skip("Implementation pending")
class TestEnhancedWorkflows:
"""Test enhanced workflow tools."""
def test_batch_commit_analysis(self):
"""Test batch commit analysis functionality."""
pytest.skip("Implementation pending")
def test_enhanced_generate_and_commit(self):
"""Test enhanced generate and commit workflow."""
pytest.skip("Implementation pending")
class TestBackwardCompatibility:
"""Test backward compatibility with existing tools."""
def test_enhanced_tools_with_basic_implementation(self):
"""Test enhanced tools fall back gracefully."""
pytest.skip("Implementation pending")
def test_existing_tools_unchanged(self):
"""Test existing tools continue to work unchanged."""
pytest.skip("Implementation pending")
```
## Success Criteria
- [ ] Enhanced repository analysis tools implemented
- [ ] Intelligent commit suggestion system working
- [ ] Detailed diff analysis with GitPython features
- [ ] Branch analysis and management tools
- [ ] Enhanced workflow tools for batch operations
- [ ] Backward compatibility maintained with existing tools
- [ ] Graceful fallback when GitPython not available
- [ ] Comprehensive test coverage for enhanced features
- [ ] Performance improvements demonstrated
- [ ] User experience enhanced with richer information
## Feature Matrix
### Enhanced vs Basic Features
| Feature | Basic (commitizen.git) | Enhanced (GitPython) |
|---------|----------------------|---------------------|
| Repository status | ✅ Basic info | ✅ Detailed stats, history |
| Commit preview | ✅ File list | ✅ Diff analysis, statistics |
| File operations | ✅ Basic staging | ✅ Detailed file info |
| Branch info | ❌ Limited | ✅ Comprehensive analysis |
| Commit history | ❌ Basic | ✅ Rich metadata, stats |
| Smart suggestions | ❌ None | ✅ AI-powered suggestions |
| Diff analysis | ❌ Basic | ✅ Line-by-line analysis |
| Repository health | ❌ None | ✅ Comprehensive metrics |
### Tool Enhancement Summary
1. **analyze_repository_health**: New comprehensive analysis tool
2. **get_detailed_diff_analysis**: New detailed diff analysis
3. **get_branch_analysis**: New branch management tool
4. **smart_commit_suggestion**: New AI-powered suggestions
5. **batch_commit_analysis**: New batch operation tool
6. **generate_commit_message**: Enhanced with GitPython features
7. **generate_and_commit**: Enhanced with better previews
## Performance Impact
### Expected Performance Improvements
- **Repository Status**: 2x faster with GitPython (no subprocess overhead)
- **Diff Analysis**: 3x more detailed information
- **Commit Operations**: 1.5x faster execution
- **Memory Usage**: +5MB for enhanced features (acceptable trade-off)
### Optimization Strategies
- **Lazy Loading**: Load enhanced features only when requested
- **Caching**: Cache repository analysis for short periods
- **Selective Enhancement**: Allow users to choose enhancement level
- **Fallback Performance**: Ensure fallback is as fast as original
## Migration and Adoption
### Gradual Enhancement Strategy
1. **Phase 1**: Add enhanced tools alongside existing ones
