← Back to Blog

If you've ever had your sales team manually sifting through 200 inbound leads on a Monday morning, you already know the problem. Half those leads are noise, and the ones that matter get buried. This tutorial shows you exactly how to build a lead scoring agent using the Claude API that evaluates, scores, and recommends action on leads automatically — without anyone touching a spreadsheet.

What You'll Build

You're going to build a Python-based lead scoring agent that uses Claude's tool use feature to research companies, apply qualification rules, and return a structured score with a sales recommendation. The agent handles the full agentic loop — calling tools, processing results, and reasoning through a final decision. By the end, you'll have something capable of qualifying 100+ leads a day without any manual input.

📦 Full Source Code
All the working code for this project is written out in the steps below. Every snippet builds on the last, so by Step 4 you'll have a complete, runnable agent. Copy them in order and you're good to go.

Prerequisites

  • Python 3.9 or higher installed
  • An Anthropic API key (get one at console.anthropic.com)
  • anthropic SDK installed: pip install anthropic
  • Basic comfort with Python classes and dictionaries
  • Optional: python-dotenv for managing your API key safely

Step 1: Set Up Your Claude API Client and Tools

First, let's get the client wired up and make sure everything connects. This is the foundation the rest of the agent sits on. Nothing fancy here — just clean initialization with your API key and the right model.

setup.py 
import os
import anthropic
from dotenv import load_dotenv

load_dotenv()  # Loads ANTHROPIC_API_KEY from your .env file

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

MODEL = "claude-sonnet-4-6"

def test_connection():
    """Quick sanity check to confirm the client and model are working."""
    response = client.messages.create(
        model=MODEL,
        max_tokens=64,
        messages=[{"role": "user", "content": "Say 'API connected' and nothing else."}]
    )
    print(response.content[0].text)

if __name__ == "__main__":
    test_connection()

Run that file. You should see API connected printed to the terminal. If you get an auth error, double-check your key is set correctly in your .env file.

Step 2: Define Lead Scoring Tool Definitions

Claude's tool use feature lets the model decide when to "call" an external function and what arguments to pass it. You describe the tools in a structured format, and Claude figures out when to use them. Here we're defining two tools: one that simulates company research and one that applies your internal qualification rules.

In a real deployment, the research tool would hit an API like Clearbit, Apollo, or LinkedIn. For this tutorial, we return realistic mock data so you can see exactly what the agent receives and reasons over.

tools.py 
import json

# Tool definitions passed to Claude — these tell the model what tools exist
# and what arguments each one expects
LEAD_SCORING_TOOLS = [
    {
        "name": "research_company",
        "description": (
            "Looks up firmographic data for a company given its name and website. "
            "Returns employee count, industry, annual revenue estimate, tech stack, "
            "and recent funding information."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "company_name": {
                    "type": "string",
                    "description": "The full legal or trade name of the company."
                },
                "website": {
                    "type": "string",
                    "description": "The company's primary website URL."
                }
            },
            "required": ["company_name", "website"]
        }
    },
    {
        "name": "apply_qualification_rules",
        "description": (
            "Scores a lead from 0 to 100 based on your internal ICP (Ideal Customer Profile) rules. "
            "Takes firmographic data and contact info, returns a numeric score, tier label, "
            "and the top reasons for that score."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "company_name": {"type": "string"},
                "employee_count": {"type": "integer"},
                "industry": {"type": "string"},
                "annual_revenue_usd": {"type": "number"},
                "has_recent_funding": {"type": "boolean"},
                "tech_stack": {
                    "type": "array",
                    "items": {"type": "string"}
                },
                "contact_title": {"type": "string"},
                "contact_email": {"type": "string"}
            },
            "required": [
                "company_name", "employee_count", "industry",
                "annual_revenue_usd", "contact_title"
            ]
        }
    }
]


def research_company(company_name: str, website: str) -> dict:
    """
    Simulates a company data enrichment API call.
    Replace this with a real API (Clearbit, Apollo, etc.) in production.
    """
    mock_data = {
        "gulf_coast_realty_group": {
            "company_name": company_name,
            "website": website,
            "employee_count": 42,
            "industry": "Real Estate",
            "annual_revenue_usd": 8_500_000,
            "has_recent_funding": False,
            "tech_stack": ["Salesforce", "Google Workspace", "Zapier"],
            "founded_year": 2014,
            "headquarters": "Naples, FL"
        }
    }
    # Normalize the key to find mock data regardless of exact input
    key = company_name.lower().replace(" ", "_")
    return mock_data.get(key, {
        "company_name": company_name,
        "website": website,
        "employee_count": 25,
        "industry": "Unknown",
        "annual_revenue_usd": 3_000_000,
        "has_recent_funding": False,
        "tech_stack": ["QuickBooks"],
        "founded_year": 2018,
        "headquarters": "Unknown"
    })


