How to Build a Real Estate Lead Qualifier AI Agent with Claude API (Step-by-Step)

mkdir real-estate-lead-qualifier
cd real-estate-lead-qualifier
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate
pip install anthropic python-dotenv

ANTHROPIC_API_KEY=sk-ant-your-key-here

import os
from dotenv import load_dotenv
import anthropic

load_dotenv()

client = anthropic.Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))

response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=64,
    messages=[{"role": "user", "content": "Say: API connected successfully."}]
)

print(response.content[0].text)

import os
from dotenv import load_dotenv

load_dotenv()

# Scoring weights must add up to 100
SCORING_CONFIG = {
    "budget": {
        "weight": 30,
        "tiers": {
            "above_500k": 30,
            "300k_to_500k": 22,
            "150k_to_300k": 14,
            "below_150k": 6,
            "unknown": 0
        }
    },
    "timeline": {
        "weight": 30,
        "tiers": {
            "immediate": 30,       # Ready to buy within 30 days
            "short_term": 22,      # 1 to 3 months
            "medium_term": 14,     # 3 to 6 months
            "long_term": 6,        # 6+ months
            "unknown": 0
        }
    },
    "property_type": {
        "weight": 20,
        "tiers": {
            "specific": 20,        # Knows exactly what they want
            "general": 12,         # Has a category in mind
            "unknown": 0
        }
    },
    "financing": {
        "weight": 20,
        "tiers": {
            "cash": 20,
            "pre_approved": 18,
            "in_progress": 10,
            "not_started": 4,
            "unknown": 0
        }
    }
}

QUALIFICATION_THRESHOLDS = {
    "hot": 75,
    "warm": 50,
    "nurture": 0
}

import os
import json
from dotenv import load_dotenv

load_dotenv()

# Tool definitions sent to Claude so it knows what structured data to extract
LEAD_EXTRACTION_TOOLS = [
    {
        "name": "extract_property_type",
        "description": (
            "Extract the property type the lead is interested in. "
            "Call this when you have enough information to categorize their interest."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "property_type": {
                    "type": "string",
                    "enum": [
                        "single_family_home",
                        "condo",
                        "townhouse",
                        "multi_family",
                        "land",
                        "commercial",
                        "luxury_waterfront",
                        "unknown"
                    ],
                    "description": "The specific property type the lead wants"
                },
                "specificity": {
                    "type": "string",
                    "enum": ["specific", "general", "unknown"],
                    "description": "How clearly the lead knows what property type they want"
                },
                "notes": {
                    "type": "string",
                    "description": "Any additional context about property preferences"
                }
            },
            "required": ["property_type", "specificity"]
        }
    },
    {
        "name": "extract_budget",
        "description": (
            "Extract the lead's budget range for their real estate purchase. "
            "Call this once you can determine their price range."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "min_budget": {
                    "type": "number",
                    "description": "Minimum budget in USD"
                },
                "max_budget": {
                    "type": "number",
                    "description": "Maximum budget in USD"
                },
                "budget_tier": {
                    "type": "string",
                    "enum": [
                        "above_500k",
                        "300k_to_500k",
                        "150k_to_300k",
                        "below_150k",
                        "unknown"
                    ],
                    "description": "Budget category based on max budget"
                },
                "financing_status": {
                    "type": "string",
                    "enum": ["cash", "pre_approved", "in_progress", "not_started", "unknown"],
                    "description": "Current financing or pre-approval status"
                }
            },
            "required": ["budget_tier", "financing_status"]
        }
    },
    {
        "name": "extract_timeline",
        "description": (
            "Extract how soon the lead wants to purchase. "
            "Call this when their buying urgency is clear from the conversation."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "timeline_tier": {
                    "type": "string",
                    "enum": ["immediate", "short_term", "medium_term", "long_term", "unknown"],
                    "description": "How soon they plan to buy"
                },
                "specific_timeframe": {
                    "type": "string",
                    "description": "Their exact stated timeframe if mentioned"
                },
                "urgency_drivers": {
                    "type": "string",
                    "description": "Reasons driving their timeline (relocation, lease ending, etc.)"
                }
            },
            "required": ["timeline_tier"]
        }
    }
]


