How to Give a CrewAI Agent B2B Data (With LeadOcean)

mkdir crewai-leadocean && cd crewai-leadocean
uv add crewai requests
export LEADOCEAN_API_KEY="paste-your-key-here"
export OPENAI_API_KEY="paste-your-model-key-here"

curl -s https://api.leadocean.io/v1/account -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 = 15   # hard cap for one run
spent = 0          # records counted so far
stopped = ""       # set once a call says the run must end


def retry_after(r):
    return int(r.headers.get("Retry-After", 1))


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)
        if r.status_code == 503:                      # store unreachable, never billed
            time.sleep(2 ** attempt)
            continue
        if r.status_code == 429 and retry_after(r) <= 5:   # per-second limit
            time.sleep(retry_after(r))
            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 retry_after(r) > 5:
        stopped = f"429: a fair-use window is reached. Retry-After is {retry_after(r)}s."
    if stopped:
        return None, f"STOP: {stopped}"
    if r.status_code == 429:
        return None, "ERROR: still rate limited after retries. Try again later."
    if r.status_code == 404:
        return None, "NOT_FOUND: nobody matched. Nothing was charged."
    if r.status_code == 403:
        return None, "ERROR: the key lacks the scope for this endpoint."
    if r.status_code >= 400:
        return None, f"ERROR {r.status_code}: {r.text[:300]}"   # 400 lists the bad fields
    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, ""

def filters(title, job_level, job_function, country):
    f = {"title": title, "jobLevel": job_level,
         "jobFunction": job_function, "country": country}
    f = {k: v for k, v in f.items() if v}
    f.update(emailStatus=["verified"], emailType=["work"])   # mailable work emails
    return f


@tool("Count people")
def count_people(title: list[str] = [], job_level: list[str] = [],
                 job_function: list[str] = [], country: list[str] = []) -> str:
    """Free. Count verified work-email people matching the filters.
    job_level: C-Team, VP, Director, Manager, Staff, Other.
    job_function example: Sales & Business Development.
    country: ISO codes such as US. Call this before searching."""
    body = {**filters(title, job_level, job_function, country), "limit": 1}
    data, err = call("POST", "/v1/people/search", params={"count": "true"}, body=body)
    return err or json.dumps({"total": data["meta"].get("total")})


@tool("Search people")
def search_people(title: list[str] = [], job_level: list[str] = [],
                  job_function: list[str] = [], country: list[str] = [],
                  limit: int = 5) -> str:
    """Costs 1 record per person returned. Same filters as Count people.
    Returns person_id, name, title and company. Ask for few rows."""
    n = min(limit, 100, MAX_RECORDS - spent)
    if n <= 0:
        return f"STOP: the run budget of {MAX_RECORDS} records is spent."
    body = {**filters(title, job_level, job_function, country), "limit": n}
    data, err = call("POST", "/v1/people/search", body=body)
    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. Pass a person_id from Search people.
    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({
        "name": (person.get("profile_data") or {}).get("profile_full_name"),
        "emails": [{"email": e.get("email"), "status": e.get("status")} for e in emails],
    })

from crewai import Agent, Crew, Task

import leadocean_tools as lo

scout = Agent(
    role="Prospect researcher",
    goal="Find named decision makers with a verified work email at the lowest record cost.",
    backstory="You size an audience before you pull it. You stop the moment a tool says STOP.",
    tools=[lo.count_people, lo.search_people, lo.enrich_person],
    max_iter=8,
    verbose=True,
)

task = Task(
    description=(
        "Find 3 VPs of Sales and Business Development in the US. "
        "Count the audience first. Search for 5 people, pick 3, enrich them. "
        "Never enrich more than 3 people."
    ),
    expected_output="A table of 3 people: name, title, company, work email, email status.",
    agent=scout,
)

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

python run_agent.py

{"total":129494,"totalIsExact":true,"totalAll":350517,"credits":0,
 "defaults":"Counted only addresses worth sending to (emailStatus: verified or catch_all_valid or catch_all), which is this tool's default.",
 "unsupportedFilters":[]}

{
  "success": true,
  "data": [{
    "profile_data": { "profile_full_name": "Jane Doe" },
    "contact_data": {
      "has_email": true, "email_status": "verified", "email_type": "work",
      "contact_current_experiences": [{ "job_title": "VP Sales", "company_name": "Acme", "company_domain": "acme.com" }]
    },
    "meta": { "source_ids": { "person_id": "123456789" } }
  }],
  "meta": { "count": 1, "credits": 1, "nextCursor": "opaque-string" }
}