Build an AI GitHub Actions Failure Analyzer with Claude API

name: Analyze CI Failure
 
on:
  workflow_run:
    workflows: ["CI"]  # Replace with your actual workflow name
    types: [completed]
 
permissions:
  pull-requests: write
  actions: read
 
jobs:
  analyze:
    if: ${{ github.event.workflow_run.conclusion == 'failure' }}
    runs-on: ubuntu-latest
    steps:
      - name: Checkout
        uses: actions/checkout@v4
 
      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.11"
 
      - name: Install dependencies
        run: pip install anthropic requests
 
      - name: Analyze failure and comment
        env:
          ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
          WORKFLOW_RUN_ID: ${{ github.event.workflow_run.id }}
          REPO: ${{ github.repository }}
          PR_NUMBER: ${{ github.event.workflow_run.pull_requests[0].number }}
        run: python .github/scripts/analyze_failure.py

import os
import re
import json
import requests
import anthropic
 
 
GITHUB_TOKEN = os.environ["GITHUB_TOKEN"]
ANTHROPIC_API_KEY = os.environ["ANTHROPIC_API_KEY"]
WORKFLOW_RUN_ID = os.environ["WORKFLOW_RUN_ID"]
REPO = os.environ["REPO"]
PR_NUMBER = os.environ.get("PR_NUMBER", "")
 
GITHUB_API = "https://api.github.com"
HEADERS = {
    "Authorization": f"Bearer {GITHUB_TOKEN}",
    "Accept": "application/vnd.github.v3+json",
    "X-GitHub-Api-Version": "2022-11-28"
}
 
 
def get_failed_jobs() -> list[dict]:
    """Get all failed jobs from the workflow run."""
    url = f"{GITHUB_API}/repos/{REPO}/actions/runs/{WORKFLOW_RUN_ID}/jobs"
    resp = requests.get(url, headers=HEADERS)
    resp.raise_for_status()
 
    jobs = resp.json()["jobs"]
    return [job for job in jobs if job["conclusion"] == "failure"]
 
 
def get_job_logs(job_id: int) -> str:
    """Download logs for a specific job."""
    url = f"{GITHUB_API}/repos/{REPO}/actions/jobs/{job_id}/logs"
    resp = requests.get(url, headers=HEADERS, allow_redirects=True)
 
    if resp.status_code == 302:
        # Follow redirect to actual log URL
        resp = requests.get(resp.headers["Location"])
 
    logs = resp.text
 
    # Truncate if too long (keep last 8000 chars — most relevant errors are at the end)
    if len(logs) > 8000:
        logs = "...[earlier output truncated]...\n\n" + logs[-8000:]
 
    return logs
 
 
def get_workflow_run_info() -> dict:
    """Get metadata about the workflow run."""
    url = f"{GITHUB_API}/repos/{REPO}/actions/runs/{WORKFLOW_RUN_ID}"
    resp = requests.get(url, headers=HEADERS)
    resp.raise_for_status()
    return resp.json()
 
 
