AI Agent for Roofing Companies: Damage Assessment, Insurance Claims, and Scheduling

The Roofing Business Workflow

Roofing companies operate in a unique space where most revenue comes through insurance claims after storm damage. The workflow is complex: inspect the roof, document damage with photos and measurements, generate a detailed scope of work using Xactimate pricing, submit the claim to the insurance carrier, negotiate supplements, schedule the repair once approved, and manage crews across multiple active projects. An AI agent that handles documentation, claim preparation, and scheduling can cut the time from inspection to repair start by 40%.

The most valuable automation is claim documentation. Insurance adjusters reject claims with insufficient or poorly organized documentation. An AI agent ensures every claim package is thorough and formatted to the carrier's requirements.

Damage Assessment from Inspection Data

Roof inspections generate photos, measurements, and field notes. The agent structures this raw data into a formal damage assessment.

flowchart LR
    CALLER(["Policyholder or Lead"])
    subgraph TEL["Telephony"]
        SIP["Twilio SIP and PSTN"]
    end
    subgraph BRAIN["Insurance AI Agent"]
        STT["Streaming STT<br/>Deepgram or Whisper"]
        NLU{"Intent and<br/>Entity Extraction"}
        TOOLS["Tool Calls"]
        TTS["Streaming TTS<br/>ElevenLabs or Rime"]
    end
    subgraph DATA["Live Data Plane"]
        CRM[("CRM and Notes")]
        CAL[("Calendar and<br/>Schedule")]
        KB[("Knowledge Base<br/>and Policies")]
    end
    subgraph OUT["Outcomes"]
        O1(["Quote captured"])
        O2(["Claim opened in core"])
        O3(["Licensed agent handoff"])
    end
    CALLER --> SIP --> STT --> NLU
    NLU -->|Lookup| TOOLS
    TOOLS <--> CRM
    TOOLS <--> CAL
    TOOLS <--> KB
    NLU --> TTS --> SIP --> CALLER
    NLU -->|Resolved| O1
    NLU -->|Schedule| O2
    NLU -->|Escalate| O3
    style CALLER fill:#f1f5f9,stroke:#64748b,color:#0f172a
    style NLU fill:#4f46e5,stroke:#4338ca,color:#fff
    style O1 fill:#059669,stroke:#047857,color:#fff
    style O2 fill:#0ea5e9,stroke:#0369a1,color:#fff
    style O3 fill:#f59e0b,stroke:#d97706,color:#1f2937

from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Optional

class DamageType(Enum):
    HAIL = "hail"
    WIND = "wind"
    FALLEN_TREE = "fallen_tree"
    AGE_WEAR = "age_wear"
    WATER = "water"
    MISSING_SHINGLES = "missing_shingles"

class DamageSeverity(Enum):
    MINOR = "minor"           # Cosmetic, no leak risk
    MODERATE = "moderate"     # Functional damage, leak possible
    SEVERE = "severe"         # Active leak or structural compromise
    TOTAL_LOSS = "total_loss" # Full replacement required

@dataclass
class DamageArea:
    area_id: str
    location: str          # "north slope", "ridge", "valley"
    damage_type: DamageType
    severity: DamageSeverity
    size_sqft: float
    photo_urls: list[str] = field(default_factory=list)
    notes: str = ""

@dataclass
class RoofAssessment:
    property_address: str
    inspection_date: datetime
    roof_type: str            # "asphalt_shingle", "metal", "tile", "flat"
    total_sqft: float
    pitch: str                # "4/12", "6/12", "8/12"
    stories: int
    damage_areas: list[DamageArea] = field(default_factory=list)
    storm_date: Optional[datetime] = None

    @property
    def total_damage_sqft(self) -> float:
        return sum(area.size_sqft for area in self.damage_areas)

    @property
    def damage_percentage(self) -> float:
        return (self.total_damage_sqft / self.total_sqft * 100) if self.total_sqft else 0

    def recommendation(self) -> str:
        if self.damage_percentage > 25 or any(
            a.severity == DamageSeverity.TOTAL_LOSS for a in self.damage_areas
        ):
            return "full_replacement"
        elif self.damage_percentage > 10:
            return "partial_replacement"
        else:
            return "repair"

