Build an AI Incident Response Agent

Introduction

30 Second Summary

When a production alert fires at 3 AM, the investigation process is often the slowest part. Engineers spend most of their time gathering evidence from logs and dashboards before they can even begin fixing the problem.

In this project, you will build an AI-powered incident response agent that automates the investigation workflow. Your agent triages alerts, analyzes logs, determines root cause, and generates a remediation plan while keeping humans in control of the final decision.

What You'll Build

A command-line AI agent that processes infrastructure alerts through a structured LangGraph workflow with autonomous tool use and human-in-the-loop approval.

By the end of this project, you'll have:

• A LangGraph stateful workflow with 4 nodes for triage, root cause analysis, remediation planning, and execution.
• Custom SRE tools for log search, metrics query, and dependency mapping that the AI agent calls autonomously.
• A human-in-the-loop approval gate that pauses the workflow for engineer review before any remediation executes.
• Secret Mission: Add a post-mortem generation step that automatically produces a structured incident report after resolution.

Are there any prerequisites?

You should have basic Python experience (variables, functions, imports). Familiarity with DevOps concepts like alerts, incidents, and on-call rotations is helpful but not required.

Before We Start

Before We Start...

Set Up Your Environment

Before your AI agent can investigate incidents, you need a working development environment with the right tools installed.

In this step, you will install Python, create an isolated project folder, and configure access to the Claude API that powers your agent's reasoning.

In this step, get ready to:

• Install and verify Python 3.10+.
• Create a project directory with a virtual environment.
• Install project dependencies and configure your API key.

Verify your Python installation

This project requires Python 3.10 or higher because LangGraph depends on modern Python features like type unions and structural pattern matching.

• Check your Python version by running this command:

macOS

python3 --version

Windows

python --version

✔️ I see version 3.10 or higher

You are good to go. Move on to the next substep.

ⓧ I see an older version

Your system has an older Python that does not meet the 3.10+ requirement.

• Download Python 3.12 from the official Python downloads page.
• Run the installer and follow the prompts.
• On macOS, the system default python3 may still point to the old version after installing. Use python3.12 directly, or install via pyenv to manage versions cleanly.
• After installing, close and reopen your terminal, then re-run the version check command above.

ⓧ Command not found

Python is not installed on your machine yet.

• Download Python 3.12 from the official Python downloads page.
• Run the installer. On Windows, check the box that says Add Python to PATH before clicking Install Now.
• Close and reopen your terminal after installation completes.
• Run the version check command above again to confirm you see 3.10 or higher.

Create your project directory and virtual environment

A virtual environment keeps your project's packages isolated from the rest of your system.

• Open your terminal.
• Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
• Type Terminal (macOS) or PowerShell (Windows) and press Enter to open it.

• Move to your Desktop by running this command:

cd ~/Desktop

• Create your project folder and move into it by running these commands:

mkdir incident-agent
cd incident-agent

• Confirm the folder exists by running:

ls

The folder will be empty for now. That is expected.

• Create a virtual environment by running this command:

macOS

python3 -m venv venv

Windows

python -m venv venv

• Activate the virtual environment:

macOS

source venv/bin/activate

Windows

venv\Scripts\activate

You should see (venv) appear at the start of your terminal prompt. This confirms the virtual environment is active.

Don't see (venv) in your prompt?

On Windows, you may need to allow script execution first. Run Set-ExecutionPolicy -Scope CurrentUser -ExecutionPolicy RemoteSigned in PowerShell, then try the activate command again.

On macOS, make sure you ran source venv/bin/activate (not just venv/bin/activate without source).

Help me activate my Python virtual environment.

• Now open your project in a code editor. Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
• Type Cursor or VS Code and press Enter to open it.
• In the editor, click File in the top menu bar.
• Click Open Folder.
• Navigate to your Desktop and select the incident-agent folder, then click Open.

Install dependencies and configure your API key

Your agent needs three Python packages to function. You also need an Anthropic API key so Claude can power the agent's reasoning.

• Install all three dependencies by running this command in your terminal (make sure (venv) is still showing):

pip install langgraph==1.2.1 anthropic==0.104.0 pydantic==2.11.9

What are these packages?

• langgraph is the workflow framework that orchestrates your agent's investigation steps as a stateful graph.
• anthropic is the official Python SDK for calling the Claude API.
• pydantic provides data validation and structured models for your incident state.

You should see a series of download and install messages ending with Successfully installed followed by the package names.

Seeing a permission error or install failure?

Make sure your virtual environment is active (you should see (venv) in your prompt). If not, re-run the activate command from the previous substep.

If you see a network error, check your internet connection and try again.

Help me fix my pip install error.

Now you need an API key from Anthropic to connect your agent to Claude.

• Open your browser and go to console.anthropic.com.
• Sign up for a new account or sign in if you already have one.
• Navigate to Settings, then API Keys.
• Click Create Key to generate a new API key.
• Copy the key immediately. You will not be able to see it again after closing the dialog.

Why do I need an API key?

The API key authenticates your agent with Anthropic's servers. Without it, your code cannot send prompts to Claude or receive responses.

Keep this key private. Never commit it to version control or share it publicly.

• Set the API key as an environment variable in your terminal:

macOS

export ANTHROPIC_API_KEY='your-api-key-here'

Windows

set ANTHROPIC_API_KEY=your-api-key-here

• Replace your-api-key-here with the actual key you just copied from the Anthropic console.

Keep your key available for this session

This environment variable only lasts for your current terminal session. If you close and reopen your terminal, you will need to set it again before running the agent.

Your environment is fully set up. Next, you will define the data models and SRE tools that give your agent the ability to investigate incidents.

Define Incident State and SRE Tools

Your incident response agent needs to track structured data as it moves through the investigation workflow. Every real SRE agent maintains state about the alert, its severity, the evidence gathered, and the remediation plan.

In this step, you will define Pydantic models that represent incident state, then build mock SRE tools that simulate log searches, metrics queries, and dependency mapping. These tools give the AI agent something to call during its investigation.

In this step, get ready to:

• Create Pydantic models for incident state (alert data, severity, findings, remediation plan).
• Build 3 mock SRE tools that simulate observability systems.
• Wire the tools into a format Claude can call via tool use.

Create Pydantic models for incident state

The agent needs a typed, validated structure to hold incident data as it flows through triage, root cause analysis, and remediation. Pydantic models enforce that every piece of state has the correct shape.

• Create a new file called models.py in your incident-agent folder.
• Add the imports, the Severity enum, and the Alert model by pasting this code:

from pydantic import BaseModel, Field
from typing import Optional
from enum import Enum


# Severity levels for incident classification
class Severity(str, Enum):
    CRITICAL = "critical"
    HIGH = "high"
    MEDIUM = "medium"
    LOW = "low"


# Incoming alert data from monitoring systems
class Alert(BaseModel):
    alert_id: str
    service: str
    description: str
    timestamp: str
    raw_severity: str

What does this code do?

Severity is a string enum that constrains severity to exactly four levels. Inheriting from both str and Enum means the value serializes cleanly to JSON.

Alert captures the raw data from a monitoring system. It holds an ID, the affected service name, a human-readable description, when it fired, and the monitoring system's own severity label.

• Below the Alert class, add the investigation and state models:

# Evidence gathered during root cause analysis
class Finding(BaseModel):
    source: str
    detail: str
    relevance: str


# A single step in the remediation plan
class RemediationStep(BaseModel):
    action: str
    target: str
    risk_level: str
    rollback: str


# The full incident state that flows through the workflow
class IncidentState(BaseModel):
    alert: Alert
    severity: Optional[Severity] = None
    findings: list[Finding] = Field(default_factory=list)
    root_cause: Optional[str] = None
    remediation_plan: list[RemediationStep] = Field(default_factory=list)
    approved: Optional[bool] = None
    execution_result: Optional[str] = None
    postmortem: Optional[str] = None

What does this code do?

Finding stores a piece of evidence the agent discovers (which tool found it, what it says, how relevant it is). RemediationStep describes a single action to fix the problem, including its risk and how to undo it.

IncidentState ties everything together. It starts with just an alert, then accumulates severity, findings, root cause, and a remediation plan as the agent works. Field(default_factory=list) ensures each instance gets its own fresh list instead of sharing one.

• Save models.py.
• Confirm Python can import it by running:

python -c "from models import IncidentState; print('Models loaded OK')"

You should see Models loaded OK printed in the terminal.

Seeing an import error?

Make sure your virtual environment is activated (you should see (venv) in your terminal prompt). Also check that models.py is saved in the incident-agent folder where your terminal is currently working.

help me fix a Pydantic import error in my models.py file

✔️ Awesome, I've got everything!

