AI Agent for Plumbing Services: Emergency Dispatch and Routine Scheduling

The Plumbing Dispatch Challenge

Plumbing companies face a unique operational pressure: a burst pipe at 2 AM demands a fundamentally different response than a dripping faucet reported on a Tuesday morning. Yet both calls come through the same phone line, handled by the same overworked dispatcher. An AI agent can classify urgency in seconds, route the right technician, provide instant pricing estimates, and schedule follow-up visits — all without human intervention.

The critical capability is urgency classification. Getting this wrong in either direction costs money: treating a slow drain as an emergency wastes premium-rate technician hours, while treating a slab leak as routine causes thousands in water damage.

Building the Urgency Classifier

Plumbing urgency depends on water flow, location, and damage potential. We build a scoring system that considers multiple factors.

from enum import Enum
from dataclasses import dataclass

class UrgencyLevel(Enum):
    EMERGENCY = "emergency"       # Active flooding, sewage backup, gas line
    URGENT = "urgent"             # No water, water heater failure, major leak
    SAME_DAY = "same_day"         # Moderate leak, clogged main drain
    SCHEDULED = "scheduled"       # Dripping faucet, running toilet, slow drain

@dataclass
class UrgencyAssessment:
    level: UrgencyLevel
    score: int
    reasoning: str
    max_response_hours: float

URGENCY_RULES = [
    {"keywords": ["flooding", "burst pipe", "sewage backup", "gas smell"],
     "level": UrgencyLevel.EMERGENCY, "score": 100, "max_hours": 1.0},
    {"keywords": ["no water", "no hot water", "major leak", "water heater"],
     "level": UrgencyLevel.URGENT, "score": 75, "max_hours": 4.0},
    {"keywords": ["clogged drain", "slow drain", "moderate leak"],
     "level": UrgencyLevel.SAME_DAY, "score": 50, "max_hours": 8.0},
    {"keywords": ["dripping", "running toilet", "faucet replacement"],
     "level": UrgencyLevel.SCHEDULED, "score": 25, "max_hours": 72.0},
]

def classify_urgency(description: str) -> UrgencyAssessment:
    description_lower = description.lower()
    for rule in URGENCY_RULES:
        if any(kw in description_lower for kw in rule["keywords"]):
            return UrgencyAssessment(
                level=rule["level"],
                score=rule["score"],
                reasoning=f"Matched: {[k for k in rule['keywords'] if k in description_lower]}",
                max_response_hours=rule["max_hours"],
            )
    return UrgencyAssessment(
        level=UrgencyLevel.SCHEDULED,
        score=10,
        reasoning="No urgent keywords detected",
        max_response_hours=72.0,
    )

Smart Dispatch Logic

Once urgency is classified, the agent selects the best technician based on proximity, current workload, and specialization.

from math import radians, sin, cos, sqrt, atan2