def apply_qualification_rules(
    company_name: str,
    employee_count: int,
    industry: str,
    annual_revenue_usd: float,
    has_recent_funding: bool,
    tech_stack: list,
    contact_title: str,
    contact_email: str = ""
) -> dict:
    """
    Applies ICP scoring rules and returns a structured score dict.
    Adjust the weights below to match your actual sales criteria.
    """
    score = 0
    reasons = []

    # Employee count scoring
    if 10 <= employee_count <= 200:
        score += 25
        reasons.append(f"Company size ({employee_count} employees) fits SMB sweet spot.")
    elif employee_count > 200:
        score += 15
        reasons.append(f"Company size ({employee_count} employees) is larger than ideal but viable.")

    # Revenue scoring
    if annual_revenue_usd >= 5_000_000:
        score += 25
        reasons.append(f"Revenue (~${annual_revenue_usd:,.0f}) indicates budget for AI investment.")
    elif annual_revenue_usd >= 1_000_000:
        score += 12
        reasons.append(f"Revenue (~${annual_revenue_usd:,.0f}) is moderate.")

    # Industry fit — industries Naples AI actively targets
    high_fit_industries = ["Real Estate", "Healthcare", "Automotive", "Manufacturing", "Restaurant"]
    if industry in high_fit_industries:
        score += 25
        reasons.append(f"Industry ({industry}) is a high-fit vertical for AI automation.")
    else:
        score += 5
        reasons.append(f"Industry ({industry}) is outside core verticals but not disqualifying.")

    # Recent funding is a strong buying signal
    if has_recent_funding:
        score += 15
        reasons.append("Recent funding round suggests active growth investment budget.")

    # Decision-maker contact title scoring
    executive_titles = ["CEO", "COO", "CTO", "Owner", "President", "VP", "Director", "Founder"]
    if any(title.lower() in contact_title.lower() for title in executive_titles):
        score += 10
        reasons.append(f"Contact ({contact_title}) is a decision-maker.")
    else:
        score += 2
        reasons.append(f"Contact ({contact_title}) may not be the final decision-maker.")

    # Assign tier label based on final score
    if score >= 75:
        tier = "HOT"
    elif score >= 50:
        tier = "WARM"
    elif score >= 25:
        tier = "COLD"
    else:
        tier = "UNQUALIFIED"

    return {
        "company_name": company_name,
        "score": min(score, 100),  # Cap at 100
        "tier": tier,
        "reasons": reasons,
        "recommended_action": (
            "Book a discovery call within 24 hours." if tier == "HOT"
            else "Add to nurture sequence." if tier == "WARM"
            else "Low priority — monitor for 90 days." if tier == "COLD"
            else "Do not pursue at this time."
        )
    }


def execute_tool(tool_name: str, tool_input: dict) -> str:
    """Routes a tool call from Claude to the correct Python function."""
    if tool_name == "research_company":
        result = research_company(**tool_input)
    elif tool_name == "apply_qualification_rules":
        result = apply_qualification_rules(**tool_input)
    else:
        result = {"error": f"Unknown tool: {tool_name}"}

    return json.dumps(result)  # Claude expects tool results as strings

Step 3: Create the Lead Scoring Agent Class

Now we pull everything together into a class. The agent holds the client, the tools, and the conversation state. Each method has a clear job — one to build the initial prompt, one to handle tool results, and one to run everything from end to end.

agent.py 
import json
import anthropic
from tools import LEAD_SCORING_TOOLS, execute_tool

MODEL = "claude-sonnet-4-6"

SYSTEM_PROMPT = """You are an expert B2B sales qualification agent working for Naples AI, 
a Southwest Florida AI agency that builds custom AI solutions for local businesses.

Your job is to evaluate inbound leads and determine which ones are worth pursuing.
Always use the research_company tool first to gather firmographic data, then use 
apply_qualification_rules to score the lead. After you have tool results, give a 
clear, concise final assessment that a sales rep can act on immediately.

Be direct. No fluff. A sales rep needs the score, the tier, and one paragraph on why."""


