All articles

What You'll Build

If you've searched for how to build AI agents tutorial material that actually ships working code, you're in the right place. By the end of this walkthrough, you'll have a fully functional multi-agent lead qualification pipeline built in Python using the Anthropic Claude API. Three specialized agents work together: one scores incoming leads, one routes them to the right sales bucket, and one enriches them with additional context before your team ever touches them.

This isn't a toy demo. The pattern here is exactly what we use at Naples AI when we build real estate and sales automation systems for Southwest Florida businesses. You can drop this into a production workflow with minimal changes.

Prerequisites

  • Python 3.10 or higher installed
  • An Anthropic API key (get one at console.anthropic.com)
  • Basic familiarity with Python classes and async concepts
  • anthropic SDK installed: pip install anthropic
  • pydantic installed: pip install pydantic
  • Optional but helpful: a free RapidAPI account for the enrichment step
Full Source Code Note: Every snippet in this tutorial builds toward a single working system. The complete, copy-paste-ready code is spread across Steps 1 through 7 below in the exact order you need to assemble it. By Step 7 you'll have everything. If you want to jump straight to running it, skim to Step 7 for the orchestration call and sample output.

Step 1: Set Up Your Claude API Client and Environment

First, let's get the foundation in place. Create a project folder and a .env file to hold your API key safely. Never hardcode credentials in your source files.

.env 
ANTHROPIC_API_KEY=your_api_key_here

Now install dependencies and create the base client module. This module gets imported by all three agents, so getting it right here saves headaches later.

client.py 
import os
import anthropic
from dotenv import load_dotenv

load_dotenv()

def get_client() -> anthropic.Anthropic:
    api_key = os.environ.get("ANTHROPIC_API_KEY")
    if not api_key:
        raise ValueError("ANTHROPIC_API_KEY not found in environment variables.")
    return anthropic.Anthropic(api_key=api_key)

MODEL = "claude-sonnet-4-6"

Install python-dotenv so the .env file loads cleanly: pip install python-dotenv. That's literally all the setup you need. The Anthropic SDK handles connection pooling and retries for you out of the box.

Step 2: Define Lead Data Schemas and Tool Specifications

Before writing any agent logic, define what a lead looks like and what tools the agents can call. Using Pydantic models here means you get free validation and clean serialization when passing data between agents.

schemas.py 
from pydantic import BaseModel, Field
from typing import Optional, List
from enum import Enum

class LeadTier(str, Enum):
    HOT = "hot"
    WARM = "warm"
    COLD = "cold"
    DISQUALIFIED = "disqualified"

class LeadSource(str, Enum):
    WEBSITE = "website"
    REFERRAL = "referral"
    PAID_ADS = "paid_ads"
    ORGANIC = "organic"
    EVENT = "event"

class RawLead(BaseModel):
    lead_id: str
    name: str
    email: str
    phone: Optional[str] = None
    source: LeadSource
    message: str
    budget: Optional[int] = None          # budget in USD
    timeline: Optional[str] = None        # e.g. "30 days", "6 months"
    property_type: Optional[str] = None   # for real estate use cases

class ScoredLead(BaseModel):
    lead_id: str
    name: str
    email: str
    score: int = Field(ge=0, le=100)
    tier: LeadTier
    score_rationale: str
    flags: List[str] = []                 # e.g. ["high budget", "urgent timeline"]

class RoutedLead(BaseModel):
    lead_id: str
    tier: LeadTier
    assigned_team: str
    priority_rank: int
    routing_notes: str

class EnrichedLead(BaseModel):
    lead_id: str
    name: str
    email: str
    score: int
    tier: LeadTier
    assigned_team: str
    priority_rank: int
    score_rationale: str
    routing_notes: str
    flags: List[str]
    enrichment_summary: str
    suggested_opener: str                 # personalized first-touch message

Notice how each schema maps to exactly one agent's output. This keeps the pipeline stages clean and debuggable. If something breaks, you know immediately which agent produced the bad data.

Step 3: Build the Lead Scorer Agent

The scorer is the first agent in the chain. It reads a raw lead, calls a scoring tool, and returns a structured score from 0 to 100 along with a tier classification. The key design choice here is using Claude's tool use feature so the output is always structured JSON, never free-form text that you'd need to parse.

scorer_agent.py 
import json
from typing import Any
import anthropic
from client import get_client, MODEL
from schemas import RawLead, ScoredLead, LeadTier