Insurance Claim Documentation Generator

Insurance claims require specific documentation formats. The agent compiles the assessment into a claim-ready package.

class ClaimDocumentGenerator:
    XACTIMATE_CODES = {
        "asphalt_shingle": {
            "tear_off": "RFG TKOF",
            "install": "RFG COMP",
            "underlayment": "RFG FELT",
            "flashing": "RFG FLSH",
            "ridge_cap": "RFG RDGC",
            "drip_edge": "RFG DRPE",
        },
        "metal": {
            "tear_off": "RFG TKOF",
            "install": "RFG MTL",
            "underlayment": "RFG SYNT",
        },
    }

    def generate_scope_of_work(self, assessment: RoofAssessment) -> dict:
        codes = self.XACTIMATE_CODES.get(assessment.roof_type, {})
        rec = assessment.recommendation()
        line_items = []

        if rec == "full_replacement":
            sqft = assessment.total_sqft
            line_items.extend([
                {"code": codes.get("tear_off", ""), "description": "Tear off existing roofing",
                 "quantity": sqft, "unit": "SF"},
                {"code": codes.get("install", ""), "description": f"Install {assessment.roof_type}",
                 "quantity": sqft, "unit": "SF"},
                {"code": codes.get("underlayment", ""), "description": "Install underlayment",
                 "quantity": sqft, "unit": "SF"},
            ])
        else:
            for area in assessment.damage_areas:
                line_items.append({
                    "code": codes.get("install", ""),
                    "description": f"Repair {area.location} — {area.damage_type.value}",
                    "quantity": area.size_sqft,
                    "unit": "SF",
                })

        # Add standard accessories
        perimeter_lf = (assessment.total_sqft ** 0.5) * 4
        line_items.append({
            "code": codes.get("drip_edge", ""),
            "description": "Install drip edge",
            "quantity": round(perimeter_lf),
            "unit": "LF",
        })
        return {
            "recommendation": rec,
            "line_items": line_items,
            "total_sqft_affected": assessment.total_damage_sqft,
            "photo_count": sum(len(a.photo_urls) for a in assessment.damage_areas),
        }

    def generate_claim_package(self, assessment: RoofAssessment) -> dict:
        scope = self.generate_scope_of_work(assessment)
        return {
            "claim_type": "property_damage",
            "date_of_loss": (
                assessment.storm_date.strftime("%Y-%m-%d")
                if assessment.storm_date else "Unknown"
            ),
            "property_address": assessment.property_address,
            "inspection_date": assessment.inspection_date.strftime("%Y-%m-%d"),
            "roof_details": {
                "type": assessment.roof_type,
                "total_sqft": assessment.total_sqft,
                "pitch": assessment.pitch,
                "stories": assessment.stories,
            },
            "damage_summary": {
                "areas_affected": len(assessment.damage_areas),
                "total_damage_sqft": assessment.total_damage_sqft,
                "damage_percentage": round(assessment.damage_percentage, 1),
                "damage_types": list({a.damage_type.value for a in assessment.damage_areas}),
            },
            "scope_of_work": scope,
            "supporting_documents": [
                "Inspection photos",
                "Measurement diagram",
                "Storm date verification (weather report)",
                "Material specification sheet",
            ],
        }

Insurance Workflow Tracker

Roofing claims go through multiple stages with the insurance carrier. The agent tracks progress and prompts action.

