How to Build a Multi-Agent Lead Scoring System with Claude API

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.getenv("ANTHROPIC_API_KEY"))

# Quick sanity check — send a minimal message to confirm auth works
response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=64,
    messages=[{"role": "user", "content": "Say: connection successful"}]
)

print(response.content[0].text)
# Expected output: connection successful

from dataclasses import dataclass, field
from typing import Optional

@dataclass
class Lead:
    id: str
    first_name: str
    last_name: str
    email: str
    company: str
    job_title: str
    industry: str
    company_size: str          # e.g., "1-10", "11-50", "51-200", "200+"
    annual_revenue: str        # e.g., "$500K", "$2M", "$10M+"
    source: str                # e.g., "website", "referral", "cold outreach"
    message: str               # Their inquiry or form submission text
    phone: Optional[str] = None
    website: Optional[str] = None

@dataclass
class LeadScore:
    lead_id: str
    total_score: int           # 0–100
    fit_score: int             # How well they match your ICP (ideal customer profile)
    intent_score: int          # Buying intent signals from their message
    urgency_score: int         # How soon they need a solution
    tier: str                  # "A", "B", "C", or "D"
    reasoning: str             # Plain-English explanation of the score
    recommended_action: str    # What the sales team should do next
    enriched_data: dict = field(default_factory=dict)  # Data added by the analyzer agent

# Scoring weights — adjust these to match your actual sales priorities
SCORING_WEIGHTS = {
    "fit": 0.40,
    "intent": 0.35,
    "urgency": 0.25
}

# Tier thresholds
TIER_THRESHOLDS = {
    "A": 80,   # Call within 1 hour
    "B": 60,   # Call within 24 hours
    "C": 40,   # Send nurture email
    "D": 0     # Add to long-term drip
}

def assign_tier(total_score: int) -> str:
    for tier, threshold in TIER_THRESHOLDS.items():
        if total_score >= threshold:
            return tier
    return "D"

import os
import json
from dotenv import load_dotenv
import anthropic
from lead_models import Lead, SCORING_WEIGHTS

load_dotenv()

# Tool definitions tell Claude exactly what functions it can call
ANALYZER_TOOLS = [
    {
        "name": "enrich_lead_data",
        "description": "Enriches a lead record with derived attributes like company tier, estimated deal size, and industry fit score based on the provided lead information.",
        "input_schema": {
            "type": "object",
            "properties": {
                "company_size": {
                    "type": "string",
                    "description": "The company size bracket of the lead"
                },
                "annual_revenue": {
                    "type": "string",
                    "description": "The estimated annual revenue of the lead's company"
                },
                "industry": {
                    "type": "string",
                    "description": "The industry the lead works in"
                },
                "job_title": {
                    "type": "string",
                    "description": "The lead's job title"
                }
            },
            "required": ["company_size", "annual_revenue", "industry", "job_title"]
        }
    },
    {
        "name": "calculate_fit_score",
        "description": "Calculates an ICP fit score (0-100) for a lead based on their company profile against the ideal customer profile.",
        "input_schema": {
            "type": "object",
            "properties": {
                "industry": {"type": "string"},
                "company_size": {"type": "string"},
                "annual_revenue": {"type": "string"},
                "job_title": {"type": "string"},
                "source": {"type": "string"}
            },
            "required": ["industry", "company_size", "annual_revenue", "job_title", "source"]
        }
    }
]

def enrich_lead_data(company_size: str, annual_revenue: str, industry: str, job_title: str) -> dict:
    """Simulates data enrichment — in production, call an API like Clearbit or Apollo."""
    seniority_map = {
        "ceo": "executive", "cto": "executive", "coo": "executive",
        "vp": "senior", "director": "senior", "head": "senior",
        "manager": "mid", "lead": "mid", "senior": "mid",
        "analyst": "junior", "coordinator": "junior"
    }

    title_lower = job_title.lower()
    seniority = "unknown"
    for keyword, level in seniority_map.items():
        if keyword in title_lower:
            seniority = level
            break

    # Estimate deal size based on company revenue
    revenue_to_deal = {
        "$500K": "$2,500–$5,000",
        "$1M": "$5,000–$10,000",
        "$2M": "$8,000–$15,000",
        "$5M": "$15,000–$30,000",
        "$10M+": "$30,000–$75,000"
    }
    estimated_deal = revenue_to_deal.get(annual_revenue, "$5,000–$15,000")

    return {
        "decision_maker_seniority": seniority,
        "estimated_deal_size": estimated_deal,
        "enriched": True
    }

