How to Do Web Scraping for LLM Pipelines (Clean Data for RAG)

Scrape pages for prose. Query an API for facts. That split is the whole design. Prose goes through fetch, extract and chunk. People and company facts come from structured records.

A pipeline that scrapes everything then fights the noise. Navigation menus, cookie text and stale firmographics end up in the vector store, and the model quotes them back.

This guide shows both halves with code you can run today. It is shorter than a full crawler tutorial on purpose.

TL;DR

• You end with a JSONL file of chunks. Each chunk carries a source URL and a fetch date, ready for any embedding model.
• It takes about 45 minutes from key to file.
• It spends about 26 LeadOcean records: 1 company enrichment and 25 people rows. The Free plan has 1,000 records for life, no card.

Prerequisites

• A free LeadOcean account and an API key from app.leadocean.io, exported as $LEADOCEAN_API_KEY.
• Python 3.10 or newer with requests. The extraction code below uses only the standard library.
• A page you are allowed to fetch. Check the site's robots.txt and terms before you crawl it.

export LEADOCEAN_API_KEY="paste-your-key-here"
curl -s https://api.leadocean.io/v1/account -H "x-api-key: $LEADOCEAN_API_KEY"

The account call is free. It confirms the key and shows what is left.

Step 1: Fetch the page and keep its metadata

Fetch with a plain GET first. Only reach for a headless browser when the text is missing from the response, because rendering is slower and heavier.

import requests, datetime

def fetch(url):
    r = requests.get(url, timeout=30, headers={"user-agent": "my-rag-bot/1.0 (you@example.com)"})
    r.raise_for_status()
    return {
        "url": url,
        "fetched_at": datetime.date.today().isoformat(),
        "html": r.text,
    }

Set a real user agent with a contact address. Store url and fetched_at now. You cannot rebuild them later, and a retrieval answer without a source is hard to trust.

Step 2: Extract the main text as Markdown-like blocks

Drop everything that is not content, then keep headings and paragraphs. This version removes script, style, nav, header, footer and aside tags and turns headings into Markdown markers.

from html.parser import HTMLParser

SKIP = {"script", "style", "nav", "header", "footer", "aside", "form", "noscript"}

class Extract(HTMLParser):
    def __init__(self):
        super().__init__()
        self.out, self.depth, self.tag = [], 0, None
    def handle_starttag(self, tag, attrs):
        if tag in SKIP: self.depth += 1
        self.tag = tag
    def handle_endtag(self, tag):
        if tag in SKIP and self.depth: self.depth -= 1
    def handle_data(self, data):
        text = " ".join(data.split())
        if not text or self.depth: return
        if self.tag in ("h1", "h2", "h3"):
            self.out.append("#" * int(self.tag[1]) + " " + text)
        else:
            self.out.append(text)

def to_text(html):
    p = Extract()
    p.feed(html)
    return "\n\n".join(p.out)

Open the result and read it before you build on it. This parser is a baseline, not a finished extractor. Pages with heavy layout tables or text inside images need more work. If you still see a menu or a banner, add its tag to SKIP. Ten minutes here saves a week of bad retrieval.

Step 3: Chunk on headings and tag every chunk

Split on headings, not on a fixed character count. A chunk that matches one idea retrieves better than one that cuts a sentence in half.

import json, re

def chunk(text, meta, max_words=300):
    parts = re.split(r"(?m)^(?=#{1,3} )", text)
    for i, part in enumerate(p for p in parts if p.strip()):
        words = part.split()
        for j in range(0, len(words), max_words):
            yield {**meta, "chunk": f"{i}.{j // max_words}", "text": " ".join(words[j:j + max_words])}

page = fetch("https://example.com/about")
with open("chunks.jsonl", "a") as f:
    for c in chunk(to_text(page["html"]), {"url": page["url"], "fetched_at": page["fetched_at"], "type": "web"}):
        f.write(json.dumps(c) + "\n")

Every row has url, fetched_at and type. Filter on those at query time, for example to drop anything older than 90 days. Keep chunks under about 300 words so one retrieved chunk fits easily beside the question in the model's context. Tune that number against your own retrieval tests, not a rule of thumb.

Step 4: Replace scraped facts with a company record

Headcount, industry and technology on a website are marketing copy and often out of date. Ask the database instead. POST /v1/companies/enrich takes a domain or a linkedin_url.

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"}'

A company we hold nothing for is a 404 and costs nothing. A company we return costs one record. The response has four blocks: company_data, company_detected_technologies, company_metrics and meta.

Turn the record into one short chunk so retrieval treats it like any other source.

def company_chunk(rec):
    return {
        "url": "https://api.leadocean.io/v1/companies/enrich",
        "fetched_at": rec["data"]["meta"].get("fetched_at"),
        "type": "record",
        "chunk": "company.0",
        "text": json.dumps(rec["data"]["company_data"]),
    }

The dataset refreshes monthly and each record carries its own fetched_at, so the chunk keeps a true date.

Step 5: Pull people records instead of scraping team pages

When the agent needs a name, do not scrape a team page. Size the audience first, because a count is free.

On 2026-10-21 the free leadocean_count_leads MCP tool returned this. The filters were jobLevel: Director, jobFunction: Engineering and country: US.

{
  "total": 64703,
  "totalIsExact": true,
  "totalAll": 129633,
  "distinctCompanies": 21626,
  "credits": 0
}

total counts people with a mailable email, the tool's default. totalAll is everyone matching the same filters. Now pull one page. The body is flat JSON, filters are arrays and limit runs from 1 to 100.

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

Search costs one record per person returned. Pass meta.nextCursor back as cursor for the next page. One search pages to 10,000 rows, then meta.depthCapped is true. To list one company's staff use GET /v1/companies/acme.com/people.

If your agent runs in Claude or another MCP client, skip the glue code. The same count and search run as tools on https://api.leadocean.io/mcp, with 13 tools in all and OAuth sign-in instead of a pasted key.

What you get

chunks.jsonl holds one JSON object per line, web chunks and record chunks side by side. This is the shape, with placeholder values.

{"url": "https://example.com/about", "fetched_at": "2026-10-21", "type": "web", "chunk": "1.0", "text": "## About Acme\n\nAcme builds ..."}
{"url": "https://api.leadocean.io/v1/companies/enrich", "fetched_at": "2026-10-01", "type": "record", "chunk": "company.0", "text": "{\"name\": \"Acme\", ...}"}

The company call returns this response shape from the API reference, with the inner fields left out and a placeholder date.

{
  "success": true,
  "data": {
    "company_data": {},
    "company_detected_technologies": {},
    "company_metrics": {},
    "meta": { "fetched_at": "2026-10-01" }
  },
  "meta": {}
}

Embed the text field with the model of your choice and store the other keys as metadata. At query time, filter by type and fetched_at, then cite url in the answer.

Troubleshooting

FAQ