Autonomous Chaos Engineering: AI Agents That Design Their Own Experiments in 2026

import anthropic
import json
 
client = anthropic.Anthropic()
 
 
def gather_system_context(service_name: str) -> dict:
    return {
        "dependency_graph": get_service_dependencies(service_name),    # from service mesh / tracing data
        "past_incidents": get_incident_history(service_name, months=12),
        "current_redundancy": get_replica_and_az_distribution(service_name),
        "existing_chaos_tests_run": get_chaos_experiment_history(service_name),
        "sla_criticality": get_service_criticality_tier(service_name),
    }

DESIGN_PROMPT = """Design chaos engineering experiment candidates for this service.
 
Service: {service_name}
Dependency graph (what it calls, what calls it): {dependency_graph}
Past incidents (last 12 months): {past_incidents}
Current redundancy (replicas, AZ distribution): {current_redundancy}
Chaos experiments already run on this service: {existing_tests}
Criticality tier: {sla_criticality}
 
Propose 3-5 experiment candidates that:
1. Target failure modes NOT already covered by existing_tests
2. Are informed by past_incidents — if a similar failure mode caused a
   real incident before, prioritize verifying it's actually fixed
3. Are appropriately scoped to criticality tier (don't propose a
   full-AZ failure test on a tier-1 payment service without extensive
   safeguards; simple pod-kill tests are fine at any tier)
 
For each candidate, specify:
- Hypothesis (what you expect to happen if the system is healthy)
- Blast radius and safety guardrails (max % of traffic affected, auto-abort conditions)
- Success criteria (what "the system handled this correctly" looks like)
 
Respond with ONLY valid JSON:
{{"candidates": [
  {{"name": "...", "hypothesis": "...", "blast_radius": "...",
    "safety_guardrails": "...", "success_criteria": "...", "priority": "high"|"medium"|"low"}}
]}}"""
 
 
def design_experiments(service_name: str, context: dict) -> dict:
    response = client.messages.create(
        model="claude-sonnet-5",
        max_tokens=2500,
        messages=[{
            "role": "user",
            "content": DESIGN_PROMPT.format(service_name=service_name, **context)
        }]
    )
    text = response.content[0].text.strip()
    if text.startswith("```"):
        text = text.split("```")[1].replace("json", "", 1).strip()
    return json.loads(text)

# Agent-proposed candidate translated into a Litmus ChaosEngine
# — a human reviews and approves this before it runs
apiVersion: litmuschaos.io/v1alpha1
kind: ChaosEngine
metadata:
  name: payments-db-pool-exhaustion
spec:
  appinfo:
    appns: production
    applabel: app=payments-api
  chaosServiceAccount: litmus-admin
  experiments:
    - name: pod-network-latency
      spec:
        components:
          env:
            - name: TARGET_CONTAINER
              value: payments-api
            - name: NETWORK_LATENCY
              value: "2000"    # simulates slow DB responses -> pool exhaustion pressure
            - name: TOTAL_CHAOS_DURATION
              value: "60"