--- title: "AI Agent for Plumbing Services: Emergency Dispatch and Routine Scheduling" description: "Build an AI agent that classifies plumbing emergencies, dispatches technicians with smart routing, estimates pricing, and handles follow-up scheduling for plumbing service companies." canonical: https://callsphere.ai/blog/ai-agent-plumbing-services-emergency-dispatch-scheduling category: "Learn Agentic AI" tags: ["Plumbing", "Emergency Dispatch", "Field Service AI", "Scheduling", "Pricing Estimation"] author: "CallSphere Team" published: 2026-03-17T00:00:00.000Z updated: 2026-05-06T01:02:44.689Z --- # AI Agent for Plumbing Services: Emergency Dispatch and Routine Scheduling > Build an AI agent that classifies plumbing emergencies, dispatches technicians with smart routing, estimates pricing, and handles follow-up scheduling for plumbing service companies. ## 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. ```mermaid flowchart TD USAGE{"Monthly call volume?"} STARTER["Starter under 500 calls per month"] GROWTH["Growth 500 to 5,000 per month"] SCALE["Scale 5,000 plus per month"] ENT["Enterprise dedicated infra and SLA"] USAGE -->|Light| STARTER USAGE -->|Mid| GROWTH USAGE -->|High| SCALE USAGE -->|Custom| ENT STARTER --> NEXT(["Pick monthly plan"]) GROWTH --> NEXT SCALE --> NEXT ENT --> NEXT style USAGE fill:#4f46e5,stroke:#4338ca,color:#fff style NEXT fill:#059669,stroke:#047857,color:#fff ``` ```python 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. ```python 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 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. ```python 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 --- Source: https://callsphere.ai/blog/ai-agent-plumbing-services-emergency-dispatch-scheduling