> ## Documentation Index
> Fetch the complete documentation index at: https://www.forager.ai/docs/llms.txt
> Use this file to discover all available pages before exploring further.

> ## Agent Instructions
> Forager is a REST API for B2B people, company and job post data: contact lookups (phone numbers, personal emails, work emails), person, company and job post search, and bulk data feeds.
> Server: https://api-v2.forager.ai. Send the API key in the X-API-KEY header. Most paths include {account_id}. Read FORAGER_API_KEY and FORAGER_ACCOUNT_ID from the environment; never put the key in client-side code, logs or prompts.
> To add Forager to a project, follow https://www.forager.ai/docs/guides/agent-quickstart.md. To pick an endpoint, read https://www.forager.ai/docs/guides/choose-an-endpoint.md. For end-to-end builds, see https://www.forager.ai/docs/recipes/overview.md.
> Most requests spend credits. Do not guess costs; use https://www.forager.ai/docs/api-overview/credit-pricing.md and ask the user before spending credits.
> Use only endpoints and fields defined in https://www.forager.ai/docs/openapi.yaml. Forager has no MCP server for the API yet; the /mcp endpoint on this site searches the documentation only.

# Enrich a CSV of LinkedIn URLs

> Sample app: a Python script that reads LinkedIn URLs from a CSV, enriches each row with a Forager person lookup, adds work emails in a batch and resumes.

```text Setup prompt theme={"theme":{"light":"github-light","dark":"github-dark"}}
Implement https://www.forager.ai/docs/samples/enrich-csv-linkedin-urls.md in my project: a command-line script that reads a CSV of LinkedIn profile URLs, enriches each row with the Forager person lookup, optionally adds work emails through one bulk work email batch, writes an output CSV, and can be re-run without repeating finished rows. Keep the API key in FORAGER_API_KEY and FORAGER_ACCOUNT_ID, print the credit estimate before spending, and add tests with stubbed HTTP.
```

