projects/unit3/github-actions-integration/solution/server.py [19:260]: - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - TEMPLATES_DIR = Path(__file__).parent.parent.parent / "templates" # Default PR templates DEFAULT_TEMPLATES = { "bug.md": "Bug Fix", "feature.md": "Feature", "docs.md": "Documentation", "refactor.md": "Refactor", "test.md": "Test", "performance.md": "Performance", "security.md": "Security" } # File where webhook server stores events EVENTS_FILE = Path(__file__).parent / "github_events.json" # Type mapping for PR templates TYPE_MAPPING = { "bug": "bug.md", "fix": "bug.md", "feature": "feature.md", "enhancement": "feature.md", "docs": "docs.md", "documentation": "docs.md", "refactor": "refactor.md", "cleanup": "refactor.md", "test": "test.md", "testing": "test.md", "performance": "performance.md", "optimization": "performance.md", "security": "security.md" } # ===== Original Tools from Module 1 (with output limiting) ===== @mcp.tool() async def analyze_file_changes( base_branch: str = "main", include_diff: bool = True, max_diff_lines: int = 500, working_directory: Optional[str] = None ) -> str: """Get the full diff and list of changed files in the current git repository. Args: base_branch: Base branch to compare against (default: main) include_diff: Include the full diff content (default: true) max_diff_lines: Maximum number of diff lines to include (default: 500) working_directory: Directory to run git commands in (default: current directory) """ try: # Try to get working directory from roots first if working_directory is None: try: context = mcp.get_context() roots_result = await context.session.list_roots() # Get the first root - Claude Code sets this to the CWD root = roots_result.roots[0] # FileUrl object has a .path property that gives us the path directly working_directory = root.uri.path except Exception: # If we can't get roots, fall back to current directory pass # Use provided working directory or current directory cwd = working_directory if working_directory else os.getcwd() # Get list of changed files files_result = subprocess.run( ["git", "diff", "--name-status", f"{base_branch}...HEAD"], capture_output=True, text=True, check=True, cwd=cwd ) # Get diff statistics stat_result = subprocess.run( ["git", "diff", "--stat", f"{base_branch}...HEAD"], capture_output=True, text=True, cwd=cwd ) # Get the actual diff if requested diff_content = "" truncated = False if include_diff: diff_result = subprocess.run( ["git", "diff", f"{base_branch}...HEAD"], capture_output=True, text=True, cwd=cwd ) diff_lines = diff_result.stdout.split('\n') # Check if we need to truncate if len(diff_lines) > max_diff_lines: diff_content = '\n'.join(diff_lines[:max_diff_lines]) diff_content += f"\n\n... Output truncated. Showing {max_diff_lines} of {len(diff_lines)} lines ..." diff_content += "\n... Use max_diff_lines parameter to see more ..." truncated = True else: diff_content = diff_result.stdout # Get commit messages for context commits_result = subprocess.run( ["git", "log", "--oneline", f"{base_branch}..HEAD"], capture_output=True, text=True, cwd=cwd ) analysis = { "base_branch": base_branch, "files_changed": files_result.stdout, "statistics": stat_result.stdout, "commits": commits_result.stdout, "diff": diff_content if include_diff else "Diff not included (set include_diff=true to see full diff)", "truncated": truncated, "total_diff_lines": len(diff_lines) if include_diff else 0 } return json.dumps(analysis, indent=2) except subprocess.CalledProcessError as e: return json.dumps({"error": f"Git error: {e.stderr}"}) 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 = [ { "filename": filename, "type": template_type, "content": (TEMPLATES_DIR / filename).read_text() } for filename, template_type in DEFAULT_TEMPLATES.items() ] return json.dumps(templates, indent=2) @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.) """ # Get available templates templates_response = await get_pr_templates() templates = json.loads(templates_response) # Find matching template template_file = TYPE_MAPPING.get(change_type.lower(), "feature.md") selected_template = next( (t for t in templates if t["filename"] == template_file), templates[0] # Default to first template if no match ) suggestion = { "recommended_template": selected_template, "reasoning": f"Based on your analysis: '{changes_summary}', this appears to be a {change_type} change.", "template_content": selected_template["content"], "usage_hint": "Claude can help you fill out this template based on the specific changes in your PR." } return json.dumps(suggestion, indent=2) # ===== New Module 2: GitHub Actions Tools ===== @mcp.tool() async def get_recent_actions_events(limit: int = 10) -> str: """Get recent GitHub Actions events received via webhook. Args: limit: Maximum number of events to return (default: 10) """ # Read events from file if not EVENTS_FILE.exists(): return json.dumps([]) with open(EVENTS_FILE, 'r') as f: events = json.load(f) # Return most recent events recent = events[-limit:] return json.dumps(recent, indent=2) @mcp.tool() async def get_workflow_status(workflow_name: Optional[str] = None) -> str: """Get the current status of GitHub Actions workflows. Args: workflow_name: Optional specific workflow name to filter by """ # Read events from file if not EVENTS_FILE.exists(): return json.dumps({"message": "No GitHub Actions events received yet"}) with open(EVENTS_FILE, 'r') as f: events = json.load(f) if not events: return json.dumps({"message": "No GitHub Actions events received yet"}) # Filter for workflow events workflow_events = [ e for e in events if e.get("workflow_run") is not None ] if workflow_name: workflow_events = [ e for e in workflow_events if e["workflow_run"].get("name") == workflow_name ] # Group by workflow and get latest status workflows = {} for event in workflow_events: run = event["workflow_run"] name = run["name"] if name not in workflows or run["updated_at"] > workflows[name]["updated_at"]: workflows[name] = { "name": name, "status": run["status"], "conclusion": run.get("conclusion"), "run_number": run["run_number"], "updated_at": run["updated_at"], "html_url": run["html_url"] } return json.dumps(list(workflows.values()), indent=2) - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - projects/unit3/slack-notification/solution/server.py [20:259]: - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - TEMPLATES_DIR = Path(__file__).parent.parent.parent / "templates" # Default PR templates DEFAULT_TEMPLATES = { "bug.md": "Bug Fix", "feature.md": "Feature", "docs.md": "Documentation", "refactor.md": "Refactor", "test.md": "Test", "performance.md": "Performance", "security.md": "Security" } # File where webhook server stores events EVENTS_FILE = Path(__file__).parent / "github_events.json" # Type mapping for PR templates TYPE_MAPPING = { "bug": "bug.md", "fix": "bug.md", "feature": "feature.md", "enhancement": "feature.md", "docs": "docs.md", "documentation": "docs.md", "refactor": "refactor.md", "cleanup": "refactor.md", "test": "test.md", "testing": "test.md", "performance": "performance.md", "optimization": "performance.md", "security": "security.md" } # ===== Tools from Modules 1 & 2 (Complete with output limiting) ===== @mcp.tool() async def analyze_file_changes( base_branch: str = "main", include_diff: bool = True, max_diff_lines: int = 500, working_directory: Optional[str] = None ) -> str: """Get the full diff and list of changed files in the current git repository. Args: base_branch: Base branch to compare against (default: main) include_diff: Include the full diff content (default: true) max_diff_lines: Maximum number of diff lines to include (default: 500) working_directory: Directory to run git commands in (default: current directory) """ try: # Try to get working directory from roots first if working_directory is None: try: context = mcp.get_context() roots_result = await context.session.list_roots() # Get the first root - Claude Code sets this to the CWD root = roots_result.roots[0] # FileUrl object has a .path property that gives us the path directly working_directory = root.uri.path except Exception: # If we can't get roots, fall back to current directory pass # Use provided working directory or current directory cwd = working_directory if working_directory else os.getcwd() # Get list of changed files files_result = subprocess.run( ["git", "diff", "--name-status", f"{base_branch}...HEAD"], capture_output=True, text=True, check=True, cwd=cwd ) # Get diff statistics stat_result = subprocess.run( ["git", "diff", "--stat", f"{base_branch}...HEAD"], capture_output=True, text=True, cwd=cwd ) # Get the actual diff if requested diff_content = "" truncated = False if include_diff: diff_result = subprocess.run( ["git", "diff", f"{base_branch}...HEAD"], capture_output=True, text=True, cwd=cwd ) diff_lines = diff_result.stdout.split('\n') # Check if we need to truncate if len(diff_lines) > max_diff_lines: diff_content = '\n'.join(diff_lines[:max_diff_lines]) diff_content += f"\n\n... Output truncated. Showing {max_diff_lines} of {len(diff_lines)} lines ..." diff_content += "\n... Use max_diff_lines parameter to see more ..." truncated = True else: diff_content = diff_result.stdout # Get commit messages for context commits_result = subprocess.run( ["git", "log", "--oneline", f"{base_branch}..HEAD"], capture_output=True, text=True, cwd=cwd ) analysis = { "base_branch": base_branch, "files_changed": files_result.stdout, "statistics": stat_result.stdout, "commits": commits_result.stdout, "diff": diff_content if include_diff else "Diff not included (set include_diff=true to see full diff)", "truncated": truncated, "total_diff_lines": len(diff_lines) if include_diff else 0 } return json.dumps(analysis, indent=2) except subprocess.CalledProcessError as e: return json.dumps({"error": f"Git error: {e.stderr}"}) 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 = [ { "filename": filename, "type": template_type, "content": (TEMPLATES_DIR / filename).read_text() } for filename, template_type in DEFAULT_TEMPLATES.items() ] return json.dumps(templates, indent=2) @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.) """ # Get available templates templates_response = await get_pr_templates() templates = json.loads(templates_response) # Find matching template template_file = TYPE_MAPPING.get(change_type.lower(), "feature.md") selected_template = next( (t for t in templates if t["filename"] == template_file), templates[0] # Default to first template if no match ) suggestion = { "recommended_template": selected_template, "reasoning": f"Based on your analysis: '{changes_summary}', this appears to be a {change_type} change.", "template_content": selected_template["content"], "usage_hint": "Claude can help you fill out this template based on the specific changes in your PR." } return json.dumps(suggestion, indent=2) @mcp.tool() async def get_recent_actions_events(limit: int = 10) -> str: """Get recent GitHub Actions events received via webhook. Args: limit: Maximum number of events to return (default: 10) """ # Read events from file if not EVENTS_FILE.exists(): return json.dumps([]) with open(EVENTS_FILE, 'r') as f: events = json.load(f) # Return most recent events recent = events[-limit:] return json.dumps(recent, indent=2) @mcp.tool() async def get_workflow_status(workflow_name: Optional[str] = None) -> str: """Get the current status of GitHub Actions workflows. Args: workflow_name: Optional specific workflow name to filter by """ # Read events from file if not EVENTS_FILE.exists(): return json.dumps({"message": "No GitHub Actions events received yet"}) with open(EVENTS_FILE, 'r') as f: events = json.load(f) if not events: return json.dumps({"message": "No GitHub Actions events received yet"}) # Filter for workflow events workflow_events = [ e for e in events if e.get("workflow_run") is not None ] if workflow_name: workflow_events = [ e for e in workflow_events if e["workflow_run"].get("name") == workflow_name ] # Group by workflow and get latest status workflows = {} for event in workflow_events: run = event["workflow_run"] name = run["name"] if name not in workflows or run["updated_at"] > workflows[name]["updated_at"]: workflows[name] = { "name": name, "status": run["status"], "conclusion": run.get("conclusion"), "run_number": run["run_number"], "updated_at": run["updated_at"], "html_url": run["html_url"] } return json.dumps(list(workflows.values()), indent=2) - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -