Build an AI Deployment Pipeline with Python

pip install anthropic gitpython

import anthropic
import json
import subprocess

client = anthropic.Anthropic(
    api_key="sk-your-ezai-key",
    base_url="https://ezaiapi.com"
)

def get_deploy_context(repo_path: str, max_diff_chars: int = 12000) -> dict:
    """Extract diff and recent commits from the repo."""

    diff = subprocess.run(
        ["git", "diff", "HEAD~1", "HEAD"],
        cwd=repo_path, capture_output=True, text=True
    ).stdout[:max_diff_chars]

    commits = subprocess.run(
        ["git", "log", "--oneline", "-10"],
        cwd=repo_path, capture_output=True, text=True
    ).stdout.strip()

    changed_files = subprocess.run(
        ["git", "diff", "--name-only", "HEAD~1", "HEAD"],
        cwd=repo_path, capture_output=True, text=True
    ).stdout.strip().split("\n")

    return {
        "diff": diff,
        "commits": commits,
        "changed_files": changed_files,
        "file_count": len(changed_files)
    }

REVIEW_PROMPT = """You are a senior SRE reviewing a deployment.
Analyze this diff and return a JSON object with:

{
  "risk_score": 0-100,
  "recommendation": "deploy" | "hold" | "block",
  "summary": "one paragraph explaining the changes",
  "risks": [
    {"category": "schema|auth|deps|config|data|infra",
     "severity": "low|medium|high|critical",
     "detail": "what specifically could go wrong"}
  ],
  "rollback_notes": "what to watch and how to revert"
}

Risk scoring guide:
- 0-25: Cosmetic, docs, tests — safe to auto-deploy
- 26-50: Feature code, new endpoints — standard review
- 51-75: DB migrations, auth changes, config — needs human approval
- 76-100: Breaking changes, data migrations, infra — block and discuss

Changed files: {changed_files}
Recent commits: {commits}

Diff:
{diff}

Return ONLY valid JSON. No markdown, no explanation outside the JSON."""

def analyze_deploy(context: dict) -> dict:
    prompt = REVIEW_PROMPT.format(
        changed_files=", ".join(context["changed_files"]),
        commits=context["commits"],
        diff=context["diff"]
    )

    response = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=1500,
        messages=[{"role": "user", "content": prompt}]
    )

    return json.loads(response.content[0].text)

import sys
import os

def run_deploy_gate():
    repo_path = os.environ.get("GITHUB_WORKSPACE", ".")
    context = get_deploy_context(repo_path)

    print(f"📋 Analyzing {context['file_count']} changed files...")

    result = analyze_deploy(context)
    score = result["risk_score"]
    rec = result["recommendation"]

    print(f"\n{'='*50}")
    print(f"Risk Score: {score}/100")
    print(f"Recommendation: {rec.upper()}")
    print(f"Summary: {result['summary']}")

    if result["risks"]:
        print(f"\n⚠️  Risks found:")
        for risk in result["risks"]:
            icon = {"critical": "🔴", "high": "🟠",
                    "medium": "🟡", "low": "🟢"}
            print(f"  {icon.get(risk['severity'], '⚪')} [{risk['category']}] {risk['detail']}")

    print(f"\n🔄 Rollback: {result['rollback_notes']}")

    if rec == "block":
        print("\n🚫 DEPLOY BLOCKED — manual review required")
        sys.exit(1)
    elif rec == "hold":
        print("\n⏸️  DEPLOY ON HOLD — flagged for review")
        # Exit 0 but post a Slack/Discord alert
    else:
        print("\n✅ DEPLOY APPROVED — proceeding")

if __name__ == "__main__":
    run_deploy_gate()

name: Deploy with AI Gate
on:
  push:
    branches: [main]

jobs:
  ai-gate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
        with:
          fetch-depth: 2  # Need parent commit for diff
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - run: pip install anthropic
      - run: python scripts/deploy_gate.py
        env:
          ANTHROPIC_API_KEY: ${{ secrets.EZAI_API_KEY }}

  deploy:
    needs: ai-gate
    runs-on: ubuntu-latest
    steps:
      - run: echo "Deploying to production..."

import urllib.request

def notify_slack(result: dict, webhook_url: str):
    score = result["risk_score"]
    emoji = "🚫" if score > 75 else "⚠️" if score > 50 else "✅"

    risks_text = "\n".join(
        f"• [{r['severity'].upper()}] {r['detail']}"
        for r in result["risks"]
    )

    payload = json.dumps({"text":
        f"{emoji} *Deploy Gate — Score: {score}/100*\n"
        f"Rec: {result['recommendation'].upper()}\n\n"
        f"{result['summary']}\n\n"
        f"*Risks:*\n{risks_text}\n\n"
        f"_Rollback: {result['rollback_notes']}_"
    })

    req = urllib.request.Request(
        webhook_url, data=payload.encode(),
        headers={"Content-Type": "application/json"}
    )
    urllib.request.urlopen(req)