Build a Lead-Enrichment Agent That Researches Companies Using Serper and Gemini

What We're Building

A command-line agent that reads a CSV of company domains, scrapes live web data for each domain, and uses Gemini to synthesize structured prospect profiles. The output is a clean, queryable Supabase table containing company descriptions, industry tags, estimated size, tech stack hints, and a custom relevance score for your ICP.

Feature list:

• Ingest any CSV with a domain column
• Concurrent web scraping via Serper's free tier (2,500 queries/month)
• Gemini Flash enrichment: company summary, industry, size, pain points, tech stack
• Structured output persisted to Supabase
• Rate limiting and retry logic baked in
• Dry-run mode to estimate API costs before firing

If you've built the cold-outreach personalizer before, this is the upstream step that generates the raw intel you'd feed into it.

Architecture Overview

The orchestrator reads one domain at a time, fires a multi-query search to Serper (company site + LinkedIn + Crunchbase), concatenates the top snippet results, and passes that context to Gemini with a strict JSON schema prompt. The parsed profile lands in Supabase. We batch-process with asyncio and a semaphore to respect free-tier rate limits.

Prerequisites (All Free-Tier)

Grab API keys for Gemini and Serper, then store them in a .env file:

GEMINI_API_KEY=your-gemini-key
SERPER_API_KEY=your-serper-key
SUPABASE_URL=https://your-project.supabase.co
SUPABASE_SERVICE_ROLE_KEY=your-service-role-key

Use the service role key for Supabase (server-side only) so we can bypass RLS. Never expose this in client code.

Step 1: Project Setup and Dependencies

mkdir lead-enrichment-agent && cd lead-enrichment-agent
python -m venv .venv && source .venv/bin/activate
pip install httpx python-dotenv supabase pydantic

Create enrich.py and import everything upfront:

import asyncio
import csv
import json
import os
from pathlib import Path
from typing import Optional

import httpx
from dotenv import load_dotenv
from pydantic import BaseModel
from supabase import create_client, Client

load_dotenv()

GEMINI_KEY = os.environ["GEMINI_API_KEY"]
SERPER_KEY = os.environ["SERPER_API_KEY"]
SUPABASE_URL = os.environ["SUPABASE_URL"]
SUPABASE_KEY = os.environ["SUPABASE_SERVICE_ROLE_KEY"]

GEMINI_URL = "https://generativelanguage.googleapis.com/v1beta/models/gemini-2.0-flash:generateContent"
SERPER_URL = "https://google.serper.dev/search"

We use Pydantic to define the enrichment schema so we can validate Gemini's output before inserting.

Step 2: Supabase Database Schema

Run this in the Supabase SQL Editor:

create table if not exists public.company_profiles (
  id uuid primary key default gen_random_uuid(),
  domain text unique not null,
  company_name text,
  description text,
  industry text,
  estimated_employees text,
  tech_stack text[],
  pain_points text[],
  relevance_score integer check (relevance_score between 1 and 10),
  raw_search_snippets text,
  created_at timestamptz default now(),
  updated_at timestamptz default now()
);

create index idx_company_profiles_domain on public.company_profiles(domain);

A unique constraint on domain lets us use upsert safely. The tech_stack and pain_points columns are arrays—Gemini returns lists, and Supabase's JSONB coercion handles the rest.

Initialize the Supabase client:

supabase: Client = create_client(SUPABASE_URL, SUPABASE_KEY)

Step 3: Domain Scraping with Serper

We'll query three search variations per domain to maximize signal: the company's own site, their LinkedIn page, and any Crunchbase profile. Serper's free tier gives us 2,500 queries/month—at 3 queries per domain, that's ~830 domains/month.

async def search_domain(client: httpx.AsyncClient, domain: str) -> str:
    queries = [
        f"site:{domain} about company",
        f"{domain} linkedin company",
        f"{domain} crunchbase",
    ]
    all_snippets = []

    for q in queries:
        try:
            resp = await client.post(
                SERPER_URL,
                json={"q": q, "num": 5},
                headers={"X-API-KEY": SERPER_KEY},
                timeout=15.0,
            )
            resp.raise_for_status()
            data = resp.json()
            for result in data.get("organic", []):
                snippet = result.get("snippet", "")
                if snippet:
                    all_snippets.append(snippet)
        except Exception as e:
            print(f"  [!] Serper query failed for '{q}': {e}")
            continue

    return "\n---\n".join(all_snippets[:15])