Great. Make sure you have saved models.py before moving on.

ⓧ I'd like to double check the full code

from pydantic import BaseModel, Field
from typing import Optional
from enum import Enum


class Severity(str, Enum):
    CRITICAL = "critical"
    HIGH = "high"
    MEDIUM = "medium"
    LOW = "low"


class Alert(BaseModel):
    alert_id: str
    service: str
    description: str
    timestamp: str
    raw_severity: str


class Finding(BaseModel):
    source: str
    detail: str
    relevance: str


class RemediationStep(BaseModel):
    action: str
    target: str
    risk_level: str
    rollback: str


class IncidentState(BaseModel):
    alert: Alert
    severity: Optional[Severity] = None
    findings: list[Finding] = Field(default_factory=list)
    root_cause: Optional[str] = None
    remediation_plan: list[RemediationStep] = Field(default_factory=list)
    approved: Optional[bool] = None
    execution_result: Optional[str] = None
    postmortem: Optional[str] = None

Build 3 mock SRE tools

In production, an SRE agent would connect to real observability platforms like Datadog, Grafana, or PagerDuty. For this project, you will build mock tools that return realistic simulated data so you can test the agent locally without any external services.

• Create a new file called tools.py in your incident-agent folder.
• Add the imports and mock data by pasting this code:

import json
from typing import Any

# Simulated log entries for three services
MOCK_LOGS = {
    "api-gateway": [
        {"timestamp": "2026-05-21T03:14:00Z", "level": "ERROR", "message": "Connection timeout to downstream service user-service after 30s"},
        {"timestamp": "2026-05-21T03:14:05Z", "level": "ERROR", "message": "Circuit breaker OPEN for user-service, 50 consecutive failures"},
        {"timestamp": "2026-05-21T03:13:55Z", "level": "WARN", "message": "Response time p99 exceeded 5000ms threshold"},
    ],
    "user-service": [
        {"timestamp": "2026-05-21T03:12:00Z", "level": "ERROR", "message": "OutOfMemoryError: Java heap space"},
        {"timestamp": "2026-05-21T03:12:01Z", "level": "ERROR", "message": "GC overhead limit exceeded, heap usage at 98%"},
        {"timestamp": "2026-05-21T03:11:30Z", "level": "WARN", "message": "Memory usage crossed 90% threshold"},
    ],
    "deploy-pipeline": [
        {"timestamp": "2026-05-21T02:45:00Z", "level": "INFO", "message": "Deployment started: user-service v2.4.1"},
        {"timestamp": "2026-05-21T02:47:00Z", "level": "ERROR", "message": "Health check failed for user-service v2.4.1, pod CrashLoopBackOff"},
        {"timestamp": "2026-05-21T02:47:30Z", "level": "ERROR", "message": "Rollback initiated: reverting to user-service v2.4.0"},
    ],
}

# Current performance metrics per service
MOCK_METRICS = {
    "api-gateway": {"latency_p99_ms": 8500, "error_rate_pct": 23.5, "requests_per_sec": 1200, "cpu_pct": 45, "memory_pct": 62},
    "user-service": {"latency_p99_ms": 30000, "error_rate_pct": 95.0, "requests_per_sec": 0, "cpu_pct": 12, "memory_pct": 98},
    "payment-service": {"latency_p99_ms": 120, "error_rate_pct": 0.1, "requests_per_sec": 800, "cpu_pct": 30, "memory_pct": 55},
}

# Service dependency graph
MOCK_DEPENDENCIES = {
    "api-gateway": {"depends_on": ["user-service", "payment-service", "inventory-service"], "depended_by": ["load-balancer"]},
    "user-service": {"depends_on": ["postgres-primary", "redis-cache"], "depended_by": ["api-gateway", "notification-service"]},
    "payment-service": {"depends_on": ["stripe-api", "postgres-primary"], "depended_by": ["api-gateway", "order-service"]},
}

What does this mock data represent?

MOCK_LOGS holds timestamped log entries for three services, simulating what you would get from a centralized logging system like Elasticsearch or CloudWatch Logs.

MOCK_METRICS represents real-time health metrics (latency, error rate, throughput, resource usage) you would normally pull from Prometheus or Datadog.

MOCK_DEPENDENCIES maps which services depend on which, so the agent can assess blast radius during an outage.

• Below the mock data, add the three tool functions:

def search_logs(service: str, severity_filter: str = "ERROR") -> str:
    """Search application logs for a specific service, filtered by log level."""
    logs = MOCK_LOGS.get(service, [])
    filtered = [log for log in logs if log["level"] == severity_filter or severity_filter == "ALL"]
    if not filtered:
        return json.dumps({"service": service, "logs": [], "message": f"No {severity_filter} logs found for {service}"})
    return json.dumps({"service": service, "log_count": len(filtered), "logs": filtered})


def query_metrics(service: str) -> str:
    """Query current performance metrics for a service including latency, error rate, CPU, and memory."""
    metrics = MOCK_METRICS.get(service)
    if not metrics:
        return json.dumps({"service": service, "error": "Service not found in metrics system"})
    return json.dumps({"service": service, "metrics": metrics})


def get_service_dependencies(service: str) -> str:
    """Get the dependency graph for a service showing upstream and downstream connections."""
    deps = MOCK_DEPENDENCIES.get(service)
    if not deps:
        return json.dumps({"service": service, "error": "Service not found in dependency graph"})
    return json.dumps({"service": service, "dependencies": deps})

What do these functions do?

Each function takes a service name, looks up its data in the mock dictionaries, and returns a JSON string. Returning JSON strings (instead of dicts) matches how Claude's tool use protocol works. The AI receives tool results as text and parses them.

search_logs filters log entries by severity level. query_metrics returns the full health dashboard for a service. get_service_dependencies reveals upstream and downstream connections so the agent can reason about cascade failures.

• Save tools.py.
• Verify the tools work by running:

python -c "from tools import search_logs; print(search_logs('user-service'))"

You should see a JSON string containing two ERROR log entries for user-service (the OutOfMemoryError and GC overhead messages).

Seeing a ModuleNotFoundError?

Check that your terminal's working directory is the incident-agent folder (not a parent or subfolder). Run ls to confirm tools.py appears in the listing.

help me debug why Python cannot find my tools module

Wire tools into Claude's tool use format

Claude's tool use API requires tools to be described in a specific JSON schema format. Each tool needs a name, description, and an input_schema that tells Claude what parameters the tool accepts.

• At the bottom of tools.py, add the tool definitions and mapping:

# Tool definitions in the format Claude's API expects
TOOL_DEFINITIONS = [
    {
        "name": "search_logs",
        "description": "Search application logs for a specific service, filtered by log level. Use this to find error patterns and recent issues.",
        "input_schema": {
            "type": "object",
            "properties": {
                "service": {"type": "string", "description": "The service name to search logs for"},
                "severity_filter": {"type": "string", "enum": ["ERROR", "WARN", "INFO", "ALL"], "description": "Filter logs by severity level"},
            },
            "required": ["service"],
        },
    },
    {
        "name": "query_metrics",
        "description": "Query current performance metrics for a service including latency, error rate, CPU, and memory usage.",
        "input_schema": {
            "type": "object",
            "properties": {
                "service": {"type": "string", "description": "The service name to query metrics for"},
            },
            "required": ["service"],
        },
    },
    {
        "name": "get_service_dependencies",
        "description": "Get the dependency graph for a service showing upstream and downstream connections. Use this to understand blast radius.",
        "input_schema": {
            "type": "object",
            "properties": {
                "service": {"type": "string", "description": "The service name to get dependencies for"},
            },
            "required": ["service"],
        },
    },
]

# Maps tool names to their Python functions for execution
TOOL_MAP = {
    "search_logs": search_logs,
    "query_metrics": query_metrics,
    "get_service_dependencies": get_service_dependencies,
}

What does this code do?

TOOL_DEFINITIONS is the list you will pass to Claude's API so it knows what tools are available. Each entry describes one tool's name, purpose, and expected inputs using JSON Schema.

TOOL_MAP connects tool names (strings that Claude returns in its responses) to the actual Python functions that execute them. When Claude decides to call search_logs, your code looks up the function in this dictionary and runs it.

• Save tools.py.
• Verify the full module loads correctly by running:

python -c "from tools import TOOL_DEFINITIONS, TOOL_MAP; print(f'{len(TOOL_DEFINITIONS)} tools defined, {len(TOOL_MAP)} mapped')"

You should see 3 tools defined, 3 mapped printed in the terminal.

Seeing a syntax error?

The TOOL_DEFINITIONS list has deeply nested dictionaries. Check that every opening brace { has a matching closing brace }, and that every list bracket [ has a matching ]. A missing comma between dictionary entries is the most common issue.