class InsuranceWorkflowTracker:
    WORKFLOW_STAGES = [
        "claim_filed", "adjuster_assigned", "inspection_scheduled",
        "inspection_complete", "estimate_received", "supplement_needed",
        "supplement_submitted", "approved", "work_authorized",
    ]

    def __init__(self, db):
        self.db = db

    async def update_claim_status(self, claim_id: str, new_status: str) -> dict:
        current = await self.db.fetchrow(
            "SELECT status, filed_date FROM insurance_claims WHERE claim_id = $1",
            claim_id,
        )
        stage_index = self.WORKFLOW_STAGES.index(new_status)
        next_action = self._get_next_action(new_status)

        await self.db.execute(
            """UPDATE insurance_claims
               SET status = $1, updated_at = NOW()
               WHERE claim_id = $2""",
            new_status, claim_id,
        )
        return {
            "claim_id": claim_id,
            "previous_status": current["status"],
            "new_status": new_status,
            "progress": f"{stage_index + 1}/{len(self.WORKFLOW_STAGES)}",
            "next_action": next_action,
            "days_since_filed": (datetime.now() - current["filed_date"]).days,
        }

    def _get_next_action(self, status: str) -> str:
        actions = {
            "claim_filed": "Wait for adjuster assignment (typical: 24 hours)",
            "adjuster_assigned": "Contact adjuster to schedule inspection",
            "inspection_scheduled": "Prepare for joint inspection — have documentation ready",
            "inspection_complete": "Wait for carrier estimate (typical: 5-10 business days)",
            "estimate_received": "Review estimate against your scope — prepare supplement if needed",
            "supplement_needed": "Submit supplement with supporting documentation",
            "supplement_submitted": "Follow up with adjuster in 7 business days",
            "approved": "Send authorization form to homeowner for signature",
            "work_authorized": "Schedule crew and order materials",
        }
        return actions.get(status, "Contact office for guidance")

Crew Scheduling for Roof Jobs

Roofing crews need specific equipment, favorable weather windows, and often work multiple jobs per week.

class RoofingCrewScheduler:
    async def schedule_job(
        self, job_id: str, assessment: RoofAssessment, weather_service,
    ) -> dict:
        duration_days = self._estimate_duration(assessment)
        min_crew_size = self._calculate_crew_size(assessment)

        # Find weather-clear windows
        forecast = await weather_service.get_extended_forecast(
            assessment.property_address, days=14
        )
        clear_windows = [
            day for day in forecast
            if day["precipitation_chance"] < 20
               and day["wind_speed_mph"] < 20
        ]

        consecutive_clear = self._find_consecutive_days(clear_windows, duration_days)
        if not consecutive_clear:
            return {
                "scheduled": False,
                "reason": f"Need {duration_days} consecutive clear days — none found in 14-day forecast",
                "next_check_date": forecast[-1]["date"],
            }

        return {
            "scheduled": True,
            "start_date": consecutive_clear[0]["date"],
            "end_date": consecutive_clear[-1]["date"],
            "crew_size": min_crew_size,
            "duration_days": duration_days,
        }

    def _estimate_duration(self, assessment: RoofAssessment) -> int:
        sqft = assessment.total_sqft if assessment.recommendation() == "full_replacement" else assessment.total_damage_sqft
        sqft_per_day = 1500 if assessment.stories <= 1 else 1000
        return max(1, round(sqft / sqft_per_day))

    def _calculate_crew_size(self, assessment: RoofAssessment) -> int:
        if assessment.total_sqft > 3000:
            return 6
        elif assessment.total_sqft > 1500:
            return 4
        return 3

    def _find_consecutive_days(self, clear_days: list, needed: int) -> list:
        for i in range(len(clear_days) - needed + 1):
            window = clear_days[i:i + needed]
            if len(window) == needed:
                return window
        return []

FAQ

How does the agent handle supplement negotiations with insurance carriers?

When the carrier's estimate is lower than the contractor's scope, the agent generates a supplement document that highlights specific line items where the carrier's pricing is below market rate or where damage areas were missed. It includes the relevant Xactimate codes, supporting photos for each disputed item, and references to the carrier's own pricing database. This structured approach increases supplement approval rates significantly compared to informal negotiations.

Can the agent verify storm dates against weather records?

Yes. The agent queries historical weather data APIs (NOAA Storm Events, Weather Underground) to verify that a hail or wind event occurred at the claimed location on the stated date. This verification is included in the claim package and strengthens the claim by providing independent corroboration of the date of loss.

What happens when a job needs to pause mid-project due to weather?

The agent monitors forecasts daily during active jobs. When rain is predicted, it alerts the crew lead to ensure tarps are properly secured on any open sections. It then recalculates the completion date and notifies the homeowner and any pending follow-on trades (gutters, siding) of the revised timeline.

---

#Roofing #DamageAssessment #InsuranceClaims #PhotoAnalysis #CrewScheduling #AgenticAI #LearnAI #AIEngineering