class LeadScoringAgent:
    """
    An agentic lead scorer that uses Claude's tool use to research
    and qualify sales leads automatically.
    """

    def __init__(self):
        self.client = anthropic.Anthropic()  # Reads ANTHROPIC_API_KEY from env
        self.model = MODEL
        self.tools = LEAD_SCORING_TOOLS
        self.system = SYSTEM_PROMPT

    def build_initial_message(self, lead: dict) -> str:
        """Formats the lead dictionary into a natural-language prompt for Claude."""
        return (
            f"Please evaluate this inbound lead:\n\n"
            f"Company: {lead.get('company_name', 'Unknown')}\n"
            f"Website: {lead.get('website', 'N/A')}\n"
            f"Contact Name: {lead.get('contact_name', 'N/A')}\n"
            f"Contact Title: {lead.get('contact_title', 'N/A')}\n"
            f"Contact Email: {lead.get('contact_email', 'N/A')}\n"
            f"Notes: {lead.get('notes', 'None provided')}\n\n"
            f"Research the company and then apply our qualification rules to score this lead."
        )

    def process_tool_calls(self, response, messages: list) -> list:
        """
        Handles all tool calls in a response and appends results to the message list.
        Returns the updated messages list ready for the next API call.
        """
        # Append Claude's full response (including tool_use blocks) to message history
        messages.append({"role": "assistant", "content": response.content})

        tool_results = []

        for block in response.content:
            if block.type == "tool_use":
                print(f"  → Tool called: {block.name}")
                print(f"    Input: {json.dumps(block.input, indent=2)}")

                # Execute the tool and get the result string
                result_str = execute_tool(block.name, block.input)
                result_data = json.loads(result_str)
                print(f"    Result: {json.dumps(result_data, indent=2)}\n")

                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,  # Must match the block's id exactly
                    "content": result_str
                })

        # All tool results go in a single user message
        messages.append({"role": "user", "content": tool_results})
        return messages

    def score_lead(self, lead: dict) -> dict:
        """
        Main entry point. Takes a lead dict, runs the full agentic loop,
        and returns a structured scoring result.
        """
        print(f"\n{'='*60}")
        print(f"Scoring lead: {lead.get('company_name', 'Unknown')}")
        print(f"{'='*60}\n")

        user_message = self.build_initial_message(lead)
        messages = [{"role": "user", "content": user_message}]

        # Run the agentic loop
        final_text, messages = self._run_agentic_loop(messages)

        # Parse structured data from the qualification tool result if present
        scoring_data = self._extract_scoring_data(messages)

        return {
            "lead": lead,
            "score": scoring_data.get("score"),
            "tier": scoring_data.get("tier"),
            "reasons": scoring_data.get("reasons", []),
            "recommended_action": scoring_data.get("recommended_action"),
            "agent_summary": final_text
        }

    def _extract_scoring_data(self, messages: list) -> dict:
        """
        Pulls the structured scoring result from the tool_result messages.
        Finds the apply_qualification_rules result specifically.
        """
        for message in messages:
            if message["role"] == "user" and isinstance(message["content"], list):
                for item in message["content"]:
                    if item.get("type") == "tool_result":
                        try:
                            data = json.loads(item["content"])
                            # Only return data that has a score field (qualification result)
                            if "score" in data:
                                return data
                        except (json.JSONDecodeError, KeyError):
                            continue
        return {}

    def _run_agentic_loop(self, messages: list) -> tuple[str, list]:
        """
        The core agentic loop. Keeps calling Claude until it stops
        requesting tool calls and returns a final text response.
        """
        max_iterations = 10  # Safety limit to prevent runaway loops
        iteration = 0

        while iteration < max_iterations:
            iteration += 1

            response = self.client.messages.create(
                model=self.model,
                max_tokens=2048,
                system=self.system,
                tools=self.tools,
                messages=messages
            )

            # If Claude is done calling tools, extract and return the final text
            if response.stop_reason == "end_turn":
                final_text = ""
                for block in response.content:
                    if hasattr(block, "text"):
                        final_text += block.text
                return final_text, messages

            # If Claude wants to use tools, process them and loop again
            if response.stop_reason == "tool_use":
                messages = self.process_tool_calls(response, messages)
                continue

            # Unexpected stop reason — break to avoid infinite loop
            print(f"Unexpected stop_reason: {response.stop_reason}")
            break

        return "Agent loop exceeded maximum iterations.", messages