# Tool definition that Claude will call to return structured scores
SCORE_LEAD_TOOL = {
    "name": "score_lead",
    "description": (
        "Score an inbound lead from 0 to 100 based on fit, intent, budget, "
        "and timeline signals. Classify into hot/warm/cold/disqualified tiers."
    ),
    "input_schema": {
        "type": "object",
        "properties": {
            "score": {
                "type": "integer",
                "minimum": 0,
                "maximum": 100,
                "description": "Lead quality score from 0 (worst) to 100 (best)"
            },
            "tier": {
                "type": "string",
                "enum": ["hot", "warm", "cold", "disqualified"],
                "description": "Qualification tier based on score and signals"
            },
            "score_rationale": {
                "type": "string",
                "description": "2-3 sentence explanation of why this score was given"
            },
            "flags": {
                "type": "array",
                "items": {"type": "string"},
                "description": "Notable signals extracted from the lead, e.g. 'urgent timeline', 'high budget'"
            }
        },
        "required": ["score", "tier", "score_rationale", "flags"]
    }
}

class LeadScorerAgent:
    def __init__(self):
        self.client = get_client()
        self.system_prompt = (
            "You are a lead qualification specialist for a real estate and business services company "
            "in Southwest Florida. Your job is to score inbound leads based on four signals: "
            "1) Budget fit (does their stated budget match our service range?), "
            "2) Timeline urgency (how soon do they need help?), "
            "3) Message intent (are they clearly ready to buy/hire, or just browsing?), "
            "4) Source quality (referrals and organic score higher than cold paid traffic). "
            "Always call the score_lead tool. Never respond with plain text."
        )

    def score(self, lead: RawLead) -> ScoredLead:
        lead_context = (
            f"Lead Name: {lead.name}\n"
            f"Source: {lead.source.value}\n"
            f"Message: {lead.message}\n"
            f"Budget: ${lead.budget:,}" if lead.budget else f"Budget: Not provided"
        )
        # Append timeline and property type if present
        if lead.timeline:
            lead_context += f"\nTimeline: {lead.timeline}"
        if lead.property_type:
            lead_context += f"\nProperty Type: {lead.property_type}"

        response = self.client.messages.create(
            model=MODEL,
            max_tokens=1024,
            system=self.system_prompt,
            tools=[SCORE_LEAD_TOOL],
            tool_choice={"type": "tool", "name": "score_lead"},  # force tool use
            messages=[
                {"role": "user", "content": f"Please score this lead:\n\n{lead_context}"}
            ]
        )

        # Extract tool use block from response
        tool_result = self._extract_tool_result(response)

        return ScoredLead(
            lead_id=lead.lead_id,
            name=lead.name,
            email=lead.email,
            score=tool_result["score"],
            tier=LeadTier(tool_result["tier"]),
            score_rationale=tool_result["score_rationale"],
            flags=tool_result.get("flags", [])
        )

    def _extract_tool_result(self, response: Any) -> dict:
        for block in response.content:
            if block.type == "tool_use" and block.name == "score_lead":
                return block.input
        raise ValueError("Score tool was not called by the model. Check your tool_choice config.")

The tool_choice={"type": "tool", "name": "score_lead"} parameter is what forces Claude to always return structured data. Without it, Claude might occasionally respond in plain text, which breaks your downstream parsing. Always force tool use when you need deterministic output format.

Pro tip: The system prompt matters a lot here. Giving Claude explicit scoring criteria (budget fit, timeline, intent, source) means you get consistent, defensible scores instead of vibes-based guesses. Treat the system prompt like a job description for the agent.

Step 4: Build the Lead Router Agent

The router takes the scored lead and decides which team should handle it. In a real estate agency, that might mean routing hot leads to senior brokers, warm leads to junior agents, and cold leads to a nurture drip. This agent reads the scoring output and applies business routing logic through Claude.

router_agent.py 
import json
from typing import Any
import anthropic
from client import get_client, MODEL
from schemas import ScoredLead, RoutedLead, LeadTier

