--- title: "AI Agent for Home Services: Plumbing, HVAC, and Electrical Inquiry Handling" description: "Build an AI agent that classifies home service requests, detects emergencies, provides preliminary quotes, and schedules technician visits for plumbing, HVAC, and electrical businesses." canonical: https://callsphere.ai/blog/ai-agent-home-services-plumbing-hvac-electrical-inquiry-handling category: "Learn Agentic AI" tags: ["Home Services", "Emergency Detection", "AI Scheduling", "Service Classification", "Python"] author: "CallSphere Team" published: 2026-03-17T00:00:00.000Z updated: 2026-05-06T17:14:36.573Z --- # AI Agent for Home Services: Plumbing, HVAC, and Electrical Inquiry Handling > Build an AI agent that classifies home service requests, detects emergencies, provides preliminary quotes, and schedules technician visits for plumbing, HVAC, and electrical businesses. ## Home Services: Where Speed and Accuracy Save Revenue When a homeowner calls a plumbing company, they are either dealing with a minor inconvenience or a full-blown emergency. The difference between "my faucet drips" and "water is flooding my basement" requires completely different response times, technician skill levels, and pricing. A human dispatcher makes these judgments intuitively. An AI agent can make them just as well — and it never takes a lunch break. This tutorial builds an agent for home service businesses that classifies inquiries, detects emergencies, provides ballpark quotes, and schedules technician visits. ## Service Classification Model Home service requests fall into categories with different urgency levels, skill requirements, and pricing. We model this as a structured service catalog. ```mermaid flowchart LR CALLER(["Homeowner"]) subgraph TEL["Telephony"] SIP["Twilio SIP and PSTN"] end subgraph BRAIN["Field Service AI Agent"] STT["Streaming STT Deepgram or Whisper"] NLU{"Intent and Entity Extraction"} TOOLS["Tool Calls"] TTS["Streaming TTS ElevenLabs or Rime"] end subgraph DATA["Live Data Plane"] CRM[("CRM and Notes")] CAL[("Calendar and Schedule")] KB[("Knowledge Base and Policies")] end subgraph OUT["Outcomes"] O1(["Service appointment booked"]) O2(["Quote sent via SMS"]) O3(["Tech dispatched today"]) 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 ``` ```python from dataclasses import dataclass, field from enum import Enum from typing import Optional from datetime import datetime, date, time, timedelta class ServiceCategory(Enum): PLUMBING = "plumbing" HVAC = "hvac" ELECTRICAL = "electrical" class UrgencyLevel(Enum): EMERGENCY = "emergency" # dispatch within 1 hour URGENT = "urgent" # same-day service STANDARD = "standard" # schedule within 2-3 days ROUTINE = "routine" # schedule within 1-2 weeks @dataclass class ServiceType: id: str name: str category: ServiceCategory base_price: float duration_hours: float emergency_surcharge: float requires_permit: bool = False description: str = "" @dataclass class ServiceRequest: id: str = "" customer_name: str = "" customer_phone: str = "" address: str = "" category: Optional[ServiceCategory] = None service_type_id: str = "" description: str = "" urgency: UrgencyLevel = UrgencyLevel.STANDARD estimated_cost: float = 0.0 scheduled_date: Optional[date] = None scheduled_window: str = "" # e.g., "8 AM - 12 PM" created_at: datetime = field(default_factory=datetime.now) ``` ## Emergency Detection Engine Detecting emergencies accurately is critical. A false negative means a flooded home sits for hours. A false positive means dispatching an expensive emergency crew for a dripping faucet. We use keyword matching combined with contextual scoring. ```python EMERGENCY_INDICATORS = { "plumbing": { "keywords": [ "flooding", "burst pipe", "sewage backup", "no water", "water everywhere", "gas smell", "water heater leaking", "overflowing", ], "questions": [ "Is water actively flowing or spreading?", "Can you locate and turn off the water shut-off valve?", "Do you smell gas?", ], }, "hvac": { "keywords": [ "no heat", "carbon monoxide", "gas smell", "furnace not working", "frozen pipes", "no cooling", "burning smell", ], "questions": [ "What is the current temperature inside your home?", "Do you have any carbon monoxide detectors going off?", "Are there vulnerable people (elderly, infants) in the home?", ], }, "electrical": { "keywords": [ "sparking", "burning smell", "power out", "exposed wires", "shock", "outlet smoking", "panel buzzing", "flickering lights", ], "questions": [ "Is there any visible smoke or fire?", "Have you turned off the breaker to the affected area?", "Is anyone injured?", ], }, } def assess_urgency(description: str, category: str) -> dict: """Analyze a service request description and determine urgency.""" desc_lower = description.lower() indicators = EMERGENCY_INDICATORS.get(category, {}) keywords = indicators.get("keywords", []) matched_keywords = [kw for kw in keywords if kw in desc_lower] if len(matched_keywords) >= 2 or any( critical in desc_lower for critical in ["gas smell", "fire", "carbon monoxide", "shock"] ): return { "urgency": UrgencyLevel.EMERGENCY, "reason": f"Emergency indicators detected: {matched_keywords}", "follow_up_questions": indicators.get("questions", []), } elif matched_keywords: return { "urgency": UrgencyLevel.URGENT, "reason": f"Urgent indicator: {matched_keywords[0]}", "follow_up_questions": indicators.get("questions", [])[:1], } return { "urgency": UrgencyLevel.STANDARD, "reason": "No emergency indicators detected.", "follow_up_questions": [], } ``` ## Quoting Engine Homeowners always want to know "how much will it cost?" before committing. The agent provides ballpark estimates based on the service catalog while making clear that final pricing requires an on-site assessment. ```python SERVICE_CATALOG = { "leaky_faucet": ServiceType("p1", "Faucet Repair", ServiceCategory.PLUMBING, 120, 1.0, 75), "clogged_drain": ServiceType("p2", "Drain Clearing", ServiceCategory.PLUMBING, 150, 1.5, 100), "water_heater": ServiceType("p3", "Water Heater Repair", ServiceCategory.PLUMBING, 250, 2.0, 150), "ac_repair": ServiceType("h1", "AC Repair", ServiceCategory.HVAC, 200, 2.0, 125), "furnace_repair": ServiceType("h2", "Furnace Repair", ServiceCategory.HVAC, 225, 2.5, 150), "outlet_install": ServiceType("e1", "Outlet Installation", ServiceCategory.ELECTRICAL, 175, 1.0, 100, requires_permit=True), "panel_upgrade": ServiceType("e2", "Panel Upgrade", ServiceCategory.ELECTRICAL, 1200, 6.0, 300, requires_permit=True), } def generate_estimate(service_id: str, urgency: UrgencyLevel) -> dict: service = SERVICE_CATALOG.get(service_id) if not service: return {"error": "Service not found"} base = service.base_price surcharge = service.emergency_surcharge if urgency == UrgencyLevel.EMERGENCY else 0 total_low = base + surcharge total_high = total_low * 1.4 # 40% range for unknowns return { "service": service.name, "base_price": base, "emergency_surcharge": surcharge, "estimate_range": f"${total_low:.0f} - ${total_high:.0f}", "note": "Final pricing determined after on-site assessment.", "permit_required": service.requires_permit, } ``` ## Agent Tools and Assembly ```python from agents import Agent, Runner, function_tool @function_tool def classify_and_assess(description: str, category: str) -> str: """Classify a home service request and assess urgency level.""" result = assess_urgency(description, category) urgency = result["urgency"].value response = f"Urgency: {urgency}\nReason: {result['reason']}" if result["follow_up_questions"]: response += "\nSafety questions to ask:\n" for q in result["follow_up_questions"]: response += f"- {q}\n" return response @function_tool def get_estimate(service_id: str, is_emergency: bool = False) -> str: """Get a price estimate for a specific service.""" urgency = UrgencyLevel.EMERGENCY if is_emergency else UrgencyLevel.STANDARD est = generate_estimate(service_id, urgency) if "error" in est: return est["error"] result = f"Service: {est['service']}\nEstimate: {est['estimate_range']}" if est["emergency_surcharge"]: result += f"\nEmergency surcharge: ${est['emergency_surcharge']}" if est["permit_required"]: result += "\nNote: This service requires a permit." return result + f"\n{est['note']}" @function_tool def schedule_visit( customer_name: str, phone: str, address: str, service_id: str, preferred_date: str, preferred_window: str ) -> str: """Schedule a technician visit for a home service request.""" service = SERVICE_CATALOG.get(service_id) return ( f"Visit scheduled: {service.name} for {customer_name}\n" f"Date: {preferred_date}, Window: {preferred_window}\n" f"Address: {address}\nConfirmation sent to {phone}" ) home_service_agent = Agent( name="Home Service Dispatcher", instructions="""You are a dispatcher for a home services company offering plumbing, HVAC, and electrical services. 1. Listen to the customer's problem and determine the category (plumbing, HVAC, or electrical). 2. Use classify_and_assess to evaluate urgency. If the system returns safety questions, ask the customer those questions. 3. For emergencies, immediately collect address and phone, then schedule an emergency visit. Advise on immediate safety steps. 4. For non-emergencies, use get_estimate to provide pricing, then offer to schedule_visit. 5. Always collect: customer name, phone, and service address. 6. Emphasize that estimates are approximate until on-site assessment.""", tools=[classify_and_assess, get_estimate, schedule_visit], ) ``` ## FAQ ### How does the agent handle after-hours emergency calls differently from daytime calls? After-hours emergency calls trigger the same urgency detection, but the scheduling tool routes to on-call technicians with emergency rates. Add a `BusinessHoursRouter` (similar to the answering service pattern) that checks the current time and applies the correct surcharge and dispatch rules. ### Can the agent handle requests for services not in the catalog? Yes. If `classify_and_assess` does not match a known service type, the agent instructions tell it to collect the problem description, customer details, and photos if possible, then flag the request for manual review by a dispatcher. This prevents the agent from inventing prices for unknown work. ### How do I integrate real technician availability into the scheduling? Replace the static `schedule_visit` tool with a lookup against your field service management system (ServiceTitan, Housecall Pro, or Jobber). These platforms expose APIs for real-time technician availability, and the tool interface stays the same — only the backend implementation changes. --- #HomeServices #EmergencyDetection #AIScheduling #ServiceClassification #Python #AgenticAI #LearnAI #AIEngineering --- Source: https://callsphere.ai/blog/ai-agent-home-services-plumbing-hvac-electrical-inquiry-handling