Step 4: Implement the Evaluation Loop with Tool Use and Parallel Processing

Here's the runner script. It takes a list of leads, scores them all, and prints a clean summary table at the end. I'm using concurrent.futures to process multiple leads in parallel, which is what makes the "100+ leads a day" claim actually realistic.

main.py 
import json
from concurrent.futures import ThreadPoolExecutor, as_completed
from agent import LeadScoringAgent

# Sample leads batch — in production, pull these from your CRM or web form database
SAMPLE_LEADS = [
    {
        "company_name": "Gulf Coast Realty Group",
        "website": "https://gulfcoastrealty.com",
        "contact_name": "Sandra Morales",
        "contact_title": "CEO",
        "contact_email": "[email protected]",
        "notes": "Reached out via contact form. Mentioned they're manually entering property listings."
    },
    {
        "company_name": "Sunshine Auto Plaza",
        "website": "https://sunshineautoplaza.com",
        "contact_name": "Mike Donato",
        "contact_title": "Marketing Coordinator",
        "contact_email": "[email protected]",
        "notes": "Downloaded our AI chatbot whitepaper."
    }
]


def score_single_lead(lead: dict) -> dict:
    """Wrapper so each thread creates its own agent instance (thread safety)."""
    agent = LeadScoringAgent()
    return agent.score_lead(lead)


def print_summary_table(results: list):
    """Prints a clean results table to the console."""
    print("\n" + "=" * 70)
    print(f"{'LEAD SCORING SUMMARY':^70}")
    print("=" * 70)
    print(f"{'Company':<30} {'Score':^7} {'Tier':^12} {'Action'}")
    print("-" * 70)
    for r in sorted(results, key=lambda x: x.get("score", 0), reverse=True):
        company = r["lead"]["company_name"][:29]
        score = r.get("score", "N/A")
        tier = r.get("tier", "N/A")
        action = r.get("recommended_action", "N/A")[:25]
        print(f"{company:<30} {str(score):^7} {tier:^12} {action}")
    print("=" * 70)


def main():
    results = []
    print(f"Processing {len(SAMPLE_LEADS)} leads with parallel scoring...\n")

    # Use a thread pool to score leads concurrently
    # max_workers=5 is a safe default given Anthropic's rate limits
    with ThreadPoolExecutor(max_workers=5) as executor:
        future_to_lead = {
            executor.submit(score_single_lead, lead): lead
            for lead in SAMPLE_LEADS
        }

        for future in as_completed(future_to_lead):
            lead = future_to_lead[future]
            try:
                result = future.result()
                results.append(result)

                # Print the agent's narrative summary for each lead
                print(f"\n📋 AGENT SUMMARY — {lead['company_name']}")
                print("-" * 50)
                print(result.get("agent_summary", "No summary returned."))

            except Exception as exc:
                print(f"Error scoring {lead['company_name']}: {exc}")
                results.append({
                    "lead": lead,
                    "score": 0,
                    "tier": "ERROR",
                    "recommended_action": "Manual review required.",
                    "agent_summary": str(exc)
                })

    print_summary_table(results)

    # Optionally save to JSON for downstream CRM import
    with open("scored_leads.json", "w") as f:
        json.dump(results, f, indent=2)
    print("\n✅ Results saved to scored_leads.json")


if __name__ == "__main__":
    main()

Here's what the terminal output looks like when you run this against the two sample leads:

sample_output.txt 
Processing 2 leads with parallel scoring...

============================================================
Scoring lead: Gulf Coast Realty Group
============================================================

  → Tool called: research_company
    Input: {
      "company_name": "Gulf Coast Realty Group",
      "website": "https://gulfcoastrealty.com"
    }
    Result: {
      "company_name": "Gulf Coast Realty Group",
      "employee_count": 42,
      "industry": "Real Estate",
      "annual_revenue_usd": 8500000,
      "has_recent_funding": false,
      "tech_stack": ["Salesforce", "Google Workspace", "Zapier"],
      "founded_year": 2014,
      "headquarters": "Naples, FL"
    }

  → Tool called: apply_qualification_rules
    Input: {
      "company_name": "Gulf Coast Realty Group",
      "employee_count": 42,
      "industry": "Real Estate",
      "annual_revenue_usd": 8500000,
      "has_recent_funding": false,
      "tech_stack": ["Salesforce", "Google Workspace", "