help me find the syntax error in my TOOL_DEFINITIONS

✔️ Awesome, I've got everything!

Great. Make sure you have saved tools.py before moving on.

ⓧ I'd like to double check the full code

import json
from typing import Any

MOCK_LOGS = {
    "api-gateway": [
        {"timestamp": "2026-05-21T03:14:00Z", "level": "ERROR", "message": "Connection timeout to downstream service user-service after 30s"},
        {"timestamp": "2026-05-21T03:14:05Z", "level": "ERROR", "message": "Circuit breaker OPEN for user-service, 50 consecutive failures"},
        {"timestamp": "2026-05-21T03:13:55Z", "level": "WARN", "message": "Response time p99 exceeded 5000ms threshold"},
    ],
    "user-service": [
        {"timestamp": "2026-05-21T03:12:00Z", "level": "ERROR", "message": "OutOfMemoryError: Java heap space"},
        {"timestamp": "2026-05-21T03:12:01Z", "level": "ERROR", "message": "GC overhead limit exceeded, heap usage at 98%"},
        {"timestamp": "2026-05-21T03:11:30Z", "level": "WARN", "message": "Memory usage crossed 90% threshold"},
    ],
    "deploy-pipeline": [
        {"timestamp": "2026-05-21T02:45:00Z", "level": "INFO", "message": "Deployment started: user-service v2.4.1"},
        {"timestamp": "2026-05-21T02:47:00Z", "level": "ERROR", "message": "Health check failed for user-service v2.4.1, pod CrashLoopBackOff"},
        {"timestamp": "2026-05-21T02:47:30Z", "level": "ERROR", "message": "Rollback initiated: reverting to user-service v2.4.0"},
    ],
}

MOCK_METRICS = {
    "api-gateway": {"latency_p99_ms": 8500, "error_rate_pct": 23.5, "requests_per_sec": 1200, "cpu_pct": 45, "memory_pct": 62},
    "user-service": {"latency_p99_ms": 30000, "error_rate_pct": 95.0, "requests_per_sec": 0, "cpu_pct": 12, "memory_pct": 98},
    "payment-service": {"latency_p99_ms": 120, "error_rate_pct": 0.1, "requests_per_sec": 800, "cpu_pct": 30, "memory_pct": 55},
}

MOCK_DEPENDENCIES = {
    "api-gateway": {"depends_on": ["user-service", "payment-service", "inventory-service"], "depended_by": ["load-balancer"]},
    "user-service": {"depends_on": ["postgres-primary", "redis-cache"], "depended_by": ["api-gateway", "notification-service"]},
    "payment-service": {"depends_on": ["stripe-api", "postgres-primary"], "depended_by": ["api-gateway", "order-service"]},
}


def search_logs(service: str, severity_filter: str = "ERROR") -> str:
    """Search application logs for a specific service, filtered by log level."""
    logs = MOCK_LOGS.get(service, [])
    filtered = [log for log in logs if log["level"] == severity_filter or severity_filter == "ALL"]
    if not filtered:
        return json.dumps({"service": service, "logs": [], "message": f"No {severity_filter} logs found for {service}"})
    return json.dumps({"service": service, "log_count": len(filtered), "logs": filtered})


def query_metrics(service: str) -> str:
    """Query current performance metrics for a service including latency, error rate, CPU, and memory."""
    metrics = MOCK_METRICS.get(service)
    if not metrics:
        return json.dumps({"service": service, "error": "Service not found in metrics system"})
    return json.dumps({"service": service, "metrics": metrics})


def get_service_dependencies(service: str) -> str:
    """Get the dependency graph for a service showing upstream and downstream connections."""
    deps = MOCK_DEPENDENCIES.get(service)
    if not deps:
        return json.dumps({"service": service, "error": "Service not found in dependency graph"})
    return json.dumps({"service": service, "dependencies": deps})


TOOL_DEFINITIONS = [
    {
        "name": "search_logs",
        "description": "Search application logs for a specific service, filtered by log level. Use this to find error patterns and recent issues.",
        "input_schema": {
            "type": "object",
            "properties": {
                "service": {"type": "string", "description": "The service name to search logs for"},
                "severity_filter": {"type": "string", "enum": ["ERROR", "WARN", "INFO", "ALL"], "description": "Filter logs by severity level"},
            },
            "required": ["service"],
        },
    },
    {
        "name": "query_metrics",
        "description": "Query current performance metrics for a service including latency, error rate, CPU, and memory usage.",
        "input_schema": {
            "type": "object",
            "properties": {
                "service": {"type": "string", "description": "The service name to query metrics for"},
            },
            "required": ["service"],
        },
    },
    {
        "name": "get_service_dependencies",
        "description": "Get the dependency graph for a service showing upstream and downstream connections. Use this to understand blast radius.",
        "input_schema": {
            "type": "object",
            "properties": {
                "service": {"type": "string", "description": "The service name to get dependencies for"},
            },
            "required": ["service"],
        },
    },
]

TOOL_MAP = {
    "search_logs": search_logs,
    "query_metrics": query_metrics,
    "get_service_dependencies": get_service_dependencies,
}

Your data layer is ready. Next up, you will wire these models and tools into a LangGraph workflow that orchestrates the full incident response process.

Build the LangGraph Workflow

Your incident data models and SRE tools are ready. Now you need something to orchestrate them into a coherent investigation workflow.

In this step, you will build the LangGraph state graph that defines how your agent moves through each phase of incident response. You will wire together four nodes with conditional routing and a human approval gate.

In this step, get ready to:

• Create the StateGraph with nodes for triage, root cause analysis, remediation, and execution.
• Add conditional edges that route based on severity.
• Compile the graph with an interrupt before the execution node for human approval.

Set up the agent state and configuration

A LangGraph workflow needs a state schema that defines what data flows between nodes. Each node reads from and writes to this shared state as the incident progresses through the pipeline.

• Create a new file called agent.py in your project folder.
• Add the imports and agent state definition by pasting this code:

import json
import os
from typing import Annotated, TypedDict

from anthropic import Anthropic
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from pydantic import BaseModel, Field

from models import Alert, Finding, RemediationStep, Severity, IncidentState
from tools import TOOL_DEFINITIONS, TOOL_MAP


# State schema that flows through every node in the graph
class AgentState(TypedDict):
    alert: dict                           # raw alert data from monitoring
    severity: str | None                  # triage result: critical/high/medium/low
    findings: list[dict]                  # evidence gathered during investigation
    root_cause: str | None                # determined root cause
    remediation_plan: list[dict]          # proposed fix steps
    approved: bool | None                 # human approval status
    execution_result: str | None          # outcome of remediation
    messages: Annotated[list, add_messages]  # conversation history for the LLM

What does this code do?

AgentState is a TypedDict that defines the data flowing between nodes. Each field tracks one aspect of the incident lifecycle.

The messages field uses LangGraph's add_messages reducer, which appends new messages to the conversation history instead of overwriting it. This lets Claude see the full investigation context as it moves through nodes.

Next, you need to configure the Anthropic client and give Claude its SRE persona. The system prompt tells Claude how to approach incident investigation.

• Add the client initialization and system prompt below the AgentState class:

# Initialize the Anthropic client using the API key from environment
client = Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))

# System prompt giving Claude its SRE persona and investigation methodology
SYSTEM_PROMPT = """You are an expert Site Reliability Engineer AI agent. Your job is to investigate infrastructure incidents methodically.

When triaging, assess severity based on: user impact, blast radius, and rate of degradation.
When investigating root cause, use available tools to gather evidence before concluding.
When planning remediation, propose safe, reversible actions with clear rollback plans.

Always explain your reasoning. Be concise but thorough."""

What does this code do?

The client reads your API key from the ANTHROPIC_API_KEY environment variable. You will set this variable before running the agent in a later step.

The system prompt establishes three clear investigation principles: assess impact during triage, gather evidence before conclusions, and propose reversible fixes. This structured approach mirrors how experienced SREs investigate incidents.

• Save agent.py.

Define the workflow nodes

Each node in the graph handles one phase of incident response. For now, you will create stub functions that return placeholder data. The next step will replace these stubs with full Claude-powered logic.

• Add the stub node functions and routing function below SYSTEM_PROMPT:

# Stub nodes - each handles one phase of incident response
# Full Claude-powered logic comes in the next step
def triage_node(state: AgentState) -> dict:
    return {"severity": "high", "messages": []}


def rca_node(state: AgentState) -> dict:
    return {"findings": [], "root_cause": "placeholder", "messages": []}


def remediation_node(state: AgentState) -> dict:
    return {"remediation_plan": [], "messages": []}


def execution_node(state: AgentState) -> dict:
    return {"execution_result": "placeholder", "messages": []}


