#AI Lead Qualification on WhatsApp with OpenClaw

Bot: What's your budget?
Lead: [leaves conversation]

Agent: What's the biggest challenge you're facing with customer support right now?
Lead: We're drowning in WhatsApp messages. Takes hours to respond to everyone.
Agent: I hear that a lot. How many messages are you handling daily?
Lead: Around 500, maybe more during busy periods.
Agent: That's a lot. Are you using any tools to manage it, or is it all manual?
Lead: All manual. We've looked at a few solutions but they're too expensive.
Agent: What's "too expensive" in this context? Just so I can recommend something in the right range.
Lead: We can't justify more than $500/month right now.

from openclaw import Agent, LeadScoringSkill, CRMIntegrationSkill

qualification_agent = Agent(
    name="WhatsApp Lead Qualifier",
    system_prompt="""You are a friendly pre-sales consultant helping potential customers
    evaluate whether our WhatsApp automation platform fits their needs.

    Your job is to understand:
    1. Their current pain points with customer communication
    2. The scale of their operation (message volume, team size)
    3. Their budget constraints
    4. Who makes purchasing decisions
    5. Their timeline for implementing a solution

    Be conversational, not interrogative. Ask open-ended questions.
    Listen to their responses and probe naturally based on what they share.

    Keep responses brief (2-3 sentences). WhatsApp conversations move fast.
    Use plain text only. No emojis unless they use them first.

    If they're clearly not a fit (e.g., student working on a class project,
    competitor doing research), politely disengage and wish them well.""",

    skills=[
        LeadScoringSkill(framework="bant", threshold=80),
        CRMIntegrationSkill(crm="hubspot")
    ],

    model="gpt-4o",
    temperature=0.7  # Higher temp for more natural conversation
)

class LeadScoringSkill:
    """Tracks qualification signals and computes lead score."""

    def __init__(self, framework: str = "bant", threshold: int = 80):
        self.framework = framework
        self.threshold = threshold
        self.scoring_rules = {
            "budget_signals": {
                "explicit_budget_shared": 25,
                "premium_tier_interest": 20,
                "price_objection": -10,
                "no_budget_mentioned": 0
            },
            "authority_signals": {
                "decision_maker": 30,
                "influencer": 15,
                "end_user_only": -5
            },
            "need_signals": {
                "urgent_pain": 25,
                "moderate_pain": 15,
                "exploring_options": 5,
                "no_clear_need": -10
            },
            "timeline_signals": {
                "immediate": 20,
                "this_quarter": 15,
                "next_quarter": 10,
                "no_timeline": 0
            }
        }

    def extract_signals(self, conversation_history: list) -> dict:
        """Parse conversation for qualification signals using LLM."""
        # Simplified—in production, use structured extraction with function calling
        signals = {}

        # Budget signals
        if "budget" in str(conversation_history).lower():
            signals["budget"] = "explicit_budget_shared"
        elif "expensive" in str(conversation_history).lower():
            signals["budget"] = "price_objection"

        # Authority signals
        if any(term in str(conversation_history).lower() for term in ["ceo", "founder", "director", "vp"]):
            signals["authority"] = "decision_maker"
        elif "manager" in str(conversation_history).lower():
            signals["authority"] = "influencer"

        # Need signals
        if any(term in str(conversation_history).lower() for term in ["urgent", "asap", "immediately"]):
            signals["need"] = "urgent_pain"
        elif any(term in str(conversation_history).lower() for term in ["problem", "issue", "struggling"]):
            signals["need"] = "moderate_pain"

        # Timeline signals
        if any(term in str(conversation_history).lower() for term in ["now", "today", "this week"]):
            signals["timeline"] = "immediate"
        elif any(term in str(conversation_history).lower() for term in ["month", "quarter"]):
            signals["timeline"] = "this_quarter"

        return signals

    def calculate_score(self, signals: dict) -> int:
        """Sum scores from all detected signals."""
        total = 0
        for category, signal in signals.items():
            category_key = f"{category}_signals"
            if category_key in self.scoring_rules:
                total += self.scoring_rules[category_key].get(signal, 0)
        return max(0, min(100, total))  # Clamp to 0-100

Agent: Hey! Thanks for reaching out. I'd love to learn more about what you're
looking for. What's the biggest challenge you're facing with customer communication right now?

Agent: That sounds overwhelming. About how many messages are we talking per day?
And how big is your support team handling them?

Agent: Got it. So you're looking at around 24,000 messages per month with a small team.
Have you looked at automation tools before, or is this your first time exploring options?

Agent: Makes sense—pricing needs to work for your budget. Are you the person
who'd make the call on a tool like this, or is there someone else involved in the decision?
And is this something you're looking to solve soon, or more of a long-term plan?

async def route_lead(lead_data: dict, score: int):
    """Route lead based on qualification score."""

    if score >= 80:
        # HOT LEAD - immediate sales notification
        await notify_sales_team(lead_data, urgency="high")
        return "hot"

    elif score >= 50:
        # WARM LEAD - schedule follow-up
        await schedule_follow_up(lead_data, days=3)
        return "warm"

    else:
        # COLD LEAD - add to nurture sequence
        await add_to_nurture(lead_data)
        return "cold"


