How to Build a Product Qualified Leads API With Company Data (PLG, API recipe)

A product qualified lead (PQL) API is a small endpoint in your own app. It takes a usage event, looks up the company, scores both and tells your CRM which accounts to call. You can build it in an hour. The only external piece is one company lookup per account.

Usage alone cannot tell a two-person agency from a 4,000-person enterprise. Eight active seats score the same for both. Company size and funding stage fix that, and they are not in your product data. Other guides on this topic, such as Crustdata's, make the same point: enrich first, then score.

TL;DR

• Enrich the account on its first meaningful usage event, score usage plus headcount plus funding, then look up buyers only for accounts that pass.
• The first run takes about 60 minutes and spends about 220 records for 100 accounts, so it fits inside the free plan's 1,000.
• One found company is one record. A company we hold nothing for is a 404 and costs nothing.

Prerequisites

• A LeadOcean API key in $LEADOCEAN_API_KEY. The free plan has the search and enrich scopes (pricing).
• A usage event stream with an account id, the signup work email and a few counters (seats, API calls, a milestone flag).
• Python 3.9 or later with requests.
• A rule for "meaningful": the first action that proves real use, such as a first project created. Not the signup.

Step 1: Choose the trigger and the payload

Look up the company on the first meaningful usage event, not at signup. Signup lookups spend records on throwaway emails and accounts that never return.

Take the domain from the work email and skip webmail. A personal Gmail has no company to read.

{
  "event": "first_project_created",
  "account_id": "acc_4821",
  "email": "jane.doe@acme.com",
  "usage": { "seats_active": 8, "monthly_api_calls": 4500, "milestone_passed": true }
}

FREE_MAIL = {"gmail.com", "yahoo.com", "outlook.com", "hotmail.com", "icloud.com", "proton.me"}

def domain_of(email):
    d = email.split("@")[-1].lower()
    return None if d in FREE_MAIL else d

Step 2: Look up the company

POST /v1/companies/enrich takes a domain or a linkedin_url and returns one company. It costs one record when the company is found. A 404 costs nothing.

curl -s -X POST "https://api.leadocean.io/v1/companies/enrich" \
  -H "x-api-key: $LEADOCEAN_API_KEY" \
  -H "content-type: application/json" \
  -d '{"domain":"acme.com"}'

The fields the score needs sit in data.company_data: company_employees.number_of_employees (and a band code), and company_funding.last_round_type and last_round_year. Either block can be null. Read them defensively.

The dataset is refreshed monthly, and each record carries data.meta.fetched_at. Cache the result by domain for 30 days so a busy account costs one record, not one per event.

Step 3: Score usage and company together

Keep the rule in code you own. This function adds company points to usage points and returns a number and the reasons. The weights are placeholders. Tune them against accounts that actually paid.

import os, time, requests

BASE = "https://api.leadocean.io"
H = {"x-api-key": os.environ["LEADOCEAN_API_KEY"], "content-type": "application/json"}
CACHE = {}  # domain -> (fetched_at, company_data); use Redis or a table in production

def company(domain):
    hit = CACHE.get(domain)
    if hit and time.time() - hit[0] < 30 * 86400:
        return hit[1]
    r = requests.post(f"{BASE}/v1/companies/enrich", headers=H, json={"domain": domain}, timeout=30)
    if r.status_code == 404:
        data = {}                      # not held, not billed
    elif r.status_code == 429:
        time.sleep(int(r.headers.get("Retry-After", "1")))
        return company(domain)
    else:
        r.raise_for_status()           # a 402 means the free records are spent
        data = r.json()["data"].get("company_data") or {}
    CACHE[domain] = (time.time(), data)
    return data