def handle_tool_call(tool_name: str, tool_input: dict) -> str:
    """
    Process tool calls from Claude and return a result string.
    In a real system, you might save these to a database here.
    """
    if tool_name == "extract_property_type":
        return json.dumps({
            "status": "extracted",
            "tool": "property_type",
            "data": tool_input
        })
    elif tool_name == "extract_budget":
        return json.dumps({
            "status": "extracted",
            "tool": "budget",
            "data": tool_input
        })
    elif tool_name == "extract_timeline":
        return json.dumps({
            "status": "extracted",
            "tool": "timeline",
            "data": tool_input
        })
    else:
        return json.dumps({"status": "error", "message": f"Unknown tool: {tool_name}"})

import os
import json
from dotenv import load_dotenv
import anthropic
from tools import LEAD_EXTRACTION_TOOLS, handle_tool_call
from scoring_config import SCORING_CONFIG, QUALIFICATION_THRESHOLDS

load_dotenv()


class LeadQualifierAgent:
    """
    AI agent that qualifies real estate leads using Claude API tool use.
    Extracts budget, property type, and timeline, then scores the lead.
    """

    SYSTEM_PROMPT = """You are an expert real estate lead qualification specialist for a 
Southwest Florida real estate agency. Your job is to analyze lead information and extract 
structured qualification data using the tools available to you.

When given a lead's information (from a form, chat, or phone notes):
1. Carefully read all available context about the lead
2. Use the extract_property_type tool to capture what kind of property they want
3. Use the extract_budget tool to capture their price range and financing status
4. Use the extract_timeline tool to capture how urgently they want to buy
5. After calling all three tools, provide a brief qualification summary

Be thorough but efficient. If information is missing, use "unknown" for that field — 
do not make up details that aren't in the lead data. Your accuracy directly affects 
which leads agents follow up with first."""

    def __init__(self):
        self.client = anthropic.Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))
        self.model = "claude-sonnet-4-6"
        self.extracted_data = {}  # Stores all tool outputs during the run

    def _calculate_score(self) -> dict:
        """Calculate a numeric score from extracted lead data."""
        score = 0
        breakdown = {}

        # Score budget
        budget_data = self.extracted_data.get("budget", {})
        budget_tier = budget_data.get("budget_tier", "unknown")
        budget_points = SCORING_CONFIG["budget"]["tiers"].get(budget_tier, 0)
        score += budget_points
        breakdown["budget"] = {"tier": budget_tier, "points": budget_points}

        # Score timeline
        timeline_data = self.extracted_data.get("timeline", {})
        timeline_tier = timeline_data.get("timeline_tier", "unknown")
        timeline_points = SCORING_CONFIG["timeline"]["tiers"].get(timeline_tier, 0)
        score += timeline_points
        breakdown["timeline"] = {"tier": timeline_tier, "points": timeline_points}

        # Score property type
        property_data = self.extracted_data.get("property_type", {})
        specificity = property_data.get("specificity", "unknown")
        property_points = SCORING_CONFIG["property_type"]["tiers"].get(specificity, 0)
        score += property_points
        breakdown["property_type"] = {"specificity": specificity, "points": property_points}

        # Score financing
        financing_status = budget_data.get("financing_status", "unknown")
        financing_points = SCORING_CONFIG["financing"]["tiers"].get(financing_status, 0)
        score += financing_points
        breakdown["financing"] = {"status": financing_status, "points": financing_points}

        # Determine qualification tier
        if score >= QUALIFICATION_THRESHOLDS["hot"]:
            qualification = "HOT"
        elif score >= QUALIFICATION_THRESHOLDS["warm"]:
            qualification = "WARM"
        else:
            qualification = "NURTURE"

        return {
            "total_score": score,
            "max_possible": 100,
            "qualification": qualification,
            "breakdown": breakdown
        }

    def run(self, lead_input: str) -> dict:
        """
        Main agent loop. Takes raw lead text and returns structured
        qualification results with a numeric score.
        """
        messages = [
            {"role": "user", "content": lead_input}
        ]

        # Agentic loop: keep going until Claude stops calling tools
        while True:
            response = self.client.messages.create(
                model=self.model,
                max_tokens=4096,
                system=self.SYSTEM_PROMPT,
                tools=LEAD_EXTRACTION_TOOLS,
                messages=messages
            )

            # Add Claude's response to conversation history
            messages.append({"role": "assistant", "content": response.content})

            # Check if Claude is done (no more tool calls)
            if response.stop_reason == "end_turn":
                # Extract the final text summary from Claude's last message
                final_text = ""
                for block in response.content:
                    if hasattr(block, "text"):
                        final_text = block.text
                        break

                score_result = self._calculate_score()

                return {
                    "qualification_summary": final_text,
                    "extracted_data": self.extracted_data,
                    "score": score_result
                }

            # Handle tool use if Claude is calling tools
            if response.stop_reason == "tool_use":
                tool_results = []

                for block in response.content:
                    if block.type == "tool_use":
                        tool_name = block.name
                        tool_input = block.input

                        # Run the tool and get a result
                        result = handle_tool_call(tool_name, tool_input)
                        result_data = json.loads(result)

                        # Store extracted data by tool category
                        if "data" in result_data:
                            category = result_data["tool"]
                            self.extracted_data[category] = result_data["data"]

                        tool_results.append({
                            "type": "tool_result",
                            "tool_use_id": block.id,
                            "content": result
                        })

                # Feed tool results back to Claude so it can continue
                messages.append({"role": "user", "content": tool_results})

