How to Build a Multi-Agent Lead Scoring System with Claude API (Step-by-Step)

What You'll Build

You're going to build a multi-agent lead scoring system in Python that automatically qualifies inbound leads using two specialized Claude agents — one that validates the data and one that scores the lead against your ideal customer profile. The whole thing runs in under 200 lines of code and integrates cleanly with a CRM and email workflow via tool definitions. By the end, you'll have a working agentic pipeline that hands leads off between agents, catches bad data early, and outputs a structured JSON score you can actually use.

📦 Full Source Code
The complete working code is built piece by piece in the steps below. Each snippet is standalone and correct — copy them in order and you'll have a fully functional multi-agent lead scoring system by the end of this tutorial. No pseudocode, no hand-waving.

Prerequisites

Python 3.10 or higher installed
An Anthropic API key (get one at console.anthropic.com)
Basic familiarity with Python classes and dictionaries
anthropic Python SDK installed (pip install anthropic)
A terminal and a text editor — that's genuinely all you need

Step 1: Set Up Claude API and Install Dependencies

Start by installing the Anthropic SDK and setting your API key. I keep the key in an environment variable so it never ends up hardcoded in the project files.

terminal

pip install anthropic python-dotenv

Create a .env file in your project root with your key:

.env

ANTHROPIC_API_KEY=sk-ant-your-key-here

Now create the main project file. This first block sets up your imports and loads the environment:

lead_scoring.py

import os
import json
from anthropic import Anthropic
from dotenv import load_dotenv

load_dotenv()

client = Anthropic()
MODEL = "claude-sonnet-4-6"

That's the entire setup. The SDK automatically reads ANTHROPIC_API_KEY from your environment, so no extra configuration is needed.

Step 2: Define the Lead Scoring Agent and Data Validation Agent

Here's where it gets interesting. Most people build one fat prompt that tries to do everything at once — validate the data, score the lead, and format the output. That approach falls apart fast when inputs get messy or edge cases pile up. Instead, we're splitting the work across two agents with distinct system prompts and clear responsibilities.

The Validation Agent checks that incoming lead data is complete and properly formatted before scoring even starts. The Scoring Agent then receives the clean data and evaluates the lead against your ideal customer profile. This separation means errors get caught early and the scoring logic stays clean.

lead_scoring.py (continued)

VALIDATION_AGENT_PROMPT = """You are a data validation agent for a CRM lead pipeline.
Your job is to inspect incoming lead data and verify it is complete and usable.

Check the following fields:
- name: must be a non-empty string
- email: must contain @ and a valid domain
- company: must be a non-empty string
- industry: must be one of: real_estate, healthcare, restaurant, automotive, manufacturing, other
- annual_revenue: must be a positive number (can be null if unknown)
- employees: must be a positive integer (can be null if unknown)
- message: must be a non-empty string

If validation passes, call the validate_lead tool with status="valid" and an empty errors list.
If validation fails, call the validate_lead tool with status="invalid" and a list of specific error messages.
Never guess or fill in missing required fields — flag them as errors instead."""

SCORING_AGENT_PROMPT = """You are a lead scoring agent for a B2B AI solutions agency.
You receive validated lead data and score the lead from 0 to 100 based on fit.

Scoring criteria:
- Industry fit (real_estate, healthcare, automotive = high value): up to 30 points
- Company size (employees 10-500 = ideal range): up to 25 points
- Revenue signal (annual_revenue > 500000 = strong signal): up to 25 points
- Message intent (mentions specific pain points, automation, AI = strong): up to 20 points

After scoring, call the score_lead tool with:
- score: integer 0-100
- tier: "hot" (80-100), "warm" (50-79), "cold" (0-49)
- reasoning: a 2-3 sentence plain-English explanation of the score
- recommended_action: one of "immediate_outreach", "nurture_sequence", "disqualify"

Be honest about low scores. A score of 20 is useful information, not a failure."""

💡 Why Separate Agents?
Giving each agent one job makes the system easier to debug and extend. When a lead gets a bad score, you know exactly which agent made which decision. It also lets you swap out the scoring criteria without touching validation logic — and vice versa.

Step 3: Create Tool Definitions for CRM Integration and Email

