How to Build an AI Sales Agent That Finds Leads and Automates Follow-Ups

ai-sales-agent/
├── app/
│   ├── __init__.py
│   ├── schemas.py
│   ├── scorer.py
│   └── outbox.py
├── tests/
│   └── test_sales_agent.py
└── requirements.txt

from enum import Enum
from typing import List, Optional
from pydantic import BaseModel, Field

class LeadTier(str, Enum):
    TIER_1_HOT = "TIER_1_HOT"
    TIER_2_WARM = "TIER_2_WARM"
    TIER_3_NURTURE = "TIER_3_NURTURE"
    DISQUALIFIED = "DISQUALIFIED"

class EnrichedLead(BaseModel):
    lead_id: str
    email: str
    full_name: str
    job_title: str
    company_name: str
    company_domain: str
    employee_count: int = Field(..., ge=1)
    industry: str
    recent_company_news: str = ""
    has_consent_or_inbound_intent: bool = True
    is_unsubscribed: bool = False

class FollowUpStep(BaseModel):
    day_offset: int = Field(..., ge=0, le=30)
    subject: str
    body_text: str
    unsubscribe_footer: str

class QualifiedLeadPlan(BaseModel):
    lead_id: str
    icp_score: int = Field(..., ge=0, le=100)
    tier: LeadTier
    qualification_reasons: List[str]
    sequence_steps: List[FollowUpStep]

from typing import Tuple, List
from app.schemas import EnrichedLead, LeadTier, FollowUpStep, QualifiedLeadPlan

TARGET_INDUSTRIES = {"saas", "fintech", "ai", "healthtech", "logistics"}
DECISION_MAKER_KEYWORDS = ("cto", "vp", "director", "head of", "founder", "chief", "principal")
FREE_WEBMAIL_DOMAINS = {"gmail.com", "yahoo.com", "hotmail.com", "outlook.com", "aol.com"}

def score_lead_icp(lead: EnrichedLead) -> Tuple[int, LeadTier, List[str]]:
    """Calculates a deterministic 0-100 ICP score and qualification tier."""
    if lead.is_unsubscribed or not lead.has_consent_or_inbound_intent:
        return 0, LeadTier.DISQUALIFIED, ["Lead is unsubscribed or lacks lawful outreach basis."]

    score = 0
    reasons: List[str] = []

    domain = lead.email.split("@")[-1].lower()
    if domain not in FREE_WEBMAIL_DOMAINS:
        score += 20
        reasons.append("Verified corporate email domain (+20)")

    if 25 <= lead.employee_count <= 2000:
        score += 30
        reasons.append(f"Ideal company size: {lead.employee_count} employees (+30)")
    elif lead.employee_count > 2000:
        score += 20
        reasons.append("Enterprise company size (+20)")

    title_lower = lead.job_title.lower()
    if any(k in title_lower for k in DECISION_MAKER_KEYWORDS):
        score += 30
        reasons.append(f"Decision-maker title '{lead.job_title}' (+30)")

    if lead.industry.strip().lower() in TARGET_INDUSTRIES:
        score += 20
        reasons.append(f"Target industry '{lead.industry}' (+20)")

    if score >= 75:
        tier = LeadTier.TIER_1_HOT
    elif score >= 50:
        tier = LeadTier.TIER_2_WARM
    elif score >= 25:
        tier = LeadTier.TIER_3_NURTURE
    else:
        tier = LeadTier.DISQUALIFIED

    return score, tier, reasons

def build_compliant_followup_step(
    day_offset: int,
    subject: str,
    body_text: str,
    unsubscribe_url: str,
    physical_address: str,
) -> FollowUpStep:
    """Appends mandatory anti-spam unsubscribe and physical address footer."""
    if len(body_text.split()) > 150:
        raise ValueError("Sales outreach body exceeds 150-word brevity guardrail.")
    footer = f"\n\n---\n{physical_address}\nOpt out of future messages: {unsubscribe_url}"
    return FollowUpStep(
        day_offset=day_offset,
        subject=subject.strip(),
        body_text=body_text.strip(),
        unsubscribe_footer=footer,
    )

import json
import sqlite3
from typing import List, Dict, Any

def init_outbox_db(conn: sqlite3.Connection) -> None:
    conn.executescript(
        """
        CREATE TABLE IF NOT EXISTS leads (
            lead_id TEXT PRIMARY KEY,
            email TEXT UNIQUE NOT NULL,
            icp_score INTEGER NOT NULL,
            tier TEXT NOT NULL
        );
        CREATE TABLE IF NOT EXISTS outbox_events (
            event_id INTEGER PRIMARY KEY AUTOINCREMENT,
            lead_id TEXT NOT NULL,
            event_type TEXT NOT NULL,
            payload_json TEXT NOT NULL,
            status TEXT NOT NULL DEFAULT 'PENDING',
            retry_count INTEGER NOT NULL DEFAULT 0
        );
        """
    )
    conn.commit()

def save_lead_with_outbox_atomic(
    conn: sqlite3.Connection,
    lead_id: str,
    email: str,
    icp_score: int,
    tier: str,
    sequence_payload: Dict[str, Any],
) -> None:
    """Atomically inserts the qualified lead and queues the outbound CRM/email sync event."""
    with conn:
        conn.execute(
            "INSERT INTO leads (lead_id, email, icp_score, tier) VALUES (?, ?, ?, ?)",
            (lead_id, email, icp_score, tier),
        )
        conn.execute(
            "INSERT INTO outbox_events (lead_id, event_type, payload_json, status) VALUES (?, ?, ?, 'PENDING')",
            (lead_id, "CRM_AND_CADENCE_SYNC", json.dumps(sequence_payload)),
        )

import sqlite3
import pytest
from app.schemas import EnrichedLead, LeadTier
from app.scorer import score_lead_icp, build_compliant_followup_step
from app.outbox import init_outbox_db, save_lead_with_outbox_atomic

def test_icp_scoring_and_suppression():
    hot_lead = EnrichedLead(
        lead_id="L-100",
        email="cto@fintech-corp.io",
        full_name="Rohan Mehta",
        job_title="VP of Engineering",
        company_name="Fintech Corp",
        company_domain="fintech-corp.io",
        employee_count=180,
        industry="Fintech",
    )
    score, tier, _ = score_lead_icp(hot_lead)
    assert score == 100
    assert tier == LeadTier.TIER_1_HOT

    unsub_lead = hot_lead.model_copy(update={"is_unsubscribed": True})
    u_score, u_tier, _ = score_lead_icp(unsub_lead)
    assert u_score == 0
    assert u_tier == LeadTier.DISQUALIFIED

def test_transactional_outbox_atomic_commit():
    conn = sqlite3.connect(":memory:")
    init_outbox_db(conn)
    save_lead_with_outbox_atomic(
        conn,
        lead_id="L-100",
        email="cto@fintech-corp.io",
        icp_score=100,
        tier="TIER_1_HOT",
        sequence_payload={"steps": 3},
    )
    rows = conn.execute("SELECT lead_id, status FROM outbox_events").fetchall()
    assert len(rows) == 1
    assert rows[0] == ("L-100", "PENDING")