Claude AI Tutorial #24: Claude in CI/CD — Automated Code Review in GitHub Actions

PR Opened/Updated → GitHub Actions Trigger → Get PR Diff → Claude API → Post Review Comments

#!/usr/bin/env python3
"""ai_review.py — AI code review for GitHub Actions."""

import anthropic
import json
import os
import sys
import requests

client = anthropic.Anthropic()

GITHUB_TOKEN = os.environ["GITHUB_TOKEN"]
REPO = os.environ["GITHUB_REPOSITORY"]
PR_NUMBER = os.environ["PR_NUMBER"]

REVIEW_PROMPT = """You are an AI code reviewer for a production codebase.

<rules>
- Focus on: bugs, security vulnerabilities, performance issues, error handling
- Ignore: formatting, naming preferences, style opinions
- Rate each issue: CRITICAL (blocks merge), WARNING (should fix), INFO (suggestion)
- If the code is good, return an empty array
- Only review added/changed lines (lines starting with +)
- Be specific about file names and line numbers
</rules>

<output_format>
Return a JSON array:
[
  {
    "file": "path/to/file.py",
    "line": 42,
    "severity": "CRITICAL",
    "title": "Brief issue title",
    "body": "Detailed explanation and suggested fix"
  }
]

If no issues: []
</output_format>"""


def get_pr_diff() -> str:
    """Get the PR diff from GitHub API."""
    response = requests.get(
        f"https://api.github.com/repos/{REPO}/pulls/{PR_NUMBER}",
        headers={
            "Authorization": f"Bearer {GITHUB_TOKEN}",
            "Accept": "application/vnd.github.diff",
        },
    )
    response.raise_for_status()
    return response.text


def review_code(diff: str) -> list[dict]:
    """Send diff to Claude for review."""
    # Skip very large diffs
    if len(diff) > 100000:
        print("Diff too large (>100K chars). Skipping AI review.")
        return []

    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=4096,
        temperature=0,
        system=[
            {
                "type": "text",
                "text": REVIEW_PROMPT,
                "cache_control": {"type": "ephemeral"},
            }
        ],
        messages=[
            {"role": "user", "content": f"<diff>\n{diff}\n</diff>\n\nReview this PR diff."}
        ],
    )

    text = response.content[0].text
    try:
        if "```json" in text:
            text = text.split("```json")[1].split("```")[0]
        elif "```" in text:
            text = text.split("```")[1].split("```")[0]
        return json.loads(text.strip())
    except json.JSONDecodeError:
        print(f"Could not parse response: {text[:200]}")
        return []


def post_review(issues: list[dict]):
    """Post review comments on the GitHub PR."""
    if not issues:
        # Post approval
        requests.post(
            f"https://api.github.com/repos/{REPO}/pulls/{PR_NUMBER}/reviews",
            headers={
                "Authorization": f"Bearer {GITHUB_TOKEN}",
                "Accept": "application/vnd.github.v3+json",
            },
            json={
                "body": "AI Review: No issues found. Looks good!",
                "event": "COMMENT",
            },
        )
        print("Posted: No issues found.")
        return

    # Build review comments
    comments = []
    for issue in issues:
        comments.append({
            "path": issue["file"],
            "line": issue["line"],
            "body": f"**[{issue['severity']}] {issue['title']}**\n\n{issue['body']}",
        })

    has_critical = any(i["severity"] == "CRITICAL" for i in issues)

    response = requests.post(
        f"https://api.github.com/repos/{REPO}/pulls/{PR_NUMBER}/reviews",
        headers={
            "Authorization": f"Bearer {GITHUB_TOKEN}",
            "Accept": "application/vnd.github.v3+json",
        },
        json={
            "body": f"AI Review: Found {len(issues)} issue(s) — "
                    f"{sum(1 for i in issues if i['severity']=='CRITICAL')} critical, "
                    f"{sum(1 for i in issues if i['severity']=='WARNING')} warnings, "
                    f"{sum(1 for i in issues if i['severity']=='INFO')} info.",
            "event": "REQUEST_CHANGES" if has_critical else "COMMENT",
            "comments": comments,
        },
    )

    if response.status_code == 200:
        print(f"Posted {len(comments)} review comments.")
    else:
        print(f"Error: {response.status_code} {response.text}")


def main():
    print(f"Reviewing PR #{PR_NUMBER} in {REPO}...")
    diff = get_pr_diff()
    print(f"Diff size: {len(diff):,} characters")

    issues = review_code(diff)
    print(f"Found {len(issues)} issues")

    post_review(issues)


if __name__ == "__main__":
    main()

name: AI Code Review

on:
  pull_request:
    types: [opened, synchronize]

permissions:
  contents: read
  pull-requests: write

