How to Build a Lead Qualification Agent (With CrewAI and LeadOcean)

ICP = {
    "buyer_titles": ["VP Sales", "Head of Sales", "Director of Revenue Operations"],
    "buyer_levels": ["VP", "Director", "C-Team"],
    "must_have": "A named sales or RevOps leader with a title that owns pipeline",
    "disqualify": "Student, intern, advisor, or a role in HR, legal or finance",
    "accounts": ["acme.com", "example.org", "sample.io"],   # placeholder domains
}

curl -s https://api.leadocean.io/v1/enums -H "x-api-key: $LEADOCEAN_API_KEY"

import json
import os
import time

import requests
from crewai.tools import tool

BASE = "https://api.leadocean.io"
KEY = os.environ["LEADOCEAN_API_KEY"]
MAX_RECORDS = 25   # hard cap for one run
spent = 0
stopped = ""


def call(method, path, params=None, body=None):
    """Return (json, "") on success or (None, message) on failure."""
    global spent, stopped
    if stopped:
        return None, f"STOP: {stopped}"
    for attempt in range(4):
        r = requests.request(method, BASE + path, params=params, json=body,
                             headers={"x-api-key": KEY}, timeout=30)
        wait = int(r.headers.get("Retry-After", 1))
        if r.status_code == 503:
            time.sleep(2 ** attempt)
            continue
        if r.status_code == 429 and wait <= 5:
            time.sleep(wait)
            continue
        break
    if r.status_code == 401:
        stopped = "401: the API key is missing, wrong or revoked."
    elif r.status_code == 402:
        stopped = "402: the free records are spent. They do not reset."
    elif r.status_code == 429 and wait > 5:
        stopped = f"429: the account is paced. Retry-After is {wait}s."
    if stopped:
        return None, f"STOP: {stopped}"
    if r.status_code == 429:
        return None, "ERROR: still rate limited after retries."
    if r.status_code == 404:
        return None, "NOT_FOUND: nobody matched. Nothing was charged."
    if r.status_code >= 400:
        return None, f"ERROR {r.status_code}: {r.text[:300]}"
    data = r.json()
    meta = data.get("meta") or {}
    rows = data.get("data")
    spent += meta.get("credits", len(rows) if isinstance(rows, list) else 1)
    return data, ""

@tool("List people at company")
def list_people(domain: str, limit: int = 5) -> str:
    """Costs 1 record per person returned. Pass a company domain such as acme.com.
    Returns person_id, name, title and job level for up to 5 people.
    Use it once per account, then judge the titles."""
    n = min(limit, 5, MAX_RECORDS - spent)
    if n <= 0:
        return f"STOP: the run budget of {MAX_RECORDS} records is spent."
    data, err = call("GET", f"/v1/companies/{domain}/people", params={"limit": n})
    if err:
        return err
    out = []
    for row in data["data"]:
        exp = ((row.get("contact_data") or {}).get("contact_current_experiences") or [{}])[0]
        out.append({
            "person_id": (row.get("meta") or {}).get("source_ids", {}).get("person_id"),
            "name": (row.get("profile_data") or {}).get("profile_full_name"),
            "title": exp.get("job_title"),
            "company": exp.get("company_name"),
        })
    return json.dumps(out)


@tool("Enrich person")
def enrich_person(person_id: str) -> str:
    """Costs 1 record. Only call it for a person who PASSED the ICP check.
    Returns the work email and its status."""
    if spent >= MAX_RECORDS:
        return f"STOP: the run budget of {MAX_RECORDS} records is spent."
    data, err = call("POST", "/v1/people/enrich",
                     body={"person_id": person_id, "reveal_email": True})
    if err:
        return err
    person = data["data"]
    emails = (person.get("contact_data") or {}).get("contact_emails") or []
    return json.dumps({
        "emails": [{"email": e.get("email"), "status": e.get("status")} for e in emails],
    })

from pydantic import BaseModel
from typing import List, Optional


class Verdict(BaseModel):
    domain: str
    qualified: bool
    reason: str                      # one sentence, quotes the title
    person_id: Optional[str] = None
    email: Optional[str] = None
    email_status: Optional[str] = None


class Verdicts(BaseModel):
    results: List[Verdict]

from crewai import Agent, Crew, Task

import leadocean_tools as lo
from icp import ICP

qualifier = Agent(
    role="Lead qualifier",
    goal="Decide which accounts have a real buyer and get that buyer's email.",
    backstory="You read titles carefully. You stop the moment a tool says STOP.",
    tools=[lo.list_people, lo.enrich_person],
    max_iter=10,
    verbose=True,
)

task = Task(
    description=(
        f"ICP: {ICP['must_have']}. Disqualify: {ICP['disqualify']}. "
        f"Good titles: {', '.join(ICP['buyer_titles'])}. "
        f"For each domain in {ICP['accounts']}: list the people once, pick the best "
        "buyer if one passes, enrich only that person, and give a one-sentence reason. "
        "If nobody passes, set qualified to false and do not enrich."
    ),
    expected_output="One verdict per domain.",
    agent=qualifier,
    output_pydantic=Verdicts,
)

result = Crew(agents=[qualifier], tasks=[task]).kickoff()
print(result.pydantic.model_dump_json(indent=2))
print(f"Records spent this run: {lo.spent} of {lo.MAX_RECORDS}")

python run_agent.py

{
  "results": [
    {
      "domain": "acme.com",
      "qualified": true,
      "reason": "Jane Doe is VP Sales, which owns pipeline.",
      "person_id": "123456789",
      "email": "jane.doe@acme.com",
      "email_status": "verified"
    }
  ]
}