AI Agents in Practice: Sales Lead Qualification & Routing Agent (Prompts + Code)

Introduction

This pattern implements a Sales Lead Qualification & Routing Agent. It ingests inbound leads (web forms, chat, events), normalizes and scores them, checks ICP fit and readiness, and routes to the right owner (AE/SDR/CSM/Partner) with a tracked handoff. It never claims success unless the downstream CRM actions return receipts (lead created/updated, task assigned, email sent), enabling full auditability and fast feedback loops.

The Use Case

Go-to-market teams often drown in raw leads with inconsistent fields and poor signal. The agent standardizes company/person data, de-duplicates against CRM, enriches from approved sources, applies scoring and rules (territory, segment, product interest), and then creates/updates CRM objects plus a next-step task. For uncertain or low-signal cases, it asks for one clarification (e.g., company size or use case) before proceeding. All factual statements cite policy claims (routing policies, SLOs, ICP criteria).

Prompt Contract (agent interface)

# file: contracts/lead_router_v1.yaml
role: "LeadRouterAgent"
scope: >
  Normalize, enrich, score, dedup, and route inbound leads to the correct owner with a next action.
  Ask once if critical fields are missing (email, full_name, company, country, interest).
  Propose tool calls; never assert success without a CRM receipt.
output:
  type: object
  required: [summary, decision, lead, routing, citations, next_steps, tool_proposals]
  properties:
    summary: {type: string, maxWords: 80}
    decision: {type: string, enum: ["route","need_more_info","reject","hold"]}
    lead:
      type: object
      required: [email, full_name, company, country, interest, score]
      properties:
        email: {type: string}
        full_name: {type: string}
        company: {type: string}
        country: {type: string}
        interest: {type: string}
        score: {type: integer}
    routing:
      type: object
      required: [owner_type, owner_id, queue, sla_minutes]
      properties:
        owner_type: {type: string, enum: ["AE","SDR","Partner","CSM"]}
        owner_id: {type: string}
        queue: {type: string}
        sla_minutes: {type: integer}
    citations: {type: array, items: {type: string}}
    next_steps: {type: array, items: {type: string}, maxItems: 6}
    tool_proposals:
      type: array
      items:
        type: object
        required: [name, args, preconditions, idempotency_key]
        properties:
          name: {type: string, enum: [Normalize, Enrich, ScoreLead, CheckDuplicate, RouteLead, CreateTask, SendHandoffEmail]}
          args: {type: object}
          preconditions: {type: string}
          idempotency_key: {type: string}
policy_id: "gtm_routing_policy.v7"
citation_rule: "1–2 minimal-span claim_ids per factual sentence"
decoding:
  narrative: {top_p: 0.9, temperature: 0.7}
  bullets:   {top_p: 0.82, temperature: 0.45}

Example claims (context given to the model)

[
  {"claim_id":"policy:icp:employee_min","text":"ICP companies have >= 100 employees.","effective_date":"2025-05-01","source_id":"doc:gtm_routing_policy_v7","span":">= 100 employees"},
  {"claim_id":"policy:routing:territory","text":"Leads route by country to regional SDR queues within 15 minutes.","effective_date":"2025-05-01","source_id":"doc:gtm_routing_policy_v7","span":"regional SDR queues within 15 minutes"},
  {"claim_id":"policy:routing:dedup","text":"If a matching contact exists, update and assign to the current account owner.","effective_date":"2025-05-01","source_id":"doc:gtm_routing_policy_v7","span":"assign to the current account owner"}
]

Tool Interfaces (typed, with receipts)

# tools.py
from pydantic import BaseModel, EmailStr
from typing import Optional, Dict
from datetime import datetime

class NormalizeArgs(BaseModel):
    email: EmailStr
    full_name: str
    company: str
    country: str
    interest: str

class EnrichArgs(BaseModel):
    email: EmailStr
    company: str

class ScoreLeadArgs(BaseModel):
    signals: Dict[str, int]  # e.g., {"employees": 200, "site_visits": 5, "use_case_match": 1}

class CheckDuplicateArgs(BaseModel):
    email: EmailStr

class RouteLeadArgs(BaseModel):
    email: EmailStr
    country: str
    score: int

class CreateTaskArgs(BaseModel):
    owner_id: str
    lead_email: EmailStr
    summary: str
    due_minutes: int

class SendHandoffEmailArgs(BaseModel):
    owner_id: str
    lead_email: EmailStr
    template_id: str

class ToolReceipt(BaseModel):
    tool: str
    ok: bool
    ref: str
    message: str = ""
    data: Optional[Dict] = None

# adapters.py (demo logic)
from tools import *
from datetime import datetime

CRM = {"contacts": {"[email protected]": {"owner_id":"AE-44", "account_id":"ACC-9"}}}
ACCOUNTS = {"ACME INC": {"employees": 420, "segment":"MM"}}
TERRITORY = {"US":"SDR-US-1","CA":"SDR-CA-2","DE":"SDR-EMEA-3"}

def normalize(a: NormalizeArgs) -> ToolReceipt:
    email = a.email.lower().strip()
    company = a.company.upper().strip()
    country = a.country.upper().strip()
    return ToolReceipt(tool="Normalize", ok=True, ref=f"norm-{email}",
                       data={"email":email,"full_name":a.full_name.strip(),"company":company,"country":country,"interest":a.interest.strip()})

def enrich(a: EnrichArgs) -> ToolReceipt:
    employees = ACCOUNTS.get(a.company.upper(), {}).get("employees", 50)
    return ToolReceipt(tool="Enrich", ok=True, ref=f"enrich-{a.email}", data={"employees": employees, "site_visits": 3})

def score_lead(a: ScoreLeadArgs) -> ToolReceipt:
    score = min(100, a.signals.get("employees",0)//3 + a.signals.get("site_visits",0)*5 + a.signals.get("use_case_match",0)*30)
    return ToolReceipt(tool="ScoreLead", ok=True, ref="score", data={"score": score})

def check_duplicate(a: CheckDuplicateArgs) -> ToolReceipt:
    if a.email in CRM["contacts"]:
        owner_id = CRM["contacts"][a.email]["owner_id"]
        return ToolReceipt(tool="CheckDuplicate", ok=True, ref="dup-found", data={"duplicate": True, "owner_id": owner_id})
    return ToolReceipt(tool="CheckDuplicate", ok=True, ref="dup-none", data={"duplicate": False})

def route_lead(a: RouteLeadArgs) -> ToolReceipt:
    owner = TERRITORY.get(a.country, "SDR-ROW-0")
    queue = f"queue-{owner}"
    sla = 15 if a.country in TERRITORY else 30
    return ToolReceipt(tool="RouteLead", ok=True, ref=f"route-{a.email}", data={"owner_id": owner, "queue": queue, "sla_minutes": sla})

def create_task(a: CreateTaskArgs) -> ToolReceipt:
    return ToolReceipt(tool="CreateTask", ok=True, ref=f"task-{a.owner_id}", data={"due_minutes": a.due_minutes})

def send_handoff_email(a: SendHandoffEmailArgs) -> ToolReceipt:
    return ToolReceipt(tool="SendHandoffEmail", ok=True, ref=f"mail-{a.owner_id}", data={"template_id": a.template_id})

Agent Loop (proposal → verification → execution → receipts)

# agent_lead_router.py
import uuid, json
from typing import Any, Dict, List
from tools import *
from adapters import *

ALLOWED_TOOLS = {"Normalize","Enrich","ScoreLead","CheckDuplicate","RouteLead","CreateTask","SendHandoffEmail"}

def new_idem(): return f"idem-{uuid.uuid4()}"

def verify_proposal(p: Dict[str, Any]) -> str:
    need = {"name","args","preconditions","idempotency_key"}
    if not need.issubset(p): return "Missing proposal fields"
    if p["name"] not in ALLOWED_TOOLS: return "Tool not allowed"
    return ""

def execute(p: Dict[str, Any]) -> ToolReceipt:
    n, a = p["name"], p["args"]
    if n=="Normalize":         return normalize(NormalizeArgs(**a))
    if n=="Enrich":            return enrich(EnrichArgs(**a))
    if n=="ScoreLead":         return score_lead(ScoreLeadArgs(**a))
    if n=="CheckDuplicate":    return check_duplicate(CheckDuplicateArgs(**a))
    if n=="RouteLead":         return route_lead(RouteLeadArgs(**a))
    if n=="CreateTask":        return create_task(CreateTaskArgs(**a))
    if n=="SendHandoffEmail":  return send_handoff_email(SendHandoffEmailArgs(**a))
    return ToolReceipt(tool=n, ok=False, ref="none", message="Unknown tool")

# --- Model shim returning a plan per contract (replace with your LLM call) ---
def call_model(contract_yaml: str, claims: List[Dict[str,Any]], lead: Dict[str,Any]) -> Dict[str,Any]:
    # pretend enrichment says employees=420 -> strong ICP fit
    return {
      "summary": f"Lead from {lead['company']} normalized, enriched, scored, and routed.",
      "decision": "route",
      "lead": {"email": lead["email"], "full_name": lead["full_name"], "company": lead["company"],
               "country": lead["country"], "interest": lead["interest"], "score": 88},
      "routing": {"owner_type":"SDR","owner_id":"SDR-US-1","queue":"queue-SDR-US-1","sla_minutes":15},
      "citations": ["policy:icp:employee_min","policy:routing:territory","policy:routing:dedup"],
      "next_steps": ["Normalize fields","Enrich company/person","Score lead","Check duplicate","Route + create task + email"],
      "tool_proposals": [
        {"name":"Normalize","args":lead, "preconditions":"Standardize email/company/country/interest.","idempotency_key": new_idem()},
        {"name":"Enrich","args":{"email":lead["email"],"company":lead["company"]}, "preconditions":"Append employees and signals.","idempotency_key": new_idem()},
        {"name":"ScoreLead","args":{"signals":{"employees":420,"site_visits":3,"use_case_match":1}}, "preconditions":"Compute score.","idempotency_key": new_idem()},
        {"name":"CheckDuplicate","args":{"email":lead["email"]}, "preconditions":"See if contact exists.","idempotency_key": new_idem()},
        {"name":"RouteLead","args":{"email":lead["email"],"country":lead["country"],"score":88}, "preconditions":"Map to territory/queue.","idempotency_key": new_idem()},
        {"name":"CreateTask","args":{"owner_id":"SDR-US-1","lead_email":lead["email"],"summary":"Call within SLA","due_minutes":15},
         "preconditions":"Ensure actionable follow-up.","idempotency_key": new_idem()},
        {"name":"SendHandoffEmail","args":{"owner_id":"SDR-US-1","lead_email":lead["email"],"template_id":"welcome-001"},
         "preconditions":"Send intro mail to lead.","idempotency_key": new_idem()}
      ]
    }

def render_response(plan: Dict[str,Any], receipts: List[ToolReceipt]) -> str:
    idx = {r.tool:r for r in receipts}
    lines = [plan["summary"], ""]
    lines.append(f"Decision: {plan['decision']}")
    lines.append(f"Lead: {plan['lead']}")
    lines.append(f"Routing: {plan['routing']}")
    lines.append("\nNext steps:")
    for s in plan["next_steps"]: lines.append(f"• {s}")
    if idx.get("CreateTask") and idx["CreateTask"].ok:
        lines.append(f"\nTask created: {idx['CreateTask'].ref}, due in {idx['CreateTask'].data['due_minutes']} minutes")
    if idx.get("SendHandoffEmail") and idx["SendHandoffEmail"].ok:
        lines.append(f"Handoff email sent: {idx['SendHandoffEmail'].ref}")
    lines.append("\nCitations: " + ", ".join(plan["citations"]))
    return "\n".join(lines)

def handle(lead: Dict[str,Any]) -> str:
    contract = open("contracts/lead_router_v1.yaml").read()
    claims: List[Dict[str,Any]] = []  # load routing + ICP claims
    plan = call_model(contract, claims, lead)
    receipts: List[ToolReceipt] = []
    for prop in plan["tool_proposals"]:
        reason = verify_proposal(prop)
        if reason:
            receipts.append(ToolReceipt(tool=prop["name"], ok=False, ref="blocked", message=reason)); continue
        r = execute(prop)
        receipts.append(r)
        # halt if routing failed before follow-ups
        if not r.ok and prop["name"] in {"RouteLead"}: break
    return render_response(plan, receipts)

if __name__ == "__main__":
    example_lead = {
      "email":"[email protected]",
      "full_name":"Alex Carter",
      "company":"ACME INC",
      "country":"US",
      "interest":"Evaluating enterprise plan"
    }
    print(handle(example_lead))

The Prompt You’d Send to the Model (concise and testable)

System:
You are LeadRouterAgent. Follow the contract:
- Ask once if email, full_name, company, country, or interest is missing.
- Cite 1–2 claim_ids per factual sentence using provided claims.
- Propose tools; never assert success without a CRM receipt.
- Output JSON with keys: summary, decision, lead{}, routing{}, citations[], next_steps[], tool_proposals[].

Claims (eligible only):
[ ... JSON array of ICP, dedup, and routing policies like above ... ]

User:
New lead:
{"email":"[email protected]","full_name":"Alex Carter","company":"ACME INC","country":"US","interest":"Evaluating enterprise plan"}

How to adapt quickly

Connect Normalize/Enrich to your sanctioned data vendors, map RouteLead to your territory/round-robin logic, and wire CreateTask/SendHandoffEmail to your CRM and messaging systems. Keep idempotency on all writes (to avoid duplicate contacts or tasks), enforce SLA receipts (timestamps), and log feature flags + canary + rollback for routing rule changes. Add a validator to check schema, lexicon, locale, and citation coverage before executing tool proposals.