# Routing function determines the next node after triage
def should_continue_after_triage(state: AgentState) -> str:
    return "rca"

What does this code do?

Each node function takes the current AgentState and returns a dictionary of fields to update. LangGraph merges these updates into the state before passing it to the next node.

The should_continue_after_triage function is a routing function. LangGraph calls it after triage completes to decide which node runs next. Right now it always returns "rca". In a production system, this could skip straight to execution for known patterns.

• Save agent.py.

Wire the graph with conditional edges and interrupt

Now you will connect the nodes into a directed graph. The graph defines the flow: alert comes in, gets triaged, goes through root cause analysis, then remediation planning, and finally execution.

Why use conditional edges?

Regular edges are fixed paths between nodes. Conditional edges let you branch the workflow based on state. After triage, the routing function inspects severity and decides where to go next.

The interrupt_before parameter pauses the workflow before a specific node runs. This is how you implement the human-in-the-loop safety gate. The agent investigates autonomously, but a human must approve before remediation executes.

• Add the build_graph function below should_continue_after_triage:

def build_graph() -> StateGraph:
    """Build the incident response LangGraph workflow."""
    graph = StateGraph(AgentState)

    # Register each node in the workflow
    graph.add_node("triage", triage_node)
    graph.add_node("rca", rca_node)
    graph.add_node("remediation", remediation_node)
    graph.add_node("execution", execution_node)

    # Define the flow with conditional routing after triage
    graph.add_edge(START, "triage")
    graph.add_conditional_edges("triage", should_continue_after_triage, {"rca": "rca"})
    graph.add_edge("rca", "remediation")
    graph.add_edge("remediation", "execution")
    graph.add_edge("execution", END)

    # Compile with interrupt_before to pause for human approval before execution
    compiled = graph.compile(interrupt_before=["execution"])
    return compiled

What does this code do?

• StateGraph(AgentState) creates a new graph that uses your state schema.
• add_node registers each function as a named node in the graph.
• add_conditional_edges calls should_continue_after_triage at runtime and routes to the node matching its return value.
• interrupt_before=["execution"] halts the workflow right before the execution node runs. The agent cannot remediate without human approval.

• Save agent.py.
• Confirm the file has no syntax errors by running:

python -c "from agent import build_graph; print('Graph built successfully')"

You should see Graph built successfully printed in the terminal.

Seeing an import error?

• Make sure your virtual environment is active. You should see (venv) at the start of your terminal prompt.
• Check that models.py and tools.py are in the same directory as agent.py.
• Verify the packages installed correctly by running pip list and checking that langgraph and anthropic appear in the list.

Why is my graph import failing?

✔️ Awesome, I've got everything!

Great. Double check you have saved agent.py.

ⓧ I'd like to double check the full code

import json
import os
from typing import Annotated, TypedDict

from anthropic import Anthropic
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from pydantic import BaseModel, Field

from models import Alert, Finding, RemediationStep, Severity, IncidentState
from tools import TOOL_DEFINITIONS, TOOL_MAP


class AgentState(TypedDict):
    alert: dict
    severity: str | None
    findings: list[dict]
    root_cause: str | None
    remediation_plan: list[dict]
    approved: bool | None
    execution_result: str | None
    messages: Annotated[list, add_messages]


client = Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))

SYSTEM_PROMPT = """You are an expert Site Reliability Engineer AI agent. Your job is to investigate infrastructure incidents methodically.

When triaging, assess severity based on: user impact, blast radius, and rate of degradation.
When investigating root cause, use available tools to gather evidence before concluding.
When planning remediation, propose safe, reversible actions with clear rollback plans.

Always explain your reasoning. Be concise but thorough."""


def triage_node(state: AgentState) -> dict:
    return {"severity": "high", "messages": []}


def rca_node(state: AgentState) -> dict:
    return {"findings": [], "root_cause": "placeholder", "messages": []}


def remediation_node(state: AgentState) -> dict:
    return {"remediation_plan": [], "messages": []}


def execution_node(state: AgentState) -> dict:
    return {"execution_result": "placeholder", "messages": []}


def should_continue_after_triage(state: AgentState) -> str:
    return "rca"


def build_graph() -> StateGraph:
    """Build the incident response LangGraph workflow."""
    graph = StateGraph(AgentState)
    graph.add_node("triage", triage_node)
    graph.add_node("rca", rca_node)
    graph.add_node("remediation", remediation_node)
    graph.add_node("execution", execution_node)
    graph.add_edge(START, "triage")
    graph.add_conditional_edges("triage", should_continue_after_triage, {"rca": "rca"})
    graph.add_edge("rca", "remediation")
    graph.add_edge("remediation", "execution")
    graph.add_edge("execution", END)
    compiled = graph.compile(interrupt_before=["execution"])
    return compiled

Why LangGraph instead of a simple function chain?

You could chain functions together with simple Python calls. LangGraph adds three capabilities that matter for production agents: typed state that persists between steps, built-in interrupts for human approval, and conditional routing that can branch based on runtime decisions.

This is the same pattern that production SRE automation tools use. The graph structure makes it easy to add new investigation paths, retry failed nodes, or insert additional approval gates later.

Your workflow skeleton is complete. The graph knows how to route incidents through four phases, and it will pause for approval before any remediation runs. Next up, you will replace those stub nodes with real Claude-powered reasoning that investigates incidents autonomously.

Connect Claude as the Reasoning Engine

Your graph has the right shape, but every node is a stub returning placeholder data. The agent needs a brain that can read alert context, call your SRE tools, and reason about what it finds.

In this step, you will replace each stub with real logic powered by Claude claude-sonnet-4-6. The key pattern is a tool-use loop: Claude receives a prompt, decides which tools to call, receives the results, and keeps investigating until it has enough evidence to respond.

In this step, get ready to:

• Build the tool-use loop function that lets Claude call SRE tools autonomously.
• Replace all stub node functions with real Claude-powered implementations.
• Connect the SRE system prompt to give Claude its investigation methodology.

Build the tool-use loop

When Claude decides it needs more information, it responds with a tool_use stop reason instead of a final text answer. Your code needs to execute the requested tool, send the result back, and let Claude continue reasoning. This loop repeats until Claude has gathered enough evidence to produce a final response.

• In agent.py, find the SYSTEM_PROMPT variable. Directly below it, add the call_claude_with_tools function by copying the following code:

def call_claude_with_tools(messages: list, system: str = SYSTEM_PROMPT) -> dict:
    """Call Claude with tool use enabled, handling the tool call loop."""
    current_messages = list(messages)

    while True:
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=4096,
            system=system,
            tools=TOOL_DEFINITIONS,
            messages=current_messages,
        )

        if response.stop_reason == "tool_use":
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    tool_fn = TOOL_MAP.get(block.name)
                    if tool_fn:
                        result = tool_fn(**block.input)
                    else:
                        result = json.dumps({"error": f"Unknown tool: {block.name}"})
                    tool_results.append(
                        {"type": "tool_result", "tool_use_id": block.id, "content": result}
                    )

            current_messages.append({"role": "assistant", "content": response.content})
            current_messages.append({"role": "user", "content": tool_results})
        else:
            return response

What does this code do?

This function is the core bridge between Claude and your SRE tools.

• It calls client.messages.create with the claude-sonnet-4-6 model and passes your TOOL_DEFINITIONS so Claude knows what tools are available.
• When stop_reason is "tool_use", Claude wants to call a tool. The loop executes it, sends the result back, and lets Claude keep thinking.
• When Claude finishes reasoning (any other stop reason), the function returns the final response.

Why a loop instead of a single call?

Real SRE investigation is iterative. Claude might check metrics for one service, discover a dependency, then check logs for that dependency too. The loop lets Claude chain as many tool calls as it needs before concluding.

• Save agent.py.

Seeing a syntax error?

• Make sure the function is placed between SYSTEM_PROMPT and the existing stub functions, not inside another function.
• Check that the while True: block is indented one level inside the function body.
• Help me debug my call_claude_with_tools function

Implement the node functions

Each node in your graph handles one phase of the incident lifecycle. The triage, RCA, and remediation nodes all call Claude through the loop you just built. The execution node simulates taking action.

• In agent.py, find the existing triage_node stub function:

def triage_node(state: AgentState) -> dict:
    return {"severity": "high", "messages": []}

• Replace the entire triage_node function with this implementation:

