Build an Autonomous Deployment Rollback Agent with Claude API

import requests
import time
from datetime import datetime, timedelta
 
PROMETHEUS_URL = "http://prometheus.monitoring:9090"
 
 
def query_prometheus(promql: str) -> float:
    resp = requests.get(f"{PROMETHEUS_URL}/api/v1/query", params={"query": promql})
    result = resp.json()["data"]["result"]
    return float(result[0]["value"][1]) if result else 0.0
 
 
def get_deployment_metrics(deployment: str, namespace: str) -> dict:
    """Pull the metrics that actually indicate a bad deploy."""
    return {
        "error_rate_5m": query_prometheus(
            f'sum(rate(http_requests_total{{deployment="{deployment}",namespace="{namespace}",status=~"5.."}}[5m])) '
            f'/ sum(rate(http_requests_total{{deployment="{deployment}",namespace="{namespace}"}}[5m]))'
        ),
        "p99_latency_ms": query_prometheus(
            f'histogram_quantile(0.99, rate(http_request_duration_seconds_bucket{{deployment="{deployment}",namespace="{namespace}"}}[5m])) * 1000'
        ),
        "pod_restarts_5m": query_prometheus(
            f'sum(increase(kube_pod_container_status_restarts_total{{namespace="{namespace}",pod=~"{deployment}-.*"}}[5m]))'
        ),
        "pods_ready": query_prometheus(
            f'sum(kube_pod_status_ready{{namespace="{namespace}",pod=~"{deployment}-.*",condition="true"}})'
        ),
    }

import anthropic
import json
 
client = anthropic.Anthropic()
 
DECISION_PROMPT = """You are an SRE deciding whether to roll back a deployment.
 
Deployment: {deployment}
Time since rollout started: {minutes_since} minutes
 
Baseline metrics (previous stable version, averaged over last 24h):
{baseline}
 
Current metrics (new version, last 5 minutes):
{current}
 
Rules of thumb, not hard cutoffs:
- A 2-3x error rate spike in the first 2 minutes is often just cache warming — usually "watch"
- A sustained error rate increase past 5 minutes that is NOT trending down is "rollback"
- Pod restarts increasing means crash-looping, not warming up — usually "rollback" even early
- Latency creeping up slowly over 10+ minutes with no restarts is often a real regression, not noise
 
Respond with ONLY valid JSON:
{{"decision": "healthy" | "watch" | "rollback", "confidence": "high" | "medium" | "low", "reason": "one sentence"}}"""
 
 
def decide(deployment: str, minutes_since: int, baseline: dict, current: dict) -> dict:
    response = client.messages.create(
        model="claude-sonnet-5",
        max_tokens=200,
        messages=[{
            "role": "user",
            "content": DECISION_PROMPT.format(
                deployment=deployment,
                minutes_since=minutes_since,
                baseline=json.dumps(baseline, indent=2),
                current=json.dumps(current, indent=2),
            )
        }]
    )
    text = response.content[0].text.strip()
    if text.startswith("```"):
        text = text.split("```")[1].replace("json", "", 1).strip()
    return json.loads(text)

import subprocess
import time
 
def watch_rollout(deployment: str, namespace: str, baseline: dict, max_minutes: int = 20):
    start = time.time()
 
    while (time.time() - start) < max_minutes * 60:
        current = get_deployment_metrics(deployment, namespace)
        minutes_since = int((time.time() - start) / 60)
 
        verdict = decide(deployment, minutes_since, baseline, current)
        print(f"[{minutes_since}m] {verdict['decision']} ({verdict['confidence']}): {verdict['reason']}")
 
        if verdict["decision"] == "rollback" and verdict["confidence"] in ("high", "medium"):
            rollback(deployment, namespace, verdict["reason"])
            return
        if verdict["decision"] == "healthy" and minutes_since >= 5:
            print(f"Deployment {deployment} confirmed healthy after {minutes_since}m — stopping watch")
            return
 
        time.sleep(30)
 
    print(f"Watch window expired for {deployment} — no rollback triggered, handing off to normal monitoring")
 
 
def rollback(deployment: str, namespace: str, reason: str):
    print(f"ROLLING BACK {deployment}: {reason}")
    subprocess.run(["kubectl", "rollout", "undo", f"deployment/{deployment}", "-n", namespace])
    notify_slack(f"🔴 Auto-rolled back `{deployment}` in `{namespace}`\nReason: {reason}")
 
 
def notify_slack(message: str):
    requests.post(SLACK_WEBHOOK_URL, json={"text": message})

# Flagger webhook that calls the Claude decision agent
webhooks:
- name: claude-rollback-check
  type: rollback
  url: http://rollback-agent.monitoring/check
  timeout: 10s