ROUTE_LEAD_TOOL = {
    "name": "route_lead",
    "description": (
        "Route a scored lead to the appropriate sales team and assign a priority rank "
        "based on tier, score, and business rules."
    ),
    "input_schema": {
        "type": "object",
        "properties": {
            "assigned_team": {
                "type": "string",
                "description": "Name of the team or rep type to handle this lead",
                "enum": [
                    "Senior Broker Team",
                    "Junior Agent Pool",
                    "Automated Nurture Sequence",
                    "Do Not Contact"
                ]
            },
            "priority_rank": {
                "type": "integer",
                "minimum": 1,
                "maximum": 5,
                "description": "Priority from 1 (contact within 1 hour) to 5 (contact within 2 weeks)"
            },
            "routing_notes": {
                "type": "string",
                "description": "Brief notes for the assigned team on how to approach this lead"
            }
        },
        "required": ["assigned_team", "priority_rank", "routing_notes"]
    }
}

# Maps tiers to routing instructions for the system prompt
ROUTING_RULES = {
    LeadTier.HOT: "Score 75-100. Route to Senior Broker Team with priority 1 or 2. Respond within 1 hour.",
    LeadTier.WARM: "Score 45-74. Route to Junior Agent Pool with priority 3. Respond within 24 hours.",
    LeadTier.COLD: "Score 20-44. Route to Automated Nurture Sequence with priority 4 or 5.",
    LeadTier.DISQUALIFIED: "Score 0-19 or spam signals. Route to Do Not Contact. Document reason."
}

class LeadRouterAgent:
    def __init__(self):
        self.client = get_client()
        routing_rules_text = "\n".join(
            f"- {tier.value.upper()}: {rule}"
            for tier, rule in ROUTING_RULES.items()
        )
        self.system_prompt = (
            "You are a sales operations specialist routing leads for a Southwest Florida real estate "
            "and business services company. Apply these routing rules strictly:\n\n"
            f"{routing_rules_text}\n\n"
            "Always call the route_lead tool. Use the flags and score rationale from the scorer "
            "to write helpful, specific notes for the assigned team."
        )

    def route(self, scored_lead: ScoredLead) -> RoutedLead:
        scored_context = (
            f"Lead Name: {scored_lead.name}\n"
            f"Score: {scored_lead.score}/100\n"
            f"Tier: {scored_lead.tier.value}\n"
            f"Score Rationale: {scored_lead.score_rationale}\n"
            f"Flags: {', '.join(scored_lead.flags) if scored_lead.flags else 'None'}"
        )

        response = self.client.messages.create(
            model=MODEL,
            max_tokens=1024,
            system=self.system_prompt,
            tools=[ROUTE_LEAD_TOOL],
            tool_choice={"type": "tool", "name": "route_lead"},
            messages=[
                {"role": "user", "content": f"Route this scored lead:\n\n{scored_context}"}
            ]
        )

        tool_result = self._extract_tool_result(response)

        return RoutedLead(
            lead_id=scored_lead.lead_id,
            tier=scored_lead.tier,
            assigned_team=tool_result["assigned_team"],
            priority_rank=tool_result["priority_rank"],
            routing_notes=tool_result["routing_notes"]
        )

    def _extract_tool_result(self, response: Any) -> dict:
        for block in response.content:
            if block.type == "tool_use" and block.name == "route_lead":
                return block.input
        raise ValueError("Route tool was not called by the model.")

Notice that the routing rules are embedded directly into the system prompt as a formatted string. This is intentional. When you need Claude to follow business logic precisely, writing it explicitly in the prompt beats hoping the model infers it. Clear rules produce consistent routing.

Step 5: Build the Lead Enrichment Agent

The enrichment agent is the most interesting one. It takes the scored and routed lead, synthesizes everything it knows, and produces two things: a plain-English enrichment summary and a personalized opener your sales rep can send in their first message. In production, you'd also call an external data API here (like Clearbit or Apollo) to append company data. I've mocked that call so the code runs without external credentials, but I've shown you exactly where to plug in a real API call.

enrichment_agent.py 
import json
import time
from typing import Any, Optional
import anthropic
from client import get_client, MODEL
from schemas import ScoredLead, RoutedLead, EnrichedLead

ENRICH_LEAD_TOOL = {
    "name": "enrich_lead",
    "description": (
        "Generate a lead enrichment summary and a personalized first-touch message opener "
        "for the sales team based on all available lead data."
    ),
    "input_schema": {
        "type": "object",
        "properties": {
            "enrichment_summary": {
                "type": "string",
                "description": (
                    "3-5 sentence summary of everything known about this lead including "
                    "intent signals, fit assessment, and any external data appended."
                )
            },
            "suggested_opener": {
                "type": "string",
                "description": (
                    "A personalized, natural-sounding first message (2-4 sentences) "
                    "the rep can send via email or text. Should reference something specific "
                    "from the lead's message to show it's not a template."
                )
            }
        },
        "required": ["enrichment_summary", "suggested_opener"]
    }
}