def triage_node(state: AgentState) -> dict:
    """Triage the alert: determine severity and initial assessment."""
    alert = state["alert"]
    messages = [
        {
            "role": "user",
            "content": f"""Triage this alert and determine its severity (critical/high/medium/low).

Alert ID: {alert['alert_id']}
Service: {alert['service']}
Description: {alert['description']}
Timestamp: {alert['timestamp']}
Raw Severity: {alert['raw_severity']}

Use the available tools to check the service metrics and dependencies to assess blast radius.
Respond with your severity assessment and reasoning.""",
        }
    ]

    response = call_claude_with_tools(messages)
    text_content = ""
    for block in response.content:
        if hasattr(block, "text"):
            text_content += block.text

    # Extract severity from Claude's response
    severity = "high"
    for level in ["critical", "high", "medium", "low"]:
        if level in text_content.lower():
            severity = level
            break

    return {
        "severity": severity,
        "messages": [{"role": "assistant", "content": f"[TRIAGE] Severity: {severity}\n{text_content}"}],
    }

What does this code do?

• It builds a prompt from the alert state and asks Claude to assess severity.
• Claude can use query_metrics and get_service_dependencies to check blast radius before deciding.
• The severity extraction loop scans Claude's text for the first severity keyword it finds. This is a simple heuristic that works because Claude's response always states the severity clearly.

• Save agent.py.

Seeing an indentation error?

• Make sure you deleted the entire old stub (both lines) before pasting the new implementation.
• The f-string uses triple quotes. Check that you have matching """ at the start and end of the prompt content.
• Help me fix my triage_node function

• Now find the rca_node stub function:

def rca_node(state: AgentState) -> dict:
    return {"findings": [], "root_cause": "placeholder", "messages": []}

• Replace the entire rca_node function with this implementation:

def rca_node(state: AgentState) -> dict:
    """Perform root cause analysis using tools."""
    alert = state["alert"]
    severity = state["severity"]
    messages = [
        {
            "role": "user",
            "content": f"""Perform root cause analysis for this incident.

Alert: {alert['description']}
Service: {alert['service']}
Severity: {severity}

Investigate by:
1. Searching logs for the affected service and related services
2. Checking metrics for anomalies
3. Mapping service dependencies to understand blast radius

Provide your root cause determination with supporting evidence.""",
        }
    ]

    response = call_claude_with_tools(messages)
    text_content = ""
    for block in response.content:
        if hasattr(block, "text"):
            text_content += block.text

    findings = [
        {"source": "ai_analysis", "detail": text_content, "relevance": "primary"}
    ]

    return {
        "findings": findings,
        "root_cause": text_content[:500],
        "messages": [{"role": "assistant", "content": f"[RCA] {text_content}"}],
    }

What does this code do?

• This is where the tool-use loop shines. The prompt instructs Claude to search logs, check metrics, and map dependencies. Claude will autonomously decide which tools to call and in what order.
• The findings list stores Claude's analysis as a structured dict, and root_cause is truncated to 500 characters for downstream prompts.

• Save agent.py.

Syntax error around the prompt?

• Check that you have no extra blank lines between the end of the triage_node function and the start of rca_node. Python needs exactly one blank line between top-level functions.
• Help me debug my rca_node function

• Find the remediation_node stub function:

def remediation_node(state: AgentState) -> dict:
    return {"remediation_plan": [], "messages": []}

• Replace the entire remediation_node function with this implementation:

def remediation_node(state: AgentState) -> dict:
    """Generate a remediation plan based on findings."""
    alert = state["alert"]
    root_cause = state["root_cause"]
    messages = [
        {
            "role": "user",
            "content": f"""Based on this root cause analysis, create a remediation plan.

Service: {alert['service']}
Root Cause: {root_cause}

Provide 2-3 concrete remediation steps. For each step include:
- Action to take
- Target (service/resource)
- Risk level (low/medium/high)
- Rollback procedure

Format each step clearly.""",
        }
    ]

    response = call_claude_with_tools(messages)
    text_content = ""
    for block in response.content:
        if hasattr(block, "text"):
            text_content += block.text

    remediation_plan = [
        {
            "action": "See detailed plan below",
            "target": alert["service"],
            "risk_level": "medium",
            "rollback": "Revert changes",
        }
    ]

    return {
        "remediation_plan": remediation_plan,
        "messages": [{"role": "assistant", "content": f"[REMEDIATION PLAN]\n{text_content}"}],
    }

What does this code do?

This node receives the root cause from the previous node and asks Claude to generate concrete, reversible remediation steps. The structured remediation_plan dict flows downstream to the execution node and the human approval gate.

• Save agent.py.

Getting a NameError for root_cause?

• Make sure the line reads root_cause = state["root_cause"] with matching quotes and bracket syntax.
• Help me fix NameError in remediation_node

• Finally, find the execution_node stub function:

def execution_node(state: AgentState) -> dict:
    return {"execution_result": "placeholder", "messages": []}

• Replace the entire execution_node function with this implementation:

def execution_node(state: AgentState) -> dict:
    """Execute the approved remediation plan (simulated)."""
    plan = state["remediation_plan"]
    service = state["alert"]["service"]

    result = f"Remediation executed for {service}. Actions completed: {len(plan)} steps. Service health restored."
    return {
        "execution_result": result,
        "messages": [{"role": "assistant", "content": f"[EXECUTION] {result}"}],
    }

Why is execution simulated?

In a production agent, this node would run actual remediation commands (restart pods, scale services, rollback deployments). For this project, simulation keeps things safe while demonstrating the full workflow pattern. The important part is the human approval gate that runs before this node.

• Save agent.py.

Review the system prompt

The SYSTEM_PROMPT you defined earlier is now active. Every time call_claude_with_tools runs, it passes this prompt as the system parameter. This is what gives Claude its SRE persona and methodology across all nodes.

How does the system prompt shape behavior?

• During triage, the instruction to assess "user impact, blast radius, and rate of degradation" guides Claude to check metrics and dependencies before concluding.
• During RCA, "use available tools to gather evidence before concluding" prevents Claude from guessing without investigating.
• During remediation planning, "propose safe, reversible actions with clear rollback plans" ensures every plan includes a rollback procedure.

• Also update the docstrings on should_continue_after_triage and build_graph to match the final code. Find should_continue_after_triage and add a docstring:

def should_continue_after_triage(state: AgentState) -> str:
    """Route based on severity: critical goes straight to RCA."""
    return "rca"

• Find build_graph and add a docstring plus spacing between the sections:

def build_graph() -> StateGraph:
    """Build the incident response LangGraph workflow."""
    graph = StateGraph(AgentState)

    graph.add_node("triage", triage_node)
    graph.add_node("rca", rca_node)
    graph.add_node("remediation", remediation_node)
    graph.add_node("execution", execution_node)

    graph.add_edge(START, "triage")
    graph.add_conditional_edges("triage", should_continue_after_triage, {"rca": "rca"})
    graph.add_edge("rca", "remediation")
    graph.add_edge("remediation", "execution")
    graph.add_edge("execution", END)

    compiled = graph.compile(interrupt_before=["execution"])
    return compiled

• Save agent.py.
• You won't see runtime output yet. The next step adds the human-in-the-loop gate, and the step after that runs the full agent end-to-end.

✔️ Awesome, I've got everything!

Great. Double check you have saved agent.py with all five functions implemented.

ⓧ I'd like to double check the full code

Here is the complete agent.py file at the end of this step. Compare yours against it:

import json
import os
from typing import Annotated, TypedDict

from anthropic import Anthropic
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from pydantic import BaseModel, Field

from models import Alert, Finding, RemediationStep, Severity, IncidentState
from tools import TOOL_DEFINITIONS, TOOL_MAP


class AgentState(TypedDict):
    alert: dict
    severity: str | None
    findings: list[dict]
    root_cause: str | None
    remediation_plan: list[dict]
    approved: bool | None
    execution_result: str | None
    messages: Annotated[list, add_messages]


client = Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))

SYSTEM_PROMPT = """You are an expert Site Reliability Engineer AI agent. Your job is to investigate infrastructure incidents methodically.

When triaging, assess severity based on: user impact, blast radius, and rate of degradation.
When investigating root cause, use available tools to gather evidence before concluding.
When planning remediation, propose safe, reversible actions with clear rollback plans.

Always explain your reasoning. Be concise but thorough."""


def call_claude_with_tools(messages: list, system: str = SYSTEM_PROMPT) -> dict:
    """Call Claude with tool use enabled, handling the tool call loop."""
    current_messages = list(messages)

    while True:
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=4096,
            system=system,
            tools=TOOL_DEFINITIONS,
            messages=current_messages,
        )

        if response.stop_reason == "tool_use":
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    tool_fn = TOOL_MAP.get(block.name)
                    if tool_fn:
                        result = tool_fn(**block.input)
                    else:
                        result = json.dumps({"error": f"Unknown tool: {block.name}"})
                    tool_results.append(
                        {"type": "tool_result", "tool_use_id": block.id, "content": result}
                    )

            current_messages.append({"role": "assistant", "content": response.content})
            current_messages.append({"role": "user", "content": tool_results})
        else:
            return response


def triage_node(state: AgentState) -> dict:
    """Triage the alert: determine severity and initial assessment."""
    alert = state["alert"]
    messages = [
        {
            "role": "user",
            "content": f"""Triage this alert and determine its severity (critical/high/medium/low).

