Spaces:
Sleeping
Sleeping
File size: 5,244 Bytes
22c4490 6910432 22c4490 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 |
#!/usr/bin/env python3
"""
Module 1: Basic MCP Server - Starter Code
TODO: Implement tools for analyzing git changes and suggesting PR templates
"""
import json
import subprocess
from pathlib import Path
from mcp.server.fastmcp import FastMCP
# Initialize the FastMCP server
mcp = FastMCP("pr-agent")
# PR template directory (shared across all modules)
TEMPLATES_DIR = Path(__file__).parent.parent.parent / "templates"
@mcp.tool()
async def analyze_file_changes(base_branch: str = "main",
include_diff: bool = True,
max_diff_lines: int = 500) -> str:
"""Analyze file changes with smart output limiting.
Args:
base_branch: Branch to compare against
include_diff: Whether to include the actual diff
max_diff_lines: Maximum diff lines to include (default 500)
"""
try:
# Get the diff
result = subprocess.run(
["git", "diff", f"{base_branch}...HEAD"],
capture_output=True,
text=True
)
diff_output = result.stdout
diff_lines = diff_output.split('\n')
# Smart truncation if needed
if len(diff_lines) > max_diff_lines:
truncated_diff = '\n'.join(diff_lines[:max_diff_lines])
truncated_diff += f"\n\n... Output truncated. Showing {max_diff_lines} of {len(diff_lines)} lines ..."
diff_output = truncated_diff
# Get summary statistics
stats_result = subprocess.run(
["git", "diff", "--stat", f"{base_branch}...HEAD"],
capture_output=True,
text=True
)
return json.dumps({
"stats": stats_result.stdout,
"total_lines": len(diff_lines),
"diff": diff_output if include_diff else "Use include_diff=true to see diff",
"files_changed": self._get_changed_files(base_branch)
})
except Exception as e:
return json.dumps({"error": str(e)})
@mcp.tool()
async def get_pr_templates() -> str:
"""List available PR templates with their content."""
templates = {}
if not TEMPLATES_DIR.exists():
return json.dumps({"error": f"Templates directory not found: {TEMPLATES_DIR}"})
try:
for template_file in TEMPLATES_DIR.glob("*.md"):
templates[template_file.stem] = template_file.read_text()
return json.dumps(templates)
except Exception as e:
return json.dumps({"error": f"Failed to read PR templates: {str(e)}"})
@mcp.tool()
async def suggest_template(changes_summary: str, change_type: str) -> str:
"""Let Claude analyze the changes and suggest the most appropriate PR template.
Args:
changes_summary: Your analysis of what the changes do
change_type: The type of change you've identified (bug, feature, docs, refactor, test, etc.)
"""
templates = json.loads(await get_pr_templates())
if "error" in templates:
return json.dumps(templates) # Propagate error from get_pr_templates
# Simple mapping logic. This can be made more sophisticated.
suggested_template_name = "default"
if change_type.lower() in ["bug", "fix"]:
suggested_template_name = "bug_fix"
elif change_type.lower() in ["feat", "feature"]:
suggested_template_name = "feature"
elif change_type.lower() in ["docs", "documentation"]:
suggested_template_name = "documentation"
elif change_type.lower() in ["refactor"]:
suggested_template_name = "refactor"
elif change_type.lower() in ["test", "tests"]:
suggested_template_name = "testing"
# Fallback if specific template not found but default exists
if suggested_template_name not in templates and "default" in templates:
suggested_template_name = "default"
if suggested_template_name in templates:
return json.dumps({"suggested_template": suggested_template_name, "content": templates[suggested_template_name]})
else:
return json.dumps({"error": f"No suitable template found for change type '{change_type}'. Available templates: {list(templates.keys())}"})
@mcp.prompt()
def format_ci_failure_alert() -> str:
"""Create a Slack alert for CI/CD failures."""
return """Format this GitHub Actions failure as a Slack message:
Use this template:
:rotating_light: *CI Failure Alert* :rotating_light:
A CI workflow has failed:
*Workflow*: workflow_name
*Branch*: branch_name
*Status*: Failed
*View Details*: <LOGS_LINK|View Logs>
Please check the logs and address any issues.
Use Slack markdown formatting and keep it concise for quick team scanning."""
@mcp.prompt()
def format_ci_success_summary() -> str:
"""Create a Slack message celebrating successful deployments."""
return """Format this successful GitHub Actions run as a Slack message:
Use this template:
:white_check_mark: *Deployment Successful* :white_check_mark:
Deployment completed successfully for [Repository Name]
*Changes:*
- Key feature or fix 1
- Key feature or fix 2
*Links:*
<PR_LINK|View Changes>
Keep it celebratory but informative. Use Slack markdown formatting."""
if __name__ == "__main__":
mcp.run() |