def calculate_fit_score(industry: str, company_size: str, annual_revenue: str,
                         job_title: str, source: str) -> dict:
    """Scores ICP fit based on target industries, company size, and decision-maker level."""
    score = 50  # Base score

    # Industries Naples AI serves get a boost
    target_industries = ["real estate", "healthcare", "restaurant", "automotive",
                          "manufacturing", "hospitality", "retail"]
    if any(ind in industry.lower() for ind in target_industries):
        score += 20

    # Larger companies with more budget score higher
    size_scores = {"1-10": -10, "11-50": 0, "51-200": 10, "200+": 20}
    score += size_scores.get(company_size, 0)

    # Executive and senior decision-makers close faster
    title_lower = job_title.lower()
    if any(t in title_lower for t in ["ceo", "cto", "coo", "president", "owner"]):
        score += 15
    elif any(t in title_lower for t in ["vp", "director", "head"]):
        score += 8

    # Referrals close at a much higher rate
    if source == "referral":
        score += 15
    elif source == "website":
        score += 5

    # Clamp to 0–100
    score = max(0, min(100, score))
    return {"fit_score": score}

class LeadAnalyzerAgent:
    def __init__(self):
        self.client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
        self.model = "claude-sonnet-4-6"

    def _process_tool_call(self, tool_name: str, tool_input: dict) -> str:
        """Routes tool calls from Claude to the correct local function."""
        if tool_name == "enrich_lead_data":
            result = enrich_lead_data(**tool_input)
        elif tool_name == "calculate_fit_score":
            result = calculate_fit_score(**tool_input)
        else:
            result = {"error": f"Unknown tool: {tool_name}"}
        return json.dumps(result)

    def analyze(self, lead: Lead) -> dict:
        """Runs the analyzer agent with tool use to enrich and score a lead."""
        system_prompt = """You are a lead analysis agent for Naples AI, an AI solutions agency in Southwest Florida.
Your job is to analyze inbound leads and use the available tools to enrich their data and calculate fit scores.
Always call enrich_lead_data first, then calculate_fit_score. Return a JSON summary at the end."""

        user_message = f"""Analyze this lead and use both tools to enrich and score them:

Name: {lead.first_name} {lead.last_name}
Email: {lead.email}
Company: {lead.company}
Job Title: {lead.job_title}
Industry: {lead.industry}
Company Size: {lead.company_size}
Annual Revenue: {lead.annual_revenue}
Lead Source: {lead.source}
Message: {lead.message}

After using both tools, provide a JSON summary with keys: enriched_data, fit_score, analysis_notes."""

        messages = [{"role": "user", "content": user_message}]
        enriched_data = {}
        fit_score = 50

        # Agentic loop — Claude may call tools multiple times before finishing
        while True:
            response = self.client.messages.create(
                model=self.model,
                max_tokens=1024,
                system=system_prompt,
                tools=ANALYZER_TOOLS,
                messages=messages
            )

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

            if response.stop_reason == "tool_use":
                tool_results = []
                for block in response.content:
                    if block.type == "tool_use":
                        result = self._process_tool_call(block.name, block.input)
                        result_dict = json.loads(result)

                        # Capture tool outputs for the final score object
                        if block.name == "enrich_lead_data":
                            enriched_data.update(result_dict)
                        elif block.name == "calculate_fit_score":
                            fit_score = result_dict.get("fit_score", 50)

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

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

            elif response.stop_reason == "end_turn":
                # Extract the final text summary from Claude
                final_text = ""
                for block in response.content:
                    if hasattr(block, "text"):
                        final_text = block.text
                        break
                return {
                    "enriched_data": enriched_data,
                    "fit_score": fit_score,
                    "analysis_notes": final_text
                }
            else:
                break

        return {"enriched_data": enriched_data, "fit_score": fit_score, "analysis_notes": ""}

import os
import json
import re
from dotenv import load_dotenv
import anthropic
from lead_models import Lead, LeadScore, assign_tier, SCORING_WEIGHTS

load_dotenv()