Alert ID: {alert['alert_id']}
Service: {alert['service']}
Description: {alert['description']}
Timestamp: {alert['timestamp']}
Raw Severity: {alert['raw_severity']}

Use the available tools to check the service metrics and dependencies to assess blast radius.
Respond with your severity assessment and reasoning.""",
        }
    ]

    response = call_claude_with_tools(messages)
    text_content = ""
    for block in response.content:
        if hasattr(block, "text"):
            text_content += block.text

    severity = "high"
    for level in ["critical", "high", "medium", "low"]:
        if level in text_content.lower():
            severity = level
            break

    return {
        "severity": severity,
        "messages": [{"role": "assistant", "content": f"[TRIAGE] Severity: {severity}\n{text_content}"}],
    }


def rca_node(state: AgentState) -> dict:
    """Perform root cause analysis using tools."""
    alert = state["alert"]
    severity = state["severity"]
    messages = [
        {
            "role": "user",
            "content": f"""Perform root cause analysis for this incident.

Alert: {alert['description']}
Service: {alert['service']}
Severity: {severity}

Investigate by:
1. Searching logs for the affected service and related services
2. Checking metrics for anomalies
3. Mapping service dependencies to understand blast radius

Provide your root cause determination with supporting evidence.""",
        }
    ]

    response = call_claude_with_tools(messages)
    text_content = ""
    for block in response.content:
        if hasattr(block, "text"):
            text_content += block.text

    findings = [
        {"source": "ai_analysis", "detail": text_content, "relevance": "primary"}
    ]

    return {
        "findings": findings,
        "root_cause": text_content[:500],
        "messages": [{"role": "assistant", "content": f"[RCA] {text_content}"}],
    }


def remediation_node(state: AgentState) -> dict:
    """Generate a remediation plan based on findings."""
    alert = state["alert"]
    root_cause = state["root_cause"]
    messages = [
        {
            "role": "user",
            "content": f"""Based on this root cause analysis, create a remediation plan.

Service: {alert['service']}
Root Cause: {root_cause}

Provide 2-3 concrete remediation steps. For each step include:
- Action to take
- Target (service/resource)
- Risk level (low/medium/high)
- Rollback procedure