def score(usage, c):
    pts, why = 0, []
    if usage.get("seats_active", 0) >= 5:         pts += 25; why.append("5+ seats")
    if usage.get("monthly_api_calls", 0) >= 1000: pts += 20; why.append("1k+ API calls")
    if usage.get("milestone_passed"):             pts += 20; why.append("milestone")
    emp = (c.get("company_employees") or {}).get("number_of_employees")
    if emp is not None and 50 <= emp <= 5000:     pts += 30; why.append(f"{emp} employees")
    f = c.get("company_funding") or {}
    if (f.get("last_round_year") or 0) >= 2025:   pts += 15; why.append(f"{f.get('last_round_type')} {f['last_round_year']}")
    return pts, why

Set the threshold at 80 to start. A 2-person agency with this usage scores 65 and misses. The same usage at a 240-person company funded in 2025 scores 110 and passes.

Step 4: Find the buyers at accounts that pass

A qualified account needs a person to call. Search by the company domain and the buying role. Search costs one record per person returned. Rows are thin, with no addresses, and each carries a person_id at meta.source_ids.person_id.

curl -s -X POST "https://api.leadocean.io/v1/people/search" \
  -H "x-api-key: $LEADOCEAN_API_KEY" \
  -H "content-type: application/json" \
  -d '{
    "domain": ["acme.com"],
    "jobLevel": ["VP", "Director"],
    "jobFunction": ["Engineering"],
    "limit": 3
  }'

Enrich only the people worth a record. person_id is one of the four accepted keys, and reveal_email returns the address.

def reveal(person_id):
    r = requests.post(f"{BASE}/v1/people/enrich", headers=H,
                      json={"person_id": person_id, "reveal_email": True}, timeout=30)
    return None if r.status_code == 404 else r.json()["data"]   # 404 costs nothing

Read email_status before anything is sent. Bounced emails are not refunded.

Step 5: Expose it as one endpoint

Wire the pieces behind a route your event pipeline calls. This version uses FastAPI. It returns the score and reasons, and pushes to your CRM only above the threshold.

from fastapi import FastAPI
app = FastAPI()

@app.post("/pql")
def pql(evt: dict):
    domain = domain_of(evt["email"])
    if not domain:
        return {"qualified": False, "reason": "webmail"}
    pts, why = score(evt["usage"], company(domain))
    return {"account_id": evt["account_id"], "score": pts, "reasons": why, "qualified": pts >= 80}

On qualified: true, run Step 4 and post the result to your CRM with your CRM's own API. Run it with uvicorn app:app.

What you get

Before you spend records on buyers, size the audience for the pattern your score rewards. This is the real result of leadocean_count_leads, run on 2026-10-21. Filters: employeeRange 51-200 and 201-500, jobLevel VP and Director, jobFunction Engineering, country US.

{
  "total": 15154,
  "totalIsExact": true,
  "totalAll": 39431,
  "distinctCompanies": 9623,
  "credits": 0,
  "defaults": "Counted only addresses worth sending to (emailStatus: verified or catch_all_valid or catch_all)",
  "source": "leadocean",
  "unsupportedFilters": []
}

The total is a mailable count of people, not companies: 15,154 with a verified or catch-all email out of 39,431, spread over 9,623 employers. It spent no records. Agents get the same call from the free leadocean_count_leads tool on the MCP server.

The company lookup returns this shape (response shape from the API reference, placeholder values).

{
  "success": true,
  "data": {
    "company_data": {
      "company_name": "Acme",
      "company_domain": "acme.com",
      "company_employees": { "number_of_employees": 240, "number_of_employees_code": "201-500" },
      "company_funding": { "last_round_type": "Series B", "last_round_year": 2025 }
    },
    "meta": { "schema": "leadocean.company.v1", "fetched_at": "2026-09-01T00:00:00Z" }
  },
  "meta": { "credits": 1 }
}

The /pql route then answers like this (placeholder usage, not a real account).

{ "account_id": "acc_4821", "score": 110, "reasons": ["5+ seats", "1k+ API calls", "milestone", "240 employees", "Series B 2025"], "qualified": true }

We have not run this on your product, so there is no real score distribution to show.

Troubleshooting

FAQ