Build an AI Incident Report Generator with Claude API

import anthropic
from datetime import datetime
import json

client = anthropic.Anthropic()

# ─── Report Schema ─────────────────────────────────────────────────────────────

INCIDENT_REPORT_TOOL = {
    "name": "generate_incident_report",
    "description": "Generates a complete, structured IT incident report from raw notes.",
    "input_schema": {
        "type": "object",
        "properties": {
            "incident_title": {
                "type": "string",
                "description": "Brief, clear title for the incident (max 100 chars)"
            },
            "severity": {
                "type": "string",
                "enum": ["P1 - Critical", "P2 - High", "P3 - Medium", "P4 - Low"],
                "description": "Severity based on impact and affected users"
            },
            "incident_type": {
                "type": "string",
                "enum": [
                    "Infrastructure Outage", "Security Incident", "Application Failure",
                    "Data Integrity Issue", "Network Disruption", "Database Issue",
                    "Authentication Failure", "Performance Degradation", "Other"
                ]
            },
            "affected_systems": {
                "type": "array",
                "items": {"type": "string"},
                "description": "List of systems, services, or components affected"
            },
            "affected_users_estimate": {
                "type": "string",
                "description": "Estimated number or percentage of users impacted"
            },
            "incident_start": {
                "type": "string",
                "description": "Best estimate of incident start time from notes"
            },
            "incident_detected": {
                "type": "string",
                "description": "When the incident was first detected or reported"
            },
            "incident_resolved": {
                "type": "string",
                "description": "When the incident was fully resolved, or 'Ongoing'"
            },
            "summary": {
                "type": "string",
                "description": "2-3 sentence factual summary of what happened and impact"
            },
            "timeline": {
                "type": "array",
                "items": {
                    "type": "object",
                    "properties": {
                        "time": {"type": "string"},
                        "event": {"type": "string"}
                    },
                    "required": ["time", "event"]
                },
                "description": "Chronological list of key events extracted from notes"
            },
            "root_cause": {
                "type": "string",
                "description": "Most probable root cause based on the notes. Use 'Under investigation' if unclear."
            },
            "contributing_factors": {
                "type": "array",
                "items": {"type": "string"},
                "description": "Secondary factors that contributed to the incident or its impact"
            },
            "resolution_steps_taken": {
                "type": "array",
                "items": {"type": "string"},
                "description": "Chronological actions taken to resolve the incident"
            },
            "recommended_follow_up_actions": {
                "type": "array",
                "items": {
                    "type": "object",
                    "properties": {
                        "action": {"type": "string"},
                        "priority": {"type": "string", "enum": ["Immediate", "This Sprint", "Next Quarter"]},
                        "owner": {"type": "string", "description": "Role or team responsible"}
                    },
                    "required": ["action", "priority", "owner"]
                }
            },
            "lessons_learned": {
                "type": "array",
                "items": {"type": "string"},
                "description": "Key learnings that should be acted on to prevent recurrence"
            },
            "requires_post_mortem": {
                "type": "boolean",
                "description": "True for P1/P2 incidents or any with significant data loss or security impact"
            }
        },
        "required": [
            "incident_title", "severity", "incident_type", "affected_systems",
            "affected_users_estimate", "incident_start", "incident_detected",
            "incident_resolved", "summary", "timeline", "root_cause",
            "contributing_factors", "resolution_steps_taken",
            "recommended_follow_up_actions", "lessons_learned", "requires_post_mortem"
        ]
    }
}

# ─── Generation Engine ─────────────────────────────────────────────────────────

SYSTEM_PROMPT = """You are an expert IT incident management specialist and technical writer.
You produce professional, accurate, and actionable incident reports from raw engineer notes.

Guidelines:
- Extract facts from the notes; do not invent information not present in the notes
- If a timeline entry has no specific time, estimate it as relative time (e.g., "T+15min")
- Severity classification: P1=total service outage or data breach, P2=major feature down or >20% users affected, P3=partial degradation, P4=cosmetic or minor
- Root cause should be factual and specific, not vague generalisations
- Follow-up actions must be specific and actionable, not generic advice
"""


def generate_incident_report(raw_notes: str, additional_context: dict = None) -> dict:
    """
    Generate a structured incident report from raw notes.
    
    Args:
        raw_notes: Raw text notes from engineers or ticket history
        additional_context: Optional dict with known fields e.g. {"reported_by": "Alice"}
    
    Returns:
        Structured incident report as a dict
    """
    context_str = ""
    if additional_context:
        context_str = "\n\nAdditional known context:\n" + json.dumps(additional_context, indent=2)
    
    message = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=4096,
        system=SYSTEM_PROMPT,
        tools=[INCIDENT_REPORT_TOOL],
        tool_choice={"type": "tool", "name": "generate_incident_report"},
        messages=[
            {
                "role": "user",
                "content": f"""RAW INCIDENT NOTES:
{raw_notes}
{context_str}

Generate a complete, structured incident report from these notes."""
            }
        ]
    )
    
    # Extract tool use result
    for block in message.content:
        if block.type == "tool_use":
            return block.input
    
    raise ValueError("Claude did not return a structured report — check tool configuration")