jobs:
  ai-review:
    runs-on: ubuntu-latest
    steps:
      - name: Checkout
        uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.12"

      - name: Install dependencies
        run: pip install anthropic requests

      - name: Run AI Review
        env:
          ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
          GITHUB_REPOSITORY: ${{ github.repository }}
          PR_NUMBER: ${{ github.event.pull_request.number }}
        run: python scripts/ai_review.py

MAX_DIFF_CHARS = 50000  # ~15,000 tokens

if len(diff) > MAX_DIFF_CHARS:
    print(f"Skipping: diff too large ({len(diff):,} chars > {MAX_DIFF_CHARS:,})")
    post_comment("AI Review: PR too large for automated review. Please request a human review.")
    sys.exit(0)

def filter_diff(diff: str) -> str:
    """Keep only code file diffs."""
    code_extensions = {".py", ".ts", ".js", ".go", ".rs", ".kt", ".java", ".tsx", ".jsx"}
    filtered_parts = []
    current_file = ""

    for line in diff.split("\n"):
        if line.startswith("diff --git"):
            current_file = line.split(" b/")[-1] if " b/" in line else ""
        if any(current_file.endswith(ext) for ext in code_extensions):
            filtered_parts.append(line)

    return "\n".join(filtered_parts)

def filter_diff_by_files(diff: str, files: list[str]) -> str:
    """Keep only diffs for the specified files."""
    filtered = []
    include = False
    for line in diff.split("\n"):
        if line.startswith("diff --git"):
            include = any(f in line for f in files)
        if include:
            filtered.append(line)
    return "\n".join(filtered)

def post_comment(message: str):
    """Post a general comment on the PR."""
    requests.post(
        f"https://api.github.com/repos/{REPO}/issues/{PR_NUMBER}/comments",
        headers={
            "Authorization": f"Bearer {GITHUB_TOKEN}",
            "Accept": "application/vnd.github.v3+json",
        },
        json={"body": message},
    )

def triage_with_haiku(diff: str) -> list[str]:
    """Quick triage with Haiku to find files worth reviewing."""
    response = client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=1024,
        temperature=0,
        messages=[{
            "role": "user",
            "content": f"""Look at this diff and list only the files that might have bugs or security issues.
Return a JSON array of file paths. If all files look safe, return [].

<diff>
{diff[:30000]}
</diff>""",
        }],
    )

    text = response.content[0].text
    try:
        return json.loads(text.strip())
    except:
        return []

# Triage first, then detailed review
flagged_files = triage_with_haiku(diff)
if flagged_files:
    # Only review flagged files with Sonnet
    filtered_diff = filter_diff_by_files(diff, flagged_files)
    issues = review_code(filtered_diff)

{
  "focus": ["security", "error-handling", "performance"],
  "ignore": ["formatting", "naming", "comments"],
  "languages": ["python", "typescript"],
  "severity_threshold": "WARNING",
  "max_diff_size": 50000,
  "model": "claude-sonnet-4-6",
  "custom_rules": [
    "All database queries must use parameterized queries",
    "All API endpoints must have authentication",
    "All error responses must include error codes"
  ]
}

import os

def load_review_config() -> dict:
    """Load review configuration from .ai-review.json."""
    config_path = ".ai-review.json"
    if os.path.exists(config_path):
        with open(config_path) as f:
            return json.load(f)
    return {}

config = load_review_config()
if config.get("custom_rules"):
    REVIEW_PROMPT += "\n\n<custom_rules>\n"
    for rule in config["custom_rules"]:
        REVIEW_PROMPT += f"- {rule}\n"
    REVIEW_PROMPT += "</custom_rules>"

def submit_batch_review(diff: str, pr_number: int):
    """Submit a review to the Batch API for 50% savings."""
    batch = client.batches.create(
        requests=[
            {
                "custom_id": f"pr-{pr_number}",
                "params": {
                    "model": "claude-sonnet-4-6",
                    "max_tokens": 4096,
                    "temperature": 0,
                    "system": REVIEW_PROMPT,
                    "messages": [
                        {"role": "user", "content": f"<diff>\n{diff}\n</diff>\n\nReview this PR diff."}
                    ],
                },
            }
        ],
    )
    print(f"Batch submitted: {batch.id}")
    return batch.id

# .gitlab-ci.yml
ai-review:
  stage: test
  image: python:3.12-slim
  script:
    - pip install anthropic requests
    - python scripts/ai_review_gitlab.py
  rules:
    - if: $CI_PIPELINE_SOURCE == "merge_request_event"
  variables:
    ANTHROPIC_API_KEY: $ANTHROPIC_API_KEY
    GITLAB_TOKEN: $CI_JOB_TOKEN
    MR_IID: $CI_MERGE_REQUEST_IID
    PROJECT_ID: $CI_PROJECT_ID

# Skip diffs containing potential secrets
secret_patterns = [".env", "credentials", "secret_key", "api_key", "password"]
for pattern in secret_patterns:
    if pattern in diff.lower():
        print(f"Warning: diff may contain secrets ({pattern}). Skipping review.")
        sys.exit(0)