AI Lead Enrichment with a Scheduled Python Agent¶
AI lead enrichment should not mean collecting everything an API can return. A production pipeline needs tighter boundaries. It should accept consented records, request only necessary fields, validate every response, and preserve where each value came from. It also needs safe retries and deterministic output.
This guide builds that pipeline as a scheduled Python agent. The result is a JSON artifact you can inspect, audit, and import into a CRM without silently overwriting good data.
Define a Consent-First AI Lead Enrichment Pipeline¶
Lead enrichment adds information to records you already hold. Depending on the provider, that information can include a job title, company size, revenue, industry, LinkedIn profile, or phone number.
Providers may aggregate data from public registries, social profiles, and company filings. Some also use AI to extract and verify information in real time. That availability does not mean your agent should request every possible field.
Start with an explicit input contract:
{
"lead_id": "lead_acme_alice",
"email": "alice@example.com",
"company_domain": "example.com",
"consent": {
"status": "granted",
"recorded_at": "2026-09-02T08:30:00Z",
"source": "product_signup"
}
}
The agent should reject a record when:
- Consent is absent or not granted.
- The consent timestamp is missing.
- There is no stable internal lead ID.
- The input contains no approved enrichment key.
- The requested operation exceeds your documented purpose.
Keep the consent check before any provider call:
from datetime import datetime
from pydantic import BaseModel, EmailStr, ValidationError
class Consent(BaseModel):
status: str
recorded_at: datetime
source: str
class LeadInput(BaseModel):
lead_id: str
email: EmailStr | None = None
company_domain: str | None = None
consent: Consent
def assert_enrichable(self) -> None:
if self.consent.status != "granted":
raise ValueError("Enrichment blocked: consent not granted")
if not self.email and not self.company_domain:
raise ValueError("Enrichment blocked: no approved lookup key")
Install the dependencies with:
Do not copy the raw provider payload into your CRM. Select the fields your workflow needs. If the sales workflow only uses company industry and employee range, discard unrelated personal attributes before storage.
Consent requirements vary by context. Treat this design as a technical control, not legal advice.
Design the Python Architecture from Input to CRM-Ready Output¶
A maintainable enrichment job is a pipeline, not one large API function:
consented lead intake
-> normalization
-> provider request
-> response validation
-> quality gate
-> provenance log
-> JSON/CSV export or CRM upsert
This separation matters because enrichment datasets can contain company hierarchies, technology stacks, revenue, buying intent, engagement signals, and risk forecasts. Each category needs its own validation and storage decision.
Represent the workflow with small functions:
def process_lead(raw_lead: dict) -> dict:
lead = LeadInput.model_validate(raw_lead)
lead.assert_enrichable()
request = build_provider_request(lead)
raw_response = call_provider(request)
enrichment = validate_provider_response(raw_response)
checked = apply_quality_gate(lead, enrichment)
return build_output_record(lead, checked)
Normalization should make equivalent inputs consistent without changing their meaning:
def normalize_domain(value: str | None) -> str | None:
if value is None:
return None
domain = value.strip().lower()
for prefix in ("https://", "http://", "www."):
if domain.startswith(prefix):
domain = domain.removeprefix(prefix)
return domain.rstrip("/")
Keep orchestration separate from provider-specific adapters. You can then replace a provider without changing consent enforcement, output formatting, or CRM upsert logic.
This is the same principle used in a broader multi-step AI agent workflow: each stage has a clear input, output, and failure mode.
Validate Every Enrichment API Response Before Saving It¶
An HTTP success response only means the provider returned something. It does not mean the payload is complete, current, or safe to write.
Define the expected schema:
from datetime import datetime
from enum import Enum
from pydantic import BaseModel, EmailStr, HttpUrl, Field
class ValidationStatus(str, Enum):
accepted = "accepted"
rejected = "rejected"
class EnrichedCompany(BaseModel):
name: str
domain: str
industry: str | None = None
employee_range: str | None = None
linkedin_url: HttpUrl | None = None
class ProviderResult(BaseModel):
provider_record_id: str
retrieved_at: datetime
company: EnrichedCompany
work_email: EmailStr | None = None
job_title: str | None = None
confidence: float | None = Field(default=None, ge=0, le=1)
evidence: list[str] = Field(default_factory=list)
Pydantic now rejects invalid email addresses, malformed URLs, unsupported confidence values, and missing required fields. Add business-level checks after schema validation:
def apply_quality_gate(
lead: LeadInput,
result: ProviderResult,
) -> ProviderResult:
expected = normalize_domain(lead.company_domain)
received = normalize_domain(result.company.domain)
if expected and received != expected:
raise ValueError(
f"Domain mismatch: expected {expected}, received {received}"
)
if not result.provider_record_id:
raise ValueError("Missing provider record identifier")
if not result.company.name.strip():
raise ValueError("Empty company name")
return result
Reject contradictory records instead of guessing. For example, do not merge a provider response for a different company just because the contact names look similar. Send ambiguous results to a review or rejection artifact.
When comparing providers, evaluate accuracy and freshness, contact and company coverage, enrichment depth, integration options, intent signals, and fit for your intended users. Keep the comparison tied to a fixed test fixture so each provider receives the same consented inputs.
Handle Rate Limits, Timeouts, Invalid Data, and Safe Retries¶
Retry only failures that may be temporary. A timeout or rate-limit response can be retried. An invalid consent record or schema mismatch should not be.
Here is a bounded HTTP client using exponential backoff, jitter, and Retry-After support:
import os
import random
import time
import httpx
RETRYABLE_STATUS = {429, 500, 502, 503, 504}
def call_provider(payload: dict, run_id: str) -> dict:
url = os.environ["ENRICHMENT_API_URL"]
token = os.environ["ENRICHMENT_API_TOKEN"]
headers = {
"Authorization": f"Bearer {token}",
"Idempotency-Key": run_id,
}
with httpx.Client(timeout=15.0) as client:
for attempt in range(4):
try:
response = client.post(url, json=payload, headers=headers)
except httpx.TimeoutException:
if attempt == 3:
raise
time.sleep((2**attempt) + random.random())
continue
if response.status_code not in RETRYABLE_STATUS:
response.raise_for_status()
return response.json()
if attempt == 3:
response.raise_for_status()
retry_after = response.headers.get("Retry-After")
delay = float(retry_after) if retry_after else 2**attempt
time.sleep(delay + random.random())
raise RuntimeError("Provider request ended unexpectedly")
The retry count and timeout above are example operating values. Tune them to the provider contract and your job’s runtime budget.
Build a stable run ID from the schedule window and lead ID:
import hashlib
def make_run_id(schedule_window: str, lead_id: str) -> str:
value = f"{schedule_window}:{lead_id}"
return hashlib.sha256(value.encode()).hexdigest()
Use that ID for the provider request, logs, artifacts, and CRM upsert. A repeated invocation can then update the same logical operation instead of creating another record. See the deeper guide to idempotent scheduled Python jobs.
After bounded retries fail, write the input reference and structured error to a dead-letter artifact. Do not discard it, and do not loop forever.
Produce Auditable Structured Records and Quality Gates¶
Enrichment tools automate research that would otherwise involve manual investigation, using proprietary databases or third-party sources. Your output should retain enough context to explain each accepted field.
A useful record looks like this:
{
"run_id": "stable-hash",
"lead_id": "lead_acme_alice",
"consent": {
"status": "granted",
"recorded_at": "2026-09-02T08:30:00Z",
"source": "product_signup"
},
"normalized": {
"company_domain": "example.com"
},
"enrichment": {
"company_name": "Example",
"industry": "Software"
},
"provenance": {
"company_name": {
"provider": "configured_provider",
"provider_record_id": "provider-record",
"retrieved_at": "2026-09-02T09:00:00Z"
}
},
"validation_status": "accepted",
"errors": []
}
Track provenance at field level when fields can come from different sources. Keep rejected records in a separate artifact with machine-readable error codes such as consent_missing, schema_invalid, domain_mismatch, or provider_timeout.
Before CRM import, calculate operational checks from the current batch:
- Accepted and rejected record counts.
- Missing-field counts.
- Records failing freshness rules.
- Conflicts between input and enriched company domains.
- Provider failures grouped by error code.
Never overwrite a trusted CRM field merely because the provider returned a non-null value. Define per-field merge rules. A common pattern is to fill empty fields automatically while routing conflicting values for review.
Structured logs should carry the run ID, lead ID, stage, provider identifier, and outcome. Avoid placing raw personal data or API tokens in those logs. The Python agent observability guide covers the surrounding logging and monitoring pattern.
Schedule, Secure, Test, and Operate the Python Agent¶
Run the pipeline from a CLI entry point:
import json
import sys
from pathlib import Path
def main() -> int:
source = Path("input/leads.json")
destination = Path("output/enriched.json")
leads = json.loads(source.read_text())
results = []
for raw_lead in leads:
try:
results.append(process_lead(raw_lead))
except Exception as exc:
results.append({
"lead_id": raw_lead.get("lead_id"),
"validation_status": "rejected",
"errors": [type(exc).__name__],
})
destination.parent.mkdir(parents=True, exist_ok=True)
destination.write_text(json.dumps(results, indent=2, default=str))
return 0
if __name__ == "__main__":
raise SystemExit(main())
For cron, prevent overlapping runs with a lock:
Inject API credentials through environment variables. Do not commit them to the repository or place them in the input file. Give the job access only to the required provider, input location, output location, and CRM operation. The production agent security guide expands on secrets, isolation, and least privilege.
Before scheduling, run fixture-based tests for:
- Missing API secrets.
- Consent that is absent or denied.
- Invalid provider schemas.
- Timeouts and rate limits.
- Duplicate run IDs.
- Domain contradictions.
- Unserializable output.
- Failed artifact writes.
Use a dry-run mode that reads fixtures and writes an artifact without calling the provider or CRM. Then inspect the exit code and output locally. This pre-scheduling Python agent test checklist provides a production-oriented test path.
Set retention rules for input, output, error artifacts, and logs. Keep only what the workflow needs. Alert on failed runs, missing output, unusual rejection patterns, and repeated provider errors.
Conclusion¶
A reliable AI lead enrichment agent is mostly controlled data engineering. Consent is checked before network access. Provider responses pass typed and business-level validation. Retries are bounded and idempotent. Every accepted field keeps its provenance. Rejected records remain inspectable.
Once the pipeline behaves correctly in local dry runs, deploy it as an isolated scheduled job with injected secrets, non-overlapping execution, persistent artifacts, and alerts. A managed agent runtime such as HollowHost can provide that operating layer while your repository remains focused on the enrichment rules and CRM-ready output.