projects/unit3/github-actions-integration/solution/server.py [130:260]: - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - ) 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/starter/server.py [109:237]: - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - ) 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) - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -