Build a Company Enrichment API on Apify in an Afternoon

You can stand up a working company enrichment API in an afternoon. Point your backend at the LinkedIn Company Enrichment actor, pass a list of LinkedIn URLs, slugs, company names or bare domains, and get back structured firmographics: industry, size band, employees on LinkedIn, HQ address, founded year, company type and specialties. Cache on the resolved slug, store source_query and scraped_at as provenance, and return a partial record with a confidence flag when a domain does not resolve.

Why build your own company enrichment endpoint?

Somebody filed a ticket asking your product to show "company details" next to each account, and the obvious answer — buy an enrichment vendor — comes with a per-seat contract, a procurement cycle, a rate-limited sandbox key and a data model you do not control. For a feature that renders six fields in a sidebar, that is a lot of ceremony.

The engineering problem is smaller than the procurement problem. You need one function that takes whatever identifier your users already have — usually a work-email domain, sometimes a pasted LinkedIn URL, occasionally a typed company name — and returns a firmographic record. Around that function you need a cache so you are not paying to look up Stripe forty times a day, provenance fields so support can answer "where did this come from", and a degradation path so a miss renders as an incomplete card instead of a 500.

Data quality is the part teams underestimate. Gartner has put the average annual cost of poor data quality at $12.9 million per organization, and enrichment written without provenance fields is exactly how a database becomes unauditable. Storing when a record was scraped and what input produced it costs two columns and saves the rewrite.

How does this compare to the alternatives?

Building the scraper yourself is the expensive option, and not because parsing is hard. Public company pages rate-limit by IP, the page structure changes without notice, and the resolution step — domain or name to the correct company — is a separate problem from extraction. A generic scraping API hands you HTML and leaves both problems with you. A purpose-built actor hands you a typed record and an input contract you can code against.

The practical difference for a developer is that only the last row lets you write your endpoint against stable field names on day one.

How to build a company enrichment API in 4 steps

How do I call the enrichment actor from my backend?

Call it like any other job: submit input, wait, read the dataset. The Apify Python client handles polling for you.

import os
from apify_client import ApifyClient

client = ApifyClient(os.environ["APIFY_TOKEN"])
ACTOR = "thirdwatch/linkedin-company-enrichment-scraper"

def enrich(companies: list[str]) -> list[dict]:
    run = client.actor(ACTOR).call(
        run_input={
            "companies": companies,
            "maxResults": len(companies),
            "concurrency": 10,
            "proxyConfiguration": {"useApifyProxy": True},
        }
    )
    return list(client.dataset(run["defaultDatasetId"]).iterate_items())

for row in enrich([
    "https://www.linkedin.com/company/hashicorp/",
    "stripe.com",
    "Datadog",
]):
    print(row["name"], row["industry"], row["employees_on_linkedin"])

companies is the only required field. It accepts LinkedIn company URLs, bare slugs, company names and website domains in the same list, which means you can pass user input through unchanged rather than normalising it first. Set maxResults to the length of your batch so a run can never quietly cost more than the list you submitted.

What should I key the cache on?

Key the cache on the resolved slug, never on the raw input string. stripe.com, Stripe, stripe and https://www.linkedin.com/company/stripe/ are four different cache keys but one company, and an input-keyed cache will happily pay four times for the same record.

Every returned row carries both slug (canonical) and source_query (the exact string that produced it), so you can store the record once and alias every input you have seen to it.

import json, sqlite3

def store(conn: sqlite3.Connection, record: dict) -> None:
    conn.execute(
        """
        INSERT INTO company_cache (slug, payload, scraped_at)
        VALUES (?, ?, ?)
        ON CONFLICT(slug) DO UPDATE SET
            payload     = excluded.payload,
            scraped_at  = excluded.scraped_at
        """,
        (record["slug"], json.dumps(record), record["scraped_at"]),
    )
    conn.execute(
        "INSERT OR REPLACE INTO query_alias (source_query, slug) VALUES (?, ?)",
        (record["source_query"].strip().lower(), record["slug"]),
    )

def lookup(conn: sqlite3.Connection, query: str) -> dict | None:
    row = conn.execute(
        """
        SELECT c.payload FROM query_alias a
        JOIN company_cache c ON c.slug = a.slug
        WHERE a.source_query = ?
          AND c.scraped_at > datetime('now', '-90 days')
        """,
        (query.strip().lower(),),
    ).fetchone()
    return json.loads(row[0]) if row else None

The scraped_at predicate is your TTL. Firmographics move slowly — headcount bands and office lists shift over quarters, not days — so a 60 to 90 day window keeps records fresh without re-paying for a catalogue that has not changed.

How do I return a partial record when a domain does not resolve?

Do not throw. A bare domain resolves to the right company roughly three times in four, so a miss is a normal outcome your endpoint should model, not an exception. Compare the returned website_domain against what the caller asked for and attach a confidence flag the frontend can act on.

def annotate(record: dict, requested: str) -> dict:
    requested = requested.strip().lower().removeprefix("www.")
    returned = (record.get("website_domain") or "").lower().removeprefix("www.")

    if requested.startswith("https://www.linkedin.com/company/"):
        confidence = "exact"           # explicit URL, deterministic resolution
    elif returned and returned == requested:
        confidence = "exact"           # domain round-tripped
    elif returned:
        confidence = "probable"        # resolved, but to a different domain
    else:
        confidence = "unverified"      # name match with nothing to check against

    return {
        **record,
        "confidence": confidence,
        "provenance": {
            "source_query": record.get("source_query"),
            "scraped_at": record.get("scraped_at"),
            "linkedin_url": record.get("linkedin_url"),
        },
    }

