AI Agent for Home Services: Plumbing, HVAC, and Electrical Inquiry Handling

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.

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.

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.

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

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