def mock_external_data_lookup(email: str) -> Optional[dict]:
    """
    In production, replace this with a real API call, e.g.:
        response = requests.get(
            f"https://api.apollo.io/v1/people/match?email={email}",
            headers={"X-Api-Key": os.environ["APOLLO_API_KEY"]}
        )
        return response.json()
    For now, we simulate a lookup with a small delay and dummy data.
    """
    time.sleep(0.3)  # simulates network latency
    mock_profiles = {
        "default": {
            "company": "Unknown",
            "title": "Unknown",
            "linkedin": None,
            "company_size": "Unknown"
        }
    }
    # Return a richer profile for demo email domains
    if "realty" in email or "properties" in email or "homes" in email:
        return {
            "company": "Gulf Coast Properties LLC",
            "title": "Principal Broker",
            "linkedin": "linkedin.com/in/sample-broker",
            "company_size": "11-50 employees"
        }
    return mock_profiles["default"]

class LeadEnrichmentAgent:
    def __init__(self):
        self.client = get_client()
        self.system_prompt = (
            "You are a lead intelligence analyst for a Southwest Florida real estate "
            "and AI services company. Your job is to synthesize everything known about a lead "
            "and help the sales team show up prepared. Write like a smart colleague briefing "
            "another colleague, not like a CRM entry. Keep the suggested opener warm and specific, "
            "never generic. Always call the enrich_lead tool."
        )

    def enrich(
        self,
        scored_lead: ScoredLead,
        routed_lead: RoutedLead,
        original_message: str
    ) -> EnrichedLead:
        # Attempt external data append
        external_data = mock_external_data_lookup(scored_lead.email)
        external_data_text = (
            f"External profile found: {json.dumps(external_data, indent=2)}"
            if external_data
            else "No external profile data available."
        )

        full_context = (
            f"Lead Name: {scored_lead.name}\n"
            f"Email: {scored_lead.email}\n"
            f"Original Message: {original_message}\n"
            f"Score: {scored_lead.score}/100 ({scored_lead.tier.value})\n"
            f"Score Rationale: {scored_lead.score_rationale}\n"
            f"Flags: {', '.join(scored_lead.flags) if scored_lead.flags else 'None'}\n"
            f"Assigned Team: {routed_lead.assigned_team}\n"
            f"Priority: {routed_lead.priority_rank}\n"
            f"Routing Notes: {routed_lead.routing_notes}\n\n"
            f"{external_data_text}"
        )

        response = self.client.messages.create(
            model=MODEL,
            max_tokens=2048,
            system=self.system_prompt,
            tools=[ENRICH_LEAD_TOOL],
            tool_choice={"type": "tool", "name": "enrich_lead"},
            messages=[
                {
                    "role": "user",
                    "content": (
                        f"Please enrich this lead and write a personalized opener:\n\n{full_context}"
                    )
                }
            ]
        )

        tool_result = self._extract_tool_result(response)

        return EnrichedLead(
            lead_id=scored_lead.lead_id,
            name=scored_lead.name,
            email=scored_lead.email,
            score=scored_lead.score,
            tier=scored_lead.tier,
            assigned_team=routed_lead.assigned_team,
            priority_rank=routed_lead.priority_rank,
            score_rationale=scored_lead.score_rationale,
            routing_notes=routed_lead.routing_notes,
            flags=scored_lead.flags,
            enrichment_summary=tool_result["enrichment_summary"],
            suggested_opener=tool_result["suggested_opener"]
        )

    def _extract_tool_result(self, response: Any) -> dict:
        for block in response.content:
            if block.type == "tool_use" and block.name == "enrich_lead":
                return block.input
        raise ValueError("Enrich tool was not called by the model.")
Where to plug in real data: The mock_external_data_lookup function is your extension point. Swap it out for Apollo.io, Clearbit, or even a Google search API call and you instantly get real company and title data appended to every lead. The rest of the agent doesn't change at all.

Step 6: Orchestrate the Multi-Agent Pipeline

This is where the three agents get wired together. The orchestrator runs each agent in sequence, passes outputs between them, handles errors at