) (1).png)
If you've searched for how to build a multi agent system in Python, you've probably found a lot of theory and very little working code. This tutorial fixes that. You'll get a complete, production-ready multi-agent lead qualification system built on the Claude API — no fluff, no pseudocode, just something you can actually run.
What You'll Build
You'll build a three-agent pipeline that takes raw lead data, validates it, scores it based on fit, and routes it to the right sales rep — all automatically. The system uses Claude's tool-use feature to wire up three specialized agents: a Lead Analyzer, a Lead Scorer, and a Lead Router. By the end, you'll have a working agentic workflow you can drop into any CRM or sales process.
Prerequisites
- Python 3.10 or higher installed
- An Anthropic API key (get one at console.anthropic.com)
- Basic Python knowledge — you don't need to be an expert
anthropicPython SDK installed (pip install anthropic)- Familiarity with Python dictionaries and functions
📦 Full Source Code
The complete working code for this tutorial is built up section by section below. Every snippet connects to the next one. By Step 7, you'll have the entire system assembled and ready to run. Copy each block in order, or scroll to Step 7 to see how they all fit together in one file.
The complete working code for this tutorial is built up section by section below. Every snippet connects to the next one. By Step 7, you'll have the entire system assembled and ready to run. Copy each block in order, or scroll to Step 7 to see how they all fit together in one file.
Step 1: Set Up Your Environment and Claude API Key
First, install the Anthropic SDK if you haven't already. Open your terminal and run pip install anthropic. Then create a .env file in your project root and add your API key — never hardcode it directly in your script.
ANTHROPIC_API_KEY=sk-ant-your-key-here
Now create your main project file. We'll build everything in a single lead_qualification.py file for clarity. Here's the setup block — this goes at the very top of the file.
import os
import json
import anthropic
from typing import Any
# Load API key from environment
api_key = os.environ.get("ANTHROPIC_API_KEY")
if not api_key:
raise ValueError("ANTHROPIC_API_KEY environment variable not set")
# Initialize the Anthropic client
client = anthropic.Anthropic(api_key=api_key)
# Model to use across all agents
MODEL = "claude-sonnet-4-5"
💡 Tip: If you're on a Mac or Linux machine, you can also export your key inline:
export ANTHROPIC_API_KEY=sk-ant-... and the os.environ.get() call will pick it up automatically without a .env file.
Step 2: Define Tool Schemas for Data Validation and CRM Integration
Claude's tool-use feature lets you define functions that the model can call during a conversation. Think of them as structured API endpoints Claude knows how to trigger. We need three: one to validate lead data, one to score a lead, and one to route it to the right rep.
Here's the tool schema definitions. These tell Claude what each tool does and what parameters it expects — it's basically a JSON contract.
lead_qualification.py (continued)# --- Tool Schema Definitions ---
TOOLS = [
{
"name": "validate_lead",
"description": (
"Validates incoming lead data for completeness and format. "
"Checks required fields like name, email, company, and budget. "
"Returns a validation result with any missing or malformed fields."
),
"input_schema": {
"type": "object",
"properties": {
"name": {
"type": "string",
"description": "Full name of the lead"
},
"email": {
"type": "string",
"description": "Email address of the lead"
},
"company": {
"type": "string",
"description": "Company or organization name"
},
"budget": {
"type": "number",
"description": "Estimated budget in USD"
},
"industry": {
"type": "string",
"description": "Industry vertical (e.g., real estate, healthcare)"
},
"use_case": {
"type": "string",
"description": "Brief description of what the lead wants to build or solve"
}
},
"required": ["name", "email", "company", "budget", "industry", "use_case"]
}
},
{
"name": "score_lead",
"description": (
"Scores a validated lead from 0 to 100 based on budget fit, "
"industry match, and use case complexity. Returns a numeric score "
"and a tier label: hot, warm, or cold."
),
"input_schema": {
"type": "object",
"properties": {
"name": {
"type": "string",
"description": "Full name of the lead"
},
"budget": {
"type": "number",
"description": "Lead's stated budget in USD"
},
"industry": {
"type": "string",
"description": "Industry vertical"
},
"use_case": {
"type": "string",
"description": "What the lead wants to build"
}
},
"required": ["name", "budget", "industry", "use_case"]
}
},
{
"name": "route_to_sales",
"description": (
"Routes a scored lead to the appropriate sales rep or pipeline stage. "
"Hot leads go to senior reps, warm leads to mid-level, cold leads to nurture. "
"Returns the assigned rep name and next action."
),
"input_schema": {
"type": "object",
"properties": {
"name": {
"type": "string",
"description": "Full name of the lead"
},
"score": {
"type": "number",
"description": "Numeric lead score from 0 to 100"
},
"tier": {
"type": "string",
"description": "Lead tier: hot, warm, or cold",
"enum": ["hot", "warm", "cold"]
},
"industry": {
"type": "string",
"description": "Industry vertical for rep matching"
}
},
"required": ["name", "score", "tier", "industry"]
}
}
]
Step 3: Create the Lead Analyzer Agent
The Lead Analyzer is the first agent in the pipeline. Its job is to take raw lead data, call the validate_lead tool, and return a structured validation result. I'm building each agent as a Python function that handles its own Claude API call and tool loop.
# --- Tool Function: validate_lead ---
def validate_lead(name: str, email: str, company: str,
budget: float, industry: str, use_case: str) -> dict[str, Any]:
"""Validates lead fields and returns a structured result."""
errors = []
# Basic email format check
if "@" not in email or "." not in email.split("@")[-1]:
errors.append(f"Invalid email format: {email}")
# Budget floor check — we don't work with leads under $5k
if budget < 5000:
errors.append(f"Budget too low: ${budget:,.0f} (minimum $5,000)")
# Use case must be substantive
if len(use_case.strip()) < 20:
errors.append("Use case description is too short — needs more detail")
is_valid = len(errors) == 0
return {
"status": "valid" if is_valid else "invalid",
"errors": errors,
"lead": {
"name": name,
"email": email,
"company": company,
"budget": budget,
"industry": industry,
"use_case": use_case
}
}
# --- Agent 1: Lead Analyzer ---
def lead_analyzer_agent(lead_data: dict[str, Any]) -> dict[str, Any]:
"""
First agent in the pipeline. Validates raw lead data using the validate_lead tool.
Returns structured validation result to pass downstream.
"""
system_prompt = (
"You are a lead validation specialist. Your only job is to call the validate_lead "
"tool with the lead data you receive. Do not skip the tool call. Do not add commentary. "
"Just call the tool and return the result."
)
user_message = (
f"Please validate this lead:\n"
f"Name: {lead_data.get('name')}\n"
f"Email: {lead_data.get('email')}\n"
f"Company: {lead_data.get('company')}\n"
f"Budget: {lead_data.get('budget')}\n"
f"Industry: {lead_data.get('industry')}\n"
f"Use Case: {lead_data.get('use_case')}"
)
messages = [{"role": "user", "content": user_message}]
# Initial API call — Claude will respond with a tool_use block
response = client.messages.create(
model=MODEL,
max_tokens=1024,
system=system_prompt,
tools=[TOOLS[0]], # Only expose validate_lead to this agent
messages=messages
)
# Handle tool call in a loop (agent loop pattern)
while 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
# Execute the actual tool function
if tool_name == "validate_lead":
result = validate_lead(**tool_input)
else:
result = {"error": f"Unknown tool: {tool_name}"}
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": json.dumps(result)
})
# Append assistant response and tool results to message history
messages.append({"role": "assistant", "content": response.content})
messages.append({"role": "user", "content": tool_results})
# Continue the loop — Claude will finish after seeing the tool result
response = client.messages.create(
model=MODEL,
max_tokens=1024,
system=system_prompt,
tools=[TOOLS[0]],
messages=messages
)
# Extract the validation result from the last tool call
for msg in messages:
if isinstance(msg["content"], list):
for item in msg["content"]:
if isinstance(item, dict) and item.get("type") == "tool_result":
return json.loads(item["content"])
return {"status": "error", "errors": ["Analyzer agent failed to produce a result"]}
Step 4: Create the Lead Scorer Agent
The Lead Scorer takes a validated lead and assigns it a numeric score plus a tier label. This is where business logic lives — higher budgets, priority industries, and complex use cases score higher. Keep this agent focused: it only calls score_lead and returns the result.
# --- Tool Function: score_lead ---
def score_lead(name: str, budget: float, industry: str, use_case: str) -> dict[str, Any]:
"""
Scores a lead from 0-100 based on budget, industry fit, and use case.
Returns score, tier, and reasoning.
"""
score = 0
reasoning = []
# Budget scoring — up to 40 points
if budget >= 50000:
score += 40
reasoning.append("Large budget (40 pts)")
elif budget >= 20000:
score += 28
reasoning.append("Mid-range budget (28 pts)")
elif budget >= 10000:
score += 16
reasoning.append("Entry budget (16 pts)")
else:
score += 5
reasoning.append("Low budget (5 pts)")
# Industry scoring — up to 35 points (these match Naples AI's core verticals)
priority_industries = {
"real estate": 35,
"healthcare": 32,
"manufacturing": 30,
"automotive": 28,
"restaurant": 25,
"retail": 20
}
industry_lower = industry.lower()
industry_score = priority_industries.get(industry_lower, 10)
score += industry_score
reasoning.append(f"Industry: {industry} ({industry_score} pts)")
# Use case complexity scoring — up to 25 points
high_value_keywords = [
"automation", "ai", "integration", "pipeline",
"chatbot", "analytics", "crm", "prediction"
]
use_case_lower = use_case.lower()
keyword_matches = sum(1 for kw in high_value_keywords if kw in use_case_lower)
use_case_score = min(keyword_matches * 6, 25)
score += use_case_score
reasoning.append(f"Use case keywords matched: {keyword_matches} ({use_case_score} pts)")
# Determine tier
if score >= 75:
tier = "hot"
elif score >= 45:
tier = "warm"
else:
tier = "cold"
return {
"name": name,
"score": score,
"tier": tier,
"reasoning": reasoning
}
# --- Agent 2: Lead Scorer ---
def lead_scorer_agent(validated_lead: dict[str, Any]) -> dict[str, Any]:
"""
Second agent in the pipeline. Scores the validated lead.
Only activated if the lead passed validation.
"""
lead = validated_lead.get("lead", {})
system_prompt = (
"You are a lead scoring specialist. Call the score_lead tool with the lead "
"data provided. Do not skip the tool. Return the score result directly."
)
user_message = (
f"Score this validated lead:\n"
f"Name: {lead.get('name')}\n"
f"Budget: {lead.get('budget')}\n"
f"Industry: {lead.get('industry')}\n"
f"Use Case: {lead.get('use_case')}"
)
messages = [{"role": "user", "content": user_message}]
response = client.messages.create(
model=MODEL,
max_tokens=1024,
system=system_prompt,
tools=[TOOLS[1]], # Only expose score_lead to this agent
messages=messages
)
while response.stop_reason == "tool_use":
tool_results = []
for block in response.content:
if block.type == "tool_use":
tool_input = block.input
result = score_lead(**tool_input)
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": json.dumps(result)
})
messages.append({"role": "assistant", "content": response.content})
messages.append({"role": "user", "content": tool_results})
response = client.messages.create(
model=MODEL,
max_tokens=1024,
system=system_prompt,
tools=[TOOLS[1]],
messages=messages
)
for msg in messages:
if isinstance(msg["content"], list):
for item in msg["content"]:
if isinstance(item, dict) and item.get("type") == "tool_result":
return json.loads(item["content"])
return {"error": "Scorer agent failed to produce a result"}
Step 5: Create the Lead Router Agent
The Lead Router is the final agent in the chain. It takes the score and tier from the previous step and decides where the lead goes. Hot leads get a senior rep and a same-day callback. Warm leads go to a mid-level rep for a discovery call. Cold leads drop into a nurture sequence.
lead_qualification.py (continued)# --- Tool Function: route_to_sales ---
def route_to_sales(name: str, score: float, tier: str, industry: str) -> dict[str, Any]:
"""
Routes a scored lead to the right rep and pipeline stage.
Returns assigned rep, pipeline stage, and recommended next action.
"""
# Rep roster — in production this would query your CRM
rep_assignments = {
"hot": {
"rep": "Chris Mejias",
"title": "Founder & Senior AI Consultant",
"pipeline_stage": "Priority Discovery",
"next_action": "Schedule same-day 30-minute strategy call",
"sla": "Contact within 2 hours"
},
"warm": {
"rep": "Solutions Team",
"title": "AI Solutions Consultant",
"pipeline_stage": "Standard Discovery",
"next_action": "Send intro email and schedule discovery call this week",
"sla": "Contact within 24 hours"
},
"cold": {
"rep": "Marketing Automation",
"title": "Nurture Sequence",
"pipeline_stage": "Lead Nurture",
"next_action": "Enroll in 5-email AI education drip sequence",
"sla": "Automated follow-up within 1 hour"
}
}
assignment = rep_assignments.get(tier, rep_assignments["cold"])
return {
"lead_name": name,
"score": score,
"tier": tier.upper(),
"industry": industry,
"assigned_rep": assignment["rep"],
"rep_title": assignment["title"],
"pipeline_stage": assignment["pipeline_stage"],
"next_action": assignment["next_action"],
"sla": assignment["sla"],
"routing_status": "success"
}
# --- Agent 3: Lead Router ---
def lead_router_agent(scored_lead: dict[str, Any]) -> dict[str, Any]:
"""
Third and final agent. Routes the lead based on score and tier.
"""
system_prompt = (
"You are a lead routing specialist. Call the route_to_sales tool with the "
"scored lead data. Always call the tool — do not make routing decisions yourself."
)
user_message = (
f"Route this scored lead:\n"
f"Name: {scored_lead.get('name')}\n"
f"Score: {scored_lead.get('score')}\n"
f"Tier: {scored_lead.get('tier')}\n"
f"Industry: {scored_lead.get('industry', 'unknown')}"
)
messages = [{"role": "user", "content": user_message}]
response = client.messages.create(
model=MODEL,
max_tokens=1024,
system=system_prompt,
tools=[TOOLS[2]], # Only expose route_to_sales to this agent
messages=messages
)
while response.stop_reason == "tool_use":
tool_results = []
for block in response.content:
if block.type == "tool_use":
tool_input = block.input
result = route_to_sales(**tool_input)
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": json.dumps(result)
})
messages.append({"role": "assistant", "content": response.content})
messages.append({"role": "user", "content": tool_results})
response = client.messages.create(
model=MODEL,
max_tokens=1024,
system=system_prompt,
tools=[TOOLS[2]],
messages=messages
)
for msg in messages:
if isinstance(msg["content"], list):
for item in msg["content"]:
if isinstance(item, dict) and item.get("type") == "tool_result":
return json.loads(item["content"])
return {"error": "Router agent failed to produce a result"}
Step 6: Build the Orchestrator Loop
The orchestrator is the brain that connects the three agents in sequence. It doesn't use Claude directly — it just passes data between agents and handles errors at each stage. This separation of concerns is what makes the system extensible. You can swap out any agent without touching the others.
lead_qualification.py (continued)# --- Orchestrator: Multi-Agent Pipeline ---
def run_lead_qualification_pipeline(lead_data: dict[str, Any]) -> dict[str, Any]:
"""
Orchestrates the full lead qualification pipeline across three agents.
Returns a complete qualification result including validation, score, and routing.
"""
print(f"\n{'='*55}")
print(f" LEAD QUALIFICATION PIPELINE")
print(f" Processing: