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.