Waterfall enrichment with Forager

Recipe prompt

Implement this recipe in my project: https://www.forager.ai/docs/recipes/waterfall-step.md

Add a Forager step to our enrichment pipeline. It submits a Forager work email lookup, polls the job until it completes, and returns a normalized result of found, not_found or error. On found, stop the waterfall. On not_found or a retryable error, pass the record to the next step. Handle 403, 429 with backoff, and polling timeouts, and never let one failed record stop the run. Keep the API key server-side in FORAGER_API_KEY with FORAGER_ACCOUNT_ID. Add tests for each outcome.

Waterfall enrichment: Forager as one step between two other data providers

What you’ll build

One step in a pipeline that tries sources in order and stops at the first answer. The Forager step takes a record, asks for a work email, and returns a clean result your pipeline can act on. Other steps in your pipeline stay yours to design.

Steps

Choose the input

The single work email lookup takes either first_name, last_name and domain, or linkedin_public_identifier alone. Never send a domain and an identifier together; the request is refused. Prefer the identifier when you have it.

Submit the lookup

The call returns a job_id at once. The lookup itself runs asynchronously.

JOB_ID=$(curl -s -X POST "https://api-v2.forager.ai/api/v2/$FORAGER_ACCOUNT_ID/datastorage/person_contacts_lookup/work_emails/" \
  -H "X-API-KEY: $FORAGER_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"first_name": "Jane", "last_name": "Doe", "domain": "example.com"}' | jq -r .job_id)

import os, requests

BASE = f"https://api-v2.forager.ai/api/v2/{os.environ['FORAGER_ACCOUNT_ID']}/datastorage/person_contacts_lookup/work_emails/"
HEADERS = {"X-API-KEY": os.environ["FORAGER_API_KEY"]}

r = requests.post(
    BASE,
    headers=HEADERS,
    json={"first_name": "Jane", "last_name": "Doe", "domain": "example.com"},
    timeout=30,
)
r.raise_for_status()
job_id = r.json()["job_id"]

const BASE = `https://api-v2.forager.ai/api/v2/${process.env.FORAGER_ACCOUNT_ID}/datastorage/person_contacts_lookup/work_emails/`;
const headers = {
  "X-API-KEY": process.env.FORAGER_API_KEY,
  "Content-Type": "application/json",
};

const res = await fetch(BASE, {
  method: "POST",
  headers,
  body: JSON.stringify({ first_name: "Jane", last_name: "Doe", domain: "example.com" }),
});
if (!res.ok) throw new Error(`Forager ${res.status}`);
const { job_id: jobId } = await res.json();

Poll until the job is terminal

Poll Single work email lookup result. status moves through queued, running and awaiting_provider to completed or failed. While the job is running, a Retry-After header suggests when to poll again. Polling is free.

curl "https://api-v2.forager.ai/api/v2/$FORAGER_ACCOUNT_ID/datastorage/person_contacts_lookup/work_emails/$JOB_ID/" \
  -H "X-API-KEY: $FORAGER_API_KEY"

import time

def poll(job_id, max_wait=120):
    deadline = time.time() + max_wait
    while time.time() < deadline:
        r = requests.get(f"{BASE}{job_id}/", headers=HEADERS, timeout=30)
        if r.status_code == 429:  # rate limited: wait, then poll again
            time.sleep(float(r.headers.get("Retry-After", 5)))
            continue
        r.raise_for_status()
        job = r.json()
        if job["status"] in ("completed", "failed"):
            return job
        time.sleep(float(r.headers.get("Retry-After", 2)))
    return None

async function poll(jobId, maxWaitMs = 120000) {
  const deadline = Date.now() + maxWaitMs;
  while (Date.now() < deadline) {
    const r = await fetch(`${BASE}${jobId}/`, { headers });
    if (r.status === 429) {
      // rate limited: wait, then poll again
      const retry = Number(r.headers.get("Retry-After") ?? 5);
      await new Promise((ok) => setTimeout(ok, retry * 1000));
      continue;
    }
    if (!r.ok) throw new Error(`Forager ${r.status}`);
    const job = await r.json();
    if (["completed", "failed"].includes(job.status)) return job;
    const wait = Number(r.headers.get("Retry-After") ?? 2);
    await new Promise((ok) => setTimeout(ok, wait * 1000));
  }
  return null;
}

Normalize the result

Map the job to the three outcomes your pipeline understands.

Job	Pipeline result
`status` is `completed`, `result` is `found`	`found`: use `emails` and stop
`status` is `completed`, `result` is `not_found`	`not_found`: go to the next step
`status` is `failed`, `error_code` is `person_not_found`, `no_current_employer` or `input_unresolved`	`not_found`: retry once with name and `domain` if you have them, otherwise go to the next step
`status` is `failed`, `error_code` is `provider_error`, `expired` or `internal_error`	`error`: retry later, then go to the next step
Polling timed out	`error`

Handle the edge cases

Not found. not_found is a normal answer, not a failure. Log it and move on.
403. The key is missing or invalid. Stop the run and alert someone, because every record will fail.
429. Back off and retry. The poll loop above waits for Retry-After on a 429. The single lookup is rate limited for interactive use, so for many records use the bulk recipe. See errors and rate limits.
Credits. A work email lookup costs 5 credits, charged only when an address is found. Polling never costs credits. See credit pricing.
Email shape. The spec describes each entry in emails as an address with a verdict, but types it as a string.

Bulk work emails

Submit many lookups in one request.

Single work email lookup

Request and response reference.

Errors and rate limits

What is documented today.

Recipes

Sales and dialer platforms

Recruiting platforms

Data and product teams

AI agents

Sample apps

More examples

Waterfall enrichment with Forager

What you’ll build

Steps

Handle the edge cases

Bulk work emails

Single work email lookup

Errors and rate limits

​What you’ll build

​Steps

​Handle the edge cases

​Related

Bulk work emails

Single work email lookup

Errors and rate limits

What you’ll build

Steps

Handle the edge cases

Related