How to Build a Lead Qualification Agent (LangChain and LeadOcean)

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

import os, time, requests

BASE = "https://api.leadocean.io"
HEADERS = {"x-api-key": os.environ["LEADOCEAN_API_KEY"]}

class OutOfRecords(Exception):
    pass

def call(method, path, params=None, body=None):
    for attempt in range(2):
        r = requests.request(method, BASE + path, headers=HEADERS,
                             params=params, json=body, timeout=30)
        if r.status_code == 429 and attempt == 0:
            time.sleep(float(r.headers.get("Retry-After", "1")))
            continue
        if r.status_code == 402:
            raise OutOfRecords("LeadOcean records are spent. Stop and tell the user.")
        if r.status_code == 404:
            return {"data": None, "note": "nobody matched"}
        r.raise_for_status()
        return r.json()
    raise RuntimeError("Rate limited twice. Slow down.")

from langchain.tools import tool

@tool
def size_list(title: list[str], country: list[str], job_level: list[str]) -> dict:
    """Count how many people match. Free. Call this before search_people.
    job_level values are exact: C-Team, VP, Director, Manager, Staff, Other.
    country is ISO alpha-2, for example US."""
    out = call("POST", "/v1/people/search", params={"count": "true"},
               body={"title": title, "country": country,
                     "jobLevel": job_level, "limit": 1})
    return {"total": out["meta"].get("total"),
            "exact": out["meta"].get("totalIsExact")}

@tool
def search_people(title: list[str], country: list[str],
                  job_level: list[str], limit: int = 10) -> list:
    """Search people. Costs 1 record per person returned. Keep limit at 10 or less.
    Returns profile rows and a person_id for enrich_person."""
    out = call("POST", "/v1/people/search",
               body={"title": title, "country": country,
                     "jobLevel": job_level, "limit": min(limit, 25)})
    return out["data"]

@tool
def enrich_person(person_id: str) -> dict:
    """Enrich one person by person_id. Costs 1 record. Returns career history
    and the best email with its status. Call it only for strong matches."""
    return call("POST", "/v1/people/enrich",
                body={"person_id": person_id, "reveal_email": True})

from langchain.agents import create_agent

SYSTEM = """You qualify B2B leads for a seller of sales-data tooling.
ICP: VP or Director in sales or revenue operations at software companies.
Process:
1. Call size_list first. If total is above 5,000, narrow the filters.
2. Call search_people with limit 10.
3. Score each person 0 to 10 on title fit and seniority.
4. Call enrich_person for at most 3 people scoring 8 or higher.
5. Return a table: name, title, score, email, email_status, one reason.
Never enrich a person scoring below 8. If a tool says records are spent, stop."""

agent = create_agent(
    model="anthropic:claude-sonnet-5",
    tools=[size_list, search_people, enrich_person],
    system_prompt=SYSTEM,
)

def run(task: str):
    try:
        result = agent.invoke({"messages": [{"role": "user", "content": task}]})
        return result["messages"][-1].content
    except OutOfRecords as e:
        return f"Stopped: {e}"

print(run("Find VP Sales contacts in the US and qualify the best three."))

SPENT = {"records": 0}
BUDGET = 25

def charge(n):
    SPENT["records"] += n
    if SPENT["records"] > BUDGET:
        raise OutOfRecords("Run budget reached.")

{
  "total": 10839,
  "totalIsExact": true,
  "totalAll": 36817,
  "distinctCompanies": 7845,
  "credits": 0
}

{
  "success": true,
  "data": [
    { "profile_data": { "profile_full_name": "Jane Doe", "...": "headline, location" },
      "meta": { "source_ids": { "person_id": "1234567890" } } }
  ],
  "meta": { "count": 1, "nextCursor": null }
}