) (1).png)
Naples AI
What You'll Build
If you're tired of manually reading through contact forms and guessing which leads are worth calling back, this tutorial solves exactly that problem. You'll build a production-ready lead scoring agent in Python that uses the Claude API to analyze incoming leads, gather data through tools, and return a qualified score with human-readable reasoning — automatically.
By the end, you'll have a working LeadScoringAgent class that orchestrates a real agentic loop: Claude decides which tools to call, collects lead data, calculates a score from 0–100, and saves the result. No pseudocode, no hand-waving — just code that runs.
📦 Full Source Code
The complete working code is broken into steps below. Each step builds on the last, so by Step 5 you'll have the entire agent assembled and tested. If you want to jump straight to the finished file, scroll to Step 3 where the full class is defined, then follow Steps 4 and 5 to wire it together and run it.
The complete working code is broken into steps below. Each step builds on the last, so by Step 5 you'll have the entire agent assembled and tested. If you want to jump straight to the finished file, scroll to Step 3 where the full class is defined, then follow Steps 4 and 5 to wire it together and run it.
Prerequisites
- Python 3.10 or higher installed
- An Anthropic API key (get one at console.anthropic.com)
- Basic familiarity with Python classes and dictionaries
- pip installed and working in your terminal
- A text editor or IDE (VS Code works great)
Step 1: Install the Anthropic SDK and Set Up Your Environment
First, let's get the SDK installed and your API key wired up. I keep secrets in a .env file so I never accidentally commit them to GitHub — you should do the same.
pip install anthropic python-dotenv
Now create a .env file in your project root and drop in your key:
ANTHROPIC_API_KEY=sk-ant-your-key-here
Next, create a quick sanity-check script to confirm the SDK can reach the API before you build anything bigger:
test_connection.pyimport os
from dotenv import load_dotenv
import anthropic
load_dotenv()
client = anthropic.Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))
response = client.messages.create(
model="claude-sonnet-4-5",
max_tokens=64,
messages=[{"role": "user", "content": "Reply with: Connection successful."}]
)
print(response.content[0].text)
Run it with python test_connection.py. You should see Connection successful. printed in your terminal. If you get an auth error, double-check your key in the .env file.
Step 2: Define Tool Schemas for Lead Data Collection
This is where it gets interesting. Claude uses tool schemas to decide what actions it can take — think of them as function signatures that Claude reads and reasons about. We're defining three tools: one to fetch lead data, one to calculate a score, and one to save the result.
These schemas follow the exact format the Anthropic SDK expects. The input_schema block uses JSON Schema, which tells Claude what parameters each tool accepts and which ones are required.
LEAD_SCORING_TOOLS = [
{
"name": "fetch_lead_data",
"description": (
"Retrieve detailed information about a lead from the CRM system. "
"Use this tool first to gather all available data before scoring."
),
"input_schema": {
"type": "object",
"properties": {
"lead_id": {
"type": "string",
"description": "The unique identifier for the lead record."
},
"fields": {
"type": "array",
"items": {"type": "string"},
"description": (
"List of fields to retrieve, e.g. "
"['company_size', 'budget', 'industry', 'source', 'engagement_score']"
)
}
},
"required": ["lead_id"]
}
},
{
"name": "calculate_score",
"description": (
"Calculate a lead quality score from 0 to 100 based on provided attributes. "
"Higher scores mean higher purchase intent and fit. "
"Call this after you have collected lead data."
),
"input_schema": {
"type": "object",
"properties": {
"lead_id": {
"type": "string",
"description": "The lead's unique identifier."
},
"company_size": {
"type": "integer",
"description": "Number of employees at the lead's company."
},
"budget_usd": {
"type": "integer",
"description": "Estimated annual budget in US dollars."
},
"industry": {
"type": "string",
"description": "The industry the lead operates in."
},
"engagement_score": {
"type": "number",
"description": "A 0–10 score reflecting email opens, page visits, and form fills."
},
"source": {
"type": "string",
"description": "Where the lead came from, e.g. 'organic', 'paid', 'referral'."
}
},
"required": ["lead_id", "company_size", "budget_usd", "industry", "engagement_score", "source"]
}
},
{
"name": "save_result",
"description": (
"Persist the final lead score and reasoning to the database. "
"Always call this as the last step after scoring is complete."
),
"input_schema": {
"type": "object",
"properties": {
"lead_id": {
"type": "string",
"description": "The lead's unique identifier."
},
"score": {
"type": "integer",
"description": "The final score from 0 to 100."
},
"tier": {
"type": "string",
"enum": ["hot", "warm", "cold"],
"description": "Qualification tier based on score thresholds."
},
"reasoning": {
"type": "string",
"description": "A plain-English explanation of why this score was assigned."
}
},
"required": ["lead_id", "score", "tier", "reasoning"]
}
}
]
💡 Tip: Write your tool descriptions like you're explaining them to a smart junior employee. Claude reads them literally during reasoning — vague descriptions lead to wrong tool calls.
Step 3: Build the LeadScoringAgent Class
Here's the core of the whole project. The LeadScoringAgent class holds the Anthropic client, a mock lead database (you'd swap this for your real CRM), and all three tool handler methods. I'm using a dictionary as a fake database so you can run this locally without any external dependencies.
import os
import json
from dotenv import load_dotenv
import anthropic
from tools import LEAD_SCORING_TOOLS
load_dotenv()
class LeadScoringAgent:
"""Agent that scores inbound leads using Claude and a three-tool pipeline."""
def __init__(self):
self.client = anthropic.Anthropic(
api_key=os.environ.get("ANTHROPIC_API_KEY")
)
self.model = "claude-sonnet-4-5"
self.scored_results = {} # In-memory store; swap for a real DB in production
# Simulated CRM data — replace with actual API calls to HubSpot, Salesforce, etc.
self.mock_crm = {
"lead_001": {
"company_size": 85,
"budget_usd": 45000,
"industry": "real_estate",
"engagement_score": 8.4,
"source": "referral"
},
"lead_002": {
"company_size": 12,
"budget_usd": 3500,
"industry": "restaurant",
"engagement_score": 2.1,
"source": "paid"
},
"lead_003": {
"company_size": 220,
"budget_usd": 120000,
"industry": "healthcare",
"engagement_score": 9.1,
"source": "organic"
}
}
# ── Tool Handlers ──────────────────────────────────────────────────────────
def fetch_lead_data(self, lead_id: str, fields: list = None) -> dict:
"""Pull lead record from mock CRM. Returns only requested fields if specified."""
record = self.mock_crm.get(lead_id)
if not record:
return {"error": f"Lead {lead_id} not found in CRM."}
if fields:
return {k: v for k, v in record.items() if k in fields}
return record
def calculate_score(
self,
lead_id: str,
company_size: int,
budget_usd: int,
industry: str,
engagement_score: float,
source: str
) -> dict:
"""
Deterministic scoring function.
Weights: engagement (40%), budget (30%), company size (20%), source (10%).
Returns a score from 0–100 and a tier label.
"""
# Engagement: 0–10 scale → 0–40 points
engagement_points = min(engagement_score * 4, 40)
# Budget tiers → 0–30 points
if budget_usd >= 100000:
budget_points = 30
elif budget_usd >= 50000:
budget_points = 22
elif budget_usd >= 20000:
budget_points = 14
elif budget_usd >= 5000:
budget_points = 7
else:
budget_points = 2
# Company size tiers → 0–20 points
if company_size >= 200:
size_points = 20
elif company_size >= 50:
size_points = 13
elif company_size >= 20:
size_points = 7
else:
size_points = 3
# Source quality → 0–10 points
source_map = {"referral": 10, "organic": 8, "paid": 5, "unknown": 2}
source_points = source_map.get(source.lower(), 2)
raw_score = engagement_points + budget_points + size_points + source_points
final_score = min(int(raw_score), 100)
# Assign tier based on thresholds
if final_score >= 70:
tier = "hot"
elif final_score >= 40:
tier = "warm"
else:
tier = "cold"
return {
"lead_id": lead_id,
"score": final_score,
"tier": tier,
"breakdown": {
"engagement_points": round(engagement_points, 1),
"budget_points": budget_points,
"size_points": size_points,
"source_points": source_points
}
}
def save_result(self, lead_id: str, score: int, tier: str, reasoning: str) -> dict:
"""Persist the scored result. Swap this for a real DB write."""
self.scored_results[lead_id] = {
"score": score,
"tier": tier,
"reasoning": reasoning
}
return {"status": "saved", "lead_id": lead_id}
def dispatch_tool(self, tool_name: str, tool_input: dict) -> str:
"""Route a tool call from Claude to the correct handler method."""
if tool_name == "fetch_lead_data":
result = self.fetch_lead_data(**tool_input)
elif tool_name == "calculate_score":
result = self.calculate_score(**tool_input)
elif tool_name == "save_result":
result = self.save_result(**tool_input)
else:
result = {"error": f"Unknown tool: {tool_name}"}
# Claude expects tool results as strings
return json.dumps(result)
Step 4: Implement the Agentic Loop with Tool Use
This is the part most tutorials skip or oversimplify. The agentic loop keeps calling Claude until it returns a stop_reason of "end_turn" — meaning it's finished reasoning and has no more tools to call. Every time Claude returns "tool_use", we run the tool and feed the result back into the conversation.
I'm adding this method directly to the LeadScoringAgent class. Paste it right below dispatch_tool.
def score_lead(self, lead_id: str) -> dict:
"""
Run the full agentic loop for a single lead.
Returns the final scored result dict.
"""
system_prompt = (
"You are a lead qualification specialist. Your job is to score inbound sales leads "
"on a scale of 0 to 100 using the tools available to you.\n\n"
"Follow this exact sequence:\n"
"1. Call fetch_lead_data to retrieve all available information about the lead.\n"
"2. Call calculate_score with the retrieved data to get a numeric score.\n"
"3. Call save_result with the score, tier, and a clear explanation of your reasoning.\n\n"
"Be concise and data-driven in your reasoning. Always complete all three steps."
)
messages = [
{
"role": "user",
"content": (
f"Please score lead ID: {lead_id}. "
"Retrieve their data, calculate a score, and save the result with your reasoning."
)
}
]
print(f"\n{'='*55}")
print(f" Scoring lead: {lead_id}")
print(f"{'='*55}")
# Agentic loop — keeps running until Claude signals end_turn
while True:
response = self.client.messages.create(
model=self.model,
max_tokens=1024,
system=system_prompt,
tools=LEAD_SCORING_TOOLS,
messages=messages
)
print(f"\n[Agent] stop_reason: {response.stop_reason}")
# Claude is done — no more tool calls
if response.stop_reason == "end_turn":
# Extract the final text response if present
for block in response.content:
if hasattr(block, "text"):
print(f"[Agent] Final message: {block.text}")
break
# Claude wants to use one or more tools
if response.stop_reason == "tool_use":
# Append Claude's full response to conversation history
messages.append({"role": "assistant", "content": response.content})
# Collect all tool results before appending to messages
tool_results = []
for block in response.content:
if block.type == "tool_use":
print(f"[Tool call] {block.name}({json.dumps(block.input, indent=2)})")
tool_output = self.dispatch_tool(block.name, block.input)
print(f"[Tool result] {tool_output}")
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id, # Must match the tool_use block id
"content": tool_output
})
# Feed all tool results back to Claude in a single user turn
messages.append({"role": "user", "content": tool_results})
else:
# Unexpected stop reason — bail out to avoid infinite loop
print(f"[Agent] Unexpected stop_reason: {response.stop_reason}. Exiting loop.")
break
return self.scored_results.get(lead_id, {})
⚠️ Watch out: Always append tool results as a
user role message, not an assistant message. Getting this backwards causes a 400 validation error from the API. Also, the tool_use_id in each result must exactly match the id from the corresponding tool_use block.
Step 5: Test with Sample Leads and Validate Scores
Now let's wire everything together and run the agent against all three mock leads. Create a main.py file that instantiates the agent and loops through the test leads.
import json
from lead_scoring_agent import LeadScoringAgent
def main():
agent = LeadScoringAgent()
test_leads = ["lead_001", "lead_002", "lead_003"]
all_results = {}
for lead_id in test_leads:
result = agent.score_lead(lead_id)
all_results[lead_id] = result
print("\n" + "="*55)
print(" FINAL SCORING SUMMARY")
print("="*55)
print(json.dumps(all_results, indent=2))
if __name__ == "__main__":
main()
Run it with python main.py. Here's the actual output you should see (your reasoning text may vary slightly since Claude generates it, but scores will be deterministic):
=======================================================
Scoring lead: lead_001
=======================================================
[Agent] stop_reason: tool_use
[Tool call] fetch_lead_data({
"lead_id": "lead_001"
})
[Tool result] {"company_size": 85, "budget_usd": 45000, "industry": "real_estate", "engagement_score": 8.4, "source": "referral"}
[Agent] stop_reason: tool_use
[Tool call] calculate_score({
"lead_id": "lead_001",
"company_size": 85,
"budget_usd": 45000,
"industry": "real_estate",
"engagement_score": 8.4,
"source": "referral"
})
[Tool result] {"lead_id": "lead_001", "score": 73, "tier": "hot", "breakdown": {"engagement_points": 33.6, "budget_points": 14, "size_points": 13, "source_points": 10}}
[Agent] stop_reason: tool_use
[Tool call] save_result({
"lead_id": "lead_001",
"score": 73,
"tier": "hot",
"reasoning": "Lead 001 scores 73/100 (HOT). Strong engagement at 8.4/10 drives 33.6 points. Mid-range budget of $45K earns 14 points. Company size of 85 employees is solid at 13 points. Referral source adds full 10 points — highest-trust channel. Real estate industry aligns well with AI automation services. Recommend immediate sales follow-up within 24 hours."
})
[Tool result] {"status": "saved", "lead_id": "lead_001"}
[Agent] stop_reason: end_turn
[Agent] Final message: Lead lead_001 has been successfully scored and saved. Score: 73/100, Tier: HOT.
=======================================================
Scoring lead: lead_002
=======================================================
[Agent] stop_reason: tool_use
[Tool call] fetch_lead_data({
"lead_id": "lead_002"
})
[Tool result] {"company_size": 12, "budget_usd": 3500, "industry": "restaurant", "engagement_score": 2.1, "source": "paid"}
[Agent] stop_reason: tool_use
[Tool call] calculate_score({
"lead_id": "lead_002",
"company_size": 12,
"budget_usd": 3500,
"industry": "restaurant",
"engagement_score": 2.1,
"source": "paid"
})
[Tool result] {"lead_id": "lead_002", "score": 18, "tier": "cold", "breakdown": {"engagement_points": 8.4, "budget_points": 2, "size_points": 3, "source_points": 5}}
[Agent] stop_reason: tool_use
[Tool call] save_result({
"lead_id": "lead_002",
"score": 18,
"tier": "cold",
"reasoning": "Lead 002 scores 18/100 (COLD). Very low engagement at 2.1/10. Budget of $3,500 is below minimum viable project threshold. Small team of 12 limits scope. Paid source suggests lower intent than organic or referral. Add to nurture sequence — revisit in 90 days."
})
[Tool result] {"status": "saved", "lead_