Claude agents take action through tools. We're defining four tools here: one for validation output, one for scoring output, one to simulate writing to a CRM, and one to trigger an email sequence. In a real deployment, the CRM and email tools would call your actual APIs — HubSpot, Salesforce, ActiveCampaign, whatever you're using.

lead_scoring.py (continued)

VALIDATION_TOOLS = [
    {
        "name": "validate_lead",
        "description": "Record the validation result for an incoming lead.",
        "input_schema": {
            "type": "object",
            "properties": {
                "status": {
                    "type": "string",
                    "enum": ["valid", "invalid"],
                    "description": "Whether the lead data passed validation."
                },
                "errors": {
                    "type": "array",
                    "items": {"type": "string"},
                    "description": "List of validation error messages. Empty if valid."
                }
            },
            "required": ["status", "errors"]
        }
    }
]

SCORING_TOOLS = [
    {
        "name": "score_lead",
        "description": "Record the lead score and recommended action.",
        "input_schema": {
            "type": "object",
            "properties": {
                "score": {
                    "type": "integer",
                    "description": "Lead score from 0 to 100."
                },
                "tier": {
                    "type": "string",
                    "enum": ["hot", "warm", "cold"],
                    "description": "Lead tier based on score."
                },
                "reasoning": {
                    "type": "string",
                    "description": "Plain-English explanation of the score."
                },
                "recommended_action": {
                    "type": "string",
                    "enum": ["immediate_outreach", "nurture_sequence", "disqualify"],
                    "description": "Recommended next action for this lead."
                }
            },
            "required": ["score", "tier", "reasoning", "recommended_action"]
        }
    },
    {
        "name": "update_crm",
        "description": "Write the lead score and tier to the CRM system.",
        "input_schema": {
            "type": "object",
            "properties": {
                "lead_email": {
                    "type": "string",
                    "description": "Lead email address used as the CRM record key."
                },
                "score": {"type": "integer"},
                "tier": {"type": "string"},
                "recommended_action": {"type": "string"}
            },
            "required": ["lead_email", "score", "tier", "recommended_action"]
        }
    },
    {
        "name": "trigger_email_sequence",
        "description": "Enroll the lead in the appropriate email nurture sequence.",
        "input_schema": {
            "type": "object",
            "properties": {
                "lead_email": {"type": "string"},
                "sequence_name": {
                    "type": "string",
                    "enum": ["hot_lead_immediate", "warm_nurture_30day", "cold_archive"],
                    "description": "The email sequence to enroll the lead in."
                }
            },
            "required": ["lead_email", "sequence_name"]
        }
    }
]


def handle_crm_update(inputs: dict) -> dict:
    """Simulate CRM write. Replace with real API call in production."""
    print(f"  [CRM] Updated record for {inputs['lead_email']}: "
          f"score={inputs['score']}, tier={inputs['tier']}")
    return {"success": True, "crm_record_id": f"CRM-{inputs['lead_email'][:8].upper()}"}


def handle_email_trigger(inputs: dict) -> dict:
    """Simulate email enrollment. Replace with ActiveCampaign/HubSpot call."""
    print(f"  [EMAIL] Enrolled {inputs['lead_email']} in '{inputs['sequence_name']}'")
    return {"success": True, "enrolled": True}

Step 4: Build the Orchestration Loop with Agent Handoffs

This is the core of a multi-agent system — the orchestrator. It holds a tool registry, knows which agent to call at each stage, and handles the handoff from validation to scoring. The loop keeps running until the agent stops calling tools, which is Claude's way of saying "I'm done."

lead_scoring.py (continued)

class LeadScoringOrchestrator:
    def __init__(self):
        # Tool registry maps tool names to their handler functions
        self.tool_registry = {
            "update_crm": handle_crm_update,
            "trigger_email_sequence": handle_email_trigger
        }
        self.validation_result = None
        self.scoring_result = None

    def run_agent(
        self,
        system_prompt: str,
        tools: list,
        initial_message: str
    ) -> dict | None:
        """
        Run a single agent in a loop until it completes its tool call.
        Returns the tool input dict from the agent's final tool call.
        """
        messages = [{"role": "user", "content": initial_message}]
        tool_result = None

        while True:
            response = client.messages.create(
                model=MODEL,
                max_tokens=1024,
                system=system_prompt,
                tools=tools,
                messages=messages
            )

            # Append assistant response to conversation history
            messages.append({"role": "assistant", "content": response.content})

            if response.stop_reason == "end_turn":
                # Agent finished without calling a tool — shouldn't happen in this design
                break

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

                for block in response.content:
                    if block.type != "tool_use":
                        continue

                    tool_name = block.name
                    tool_input = block.input

                    # If it's a side-effect tool (CRM, email), execute it
                    if tool_name in self.tool_registry:
                        result = self.tool_registry[tool_name](tool_input)
                        tool_results_content.append({
                            "type": "tool_result",
                            "tool_use_id": block.id,
                            "content": json.dumps(result)
                        })
                    else:
                        # It's a structured output tool (validate_lead, score_lead)
                        tool_result = tool_input
                        tool_results_content.append({
                            "type": "tool_result",
                            "tool_use_id": block.id,
                            "content": json.dumps({"received": True})
                        })

                # Feed tool results back to continue the conversation
                messages.append({"role": "user", "content": tool_results_content})

                # If we got a structured output result, we're done with this agent
                if tool_result is not None:
                    break

        return tool_result

    def process_lead(self, lead_data: dict) -> dict:
        """
        Full pipeline: validate -> score -> CRM update -> email enrollment.
        Returns the complete scored lead result.
        """
        lead_json = json.dumps(lead_data, indent=2)
        print(f"\n{'='*50}")
        print(f"Processing lead: {lead_data.get('name', 'Unknown')} "
              f"<{lead_data.get('email', 'no-email')}>")
        print(f"{'='*50}")

        # Stage 1: Validation Agent
        print("\n[Stage 1] Running Validation Agent...")
        validation_message = f"Please validate this incoming lead data:\n\n{lead_json}"
        self.validation_result = self.run_agent(
            VALIDATION_AGENT_PROMPT,
            VALIDATION_TOOLS,
            validation_message
        )

        if not self.validation_result:
            return {"error": "Validation agent failed to return a result."}

        print(f"  Validation status: {self.validation_result['status']}")

        if self.validation_result["status"] == "invalid":
            print(f"  Errors: {self.validation_result['errors']}")
            return {
                "lead": lead_data,
                "validation": self.validation_result,
                "scoring": None,
                "final_status": "rejected_invalid_data"
            }

        # Stage 2: Scoring Agent (only runs on valid leads)
        print("\n[Stage 2] Running Scoring Agent...")
        scoring_message = (
            f"Please score this validated lead and update the CRM and email system:\n\n"
            f"{lead_json}\n\n"
            f"After scoring, also call update_crm and trigger_email_sequence "
            f"using the lead's email and your scoring results."
        )
        self.scoring_result = self.run_agent(
            SCORING_AGENT_PROMPT,
            SCORING_TOOLS,
            scoring_message
        )

        if not self.scoring_result:
            return {"error": "Scoring agent failed to return a result."}

        print(f"\n  Score: {self.scoring_result['score']}/100 "
              f"({self.scoring_result['tier'].upper()})")
        print(f"  Action: {self.scoring_result['recommended_action']}")
        print(f"  Reasoning: {self.scoring_result['reasoning']}")

        return {
            "lead": lead_data,
            "validation": self.validation_result,
            "scoring": self.scoring_result,
            "final_status": "scored"
        }

The orchestrator class holds state across both agent runs and keeps the tool registry in one place. Adding a new tool later just means adding a function and registering it in __init__ — the loop handles everything else automatically.

Example Lead Input and Running the System

Here's how you wire it all up and run it with a real sample lead. I'm including both a high-quality lead and a bad one so you can see the validation path in action.

lead_scoring.py (continued)