def placeholder(requested: str) -> dict:
    return {"source_query": requested, "confidence": "unresolved", "name": None}

Three states beat a boolean. exact can auto-populate a CRM field, probable can render with a "verify" affordance, and unresolved can queue for a human who pastes a LinkedIn URL — which always resolves deterministically.

How do I batch lookups instead of one run per company?

Batch on the cache misses only, and let concurrency do the parallelism inside a single run rather than firing one run per company. Runs have fixed overhead; a hundred single-company runs cost far more wall-clock time than one hundred-company run.

def enrich_batch(conn, queries: list[str], chunk_size: int = 200) -> dict[str, dict]:
    out = {q: lookup(conn, q) for q in queries}
    misses = [q for q, hit in out.items() if hit is None]

    for i in range(0, len(misses), chunk_size):
        chunk = misses[i:i + chunk_size]
        run = client.actor(ACTOR).call(
            run_input={
                "companies": chunk,
                "maxResults": len(chunk),
                "concurrency": 20,
                "proxyConfiguration": {"useApifyProxy": True},
            }
        )
        for record in client.dataset(run["defaultDatasetId"]).iterate_items():
            store(conn, record)
            out[record["source_query"]] = record
        conn.commit()

    return {
        q: annotate(rec, q) if rec else placeholder(q)
        for q, rec in out.items()
    }

Chunking at a couple of hundred keeps blast radius small: if one run fails, you retry two hundred lookups, not twenty thousand. Raise concurrency toward its ceiling of 30 for large backfills and drop it back toward the default of 10 if the run log shows repeated retries.

How do I expose it as an HTTP endpoint?

Wrap the batch function in a single route. Accept a list, return a map, and never make the caller poll.

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class EnrichRequest(BaseModel):
    companies: list[str]

@app.post("/v1/enrich")
def enrich_endpoint(req: EnrichRequest):
    with sqlite3.connect("cache.db") as conn:
        return {"results": enrich_batch(conn, req.companies[:500])}

Cap the request size, return confidence on every row, and your callers get a company enrichment API with cache hits served in single-digit milliseconds and misses filled on demand.

Sample output

Each resolved company produces exactly one record. Fields absent on LinkedIn itself come back missing rather than guessed, which is what makes the confidence flag above meaningful.

[
  {
    "name": "HashiCorp",
    "slug": "hashicorp",
    "company_id": "3160246",
    "linkedin_url": "https://www.linkedin.com/company/hashicorp",
    "tagline": "Infrastructure enables innovation",
    "industry": "Software Development",
    "company_size": "1001-5000 employees",
    "employee_count_min": 1001,
    "employee_count_max": 5000,
    "employees_on_linkedin": 2184,
    "follower_count": 231470,
    "founded_year": 2012,
    "company_type": "Public Company",
    "specialties": ["Terraform", "Vault", "Consul", "Nomad"],
    "website": "https://www.hashicorp.com",
    "website_domain": "hashicorp.com",
    "headquarters": "San Francisco, California, US",
    "hq_city": "San Francisco",
    "hq_region": "California",
    "hq_country": "US",
    "locations_count": 4,
    "logo_url": "https://media.licdn.com/dms/image/...",
    "scraped_at": "2026-07-28T09:14:22Z",
    "source_query": "https://www.linkedin.com/company/hashicorp/"
  },
  {
    "name": "Stripe",
    "slug": "stripe",
    "industry": "Financial Services",
    "company_size": "5001-10000 employees",
    "employees_on_linkedin": 8926,
    "founded_year": 2010,
    "company_type": "Privately Held",
    "website_domain": "stripe.com",
    "hq_country": "US",
    "scraped_at": "2026-07-28T09:14:23Z",
    "source_query": "stripe.com"
  }
]

source_query is your join key back to the caller's input, scraped_at is your TTL and audit column, and website_domain is what you diff against a requested domain to set confidence. employee_count_min and employee_count_max are the size band parsed into integers, which is what you want for range filters — company_size is the display string.

Common pitfalls

Domain resolution is the honest weak point. Bare website domains land on the correct LinkedIn company roughly three times in four; heavily branded or ambiguous domains can miss, and plain company names are fuzzier still — a common name may resolve to a larger company that shares it. If your product ever has a LinkedIn URL or slug on hand, pass that instead, because those resolve deterministically.

Second, optional fields are genuinely optional. founded_year, specialties and employees_on_linkedin are absent on pages where the company never filled them in, so any code that indexes them directly will break on a real list. Use .get() and render nulls.

Third, LinkedIn rate-limits by IP, which is why the run keeps proxy rotation on by default. If the log shows repeated retries, lower concurrency before assuming a bug. Finally, only public company pages are in scope — there is no member-gated or private data behind this. The actor handles resolution, rotation, retries and parsing; your job is the cache, the confidence flag and the endpoint contract.

Related use cases

Once the endpoint exists, the same records feed several adjacent jobs:

• Backfill firmographics on CRM accounts — bulk-fill blank Industry, Employee Count and HQ Country fields.
• Route inbound signups by company size and industry — score a work-email signup within seconds of capture.
• Screen startup dealflow with company firmographics — pre-filter an inbound list against mandate criteria.
• Size a B2B market with LinkedIn company firmographics — bottom-up TAM from named companies.
• Guide to scraping business data — the category pillar, plus the full blog index and the actor page.