async def notify_sales_team(lead_data: dict, urgency: str):
    """Send lead details to sales team WhatsApp group."""

    summary = f"""🔥 HOT LEAD ALERT

Name: {lead_data['name']}
Contact: {lead_data['phone']}
Company: {lead_data.get('company', 'N/A')}
Score: {lead_data['score']}/100

Pain Point: {lead_data['pain_point']}
Budget: {lead_data.get('budget', 'Not disclosed')}
Timeline: {lead_data['timeline']}
Authority: {lead_data['authority_level']}

Conversation:
{lead_data['conversation_summary']}

🎯 ACTION: Reach out within 1 hour"""

    async with httpx.AsyncClient() as client:
        await client.post(
            f"https://apiv2.waiflow.app/api/v2/sessions/sales-bot/messages",
            headers={"Authorization": f"Bearer {MOLTFLOW_API_TOKEN}"},
            json={
                "chatId": SALES_GROUP_ID,  # Your sales team's WhatsApp group
                "text": summary
            }
        )

async def push_to_crm(lead_data: dict):
    """Send lead to HubSpot (or your CRM of choice)."""

    hubspot_payload = {
        "properties": {
            "firstname": lead_data.get("name", "").split()[0],
            "lastname": " ".join(lead_data.get("name", "").split()[1:]) or "Unknown",
            "phone": lead_data["phone"],
            "company": lead_data.get("company"),
            "lead_score": lead_data["score"],
            "lead_source": "WhatsApp Bot",
            "pain_point": lead_data["pain_point"],
            "qualification_status": lead_data["qualification_tier"],  # hot/warm/cold
            "notes": lead_data["conversation_summary"]
        }
    }

    async with httpx.AsyncClient() as client:
        response = await client.post(
            "https://api.hubspot.com/crm/v3/objects/contacts",
            headers={
                "Authorization": f"Bearer {HUBSPOT_API_KEY}",
                "Content-Type": "application/json"
            },
            json=hubspot_payload
        )

    if response.status_code == 201:
        print(f"Lead {lead_data['phone']} pushed to HubSpot")
    else:
        print(f"CRM push failed: {response.text}")

system_prompt="""...
If you determine this is NOT a good fit, respond warmly:
'Thanks for reaching out! Based on what you've shared, it sounds like
our platform might not be the best fit for your needs right now.
I appreciate you taking the time to chat. Best of luck with your project!'

Disqualify if:
- They're a student working on a school project
- They represent a direct competitor doing research
- Their use case doesn't align with our product (e.g., personal use, non-business)
- Budget is less than $50/month (below our minimum plan)
"""

from fastapi import FastAPI, Request
import httpx
from openclaw import Agent

app = FastAPI()

@app.post("/webhook/whatsapp")
async def qualify_lead(request: Request):
    """Receives inbound lead messages, qualifies via conversation, routes accordingly."""

    payload = await request.json()
    message = payload["data"]["body"]
    sender = payload["data"]["from"]
    session = payload["session"]

    # Retrieve or initialize conversation state
    conversation = get_conversation_state(sender)
    conversation["messages"].append({"role": "user", "content": message})

    # Process with qualification agent
    response = await qualification_agent.process(
        input_text=message,
        conversation_history=conversation["messages"],
        context={"sender_id": sender}
    )

    # Update conversation state
    conversation["messages"].append({"role": "assistant", "content": response.text})
    save_conversation_state(sender, conversation)

    # Check if qualification is complete
    if response.metadata.get("qualification_complete"):
        lead_data = extract_lead_data(conversation, sender)
        score = response.metadata["lead_score"]

        # Route based on score
        tier = await route_lead(lead_data, score)

        # Push to CRM
        await push_to_crm({**lead_data, "score": score, "qualification_tier": tier})

        # Send confirmation to lead
        if tier == "hot":
            confirmation = "Thanks for all that info! Someone from our sales team will reach out shortly to discuss next steps."
        elif tier == "warm":
            confirmation = "Appreciate you sharing that. I'll have someone follow up with you in the next few days."
        else:
            confirmation = "Thanks for chatting! We'll add you to our mailing list for updates."

        # Send response
        async with httpx.AsyncClient() as client:
            await client.post(
                f"https://apiv2.waiflow.app/api/v2/sessions/{session}/messages",
                headers={"Authorization": f"Bearer {MOLTFLOW_API_TOKEN}"},
                json={"chatId": sender, "text": confirmation}
            )

    else:
        # Qualification ongoing—continue conversation
        async with httpx.AsyncClient() as client:
            await client.post(
                f"https://apiv2.waiflow.app/api/v2/sessions/{session}/messages",
                headers={"Authorization": f"Bearer {MOLTFLOW_API_TOKEN}"},
                json={"chatId": sender, "text": response.text}
            )

    return {"status": "ok"}

# Log each question and the eventual outcome
qualification_log = {
    "question": "How many messages do you handle daily?",
    "lead_id": sender,
    "answer": message,
    "eventual_outcome": "closed_deal"  # or "lost" or "no_response"
}

# After 100+ leads, analyze
# Which questions appear in 90% of closed deals?
# Which questions don't correlate with outcomes at all?

behavioral_score = 0

# Fast responders are hot
response_time_seconds = time.time() - conversation["last_bot_message_time"]
if response_time_seconds < 300:  # 5 minutes
    behavioral_score += 15

# Detailed responses show engagement
if len(message) > 100:  # characters
    behavioral_score += 10

# Technical questions indicate buying intent
if any(term in message.lower() for term in ["api", "integration", "implement", "deploy", "security"]):
    behavioral_score += 10

total_score = bant_score + behavioral_score