The Problem With Selector-Based Scraping
Traditional scrapers break the moment a site ships a redesign — every CSS selector is a brittle contract with someone else's markup. CRW's /v1/extract endpoint flips this: you describe the shape of the data you want with a JSON schema, and an LLM reads the page semantically. One schema works across thousands of differently-built pages. This tutorial shows how to do it reliably, with validation and batching.
Prerequisites
- CRW running:
docker run -p 3000:3000 ghcr.io/us/crw:latest - Python 3.10+ and an OpenAI API key (CRW uses it for the extraction LLM)
pip install firecrawl-py pydantic
Step 1: SDK Setup
from firecrawl import FirecrawlApp
app = FirecrawlApp(api_key="fc-YOUR-KEY", api_url="http://localhost:3000")
# fastCRW cloud: api_url="https://api.fastcrw.com"
Step 2: Define the Schema From a Pydantic Model
Keep one source of truth. Define a Pydantic model, derive the JSON schema from it, and reuse the same model to validate the response:
from pydantic import BaseModel, Field, HttpUrl
from typing import Optional
class CompanyProfile(BaseModel):
name: str = Field(description="The company's official name")
tagline: Optional[str] = Field(None, description="Short marketing tagline")
founded_year: Optional[int] = Field(None, description="Year the company was founded")
employee_count: Optional[str] = Field(
None, description="Headcount or range, e.g. '11-50'")
industry: Optional[str] = None
headquarters: Optional[str] = Field(None, description="City, country")
pricing_model: Optional[str] = Field(
None, description="e.g. 'subscription', 'usage-based', 'free'")
SCHEMA = CompanyProfile.model_json_schema()
Step 3: Extract and Validate One Page
from pydantic import ValidationError
def extract_profile(url: str) -> CompanyProfile | None:
result = app.extract(
urls=[url],
params={
"prompt": "Extract the company profile from this page. Use null for any field you cannot find — do not guess.",
"schema": SCHEMA,
},
)
if not result or "data" not in result:
print(f" no data for {url}")
return None
try:
return CompanyProfile.model_validate(result["data"])
except ValidationError as e:
print(f" validation failed for {url}: {e}")
return None
profile = extract_profile("https://example-startup.com/about")
if profile:
print(profile.model_dump_json(indent=2))
The "use null, do not guess" instruction matters. Without it, LLMs hallucinate plausible values for missing fields. Pydantic validation then catches type mismatches before bad data reaches your database.
Step 4: Nested and Array Schemas
Schemas compose. Extract a list of structured items — for example, a pricing table — with nested objects and arrays:
class Plan(BaseModel):
name: str
monthly_price: Optional[float] = Field(None, description="USD per month, null if custom")
is_free: bool = Field(description="True if this plan costs nothing")
features: list[str] = Field(default_factory=list)
class PricingPage(BaseModel):
currency: str = Field(description="Currency code, e.g. USD")
plans: list[Plan]
def extract_pricing(url: str) -> PricingPage | None:
result = app.extract(
urls=[url],
params={
"prompt": "Extract every pricing plan with its price and feature list.",
"schema": PricingPage.model_json_schema(),
},
)
if result and "data" in result:
try:
return PricingPage.model_validate(result["data"])
except ValidationError as e:
print(f" invalid pricing: {e}")
return None
Step 5: Batch Extraction With Backoff
Production extraction runs over many URLs. Add retry with exponential backoff so a transient failure does not lose a record:
import time
def extract_with_retry(fn, url: str, attempts: int = 3):
delay = 2.0
for i in range(1, attempts + 1):
try:
out = fn(url)
if out is not None:
return out
except Exception as e: # network / API hiccup
print(f" attempt {i} failed for {url}: {e}")
time.sleep(delay)
delay *= 2
return None
def batch_extract(urls: list[str]) -> dict[str, CompanyProfile]:
results: dict[str, CompanyProfile] = {}
for url in urls:
print(f"Extracting {url}")
profile = extract_with_retry(extract_profile, url)
if profile:
results[url] = profile
print(f"Succeeded on {len(results)}/{len(urls)}")
return results
Step 6: Export to CSV
import csv
def export_csv(profiles: dict[str, CompanyProfile], path: str):
if not profiles:
return
fields = list(CompanyProfile.model_fields.keys()) + ["source_url"]
with open(path, "w", newline="") as f:
w = csv.DictWriter(f, fieldnames=fields)
w.writeheader()
for url, p in profiles.items():
row = p.model_dump()
row["source_url"] = url
w.writerow(row)
if __name__ == "__main__":
urls = [
"https://example-startup.com/about",
"https://another-co.com/company",
]
data = batch_extract(urls)
export_csv(data, "companies.csv")
Why Schemas Beat Selectors at Scale
The economic argument for schema extraction is about marginal cost. With CSS or XPath selectors, the cost of supporting one more site is roughly constant — you inspect the markup, write rules, and add a test. Across fifty sites that is fifty parsers to write and, worse, fifty parsers to keep alive as those sites redesign on their own schedules. The maintenance cost grows with the number of sources and never goes away. With a JSON schema the marginal cost of a new site is approximately zero: the same schema and prompt apply because the model reads meaning, not structure. Your engineering effort shifts from "fifty brittle parsers" to "one schema you refine occasionally," and that schema improves every site at once instead of one.
This does not mean selectors are never right. For a single high-volume page whose structure you control or that never changes, a selector is faster and cheaper per request. The schema approach wins decisively when the surface is heterogeneous and evolving — exactly the situation in lead enrichment, competitive monitoring, catalog aggregation, and any "pull this field from arbitrary sites" task. Choose based on the shape of the problem, not dogma.
Constraining the Model With JSON Schema Features
Descriptions are the primary lever, but JSON Schema has more tools to keep output clean, and CRW honors them. Use enum for any categorical field so values are consistent across pages instead of a dozen synonyms for the same concept. Use format hints (e.g. "format": "date") and explicit numeric types so downstream parsing does not have to guess. Keep arrays typed end-to-end so a list never comes back as a comma-joined string:
STRICT_SCHEMA = {
"type": "object",
"properties": {
"company_stage": {
"type": "string",
"enum": ["pre-seed", "seed", "series-a", "series-b", "growth", "public", "unknown"],
"description": "Funding stage; use 'unknown' if not stated",
},
"founded": {"type": "integer", "description": "4-digit founding year"},
"categories": {
"type": "array",
"items": {"type": "string"},
"description": "Industry tags, lowercase, no duplicates",
},
},
"required": ["company_stage"],
}
The enum here is doing real work: without it the model returns "Series A", "series-a", "A round", and "Seed/Series A" interchangeably and your analytics fracture. With it, the output is one of seven known tokens you can group on directly. Constrain aggressively wherever the value space is known — it is cheaper than cleaning the data later.
Failure Handling in Practice
At volume, three failure classes dominate, and each has a distinct response. A transient failure (timeout, momentary block) should be retried with backoff — the extract_with_retry helper handles this. A validation failure (the model returned data that violates the schema) should be logged with the offending payload and the record skipped, never silently coerced — coercing bad data into your database is worse than dropping it. A systematic failure (a whole class of pages returns empty) is a signal to revise the prompt or schema, not to retry harder. Wiring these three responses explicitly, rather than wrapping everything in a bare except, is what makes a batch extractor trustworthy: you end up with a clean dataset plus an honest log of exactly what could not be extracted and why.
Schema Design Tips
- Describe every field — the LLM uses
descriptionas the extraction instruction. Vague descriptions produce vague data. - Make most fields optional — required fields force the model to invent values when they are absent.
- Use enums for categoricals — constrain
pricing_modelwith a JSON Schemaenumso values are consistent across pages. - Validate downstream — never trust raw LLM output; round-trip it through Pydantic before persisting.
Why CRW for Structured Extraction
- Selector-free — one schema generalizes across site redesigns and thousands of differently-built pages.
- Open-core Rust — small single binary, lower-latency, local-first, AGPL-3.0 + Managed Cloud.
- Firecrawl-compatible — the SDK works by changing one URL; no rewrite to adopt.
A Reusable Extraction Wrapper
Tie the pieces together into one helper you can reuse across projects: it takes any Pydantic model and a URL, runs extraction, validates, and returns either a typed object or a structured error you can log. This is the shape you actually want in a codebase rather than scattered app.extract calls:
from typing import TypeVar, Type
from pydantic import BaseModel, ValidationError
T = TypeVar("T", bound=BaseModel)
def extract_as(model: Type[T], url: str,
prompt: str | None = None) -> T | dict:
"""Return a validated model instance, or {'error': ...} on failure."""
res = app.extract(urls=[url], params={
"prompt": prompt or f"Extract the {model.__name__} fields. "
f"Use null for anything missing; do not guess.",
"schema": model.model_json_schema(),
})
if not res or "data" not in res:
return {"error": "no_data", "url": url}
try:
return model.model_validate(res["data"])
except ValidationError as e:
return {"error": "validation", "url": url, "detail": e.errors()}
profile = extract_as(CompanyProfile, "https://example-startup.com/about")
if isinstance(profile, CompanyProfile):
print(profile.name, profile.pricing_model)
else:
print("extraction failed:", profile)
One function, generic over any schema, with the "null not guess" instruction and validation baked in. New extraction tasks become "define a model, call extract_as" — which is the entire point of the schema approach distilled into an API your team will actually use consistently.
Next Steps
- See Build an AI Price Tracker for extraction in a scheduled pipeline
- Read Lead Enrichment with CRW
Self-host CRW from GitHub for free, or use fastCRW for managed cloud scraping.
