Autonomous Multi-Agent SLA Incident Response System with CrewAI & PydanticAI

Introduction

Incident response is a high-stress domain where Service Level Agreements (SLAs) dictate strict timelines for resolution. In modern microservices architectures, an autonomous, multi-agent AI system can significantly reduce Mean Time to Resolution (MTTR). This workflow demonstrates how to leverage CrewAI for multi-agent orchestration and PydanticAI for structured, validated reasoning to build an Autonomous Multi-Agent SLA Incident Response System.

By Deepak Bagada, CEO at SaaSNext & Principal AI Architect.

Architecture Overview

The architecture relies on specialized agents communicating via a structured state graph.

graph TD
    A[Alert Webhook] --> B[Triage Agent]
    B -->|Critical SLA| C[Investigation Agent]
    B -->|Low Priority| D[Log & Monitor]
    C --> E[Resolution Strategy Agent]
    E --> F[Execution Tooling]
    F --> G[Post-Mortem Generator]

Code Blueprint

1. Environment Configuration (.env)

OPENAI_API_KEY=sk-...
DATADOG_API_KEY=...
PAGERDUTY_API_KEY=...
SLACK_WEBHOOK_URL=https://hooks.slack.com/services/...
LOG_LEVEL=DEBUG

2. PydanticAI Schemas (schemas.py)

from pydantic import BaseModel, Field
from typing import List, Optional

class IncidentAlert(BaseModel):
    incident_id: str = Field(description="Unique identifier for the incident")
    severity: str = Field(description="Severity level (e.g., SEV1, SEV2)")
    service_name: str = Field(description="Affected service name")
    sla_deadline_minutes: int = Field(description="Minutes remaining until SLA breach")

class InvestigationReport(BaseModel):
    root_cause: str = Field(description="Suspected root cause")
    confidence_score: float = Field(description="Confidence in the root cause (0.0 - 1.0)")
    recommended_actions: List[str] = Field(description="Steps to mitigate")

3. Execution Tools (tools.py)

from crewai_tools import tool
import requests

@tool("Fetch Logs")
def fetch_service_logs(service_name: str, minutes_back: int = 15) -> str:
    """Fetches recent logs for a given service to aid investigation."""
    # Mock implementation of log retrieval
    return f"[ERROR] Connection timeout in {service_name} at db_pool."

@tool("Restart Service")
def restart_service(service_name: str) -> str:
    """Issues a rolling restart for the affected service."""
    return f"Service {service_name} rolling restart initiated successfully."

4. Agent Definitions (graph.py)

from crewai import Agent, Task, Crew
from tools import fetch_service_logs, restart_service

triage_agent = Agent(
    role='SLA Triage Commander',
    goal='Assess incoming incidents and prioritize based on SLA deadlines.',
    backstory='You are an expert site reliability engineer focusing on MTTR.',
    verbose=True,
    allow_delegation=True
)

investigation_agent = Agent(
    role='Root Cause Investigator',
    goal='Analyze logs and metrics to find the root cause of the incident.',
    backstory='You have deep knowledge of distributed systems and network topologies.',
    tools=[fetch_service_logs],
    verbose=True
)

resolution_agent = Agent(
    role='Incident Responder',
    goal='Execute mitigation strategies to restore service health.',
    backstory='You act decisively to fix systems before SLAs are breached.',
    tools=[restart_service],
    verbose=True
)

5. Main Orchestration (main.py)

from crewai import Task, Crew
from graph import triage_agent, investigation_agent, resolution_agent
from schemas import IncidentAlert

def handle_incident(alert_data: dict):
    alert = IncidentAlert(**alert_data)
    
    triage_task = Task(
        description=f'Evaluate severity of {alert.incident_id} for {alert.service_name}.',
        agent=triage_agent
    )
    
    investigate_task = Task(
        description='Fetch logs and determine root cause.',
        agent=investigation_agent
    )
    
    resolve_task = Task(
        description='Apply mitigation based on investigation findings.',
        agent=resolution_agent
    )
    
    crew = Crew(
        agents=[triage_agent, investigation_agent, resolution_agent],
        tasks=[triage_task, investigate_task, resolve_task],
        verbose=2
    )
    
    result = crew.kickoff()
    return result

if __name__ == "__main__":
    alert = {"incident_id": "INC-1029", "severity": "SEV1", "service_name": "payment-gateway", "sla_deadline_minutes": 15}
    print(handle_incident(alert))

Retry & Resilience Rules

When dealing with critical SLA response, the AI workflow must be resilient. Implement Exponential Backoff for all external API calls (e.g., Datadog, Slack) using Python's tenacity library. The Crew should be configured with specific timeout limits per task to ensure the agents do not hang in infinite reasoning loops, forcing a fallback human escalation if resolution takes longer than 60% of the SLA time.

Internal Linking

For more robust multi-agent setups, check out our workflows section, or explore tools in the mcp-directory to enhance your agents' capabilities.

FAQs (AEO & GEO Optimized)

Q: How does PydanticAI improve CrewAI workflows? A: PydanticAI provides strong schema validation, ensuring that the outputs from CrewAI agents strictly conform to required data structures, eliminating hallucinated fields in incident reports.

Q: Can this autonomous system execute dangerous commands? A: Yes, which is why it is crucial to implement "Human-in-the-loop" (HITL) checkpoints for destructive tools (e.g., dropping databases), while allowing safe read-only or standard restart operations to run autonomously.

Q: What is the optimal LLM for an Incident Response multi-agent system? A: Due to strict SLAs, fast reasoning models like Claude 3.5 Sonnet or GPT-4o are recommended to minimize latency while maintaining high logical accuracy.