Format each step clearly.""",
        }
    ]

    response = call_claude_with_tools(messages)
    text_content = ""
    for block in response.content:
        if hasattr(block, "text"):
            text_content += block.text

    remediation_plan = [
        {
            "action": "See detailed plan below",
            "target": alert["service"],
            "risk_level": "medium",
            "rollback": "Revert changes",
        }
    ]

    return {
        "remediation_plan": remediation_plan,
        "messages": [{"role": "assistant", "content": f"[REMEDIATION PLAN]\n{text_content}"}],
    }


def execution_node(state: AgentState) -> dict:
    """Execute the approved remediation plan (simulated)."""
    plan = state["remediation_plan"]
    service = state["alert"]["service"]

    result = f"Remediation executed for {service}. Actions completed: {len(plan)} steps. Service health restored."
    return {
        "execution_result": result,
        "messages": [{"role": "assistant", "content": f"[EXECUTION] {result}"}],
    }


def should_continue_after_triage(state: AgentState) -> str:
    """Route based on severity: critical goes straight to RCA."""
    return "rca"


def build_graph() -> StateGraph:
    """Build the incident response LangGraph workflow."""
    graph = StateGraph(AgentState)

    graph.add_node("triage", triage_node)
    graph.add_node("rca", rca_node)
    graph.add_node("remediation", remediation_node)
    graph.add_node("execution", execution_node)

    graph.add_edge(START, "triage")
    graph.add_conditional_edges("triage", should_continue_after_triage, {"rca": "rca"})
    graph.add_edge("rca", "remediation")
    graph.add_edge("remediation", "execution")
    graph.add_edge("execution", END)

    compiled = graph.compile(interrupt_before=["execution"])
    return compiled

Your agent can now think. It reads alerts, calls tools autonomously, and produces structured outputs at every stage. Next up, you will add the human-in-the-loop gate that keeps engineers in control before any remediation executes.

Add the Human-in-the-Loop Approval Gate

Your agent can now investigate incidents from start to finish. But there is a critical gap: it would execute remediation automatically without any human oversight.

In production SRE workflows, autonomous investigation is fine. Autonomous remediation is dangerous. In this step, you will build the CLI runner that streams the agent through triage, RCA, and remediation planning, then pauses at the interrupt_before gate for engineer approval before executing anything.

In this step, get ready to:

• Implement the interrupt handler that pauses before remediation execution.
• Build the CLI approval interface that shows the engineer what the agent found.
• Handle rejection so the incident remains open for manual investigation.

Create the sample incidents and imports

The human-in-the-loop pattern in LangGraph requires a checkpointer to persist state between the pause and resume. Without saved state, the graph cannot pick up where it left off after the engineer approves.

• Create a new file called main.py in your project folder.
• Add the imports and sample alert data by pasting this code:

import json
from langgraph.checkpoint.memory import InMemorySaver

from agent import build_graph, AgentState
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from typing import Annotated, TypedDict
import anthropic
import os

from tools import TOOL_DEFINITIONS, TOOL_MAP


# Three simulated incidents covering different failure modes
SAMPLE_ALERTS = [
    {
        "alert_id": "INC-2026-0521-001",
        "service": "api-gateway",
        "description": "Latency spike: p99 response time exceeded 5000ms threshold for api-gateway. Error rate rising.",
        "timestamp": "2026-05-21T03:15:00Z",
        "raw_severity": "warning",
    },
    {
        "alert_id": "INC-2026-0521-002",
        "service": "user-service",
        "description": "Service down: user-service returning 503 on all endpoints. Memory usage at 98%.",
        "timestamp": "2026-05-21T03:12:30Z",
        "raw_severity": "critical",
    },
    {
        "alert_id": "INC-2026-0521-003",
        "service": "deploy-pipeline",
        "description": "Deployment failure: user-service v2.4.1 health check failed, pods in CrashLoopBackOff.",
        "timestamp": "2026-05-21T02:47:00Z",
        "raw_severity": "high",
    },
]

What does this code do?

InMemorySaver is LangGraph's development checkpointer. It saves graph state in memory so the workflow can pause at the interrupt gate and resume later when the engineer responds.

SAMPLE_ALERTS gives you three realistic incidents to test: a latency spike on the API gateway, a memory exhaustion on the user service, and a failed deployment. Each has the same shape your agent expects from the Alert model.

• Save main.py.

Build the CLI agent runner with approval gate

The run_agent() function is the heart of your human-in-the-loop pattern. It builds the graph with a checkpointer, streams the agent through triage and RCA, then halts at the execution node for engineer approval.

• Add the run_agent() function below SAMPLE_ALERTS in main.py:

def run_agent():
    # Create a checkpointer so the graph can pause and resume
    from langgraph.checkpoint.memory import InMemorySaver
    checkpointer = InMemorySaver()

    # Import node functions to rebuild the graph with checkpointing
    from agent import AgentState, call_claude_with_tools, triage_node, rca_node, remediation_node, execution_node, should_continue_after_triage, SYSTEM_PROMPT

    # Build the graph with checkpointer for interrupt support
    graph_builder = StateGraph(AgentState)
    graph_builder.add_node("triage", triage_node)
    graph_builder.add_node("rca", rca_node)
    graph_builder.add_node("remediation", remediation_node)
    graph_builder.add_node("execution", execution_node)

    graph_builder.add_edge(START, "triage")
    graph_builder.add_conditional_edges("triage", should_continue_after_triage, {"rca": "rca"})
    graph_builder.add_edge("rca", "remediation")
    graph_builder.add_edge("remediation", "execution")
    graph_builder.add_edge("execution", END)

    # Compile with interrupt_before to pause before execution
    graph = graph_builder.compile(checkpointer=checkpointer, interrupt_before=["execution"])

    # Display incident menu
    print("=" * 60)
    print("  AI INCIDENT RESPONSE AGENT")
    print("=" * 60)
    print("\nAvailable incidents:")
    for i, alert in enumerate(SAMPLE_ALERTS):
        print(f"  [{i + 1}] {alert['alert_id']}: {alert['description'][:60]}...")

    # Get engineer's incident selection
    choice = input("\nSelect an incident (1-3): ").strip()
    try:
        alert = SAMPLE_ALERTS[int(choice) - 1]
    except (ValueError, IndexError):
        alert = SAMPLE_ALERTS[0]
        print(f"Invalid choice, using incident 1.")

    print(f"\n{'=' * 60}")
    print(f"Processing: {alert['alert_id']}")
    print(f"Service: {alert['service']}")
    print(f"Description: {alert['description']}")
    print(f"{'=' * 60}\n")

    # Thread ID lets the checkpointer track this specific run
    config = {"configurable": {"thread_id": alert["alert_id"]}}
    initial_state = {"alert": alert, "severity": None, "findings": [], "root_cause": None, "remediation_plan": [], "approved": None, "execution_result": None, "messages": []}

    # Stream the graph until it hits the interrupt_before gate
    print("[*] Running triage and investigation...\n")
    for event in graph.stream(initial_state, config, stream_mode="values"):
        if "messages" in event and event["messages"]:
            last_msg = event["messages"][-1]
            if isinstance(last_msg, dict) and "content" in last_msg:
                content = last_msg["content"]
            elif hasattr(last_msg, "content"):
                content = last_msg.content
            else:
                continue
            if content.startswith("["):
                print(f"\n{content[:200]}...")
                print()

    # Check if the graph paused at the execution interrupt
    snapshot = graph.get_state(config)
    if snapshot.next and "execution" in snapshot.next:
        print("\n" + "=" * 60)
        print("  HUMAN APPROVAL REQUIRED")
        print("=" * 60)
        print(f"\n  Severity: {snapshot.values.get('severity', 'unknown')}")
        print(f"  Root Cause: {snapshot.values.get('root_cause', 'unknown')[:150]}...")
        print(f"  Remediation Steps: {len(snapshot.values.get('remediation_plan', []))}")

        # The human-in-the-loop gate
        approval = input("\n  Approve remediation? (yes/no): ").strip().lower()

        if approval in ("yes", "y"):
            # Resume the graph from where it paused
            print("\n[*] Executing remediation...\n")
            for event in graph.stream(None, config, stream_mode="values"):
                if "messages" in event and event["messages"]:
                    last_msg = event["messages"][-1]
                    if isinstance(last_msg, dict) and "content" in last_msg:
                        content = last_msg["content"]
                    elif hasattr(last_msg, "content"):
                        content = last_msg.content
                    else:
                        continue
                    if content.startswith("["):
                        print(f"\n{content}")
        else:
            # Rejection keeps the incident open for manual investigation
            print("\n[!] Remediation rejected. Incident remains open for manual investigation.")

    print(f"\n{'=' * 60}")
    print("  INCIDENT RESPONSE COMPLETE")
    print(f"{'=' * 60}")

How does the interrupt-and-resume pattern work?

The key line is graph.compile(checkpointer=checkpointer, interrupt_before=["execution"]). This tells LangGraph to pause the workflow right before the execution node runs.

After the pause, graph.get_state(config) retrieves the saved state so you can display the severity, root cause, and remediation plan to the engineer.

When the engineer approves, graph.stream(None, config) resumes from the saved checkpoint. Passing None as input tells LangGraph to continue where it left off rather than starting fresh.

If the engineer rejects, the graph simply stops. The incident stays open for manual investigation or re-triage.

• Save main.py.

Add the entry point with API key validation

The script needs a guard that checks for the ANTHROPIC_API_KEY environment variable before running. Without it, the agent would crash mid-investigation with an authentication error.

• Add the __main__ guard at the bottom of main.py:

if __name__ == "__main__":
    # Fail fast if the API key is missing
    if not os.environ.get("ANTHROPIC_API_KEY"):
        print("Error: Set ANTHROPIC_API_KEY environment variable")
        print("  set ANTHROPIC_API_KEY=your-api-key-here")
        exit(1)
    run_agent()

What does this code do?

The if __name__ == "__main__" guard ensures run_agent() only runs when you execute main.py directly. It checks for the API key first so you get a clear error message instead of a confusing traceback deep inside the agent.

• Save main.py.
• Confirm the file exists by running:

ls main.py

You should see main.py listed in the output.

Don't see main.py?

Make sure you saved the file in the same directory as agent.py and tools.py. All four Python files (models.py, tools.py, agent.py, main.py) must be in the same folder.

help me figure out why main.py isn't showing up in my project folder

✔️ Awesome, I've got everything!

Great. Double check you have saved main.py with all three sections: imports and sample alerts, the run_agent() function, and the __main__ guard.

ⓧ I'd like to double check the full code

import json
from langgraph.checkpoint.memory import InMemorySaver

from agent import build_graph, AgentState
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from typing import Annotated, TypedDict
import anthropic
import os

from tools import TOOL_DEFINITIONS, TOOL_MAP


SAMPLE_ALERTS = [
    {
        "alert_id": "INC-2026-0521-001",
        "service": "api-gateway",
        "description": "Latency spike: p99 response time exceeded 5000ms threshold for api-gateway. Error rate rising.",
        "timestamp": "2026-05-21T03:15:00Z",
        "raw_severity": "warning",
    },
    {
        "alert_id": "INC-2026-0521-002",
        "service": "user-service",
        "description": "Service down: user-service returning 503 on all endpoints. Memory usage at 98%.",
        "timestamp": "2026-05-21T03:12:30Z",
        "raw_severity": "critical",
    },
    {
        "alert_id": "INC-2026-0521-003",
        "service": "deploy-pipeline",
        "description": "Deployment failure: user-service v2.4.1 health check failed, pods in CrashLoopBackOff.",
        "timestamp": "2026-05-21T02:47:00Z",
        "raw_severity": "high",
    },
]


def run_agent():
    from langgraph.checkpoint.memory import InMemorySaver
    checkpointer = InMemorySaver()
    from agent import AgentState, call_claude_with_tools, triage_node, rca_node, remediation_node, execution_node, should_continue_after_triage, SYSTEM_PROMPT
    graph_builder = StateGraph(AgentState)
    graph_builder.add_node("triage", triage_node)
    graph_builder.add_node("rca", rca_node)
    graph_builder.add_node("remediation", remediation_node)
    graph_builder.add_node("execution", execution_node)
    graph_builder.add_edge(START, "triage")
    graph_builder.add_conditional_edges("triage", should_continue_after_triage, {"rca": "rca"})
    graph_builder.add_edge("rca", "remediation")
    graph_builder.add_edge("remediation", "execution")
    graph_builder.add_edge("execution", END)
    graph = graph_builder.compile(checkpointer=checkpointer, interrupt_before=["execution"])

    print("=" * 60)
    print("  AI INCIDENT RESPONSE AGENT")
    print("=" * 60)
    print("\nAvailable incidents:")
    for i, alert in enumerate(SAMPLE_ALERTS):
        print(f"  [{i + 1}] {alert['alert_id']}: {alert['description'][:60]}...")

    choice = input("\nSelect an incident (1-3): ").strip()
    try:
        alert = SAMPLE_ALERTS[int(choice) - 1]
    except (ValueError, IndexError):
        alert = SAMPLE_ALERTS[0]
        print(f"Invalid choice, using incident 1.")

    print(f"\n{'=' * 60}")
    print(f"Processing: {alert['alert_id']}")
    print(f"Service: {alert['service']}")
    print(f"Description: {alert['description']}")
    print(f"{'=' * 60}\n")

    config = {"configurable": {"thread_id": alert["alert_id"]}}
    initial_state = {"alert": alert, "severity": None, "findings": [], "root_cause": None, "remediation_plan": [], "approved": None, "execution_result": None, "messages": []}

    print("[*] Running triage and investigation...\n")
    for event in graph.stream(initial_state, config, stream_mode="values"):
        if "messages" in event and event["messages"]:
            last_msg = event["messages"][-1]
            if isinstance(last_msg, dict) and "content" in last_msg:
                content = last_msg["content"]
            elif hasattr(last_msg, "content"):
                content = last_msg.content
            else:
                continue
            if content.startswith("["):
                print(f"\n{content[:200]}...")
                print()

    snapshot = graph.get_state(config)
    if snapshot.next and "execution" in snapshot.next:
        print("\n" + "=" * 60)
        print("  HUMAN APPROVAL REQUIRED")
        print("=" * 60)
        print(f"\n  Severity: {snapshot.values.get('severity', 'unknown')}")
        print(f"  Root Cause: {snapshot.values.get('root_cause', 'unknown')[:150]}...")
        print(f"  Remediation Steps: {len(snapshot.values.get('remediation_plan', []))}")

        approval = input("\n  Approve remediation? (yes/no): ").strip().lower()

        if approval in ("yes", "y"):
            print("\n[*] Executing remediation...\n")
            for event in graph.stream(None, config, stream_mode="values"):
                if "messages" in event and event["messages"]:
                    last_msg = event["messages"][-1]
                    if isinstance(last_msg, dict) and "content" in last_msg:
                        content = last_msg["content"]
                    elif hasattr(last_msg, "content"):
                        content = last_msg.content
                    else:
                        continue
                    if content.startswith("["):
                        print(f"\n{content}")
        else:
            print("\n[!] Remediation rejected. Incident remains open for manual investigation.")

    print(f"\n{'=' * 60}")
    print("  INCIDENT RESPONSE COMPLETE")
    print(f"{'=' * 60}")


if __name__ == "__main__":
    if not os.environ.get("ANTHROPIC_API_KEY"):
        print("Error: Set ANTHROPIC_API_KEY environment variable")
        print("  set ANTHROPIC_API_KEY=your-api-key-here")
        exit(1)
    run_agent()

Why rebuild the graph inside run_agent()?

The build_graph() helper in agent.py compiles a graph without a checkpointer. To support the interrupt-and-resume pattern, you need to pass a checkpointer at compile time. Rebuilding the graph inside run_agent() gives you control over the checkpointer lifecycle.

In production, you would use a persistent checkpointer (like PostgreSQL or Redis) so the approval gate can span minutes or hours while an engineer reviews the plan.

Your approval gate is in place. Next up, you will run the agent against all three sample incidents to verify the full workflow from triage to execution.

Test with Simulated Incidents

You have all four files wired together: models, tools, agent logic, and the CLI runner. Now it's time to feed real incident scenarios into the agent and watch the full triage-to-remediation workflow in action.

The SAMPLE_ALERTS list in main.py contains three different failure patterns. Each one tests a different aspect of the agent's reasoning: a latency spike, a service-down event, and a failed deployment.

In this step, get ready to:

• Set your API key and run the agent for the first time.
• Process all three incident types and observe the workflow output.
• Test both approving and rejecting remediation plans at the approval gate.

Set your API key and run the first incident

The agent needs your Anthropic API key to call Claude. The __main__ guard in main.py checks for this environment variable before starting.

• Set your ANTHROPIC_API_KEY environment variable by running this command:

macOS

export ANTHROPIC_API_KEY=your-api-key-here

Windows

set ANTHROPIC_API_KEY=your-api-key-here

• Replace your-api-key-here with your actual Anthropic API key from the Anthropic Console.

Seeing an API key error?

If you see the message about setting ANTHROPIC_API_KEY, make sure you exported it in the same terminal session where you run python main.py. Opening a new terminal window resets environment variables.

help me debug my ANTHROPIC_API_KEY environment variable not being found

• Run the agent by executing this command:

python main.py

You should see the incident selection menu appear with three numbered options.

• Type 1 and press Enter to select the latency spike incident.

The agent now begins its investigation. You should see output prefixed with [TRIAGE], [RCA], and [REMEDIATION PLAN] as each workflow node completes.

What should I see?

The agent streams its progress as it moves through each node. First, Claude triages the alert by calling query_metrics and get_service_dependencies to assess blast radius.

Then it performs root cause analysis by searching logs and correlating metrics. Finally, it generates a remediation plan and pauses at the HUMAN APPROVAL REQUIRED gate.

• When the approval prompt appears, type yes and press Enter to approve the remediation.

You should see an [EXECUTION] message confirming the remediation ran, followed by INCIDENT RESPONSE COMPLETE.

Run the remaining incidents

Each incident exercises different mock data in your tools. The memory leak incident hits user-service logs showing OOM errors. The deployment failure hits deploy-pipeline logs with CrashLoopBackOff messages.

• Run the agent again by executing:

python main.py

• Type 2 and press Enter to select the service-down incident.
• Observe the triage output. Claude should assess this as critical severity since the service is completely down with 98% memory usage.
• When the approval prompt appears, type yes to approve.

Why does each incident produce different output?

Each sample alert points to a different service name. When Claude calls search_logs or query_metrics, it gets different mock data from tools.py. This simulates what a real SRE agent would see querying actual observability systems.

• Run the agent one more time:

python main.py

• Type 3 and press Enter to select the deployment failure incident.
• Observe the RCA output. Claude should identify the failed health check and CrashLoopBackOff from the deploy-pipeline logs.
• When the approval prompt appears, type yes to approve.

All three incidents should complete with the INCIDENT RESPONSE COMPLETE banner after approval.

Agent seems stuck or slow?

Claude makes multiple tool calls during investigation, so each run involves several API round-trips. The RCA node especially takes longer because it searches logs, checks metrics, and maps dependencies. Wait up to 30 seconds per node.

If you see a timeout or connection error, check your internet connection and try running again.

help me debug why my LangGraph agent is timing out during tool calls

Verify the approval gate rejects correctly

The human-in-the-loop gate is the safety mechanism that prevents the agent from executing remediation without engineer consent. You need to verify it actually stops when you say no.

• Run the agent one final time:

python main.py

• Select any incident (type 1, 2, or 3).
• Wait for the HUMAN APPROVAL REQUIRED banner to appear with the severity and root cause summary.
• Type no and press Enter to reject the remediation plan.

You should see the rejection message confirming the incident remains open for manual investigation. The agent does NOT execute the remediation steps.

Why is this gate so important?

In production SRE, automated remediation without human approval is dangerous. Restarting a service, scaling pods, or rolling back a deployment can cause data loss or cascading failures if the root cause analysis was wrong.

The interrupt_before pattern in LangGraph gives you this safety gate for free. The graph pauses its state, waits for external input, and resumes only when instructed.

Approval gate not appearing?

If the agent goes straight to INCIDENT RESPONSE COMPLETE without pausing, the interrupt_before=["execution"] parameter may be missing from the graph_builder.compile() call in main.py.

Also confirm you included checkpointer=checkpointer in the same compile() call. The interrupt mechanism requires a checkpointer to persist state while waiting for human input.

my LangGraph agent skips the human approval step and goes straight to execution

Your AI incident response agent is fully operational. It autonomously investigates incidents using your SRE tools, produces structured analysis, and respects human decisions at the approval gate. Next up, take on the Secret Mission to add automatic post-mortem generation.

Secret mission

Post-Mortem Generation

Your agent resolves incidents, but what about the paperwork? In production SRE, every resolved incident needs a structured post-mortem report for future learning. Can you add an automatic post-mortem generation step to the workflow?

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally with pay-per-call API usage to Anthropic's Claude. There are no ongoing costs since you only pay when you actively run the agent.

Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

• Project folder containing agent.py, main.py, models.py, and tools.py.
• Python virtual environment and installed packages (langgraph, anthropic, pydantic).
• Generated post-mortem files (postmortem_INC-2026-0521-001.md and any others).

Keep everything running

No action needed. Choose this if you plan to keep experimenting with your incident response agent or want to modify the workflow.

• Your project files remain on your Desktop ready to run anytime.
• Your virtual environment stays intact with all dependencies installed.
• Your ANTHROPIC_API_KEY environment variable only lasts for your current terminal session. You will need to export it again next time you open a new terminal.

Pause - I'll come back to this later

Shut down any running processes to free up system resources. Your files stay on disk for when you return.

• Close the terminal window where you ran the agent.
• Deactivate the virtual environment if it is still active by running:

deactivate

• Your project files and virtual environment remain on disk. Next time you return, reactivate the virtual environment and re-export your ANTHROPIC_API_KEY.

Delete - I don't want to use this again

Remove all project files and the virtual environment from your machine.

• Close any terminal windows running the agent.
• Delete the project folder (which contains the virtual environment, all Python files, and generated post-mortem files) by running:

macOS

cd ~/Desktop
rm -rf incident-response-agent

Windows

cd ~\Desktop
rmdir /s /q incident-response-agent

• Confirm the folder is gone by running:

ls ~/Desktop

• You should no longer see the project folder listed.
• If you added ANTHROPIC_API_KEY to a shell profile file (like .bashrc or .zshrc), open that file and remove the export line. If you only exported it in the terminal session, it is already gone once you close the terminal.

Nice Work!

Nice Work!

You just built an AI-powered incident response agent that automates the SRE investigation workflow from alert to resolution.

You've learned how to:

• Build a LangGraph stateful workflow that processes infrastructure alerts through triage, root cause analysis, and remediation planning.
• Create custom SRE tools (log search, metrics query, dependency mapping) that an AI agent calls autonomously to gather evidence.
• Implement a human-in-the-loop approval gate using LangGraph's interrupt_before pattern to keep engineers in control of remediation execution.
• An automatic post-mortem generator that documents resolved incidents as structured markdown files for future learning.

Ready to quiz yourself?