def haversine_distance(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
    R = 3959  # Earth radius in miles
    dlat = radians(lat2 - lat1)
    dlon = radians(lon2 - lon1)
    a = sin(dlat/2)**2 + cos(radians(lat1)) * cos(radians(lat2)) * sin(dlon/2)**2
    return R * 2 * atan2(sqrt(a), sqrt(1 - a))

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

    async def find_best_technician(
        self, urgency: UrgencyLevel, job_lat: float, job_lon: float,
        specialization: str = None,
    ) -> dict:
        techs = await self.db.fetch("""
            SELECT t.id, t.name, t.current_lat, t.current_lon,
                   t.active_jobs, t.specializations, t.rating
            FROM technicians t
            WHERE t.status = 'available'
              AND t.active_jobs < t.max_concurrent_jobs
            ORDER BY t.rating DESC
        """)

        scored = []
        for tech in techs:
            distance = haversine_distance(
                job_lat, job_lon,
                tech["current_lat"], tech["current_lon"]
            )
            distance_score = max(0, 100 - (distance * 5))
            workload_score = (5 - tech["active_jobs"]) * 20
            spec_score = 30 if specialization in tech["specializations"] else 0
            total = distance_score + workload_score + spec_score

            if urgency == UrgencyLevel.EMERGENCY:
                total = distance_score * 2 + spec_score  # Proximity dominates
            scored.append({**dict(tech), "score": total, "distance_miles": round(distance, 1)})

        scored.sort(key=lambda t: t["score"], reverse=True)
        return scored[0] if scored else None

Pricing Estimation Engine

Customers want to know costs upfront. The agent builds estimates from a service catalog with labor and materials.

SERVICE_CATALOG = {
    "faucet_repair": {"base_labor": 85, "parts_avg": 25, "hours": 1.0},
    "toilet_repair": {"base_labor": 95, "parts_avg": 35, "hours": 1.0},
    "drain_clearing": {"base_labor": 150, "parts_avg": 0, "hours": 1.5},
    "water_heater_replace": {"base_labor": 450, "parts_avg": 800, "hours": 4.0},
    "pipe_repair": {"base_labor": 200, "parts_avg": 50, "hours": 2.0},
    "slab_leak_repair": {"base_labor": 1200, "parts_avg": 300, "hours": 8.0},
}

def estimate_price(
    service_type: str, urgency: UrgencyLevel, after_hours: bool = False,
) -> dict:
    service = SERVICE_CATALOG.get(service_type)
    if not service:
        return {"error": f"Unknown service type: {service_type}"}

    labor = service["base_labor"]
    multiplier = {
        UrgencyLevel.EMERGENCY: 1.75,
        UrgencyLevel.URGENT: 1.35,
        UrgencyLevel.SAME_DAY: 1.15,
        UrgencyLevel.SCHEDULED: 1.0,
    }
    labor *= multiplier[urgency]
    if after_hours:
        labor *= 1.5

    total = labor + service["parts_avg"]
    return {
        "service": service_type,
        "labor_estimate": round(labor, 2),
        "parts_estimate": service["parts_avg"],
        "total_range": f"${round(total * 0.85, 0):.0f} - ${round(total * 1.15, 0):.0f}",
        "urgency_surcharge": multiplier[urgency] > 1.0,
        "after_hours_surcharge": after_hours,
    }

Follow-Up Scheduling

After the initial service call, the agent automatically schedules follow-up visits for warranty checks or ongoing issues.

from datetime import datetime, timedelta

class FollowUpScheduler:
    async def create_follow_up(
        self, job_id: str, service_type: str, customer_id: str
    ) -> dict:
        follow_up_rules = {
            "water_heater_replace": {"days": 30, "reason": "Installation warranty check"},
            "pipe_repair": {"days": 14, "reason": "Leak recheck"},
            "slab_leak_repair": {"days": 7, "reason": "Pressure test verification"},
        }
        rule = follow_up_rules.get(service_type)
        if not rule:
            return {"follow_up_needed": False}

        follow_up_date = datetime.now() + timedelta(days=rule["days"])
        return {
            "follow_up_needed": True,
            "scheduled_date": follow_up_date.strftime("%Y-%m-%d"),
            "reason": rule["reason"],
            "job_reference": job_id,
            "customer_id": customer_id,
        }

FAQ

How does the agent handle multiple emergencies at the same time?

The dispatcher maintains a priority queue. When multiple emergencies arrive simultaneously, it scores each technician against each job and solves the assignment problem to minimize total response time. If all technicians are occupied, it escalates to the on-call manager and provides the customer with an honest ETA rather than a false promise.

Should the pricing estimates be binding?

No. The agent always presents estimates as ranges with clear disclaimers. The final price depends on on-site conditions. However, tracking estimate-to-invoice variance helps you calibrate the model over time — most well-tuned systems achieve 80-90% accuracy.

How do you handle after-hours calls differently?

After-hours logic checks the current time against business hours and automatically applies the surcharge multiplier. The agent also adjusts the available technician pool to only show on-call staff and sets customer expectations about response times.

---

#Plumbing #EmergencyDispatch #FieldServiceAI #Scheduling #PricingEstimation #AgenticAI #LearnAI #AIEngineering