def clean_logs(raw_logs: str) -> str:
    """
    Remove ANSI escape codes and timestamp prefixes from GitHub Actions logs.
    GitHub logs have a format like: 2026-07-05T10:23:45.123Z ##[section]Starting job
    """
    # Remove ANSI escape codes
    ansi_escape = re.compile(r"\x1b\[[0-9;]*[mGKH]")
    logs = ansi_escape.sub("", raw_logs)
 
    # Remove GitHub Actions timestamps but keep the content
    logs = re.sub(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d+Z ", "", logs, flags=re.MULTILINE)
 
    # Remove GitHub Actions command markers
    logs = re.sub(r"##\[.*?\]", "", logs)
 
    return logs.strip()
 
 
def analyze_with_claude(job_name: str, logs: str, run_info: dict) -> str:
    """Send logs to Claude for analysis."""
    client = anthropic.Anthropic(api_key=ANTHROPIC_API_KEY)
 
    log_fence = "```"
    prompt = (
        "You are a CI/CD expert helping diagnose a GitHub Actions workflow failure.\n\n"
        f"## Failed Job: {job_name}\n"
        f"## Repository: {REPO}\n"
        f"## Workflow: {run_info.get('name', 'unknown')}\n"
        f"## Branch: {run_info.get('head_branch', 'unknown')}\n"
        f"## Triggered by: {run_info.get('event', 'unknown')}\n\n"
        f"## Job Logs\n{log_fence}\n{logs}\n{log_fence}\n\n"
        "Analyze this CI failure and provide:\n\n"
        "1. **Root Cause** (1-2 sentences): What exactly failed and why?\n\n"
        "2. **Error Type**: Code/test failure, infrastructure issue, configuration error, or flaky test?\n\n"
        "3. **Specific Fix**: Exact commands or code changes to fix this.\n\n"
        "4. **Is this blocking?** Should the PR be blocked until fixed?\n\n"
        "Keep the response concise and actionable. Use markdown formatting."
    )
 
    message = client.messages.create(
        model="claude-sonnet-5",
        max_tokens=1000,
        messages=[{"role": "user", "content": prompt}]
    )
 
    return message.content[0].text
 
 
def post_pr_comment(analysis: str, failed_jobs: list[dict], run_url: str):
    """Post the analysis as a PR comment."""
    if not PR_NUMBER:
        print("No PR number found, skipping comment")
        return
 
    job_list = "\n".join([f"- `{job['name']}`" for job in failed_jobs])
 
    comment_body = f"""## CI Failure Analysis 🤖
 
**Failed Jobs:**
{job_list}
 
---
 
{analysis}
 
---
 
<details>
<summary>View full workflow run</summary>
 
[Workflow Run #{WORKFLOW_RUN_ID}]({run_url})
</details>
 
*Analysis by [Claude API](https://anthropic.com) + DevOpsBoys CI Analyzer*"""
 
    url = f"{GITHUB_API}/repos/{REPO}/issues/{PR_NUMBER}/comments"
    resp = requests.post(url, headers=HEADERS, json={"body": comment_body})
 
    if resp.status_code == 201:
        print(f"Posted analysis comment to PR #{PR_NUMBER}")
    else:
        print(f"Failed to post comment: {resp.status_code} {resp.text}")
 
 
def main():
    print(f"Analyzing workflow run {WORKFLOW_RUN_ID}...")
 
    # Get run metadata
    run_info = get_workflow_run_info()
    run_url = run_info.get("html_url", "")
 
    # Get failed jobs
    failed_jobs = get_failed_jobs()
    if not failed_jobs:
        print("No failed jobs found")
        return
 
    print(f"Found {len(failed_jobs)} failed job(s)")
 
    # Analyze the first failed job (or all if you prefer)
    analyses = []
    for job in failed_jobs[:2]:  # Limit to 2 jobs to control API costs
        print(f"Fetching logs for: {job['name']}")
        raw_logs = get_job_logs(job["id"])
        clean = clean_logs(raw_logs)
 
        print(f"Analyzing with Claude API...")
        analysis = analyze_with_claude(job["name"], clean, run_info)
        analyses.append(f"### Job: `{job['name']}`\n\n{analysis}")
 
    combined_analysis = "\n\n---\n\n".join(analyses)
 
    # Post to PR
    post_pr_comment(combined_analysis, failed_jobs, run_url)
    print("Done!")
 
 
if __name__ == "__main__":
    main()

def has_existing_analysis_comment() -> bool:
    url = f"{GITHUB_API}/repos/{REPO}/issues/{PR_NUMBER}/comments"
    resp = requests.get(url, headers=HEADERS)
    comments = resp.json()
    return any("CI Failure Analysis" in c["body"] for c in comments)

def count_failure_comments() -> int:
    url = f"{GITHUB_API}/repos/{REPO}/issues/{PR_NUMBER}/comments"
    resp = requests.get(url, headers=HEADERS)
    return sum(1 for c in resp.json() if "CI Failure Analysis" in c.get("body", ""))