This sample is one Python file, `enrich_csv.py`, that turns a CSV of LinkedIn profile URLs into a CSV of names, titles, companies and, if you ask for them, work emails. It uses two Forager endpoints: the [person lookup](https://www.forager.ai/docs/openapi/people/datastorage_person_detail_lookup_create.md) (1 credit per row) and the [bulk work email lookup](https://www.forager.ai/docs/openapi/work-emails/v2_datastorage_person_contacts_lookup_work_emails_bulk_create.md) (5 credits per address found, one batch for the whole file). It prints the credit estimate and asks before it spends anything, and a second run skips rows that are already in the output file.

## What you need

* Python 3.9 or newer and `pip install requests`.
* `FORAGER_API_KEY` and `FORAGER_ACCOUNT_ID` in the environment, from the [Web App](https://www.forager.ai/docs/api-overview/create-an-account.md).
* A CSV with a `linkedin_url` column. The fictional sample below is what the walkthrough uses.

```csv input.csv theme={"theme":{"light":"github-light","dark":"github-dark"}}
linkedin_url,note
https://www.linkedin.com/in/jane-doe-example/,inbound lead
https://www.linkedin.com/in/john-roe-example,conference
linkedin.com/in/alex-example-smith/,referral
```

## The script

**1. Parse the URLs and estimate the cost**

The person lookup takes the LinkedIn public identifier, the slug after `/in/`. Everything else in the URL (scheme, `www`, trailing slash, query string) is dropped. Rows that already have an `id` in the output file are skipped, so a crashed run can be resumed.

```python enrich_csv.py (part 1 of 3) theme={"theme":{"light":"github-light","dark":"github-dark"}}
#!/usr/bin/env python3
"""Enrich a CSV of LinkedIn profile URLs with the Forager API.

usage: enrich_csv.py input.csv output.csv [--work-emails] [--yes]
"""
import argparse
import csv
import os
import re
import sys
import time
from urllib.parse import urlparse

import requests

BASE = "https://api-v2.forager.ai/api"
ACCOUNT = os.environ["FORAGER_ACCOUNT_ID"]
HEADERS = {"X-API-KEY": os.environ["FORAGER_API_KEY"]}
OUT_FIELDS = ["linkedin_url", "handle", "id", "full_name", "headline", "current_title",
              "current_company", "current_domain", "work_email", "work_email_result", "error"]
PERSON_LOOKUP_CREDITS = 1   # https://www.forager.ai/docs/api-overview/credit-pricing
WORK_EMAIL_CREDITS = 5      # per address found; nothing when not found


def handle_from_url(url):
    """'https://www.linkedin.com/in/jane-doe-example/?x=1' -> 'jane-doe-example'."""
    url = url.strip()
    if "://" not in url:
        url = "https://" + url
    match = re.search(r"/in/([^/?#]+)", urlparse(url).path + "?")
    return match.group(1) if match else None


def read_rows(path):
    with open(path, newline="", encoding="utf-8") as f:
        return list(csv.DictReader(f))


def done_handles(path):
    if not os.path.exists(path):
        return set()
    with open(path, newline="", encoding="utf-8") as f:
        return {row["handle"] for row in csv.DictReader(f) if row.get("id")}
```

**2. Call the API with retries**

One helper sends every request. A `403` stops the run (the key or Account ID is wrong, retrying cannot help), a `429` waits and retries, and anything else is recorded in the row's `error` column so one bad row does not stop the file. See [errors and rate limits](https://www.forager.ai/docs/errors-and-rate-limits.md).

```python enrich_csv.py (part 2 of 3) theme={"theme":{"light":"github-light","dark":"github-dark"}}
class AuthError(Exception):
    pass


def forager(method, path, json=None, retries=3):
    wait = 2.0
    for attempt in range(retries + 1):
        resp = requests.request(method, f"{BASE}/{path}", headers=HEADERS, json=json, timeout=60)
        if resp.status_code == 403:
            raise AuthError("Forager rejected FORAGER_API_KEY / FORAGER_ACCOUNT_ID: " + resp.text)
        if resp.status_code == 429 and attempt < retries:
            time.sleep(float(resp.headers.get("Retry-After", wait)))
            wait *= 2
            continue
        return resp
    return resp


def lookup_person(handle):
    """Return (row fields, error). 1 credit."""
    resp = forager("POST", f"{ACCOUNT}/datastorage/person_detail_lookup/",
                   json={"linkedin_public_identifier": handle})
    if resp.status_code != 200:
        return {}, f"HTTP {resp.status_code}: {resp.text[:200]}"
    person = resp.json()
    current = next((r for r in person.get("roles", []) if r.get("is_current")), None)
    org = (current or {}).get("organization") or {}
    return {
        "id": person["id"],
        "full_name": person.get("full_name", ""),
        "headline": person.get("headline") or "",
        "current_title": (current or {}).get("role_title") or "",
        "current_company": org.get("name") or (current or {}).get("organization_name") or "",
        "current_domain": org.get("domain") or "",
    }, ""


def bulk_work_emails(handles, max_wait=900):
    """Submit one batch for every handle, poll until it completes, return {handle: (email, result)}.
    5 credits per address found. The batch is rejected, and nothing is charged, if the balance
    cannot cover it."""
    resp = forager("POST", f"v2/{ACCOUNT}/datastorage/person_contacts_lookup/work_emails/bulk/",
                   json={"email_lookups": [{"linkedin_public_identifier": h} for h in handles]})
    if resp.status_code != 202:
        raise RuntimeError(f"bulk submit failed: HTTP {resp.status_code}: {resp.text[:300]}")
    batch_id = resp.json()["batch_id"]
    deadline = time.monotonic() + max_wait
    while time.monotonic() < deadline:
        poll = forager("GET", f"v2/{ACCOUNT}/datastorage/person_contacts_lookup/work_emails/bulk/{batch_id}/")
        batch = poll.json()
        if batch["status"] in ("completed", "failed"):
            break
        time.sleep(float(poll.headers.get("Retry-After", 10)))   # polling is free
    else:
        raise TimeoutError(f"batch {batch_id} still {batch['status']} after {max_wait}s")
    out = {}
    for item in batch.get("results", []):                        # one per submitted row, in order
        handle = handles[item["submitted_index"]]
        if item["status"] == "failed":
            out[handle] = ("", f"failed:{item.get('error_code')}")
        else:
            emails = item.get("emails") or []
            out[handle] = (emails[0] if emails else "", item.get("result") or "")
    return out
```

**3. Run the file end to end**

The main function prints how many rows are new and what they cost, waits for a `y` unless you pass `--yes`, enriches row by row (appending to the output as it goes), and runs the work email batch last so a crash mid-file never loses the person data already fetched.

```python enrich_csv.py (part 3 of 3) theme={"theme":{"light":"github-light","dark":"github-dark"}}
def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("input")
    ap.add_argument("output")
    ap.add_argument("--work-emails", action="store_true", help="also find work emails (bulk, 5 credits per address found)")
    ap.add_argument("--yes", action="store_true", help="skip the cost confirmation")
    args = ap.parse_args()

    rows = read_rows(args.input)
    skip = done_handles(args.output)
    todo = []
    for row in rows:
        handle = handle_from_url(row.get("linkedin_url", ""))
        if handle and handle not in skip:
            todo.append((row["linkedin_url"], handle))
    bad = sum(1 for row in rows if not handle_from_url(row.get("linkedin_url", "")))

    print(f"{len(rows)} rows, {len(skip)} already done, {bad} without a usable /in/ URL, {len(todo)} to enrich")
    estimate = len(todo) * PERSON_LOOKUP_CREDITS
    if args.work_emails:
        estimate += len(todo) * WORK_EMAIL_CREDITS
        print(f"up to {estimate} credits ({PERSON_LOOKUP_CREDITS} per lookup, plus {WORK_EMAIL_CREDITS} per work email found)")
    else:
        print(f"{estimate} credits ({PERSON_LOOKUP_CREDITS} per lookup)")
    if not todo:
        return
    if not args.yes and input("continue? [y/N] ").strip().lower() != "y":
        sys.exit(1)

    new_file = not os.path.exists(args.output)
    enriched = []
    with open(args.output, "a", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=OUT_FIELDS)
        if new_file:
            writer.writeheader()
        for url, handle in todo:
            fields, error = lookup_person(handle)
            record = {"linkedin_url": url, "handle": handle, "error": error, **fields}
            enriched.append(record)
            writer.writerow({k: record.get(k, "") for k in OUT_FIELDS})
            f.flush()
            print(f"{handle}: {record.get('full_name') or error}")

    if args.work_emails:
        found = [r["handle"] for r in enriched if r.get("id")]
        if found:
            emails = bulk_work_emails(found)
            for record in enriched:
                record["work_email"], record["work_email_result"] = emails.get(record["handle"], ("", ""))
            # Rewrite the file so the email columns land on the rows written above.
            existing = [r for r in read_rows(args.output) if r["handle"] not in found]
            with open(args.output, "w", newline="", encoding="utf-8") as f:
                writer = csv.DictWriter(f, fieldnames=OUT_FIELDS)
                writer.writeheader()
                for record in existing + enriched:
                    writer.writerow({k: record.get(k, "") for k in OUT_FIELDS})
            print(f"work emails: {sum(1 for r in enriched if r.get('work_email'))} found")


if __name__ == "__main__":
    try:
        main()
    except AuthError as e:
        sys.exit(str(e))
```

**4. Run it**

```bash theme={"theme":{"light":"github-light","dark":"github-dark"}}
python3 enrich_csv.py input.csv output.csv --work-emails
```

```text Output (illustrative) theme={"theme":{"light":"github-light","dark":"github-dark"}}
3 rows, 0 already done, 0 without a usable /in/ URL, 3 to enrich
up to 18 credits (1 per lookup, plus 5 per work email found)
continue? [y/N] y
jane-doe-example: Jane Doe
john-roe-example: John Roe
alex-example-smith: Alex Smith
work emails: 2 found
```

```csv output.csv (illustrative) theme={"theme":{"light":"github-light","dark":"github-dark"}}
linkedin_url,handle,id,full_name,headline,current_title,current_company,current_domain,work_email,work_email_result,error
https://www.linkedin.com/in/jane-doe-example/,jane-doe-example,48213,Jane Doe,Head of Data at Acme Example,Head of Data,Acme Example,acme.com,jane.doe@acme.com,found,
https://www.linkedin.com/in/john-roe-example,john-roe-example,48377,John Roe,VP Sales,VP Sales,Example Labs,example.com,john.roe@example.com,found,
linkedin.com/in/alex-example-smith/,alex-example-smith,48901,Alex Smith,Founder,Founder,Smith Example Co,,,not_found,
```

## Edge cases the script handles

* **A URL that is not a profile URL** (a company page, an empty cell) is counted and skipped, never sent.
* **A person the lookup cannot resolve** gets the status code and body in `error` and the run continues.&#x20;
* **Duplicate URLs** in the input are collapsed by the bulk endpoint, which charges one lookup for identical rows.
* **Not enough credits for the batch**: the bulk endpoint rejects the whole batch and charges nothing, so the person data already written is safe. Buy credits, then re-run with `--work-emails`; the person lookups are skipped and only the batch runs again.&#x20;
* **Resuming**: rows with an `id` in the output file are skipped. Delete a row from the output to redo it.

## Adapt it

* Need phone numbers instead? Replace `bulk_work_emails` with one [phone number lookup](https://www.forager.ai/docs/openapi/phones/datastorage_person_contacts_lookup_phone_numbers_create.md) per row (`person_id` from the person lookup, 15 credits per lookup). There is no bulk phone endpoint.
* Thousands of rows? Run the batch in chunks and send a `webhook_url` instead of polling. See [Bulk work email lookup](https://www.forager.ai/docs/guides/bulk-work-emails.md).
* Whole datasets are cheaper as a [data feed](https://www.forager.ai/docs/data-license/v2/data-feed-overview.md) than as API calls.

## Related

> [card] [Python quickstart](https://www.forager.ai/docs/quickstarts/python.md)
>
> The client this script is built on, step by step.

> [card] [Contact enrichment guide](https://www.forager.ai/docs/guides/enrich-a-contact.md)
>
> Every field in the person record.

> [card] [Credit pricing](https://www.forager.ai/docs/api-overview/credit-pricing.md)
>
> What each call costs.

> [card] [Report bad data](https://www.forager.ai/docs/guides/report-bad-data.md)
>
> Send back invalid addresses with the feedback endpoints.


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.