import os
import json
from dotenv import load_dotenv
import anthropic
from tools import LEAD_EXTRACTION_TOOLS, handle_tool_call
from scoring_config import SCORING_CONFIG, QUALIFICATION_THRESHOLDS

load_dotenv()


class InteractiveLeadQualifier:
    """
    Multi-turn version of the lead qualifier. Conducts a conversation
    with the lead (or agent entering notes) to gather missing details.
    """

    SYSTEM_PROMPT = """You are a friendly real estate intake specialist for a Naples, 
Florida real estate agency. You're having a conversation to understand what a potential 
buyer is looking for.

Ask one focused question at a time. Keep your tone warm but efficient — buyers appreciate 
feeling heard without being interrogated. Collect information about:
- What type of property they want
- Their budget range  
- When they want to buy
- Their financing situation

Once you have enough information to fill in all fields (even if some are "unknown"), 
use the available tools to extract and record the structured data, then give them 
a warm closing message."""

    def __init__(self):
        self.client = anthropic.Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))
        self.model = "claude-sonnet-4-6"
        self.messages = []
        self.extracted_data = {}
        self.tools_called = set()  # Track which tools have been used

    def chat(self, user_message: str) -> str:
        """
        Send a message and get Claude's response.
        Automatically handles tool calls in the background.
        Returns only the text response for display.
        """
        self.messages.append({"role": "user", "content": user_message})

        while True:
            response = self.client.messages.create(
                model=self.model,
                max_tokens=1024,
                system=self.SYSTEM_PROMPT,
                tools=LEAD_EXTRACTION_TOOLS,
                messages=self.messages
            )

            self.messages.append({"role": "assistant", "content": response.content})

            if response.stop_reason == "end_turn":
                # Return the visible text to the user
                for block in response.content:
                    if hasattr(block, "text"):
                        return block.text
                return ""

            if response.stop_reason == "tool_use":
                tool_results = []

                for block in response.content:
                    if block.type == "tool_use":
                        self.tools_called.add(block.name)
                        result = handle_tool_call(block.name, block.input)
                        result_data = json.loads(result)

                        if "data" in result_data:
                            self.extracted_data[result_data["tool"]] = result_data["data"]

                        tool_results.append({
                            "type": "tool_result",
                            "tool_use_id": block.id,
                            "content": result
                        })

                self.messages.append({"role": "user", "content": tool_results})

    def is_qualification_complete(self) -> bool:
        """Check if all three tools have been called."""
        required_tools = {
            "extract_property_type",
            "extract_budget",
            "extract_timeline"
        }
        return required_tools.issubset(self.tools_called)


def run_interactive_session():
    """Demo of interactive lead qualification via terminal."""
    qualifier = InteractiveLeadQualifier()

    print("=== Interactive Lead Qualifier ===")
    print("Type 'quit' to exit, 'score' to see current score\n")

    # Start the conversation
    opening = qualifier.chat("Hello, I'm interested in buying a property.")
    print(f"Agent: {opening}\n")

    while not qualifier.is_qualification_complete():
        user_input = input("Lead: ").strip()

        if user_input.lower() == "quit":
            break

        response = qualifier.chat(user_input)
        print(f"Agent: {response}\n")

    if qualifier.is_qualification_complete():
        print("\n--- Qualification Complete ---")
        print(f"Extracted Data: {json.dumps(qualifier.extracted_data, indent=2)}")


if __name__ == "__main__":
    run_interactive_session()