server.py

#!/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()