Customer Scheduling Agent (small Business)

For the roofer / plumber / cleaning service / any home-service business.

"""
Customer Scheduling Agent
- Reads new lead form submissions (from a webhook-fed JSON file).
- Replies to qualified leads with available estimate slots.
- Books the appointment in the business calendar when the lead
  responds.

Prerequisites:
- Same Google auth setup as 8.7.
- Twilio account for SMS (optional; can also use email-only).
- A simple inbox for incoming form submissions: a JSON file at
  `inbox/leads.jsonl`, one lead per line, written by your website's
  form handler.
"""
import json
from datetime import datetime, timedelta
from pathlib import Path

from crewai import Agent, Task, Crew, Process
from crewai_tools import tool


# ---------- Stub tools (replace with real implementations) ----------
# Real implementations would call Gmail/Twilio/Calendar APIs.
# These are here to show the interface the agents expect.

@tool("read_new_leads")
def read_new_leads() -> str:
    """Read pending leads from the inbox."""
    inbox = Path("inbox/leads.jsonl")
    if not inbox.exists():
        return "[]"
    leads = []
    with open(inbox) as f:
        for line in f:
            if line.strip():
                leads.append(json.loads(line))
    return json.dumps(leads)


@tool("get_open_estimate_slots")
def get_open_estimate_slots(days_ahead: int = 7) -> str:
    """Return open 1-hour estimate slots between 9am-5pm in the
    next N days."""
    # Stub: in a real implementation, query the calendar.
    slots = []
    base = datetime.now().replace(hour=9, minute=0, second=0,
                                   microsecond=0)
    for d in range(1, days_ahead + 1):
        for h in [10, 14, 16]:
            slots.append((base + timedelta(days=d))
                         .replace(hour=h).isoformat())
    return json.dumps(slots[:9])


@tool("send_lead_reply")
def send_lead_reply(
    lead_id: str, channel: str, message: str
) -> str:
    """Send a reply to the lead via 'email' or 'sms'."""
    print(f"\n[WOULD SEND] {channel.upper()} to lead {lead_id}:")
    print(message)
    print("[END]\n")
    return "sent"


@tool("book_estimate")
def book_estimate(
    lead_id: str, slot_iso: str, customer_name: str,
    customer_address: str
) -> str:
    """Create the calendar event and mark the lead as booked."""
    print(f"\n[WOULD BOOK] {customer_name} at {customer_address} "
          f"on {slot_iso} (lead {lead_id})\n")
    return f"booked at {slot_iso}"


# ---------- Agents ----------

receptionist = Agent(
    role="Customer Service Receptionist",
    goal=(
        "Welcome new leads warmly, qualify them by service area and "
        "service type, and offer estimate appointment slots."
    ),
    backstory=(
        "You work for a friendly local service business. You're "
        "responsive, clear, and never pushy. You always confirm "
        "details before booking."
    ),
    tools=[read_new_leads, get_open_estimate_slots, send_lead_reply],
    verbose=True,
)

booker = Agent(
    role="Appointment Booker",
    goal=(
        "When a lead has confirmed a slot, create the calendar "
        "event and notify the owner."
    ),
    backstory=(
        "You are detail-oriented and never double-book. You always "
        "verify the address before creating the event."
    ),
    tools=[book_estimate],
    verbose=True,
)


# ---------- Tasks ----------

intake_task = Task(
    description=(
        "Read all new leads. For each lead in the service area "
        "(zipcode in [22301, 22302, 22303, 22304, 22305]) and "
        "matching our service category (residential roofing): "
        "1. Get 3 open estimate slots in the next 7 days.\n"
        "2. Compose a friendly reply offering those 3 slots.\n"
        "3. Send the reply via the channel they used (email or sms)."
        "\n\nFor leads outside the service area: send a polite "
        "message that we don't service that area, with a referral "
        "to a partner if known. For leads in unsupported categories "
        "(commercial, industrial): send a polite redirect.\n\n"
        "Output: a JSON list of {lead_id, action_taken, "
        "slots_offered}."
    ),
    agent=receptionist,
    expected_output="JSON list of intake actions.",
)


crew = Crew(
    agents=[receptionist, booker],
    tasks=[intake_task],
    process=Process.sequential,
    verbose=True,
)


if __name__ == "__main__":
    result = crew.kickoff()
    print("\n=== INTAKE RESULTS ===")
    print(result)

Notes for adapting:

• Replace the stubbed tools with real implementations against your actual Gmail, Twilio, and Calendar.
• Change the service area zipcodes and service category to match your business.
• Schedule with launchd/systemd to run every 5-10 minutes during business hours.
• Add a second-step task for booking once a customer replies "I'll take the 2pm Tuesday slot." That task should parse the reply, match it to one of the offered slots, and call book_estimate.