Swarms of on-call agents

from autonomy import Agent, Model, Tool
# ...

# 3 regions × 10 services × 5 specialists = 150 diagnostic agents
regions = ["us-east-1", "us-west-2", "eu-west-1", ...]
services = ["api-gateway", "user-service", "order-service", ...]
specialists = ["database", "cache", "network", "resources", "logs"]

# Each specialist type has its own diagnostic tools
def tools_for(specialist):
  # database → query connections, find slow queries
  # cache → check hit rates, memory pressure  
  # network → check latency, packet loss
  # resources → check CPU, memory, disk
  # logs → search for errors, trace requests
  return [...]

# Define an agent and start investigation
async def investigator_agent(specialist, service, region, incident):
  agent = await Agent.start(
    node,
    instructions=f"You're a {specialist} diagnostic agent for {service} in {region}...",
    model=Model("claude-sonnet-4-5"),
    tools=tools_for(specialist),
  )
  return await agent.send(f"Investigate {incident}")

# 150 specialized agents, operating in a parallel scatter/gather pattern
findings = await asyncio.gather(*[
  investigator_agent(specialist, service, region, incident)
  for region in regions for service in services for specialist in specialists
])

class Monitor:
  """Watches for anomalies and spawns diagnostic investigations."""

  async def start(self, node: Node):
    self.node = node
    self.running = True
    self._monitor_task = asyncio.create_task(self._monitor_loop())

  async def _monitor_loop(self):
    """Background loop that checks metrics. In production, polls Prometheus/CloudWatch."""
    while self.running:
      # In production: check metrics here and call trigger_anomaly() when thresholds exceeded
      await asyncio.sleep(30)

# READ credentials needed for monitoring (requested at activation)
MONITORING_READ_CREDENTIALS = [
  "op://Infrastructure/prod-db-readonly/password",
  "op://Infrastructure/prod-db-readonly/username",
  "op://Infrastructure/prod-db-readonly/server",
  "op://Infrastructure/aws-cloudwatch/credential",
  "op://Infrastructure/k8s-prod-readonly/credential",
]

# WRITE credentials - require separate approval per action
CREDENTIAL_CATEGORIES: Dict[str, CredentialCategory] = {
  # READ credentials - approved once, shared across investigation
  "op://Infrastructure/prod-db-readonly/password": CredentialCategory.READ,
  # ...
  # WRITE credentials - require separate approval per action
  "op://Infrastructure/prod-db-rwaccess/username": CredentialCategory.WRITE,
  "op://Infrastructure/prod-db-rwaccess/password": CredentialCategory.WRITE,
  "op://Infrastructure/config-service/credential": CredentialCategory.WRITE,
}

async def run_swarm_diagnosis(node, problem, session_id, root_id, credentials):
  """Run parallel diagnosis across all regions and services.

  Spawns 150+ diagnostic agents (3 regions × 10 services × 5 agent types)
  all running in parallel. Agents are actors, so they don't block.
  """
  targets = []
  for region in REGIONS:
    for service in SERVICES:
      targets.append({"service": service, "region": region, ...})

  # Run ALL targets in parallel - agents are actors, they don't block
  results = await asyncio.gather(*[run_target(t) for t in targets])
  return results

async def run_single_agent(self, agent_type, instructions, service_name, region, model):
  """Run a single diagnostic agent."""
  agent = await Agent.start(
    node=self.node,
    instructions=f"""You are a {agent_type} diagnostic agent for {service_name} in {region}.
      {instructions}
      Provide a brief assessment of the {agent_type} health for this service.""",
    model=model,
  )

  message = f"Diagnose {agent_type} health for {service_name} in {region}."
  responses = await agent.send(message)

  return {
    "agent_type": agent_type,
    "status": "completed",
    "finding": responses[-1].content.text
  }

synthesis_agent = await Agent.start(
  node=self.node,
  instructions="""You are a senior SRE synthesizing diagnostic findings.
    Identify common patterns, determine root cause, and prioritize remediation.
    List any actions requiring WRITE access as "Actions Requiring Approval".""",
  model=Model("claude-sonnet-4-5"),
)

diagnosis = await synthesis_agent.send(
  f"Synthesize findings: {findings_summary}"
)

{
  "anomaly_type": "cascading_failure",
  "severity": "critical",
  "message": "Optional custom message"
}

{
  "status": "pending",
  "message": "Anomaly reported - monitoring agent analyzing metrics...",
  "anomaly_type": "cascading_failure"
}

{
  "active": true,
  "investigation": {
    "investigation_id": "abc123",
    "status": "running",
    "anomaly_type": "cascading_failure",
    "anomaly_message": "Detected cascading failure...",
    "agents_count": 150,
    "session_id": "def456",
    "created_at": "2024-01-15T10:30:00Z"
  },
  "message": "Investigation abc123 is running"
}

{
  "approved": true
}