Build an Autonomous Cost Anomaly Remediation Agent with Claude API

import anthropic
import boto3
import json
from datetime import datetime, timedelta
 
client = anthropic.Anthropic()
ce = boto3.client("ce")    # Cost Explorer
 
 
def get_cost_breakdown(days: int = 2) -> dict:
    """Pull cost by service, broken down by day, to spot what actually changed."""
    end = datetime.utcnow().date()
    start = end - timedelta(days=days)
 
    response = ce.get_cost_and_usage(
        TimePeriod={"Start": str(start), "End": str(end)},
        Granularity="DAILY",
        Metrics=["UnblendedCost"],
        GroupBy=[{"Type": "DIMENSION", "Key": "SERVICE"}]
    )
    return response["ResultsByTime"]
 
 
def find_recent_resource_changes(hours: int = 48) -> list[dict]:
    """CloudTrail events that could explain a cost spike — new resources,
    instance type changes, or anything a runaway script might have created."""
    cloudtrail = boto3.client("cloudtrail")
    start_time = datetime.utcnow() - timedelta(hours=hours)
 
    events = cloudtrail.lookup_events(
        LookupAttributes=[{"AttributeKey": "EventName", "AttributeValue": "RunInstances"}],
        StartTime=start_time
    )
    # Repeat for other cost-relevant event names: CreateDBInstance, CreateVolume, etc.
    return [{"event": e["EventName"], "time": str(e["EventTime"]), "user": e.get("Username")}
            for e in events["Events"]]

DIAGNOSE_PROMPT = """A cost anomaly was detected. Diagnose the likely root cause.
 
Cost breakdown by service, last {days} days:
{cost_breakdown}
 
Recent resource creation events (CloudTrail):
{recent_events}
 
Identify:
1. Which service/resource type is driving the spike
2. The most likely specific cause (e.g. "forgotten dev instance running since
   last Friday", "runaway auto-scaling from a bad HPA config", "a script
   left NAT gateways provisioned in an unused VPC")
3. Whether this looks safe to auto-remediate (e.g. an obviously orphaned
   resource with no traffic) or needs human judgment (e.g. could be
   legitimate new production load)
 
Respond with ONLY valid JSON:
{{"likely_cause": "...", "affected_resources": ["..."], "safe_to_auto_remediate": true/false, "confidence": "high"|"medium"|"low"}}"""
 
 
def diagnose_anomaly(cost_breakdown: dict, recent_events: list) -> dict:
    response = client.messages.create(
        model="claude-sonnet-5",
        max_tokens=800,
        messages=[{
            "role": "user",
            "content": DIAGNOSE_PROMPT.format(
                days=2, cost_breakdown=json.dumps(cost_breakdown, indent=2)[:4000],
                recent_events=json.dumps(recent_events, indent=2)[:2000],
            )
        }]
    )
    text = response.content[0].text.strip()
    if text.startswith("```"):
        text = text.split("```")[1].replace("json", "", 1).strip()
    return json.loads(text)

SAFE_REMEDIATIONS = {
    "unattached_ebs_volume": "snapshot_then_delete",
    "idle_load_balancer_no_targets": "delete",
    "stopped_instance_over_30_days": "notify_only",    # never auto-delete, just flag
    "unused_elastic_ip": "release",
}
 
 
def remediate(diagnosis: dict):
    if not diagnosis["safe_to_auto_remediate"] or diagnosis["confidence"] != "high":
        notify_finops_channel(
            f"Cost anomaly diagnosed but needs review: {diagnosis['likely_cause']}\n"
            f"Affected: {diagnosis['affected_resources']}"
        )
        return
 
    for resource in diagnosis["affected_resources"]:
        resource_type = classify_resource(resource)
        action = SAFE_REMEDIATIONS.get(resource_type)
 
        if action == "snapshot_then_delete":
            snapshot_id = create_snapshot(resource)
            delete_volume(resource)
            log_remediation(resource, f"Snapshotted as {snapshot_id}, then deleted")
        elif action == "delete":
            delete_resource(resource)
            log_remediation(resource, "Deleted — no dependents found")
        elif action == "release":
            release_resource(resource)
            log_remediation(resource, "Released")
        else:    # notify_only, or unrecognized type
            notify_finops_channel(f"Flagged for manual review: {resource}")

# Every remediation gets logged with full context for audit
def log_remediation(resource: str, action: str):
    audit_log.write({
        "timestamp": datetime.utcnow().isoformat(),
        "resource": resource,
        "action": action,
        "reversible": True,
        "triggered_by": "cost-anomaly-agent-v1",
    })