We cap at 15 snippets to stay under Gemini's context limits while keeping the prompt tight.

Step 4: Enrichment Agent with Gemini

This is where we convert raw search snippets into structured intel. The prompt enforces a JSON schema and asks Gemini to infer what it can, flagging null for anything it can't determine.

ENRICHMENT_PROMPT = """You are a B2B lead researcher. Given search snippets about a company, return a JSON object with these fields:
- company_name: string or null
- description: 2-3 sentence summary
- industry: string or null
- estimated_employees: string (e.g. "1-10", "11-50", "51-200", "201-1000", "1000+") or null
- tech_stack: array of strings (technologies mentioned) or empty array
- pain_points: array of strings (challenges the company likely faces given their industry/size) or empty array
- relevance_score: integer 1-10 based on how likely they need automation/engineering talent (10 = very likely)

Return ONLY valid JSON, no markdown fences.

Search snippets:
{snippets}"""


async def enrich_with_gemini(client: httpx.AsyncClient, domain: str, snippets: str) -> dict:
    prompt = ENRICHMENT_PROMPT.format(snippets=snippets[:6000])  # truncate for safety
    payload = {
        "contents": [{"parts": [{"text": prompt}]}],
        "generationConfig": {"temperature": 0.2, "maxOutputTokens": 1024},
    }

    try:
        resp = await client.post(
            f"{GEMINI_URL}?key={GEMINI_KEY}",
            json=payload,
            timeout=30.0,
        )
        resp.raise_for_status()
        data = resp.json()
        text = data["candidates"][0]["content"]["parts"][0]["text"]
        # Strip accidental markdown fences
        text = text.strip().removeprefix("```json").removesuffix("```").strip()
        return json.loads(text)
    except Exception as e:
        print(f"  [!] Gemini enrichment failed for {domain}: {e}")
        return {}

Temperature 0.2 keeps outputs deterministic. The maxOutputTokens of 1024 is plenty for a single profile.

Step 5: The Orchestration Script

This ties everything together with concurrency control. We use a semaphore to limit parallel requests and a small delay between domains to stay friendly with free-tier rate limits.

async def process_domain(
    client: httpx.AsyncClient,
    sem: asyncio.Semaphore,
    domain: str,
    dry_run: bool = False,
):
    async with sem:
        print(f"\n[*] Processing: {domain}")
        snippets = await search_domain(client, domain)
        if not snippets:
            print(f"  [-] No snippets found for {domain}, skipping.")
            return

        if dry_run:
            print(f"  [dry-run] Would enrich {domain} with {len(snippets)} chars of snippets.")
            return

        profile = await enrich_with_gemini(client, domain, snippets)
        if not profile:
            return

        row = {
            "domain": domain,
            "company_name": profile.get("company_name"),
            "description": profile.get("description"),
            "industry": profile.get("industry"),
            "estimated_employees": profile.get("estimated_employees"),
            "tech_stack": profile.get("tech_stack", []),
            "pain_points": profile.get("pain_points", []),
            "relevance_score": profile.get("relevance_score"),
            "raw_search_snippets": snippets,
        }

        result = supabase.table("company_profiles").upsert(row, on_conflict="domain").execute()
        if hasattr(result, "error") and result.error:
            print(f"  [!] Supabase insert failed: {result.error}")
        else:
            print(f"  [+] Stored profile for {profile.get('company_name') or domain}")


async def main(csv_path: str, concurrency: int = 3, dry_run: bool = False):
    domains = []
    with open(csv_path, newline="") as f:
        reader = csv.DictReader(f)
        for row in reader:
            domain = row.get("domain", "").strip().lower()
            if domain:
                domains.append(domain)

    print(f"Loaded {len(domains)} domains from {csv_path}")
    if dry_run:
        estimated_serper_calls = len(domains) * 3
        print(f"[dry-run] Estimated Serper queries: {estimated_serper_calls}")
        print(f"[dry-run] Estimated Gemini calls: {len(domains)}")

    sem = asyncio.Semaphore(concurrency)
    async with httpx.AsyncClient() as client:
        tasks = [process_domain(client, sem, d, dry_run) for d in domains]
        await asyncio.gather(*tasks)

    print("\nDone.")