class LeadQualifierAgent:
    def __init__(self):
        self.client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
        self.model = "claude-sonnet-4-6"

    def qualify(self, lead: Lead, analyzer_output: dict) -> LeadScore:
        """Uses multi-turn conversation to deeply qualify a lead and produce a final score."""
        system_prompt = """You are a senior sales qualification agent for Naples AI, an AI solutions agency in Southwest Florida.
Naples AI builds custom AI systems for real estate, healthcare, restaurants, car dealerships, and manufacturing businesses.
Your job is to assess buying intent and urgency from a lead's message and profile.
Think through this step by step. Be specific. Avoid vague language."""

        # Turn 1: Analyze intent signals from the message
        messages = [
            {
                "role": "user",
                "content": f"""Review this lead's message and identify specific buying intent signals.

Lead: {lead.first_name} {lead.last_name}, {lead.job_title} at {lead.company}
Industry: {lead.industry}
Message: "{lead.message}"
Fit Score from Analyzer: {analyzer_output.get('fit_score', 50)}/100
Enriched Data: {json.dumps(analyzer_output.get('enriched_data', {}))}

List the specific intent signals you see, then give an intent score from 0-100."""
            }
        ]

        turn1_response = self.client.messages.create(
            model=self.model,
            max_tokens=512,
            system=system_prompt,
            messages=messages
        )
        intent_analysis = turn1_response.content[0].text
        messages.append({"role": "assistant", "content": intent_analysis})

        # Turn 2: Assess urgency based on language and context
        messages.append({
            "role": "user",
            "content": """Now assess the urgency of this lead. Look for time-based language, pain points,
or competitive pressure signals in their message. Give an urgency score from 0-100 and explain why."""
        })

        turn2_response = self.client.messages.create(
            model=self.model,
            max_tokens=512,
            system=system_prompt,
            messages=messages
        )
        urgency_analysis = turn2_response.content[0].text
        messages.append({"role": "assistant", "content": urgency_analysis})

        # Turn 3: Produce the final structured score
        messages.append({
            "role": "user",
            "content": f"""Based on your intent and urgency analysis, produce a final JSON score object.
The fit score is already calculated as {analyzer_output.get('fit_score', 50)}.

Return ONLY valid JSON with these exact keys:
{{
  "intent_score": ,
  "urgency_score": ,
  "reasoning": "<2-3 sentence plain English explanation>",
  "recommended_action": ""
}}"""
        })

        turn3_response = self.client.messages.create(
            model=self.model,
            max_tokens=512,
            system=system_prompt,
            messages=messages
        )

        raw_json = turn3_response.content[0].text.strip()

        # Strip markdown fences if Claude wraps the JSON
        raw_json = re.sub(r"^```(?:json)?\s*", "", raw_json)
        raw_json = re.sub(r"\s*```$", "", raw_json)

        scores = json.loads(raw_json)

        fit_score = analyzer_output.get("fit_score", 50)
        intent_score = scores.get("intent_score", 50)
        urgency_score = scores.get("urgency_score", 50)

        # Weighted total score calculation
        total_score = int(
            fit_score * SCORING_WEIGHTS["fit"] +
            intent_score * SCORING_WEIGHTS["intent"] +
            urgency_score * SCORING_WEIGHTS["urgency"]
        )

        return LeadScore(
            lead_id=lead.id,
            total_score=total_score,
            fit_score=fit_score,
            intent_score=intent_score,
            urgency_score=urgency_score,
            tier=assign_tier(total_score),
            reasoning=scores.get("reasoning", ""),
            recommended_action=scores.get("recommended_action", ""),
            enriched_data=analyzer_output.get("enriched_data", {})
        )

import os
from dotenv import load_dotenv
from lead_models import Lead, LeadScore
from lead_analyzer import LeadAnalyzerAgent
from lead_qualifier import LeadQualifierAgent

load_dotenv()

class LeadScoringOrchestrator:
    def __init__(self):
        self.analyzer = LeadAnalyzerAgent()
        self.qualifier = LeadQualifierAgent()

    def score_lead(self, lead: Lead) -> LeadScore:
        """Runs a lead through the full two-agent pipeline and returns a scored result."""
        print(f"[Orchestrator] Analyzing lead: {lead.first_name} {lead.last_name}...")

        # Agent 1: Enrich and calculate fit score
        analyzer_output = self.analyzer.analyze(lead)
        print(f"[Orchestrator] Analyzer complete. Fit score: {analyzer_output.get('fit_score')}")

        # Agent 2: Qualify intent and urgency, produce final score
        lead_score = self.qualifier.qualify(lead, analyzer_output)
        print(f"[Orchestrator] Qualifier complete. Total score: {lead_score.total_score} | Tier: {lead_score.tier}")

        return lead_score

    def score_batch(self, leads: list[Lead]) -> list[LeadScore]:
        """Processes a list of leads and returns them sorted by score descending."""
        results = []
        for lead in leads:
            try:
                score = self.score_lead(lead)
                results.append(score)
            except Exception as e:
                # Log the error but keep processing remaining leads
                print(f"[Orchestrator] Error scoring lead {lead.id}: {e}")

        # Sort by total score so highest-priority leads appear first
        results.sort(key=lambda s