if __name__ == "__main__":
    orchestrator = LeadScoringOrchestrator()

    # Example 1: A strong lead — real estate company, clear intent
    hot_lead = {
        "name": "Maria Gonzalez",
        "email": "[email protected]",
        "company": "Suncoast Realty Group",
        "industry": "real_estate",
        "annual_revenue": 2400000,
        "employees": 45,
        "message": (
            "We manually process about 300 listing inquiries a month. "
            "Our team spends 2-3 hours a day just on data entry and follow-ups. "
            "We heard AI can automate this — we want to talk to someone this week."
        )
    }

    # Example 2: An incomplete lead — will fail validation
    bad_lead = {
        "name": "",
        "email": "notavalidemail",
        "company": "Unknown LLC",
        "industry": "fintech",  # not in allowed list
        "annual_revenue": -500,
        "employees": 12,
        "message": "Interested in your services."
    }

    result_1 = orchestrator.process_lead(hot_lead)
    print("\n--- FINAL OUTPUT (Lead 1) ---")
    print(json.dumps(result_1, indent=2))

    print("\n\n")
    orchestrator2 = LeadScoringOrchestrator()
    result_2 = orchestrator2.process_lead(bad_lead)
    print("\n--- FINAL OUTPUT (Lead 2) ---")
    print(json.dumps(result_2, indent=2))

Here's what the output looks like when you run this against the strong lead:

sample output

==================================================
Processing lead: Maria Gonzalez 
==================================================

[Stage 1] Running Validation Agent...
  Validation status: valid

[Stage 2] Running Scoring Agent...
  [CRM] Updated record for maria@su: score=91, tier=hot
  [EMAIL] Enrolled [email protected] in 'hot_lead_immediate'

  Score: 91/100 (HOT)
  Action: immediate_outreach
  Reasoning: This lead scores exceptionally well across all criteria. Real estate
  is a high-value industry for AI automation, the company size of 45 employees
  is in the ideal range, and the $2.4M revenue signals a serious business.
  The message demonstrates a specific, quantified pain point with clear buying
  intent and urgency.

--- FINAL OUTPUT (Lead 1) ---
{
  "lead": {
    "name": "Maria Gonzalez",
    "email": "[email protected]",
    "company": "Suncoast Realty Group",
    "industry": "real_estate",
    "annual_revenue": 2400000,
    "employees": 45,
    "message": "We manually process about 300 listing inquiries a month..."
  },
  "validation": {
    "status": "valid",
    "errors": []
  },
  "scoring": {
    "score": 91,
    "tier": "hot",
    "reasoning": "This lead scores exceptionally well across all criteria...",
    "recommended_action": "immediate_outreach"
  },
  "final_status": "scored"
}

How It Works: Agent Collaboration Flow

Here's the plain-English version of what's happening when you call process_lead(). The orchestrator takes your raw lead dictionary and passes it to the Validation Agent first. That agent reads the system prompt, checks every field, and calls the validate_lead tool with a pass/fail result.

If validation fails, the orchestrator stops right there and returns a rejection result with specific error messages — no API call wasted on scoring a garbage lead. If it passes, the orchestrator hands the clean data to the Scoring Agent along with permission to also fire the CRM and email tools.

The Scoring Agent evaluates the lead, calls score_lead to return its structured assessment, then calls update_crm and trigger_email_sequence as side effects. The run_agent loop handles all of this automatically — it keeps feeding tool results back to Claude until the agent stops calling tools. The whole pipeline takes 3-6 seconds on average.

🔄 The Handoff Pattern
The key design decision here is that the orchestrator owns the handoff logic, not the agents. The agents don't know about each other — they just receive a prompt and call their tools. This makes it trivial to add a third agent (like an enrichment agent that pulls LinkedIn data) without touching any existing code.

Common Errors and Fixes

Error 1: AuthenticationError on first run

Exact error: anthropic.AuthenticationError: Error code: 401 - {'type': 'error', 'error': {'type': 'authentication_error', 'message': 'invalid x-api-key'}}

This almost always means the environment variable isn't loaded. Check that your .env file is in the same directory you're running the script from, and that you have load_dotenv() called before the Anthropic() client is instantiated. You can also test it directly: print(os.environ.get("ANTHROPIC_API_KEY")) — if it prints None, the file isn't loading.

Error 2: Agent loops indefinitely without returning

Symptom: The script hangs, or you see repeated API calls with no output. This happens when the tool result you feed back to Claude doesn't match the expected format. Make sure every tool result message uses the exact structure: {"type": "tool_result", "tool_use_id": block.id,