if __name__ == "__main__":
    import argparse
    parser = argparse.ArgumentParser()
    parser.add_argument("csv", help="Path to CSV with a 'domain' column")
    parser.add_argument("--concurrency", type=int, default=3, help="Max parallel domains")
    parser.add_argument("--dry-run", action="store_true", help="Estimate API usage without calling")
    args = parser.parse_args()
    asyncio.run(main(args.csv, args.concurrency, args.dry_run))

How to Run It

1. Create a CSV file domains.csv:

domain
stripe.com
vercel.com
linear.app

1. Dry-run to sanity-check:

python enrich.py domains.csv --dry-run

1. Fire for real:

python enrich.py domains.csv --concurrency 2

Start with --concurrency 2 to avoid hitting Gemini's 15 RPM limit. Bump to 3 once you've confirmed the rhythm.

1. Query your profiles in Supabase:

select domain, company_name, industry, relevance_score
from public.company_profiles
order by relevance_score desc;

Sensible Extensions

• ICP filtering: Add a relevance_score threshold and only insert profiles scoring 7+. This keeps your DB clean for sales workflows.
• Webhook trigger: Pipe new profiles into Slack or Discord. If you've built the Discord FAQ bot, you can reuse that webhook pattern.
• Cron deployment: Ship this to GitHub Actions with a scheduled workflow. The free tier gives you 2,000 minutes/month—plenty for a daily enrichment run on a few hundred domains.
• Link to outreach: Feed these profiles directly into the cold-outreach personalizer to generate hyper-personalized emails from the same enriched data.
• Multi-source scraping: Add Firecrawl's free tier to scrape actual page text instead of relying solely on search snippets. Better fidelity, same zero-cost stack.

Common Pitfalls

1. Gemini returns markdown-wrapped JSON. The .removeprefix("```json") handles most cases, but occasionally Gemini adds a trailing explanation. If you see parse failures, add a regex extraction step that grabs the first {...} block.

2. Serper rate limits. The free tier doesn't publish a strict RPM limit, but I've seen 429s above 10 concurrent requests. Stick to --concurrency 3 and add a 1-second delay between Serper calls if you hit issues.

3. Empty snippets. Some domains (especially stealth startups) return zero useful snippets. The script skips them gracefully, but you might want to log these to a failed_domains.csv for manual review.

4. Supabase connection leaks. Always use a single httpx.AsyncClient session and let the context manager close it. Recreating the Supabase client per domain will exhaust connection pools on large batches.

5. Gemini free-tier quota. The 1,500 requests/day limit resets at midnight Pacific. If you're processing a large list, spread it across days or upgrade to pay-as-you-go (still dirt cheap).

FAQ

Q: Why Serper instead of direct scraping? A: Serper aggregates Google results, giving us structured snippets from multiple sources (company site, LinkedIn, Crunchbase) in one API call. Direct scraping requires per-site parsers and breaks constantly. Serper's free tier is generous enough for most outbound workflows.

Q: Can I use GPT-4o-mini instead of Gemini? A: Absolutely. Swap the endpoint and payload shape. GPT-4o-mini's free tier on OpenAI is limited, but the paid tier is ~$0.15/1M input tokens. The architecture doesn't care which LLM sits behind the enrich_with_* function.

Q: How do I handle companies with no web presence? A: The script returns an empty profile dict and skips insertion. You can extend the process_domain function to mark these in a separate table or push them to a manual review queue.

Q: What if my CSV has 10,000 domains? A: The free tiers won't support that in one run. Batch them into 500-domain chunks, run one chunk per day, and use --dry-run first to confirm your estimates. For production-scale enrichment, you'd move to paid API tiers and add a Redis queue—but the code structure stays identical.

Q: How is this different from the cold-outreach personalizer? A: This agent researches the company (top-of-funnel intel). The cold-outreach personalizer researches the person and generates email copy. They're designed to chain together: enrichment first, personalization second." }