# ─── Report Formatter ──────────────────────────────────────────────────────────

def format_report_markdown(report: dict, report_id: str = None) -> str:
    """Format the structured report as a Markdown document."""
    
    rid = report_id or f"INC-{datetime.now().strftime('%Y%m%d-%H%M')}"
    generated_at = datetime.now().strftime("%Y-%m-%d %H:%M UTC")
    
    post_mortem_flag = "**Post-Mortem Required**" if report.get("requires_post_mortem") else "Not required"
    
    # Timeline block
    timeline_lines = []
    for entry in report.get("timeline", []):
        timeline_lines.append(f"| {entry.get('time', '?')} | {entry.get('event', '')} |")
    timeline_table = (
        "| Time | Event |\n|------|-------|\n" + "\n".join(timeline_lines)
        if timeline_lines else "_No timeline extracted._"
    )
    
    # Follow-up actions
    follow_up_lines = []
    for action in report.get("recommended_follow_up_actions", []):
        follow_up_lines.append(
            f"- [{action.get('priority')}] **{action.get('action')}** — Owner: {action.get('owner')}"
        )
    follow_up_str = "\n".join(follow_up_lines) or "None identified."
    
    affected_systems = ", ".join(report.get("affected_systems", [])) or "Unknown"
    contributing = "\n".join(f"- {f}" for f in report.get("contributing_factors", [])) or "- None identified"
    resolution_steps = "\n".join(f"- {s}" for s in report.get("resolution_steps_taken", [])) or "- None documented"
    lessons = "\n".join(f"- {l}" for l in report.get("lessons_learned", [])) or "- None identified"
    
    return f"""# Incident Report: {rid}

**Generated:** {generated_at}  
**Title:** {report.get("incident_title")}  
**Severity:** {report.get("severity")}  
**Type:** {report.get("incident_type")}  
**Post-Mortem:** {post_mortem_flag}

---

## Impact

| Field | Value |
|-------|-------|
| Affected Systems | {affected_systems} |
| Estimated Users Affected | {report.get("affected_users_estimate", "Unknown")} |
| Incident Start | {report.get("incident_start", "Unknown")} |
| Detected At | {report.get("incident_detected", "Unknown")} |
| Resolved At | {report.get("incident_resolved", "Unknown")} |

---

## Summary

{report.get("summary", "")}

---

## Timeline

{timeline_table}

---

## Root Cause Analysis

**Root Cause:** {report.get("root_cause", "Under investigation")}

**Contributing Factors:**
{contributing}

---

## Resolution

**Steps Taken:**
{resolution_steps}

---

## Follow-Up Actions

{follow_up_str}

---

## Lessons Learned

{lessons}
"""


def export_report_json(report: dict, output_path: str):
    """Export report as JSON for ITSM integration."""
    with open(output_path, "w") as f:
        json.dump(report, f, indent=2)
    print(f"Report exported to {output_path}")

# ─── Example Usage ─────────────────────────────────────────────────────────────

SAMPLE_NOTES = """
Started getting alerts around 14:22 - database connection pool exhausted on prod-db-01.
App servers started returning 503s to users. Estimated about 3000 users affected based 
on Cloudwatch error rate. 

Sarah noticed that a new deployment went out at 14:15 - the new order processing service. 
Jon checked the slow query log and found the new service was running a full table scan 
on the orders table on every request - no index on the customer_id column.

We rolled back the deployment at 14:48. Services recovered within 2 minutes of rollback.
Db connection pool back to normal by 14:51.

The orders table has 12 million rows. The query was running in ~4 seconds each time. 
Under normal load that was fine in staging but prod traffic is 50x higher.

We need to: add the index, fix the query, update the staging load test to match prod volume,
and check if any other new queries have similar issues.
"""

if __name__ == "__main__":
    print("Generating incident report...")
    
    report_data = generate_incident_report(
        raw_notes=SAMPLE_NOTES,
        additional_context={
            "reported_by": "Monitoring System (PagerDuty)",
            "team": "Platform Engineering",
            "environment": "Production"
        }
    )
    
    # Format as Markdown
    markdown_report = format_report_markdown(report_data, report_id="INC-2026041401")
    print(markdown_report)
    
    # Export as JSON for ITSM
    export_report_json(report_data, "incident_report.json")