How to Build a Lead Qualification Agent (With the OpenAI Agents SDK)

pip install openai-agents httpx

export LEADOCEAN_API_KEY="paste-your-key-here"

import os, time, httpx

BASE = "https://api.leadocean.io"
HEADERS = {"x-api-key": os.environ["LEADOCEAN_API_KEY"], "content-type": "application/json"}
SPENT = {"records": 0}   # running total for this process
RECORD_CAP = 20          # hard ceiling for one run

def call(method, path, *, params=None, body=None, cost=0):
    if SPENT["records"] + cost > RECORD_CAP:
        return {"error": "Record cap reached for this run. Do not call this tool again."}
    for attempt in range(2):
        r = httpx.request(method, BASE + path, headers=HEADERS,
                          params=params, json=body, timeout=30)
        if r.status_code == 429 and attempt == 0:
            time.sleep(int(r.headers.get("Retry-After", "1")))
            continue
        break
    if r.status_code == 402:
        return {"error": "Out of records on this plan. Stop and tell the user."}
    if r.status_code == 429:
        return {"error": "Rate limited twice. Stop and report what you have."}
    if r.status_code == 404:
        return {"error": "Nobody matched. This lookup cost nothing."}
    if r.status_code >= 400:
        return {"error": f"HTTP {r.status_code}: {r.text[:300]}"}
    return r.json()

from agents import function_tool

@function_tool
def count_audience(job_level: list[str] | None = None,
                   job_function: list[str] | None = None,
                   country: list[str] | None = None,
                   industry: list[str] | None = None,
                   employee_range: list[str] | None = None) -> dict:
    """Count mailable people for an ICP. Free. Call this first.

    Args:
        job_level: Exact values such as "VP", "Director", "C-Team".
        job_function: Exact values such as "Sales & Business Development".
        country: ISO codes such as "US".
        industry: Exact industry labels such as "Software Development".
        employee_range: Brackets such as "51-200".
    """
    body = {"limit": 1, "jobLevel": job_level, "jobFunction": job_function,
            "country": country, "industry": industry, "employeeRange": employee_range}
    body = {k: v for k, v in body.items() if v}
    data = call("POST", "/v1/people/search", params={"count": "true"}, body=body)
    return data if "error" in data else data["meta"]

@function_tool
def search_people(limit: int,
                  job_level: list[str] | None = None,
                  job_function: list[str] | None = None,
                  country: list[str] | None = None,
                  industry: list[str] | None = None,
                  employee_range: list[str] | None = None) -> dict:
    """Search people for an ICP. Costs 1 record per person returned. Max 10.

    Args:
        limit: How many people to return, 1 to 10.
    """
    n = max(1, min(limit, 10))
    body = {"limit": n, "jobLevel": job_level, "jobFunction": job_function,
            "country": country, "industry": industry, "employeeRange": employee_range}
    body = {k: v for k, v in body.items() if v}
    data = call("POST", "/v1/people/search", body=body, cost=n)
    if "error" not in data:
        SPENT["records"] += len(data["data"])
    return data

@function_tool
def enrich_person(person_id: str) -> dict:
    """Full lookup for one person by person_id. Costs 1 record.

    Args:
        person_id: The meta.source_ids.person_id value from a search row.
    """
    data = call("POST", "/v1/people/enrich",
                body={"person_id": person_id, "reveal_email": True}, cost=1)
    if "error" not in data:
        SPENT["records"] += 1
    return data

INSTRUCTIONS = """You qualify leads for Acme, which sells project management software.

ICP: VP of Sales at US software companies with 51 to 200 staff.

Procedure:
1. Call count_audience with the ICP filters. Report the total.
2. Call search_people with the same filters and limit 10.
3. Score each person 0 to 3: +1 title matches "VP Sales" or "VP of Sales",
   +1 contact_data.email_status is verified, +1 the company is a software company.
4. Call enrich_person for the 3 highest scores only.
5. Return a ranked list: name, title, company, score, one-line reason.

Rules: use only fields that came back in a tool result. If a field is missing,
write "unknown". Never call a tool more often than the procedure says.

Filter values are exact strings: job_level "VP", job_function "Sales & Business Development",
country "US", industry "Software Development", employee_range "51-200"."""

import asyncio
from agents import Agent, Runner
from agents.exceptions import MaxTurnsExceeded

agent = Agent(
    name="Lead qualifier",
    instructions=INSTRUCTIONS,
    tools=[count_audience, search_people, enrich_person],
)

async def main():
    try:
        result = await Runner.run(agent, "Qualify this week's leads.", max_turns=12)
        print(result.final_output)
        usage = result.context_wrapper.usage
        print(f"OpenAI requests: {usage.requests}, tokens: {usage.total_tokens}")
    except MaxTurnsExceeded:
        print("Agent hit max_turns. Check the instructions for a loop.")
    print(f"LeadOcean records spent: {SPENT['records']}")

asyncio.run(main())

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

{
  "total": 1649,
  "totalIsExact": true,
  "distinctCompanies": 1125,
  "totalAll": 4141,
  "credits": 0,
  "defaults": "Counted only addresses worth sending to (emailStatus: verified or catch_all_valid or catch_all)"
}

{
  "success": true,
  "data": [
    {
      "profile_data": { "...": "name, headline, location" },
      "contact_data": { "has_email": true, "email_status": "verified", "email_type": "work",
                        "contact_current_experiences": [ { "...": "current role and company" } ] },
      "meta": { "source_ids": { "person_id": "<digits>" } }
    }
  ],
  "meta": { "count": 10, "limit": 10, "nextCursor": "<opaque cursor>" }
}