Turning Messy SERP Data Into Clean JSON Your Code Can Trust

The gap between a search response and a typed record is where most extraction pipelines quietly break. Schemas plus validation close it.

Profile picture of Serply
Serply
Search results being transformed into validated structured records

Extraction pipelines that “work” in a notebook fall over in production for one reason: the model returns a shape you didn’t expect, and nothing checks. A missing key becomes a KeyError three functions later, or worse, a .get() default that silently writes zeros into your database.

The fix is unglamorous. Define the schema, validate every record, and route failures somewhere you’ll see them.

The pipeline shape

Search → read pages → extract per page → validate → merge. Each stage is independently testable, which matters because the failure modes are different at each one.

import os
import json
import requests
from urllib.parse import quote_plus
from concurrent.futures import ThreadPoolExecutor

API_KEY = os.environ["SERPLY_API_KEY"]
BASE = "https://api.serply.io/v1"


def search(query: str, num: int = 10) -> list[dict]:
    resp = requests.get(
        f"{BASE}/search/q={quote_plus(query)}&num={num}",
        headers={"X-Api-Key": API_KEY, "X-Proxy-Location": "US"},
        timeout=30,
    )
    resp.raise_for_status()
    return resp.json().get("results", [])


def read(url: str) -> str | None:
    try:
        resp = requests.post(
            f"{BASE}/request",
            headers={"X-Api-Key": API_KEY, "Content-Type": "application/json"},
            json={"url": url, "response_type": "markdown"},
            timeout=60,
        )
        resp.raise_for_status()
        return resp.text
    except requests.RequestException:
        return None

Markdown mode returns text as the body, so .text rather than .json(). Markdown is also the better input for extraction than raw HTML — it keeps heading structure and tables while dropping the nav, scripts, and cookie banners that would otherwise eat most of your context window.

Define the schema first

from pydantic import BaseModel, Field, HttpUrl, field_validator
from typing import Literal


class CompanyRecord(BaseModel):
    name: str
    website: HttpUrl | None = None
    headquarters: str | None = Field(
        None, description="City and country, e.g. 'Berlin, Germany'"
    )
    employee_range: Literal[
        "1-10", "11-50", "51-200", "201-500", "501-1000", "1000+"
    ] | None = None
    founded_year: int | None = None
    description: str | None = Field(None, max_length=500)

    source_url: str
    confidence: Literal["high", "medium", "low"]

    @field_validator("founded_year")
    @classmethod
    def plausible_year(cls, v: int | None) -> int | None:
        if v is not None and not (1600 <= v <= 2030):
            raise ValueError(f"implausible founding year: {v}")
        return v

Three things this buys you.

Every business field is Optional. Nothing forces the model to guess, which is the main cause of fabricated extractions — a required field with no supporting text gets filled with something plausible.

employee_range is an enum rather than an integer. Asking for a headcount produces confident invented numbers; asking which bucket a company falls in produces either a correct bucket or a null.

source_url and confidence are required. Every record can be traced back and triaged.

Constrain the extraction

from anthropic import Anthropic

client = Anthropic()

EXTRACT_TOOL = {
    "name": "record_company",
    "description": "Record structured facts about a company from source text.",
    "input_schema": CompanyRecord.model_json_schema(),
}

SYSTEM = """Extract company facts from the provided page text.

Rules:
- Use null for anything the text does not state. Never infer, estimate, or
  fill from background knowledge.
- Copy values as written. Do not normalise "SF" to "San Francisco" unless
  the text says San Francisco.
- confidence: "high" if the page is the company's own site or an official
  profile; "medium" for a reputable third party; "low" for an aggregator,
  directory, or anything ambiguous.
- If the page is not about a company at all, still call the tool with the
  name you can determine and everything else null, confidence "low"."""


def extract(text: str, url: str) -> CompanyRecord | None:
    msg = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=1500,
        system=SYSTEM,
        tools=[EXTRACT_TOOL],
        tool_choice={"type": "tool", "name": "record_company"},
        messages=[{"role": "user",
                   "content": f"SOURCE URL: {url}\n\nPAGE TEXT:\n{text[:20000]}"}],
    )
    for block in msg.content:
        if block.type == "tool_use":
            data = {**block.input, "source_url": url}
            try:
                return CompanyRecord(**data)
            except Exception as e:
                log_rejected(url, block.input, str(e))
                return None
    return None

tool_choice forced to a specific tool is what makes this reliable. Without it the model sometimes answers in prose, and you’re back to regex-hunting for a JSON block. Generating the schema from the Pydantic model means the tool definition and the validator can’t drift apart.

The try/except around construction is the part people skip. A schema the model usually satisfies will be violated eventually, and you want that record in a rejects table, not raising in the middle of a batch.

Merging across sources

Multiple pages about the same company will disagree. Resolve by confidence, then by agreement:

from collections import Counter


def merge(records: list[CompanyRecord]) -> dict:
    if not records:
        return {}

    rank = {"high": 3, "medium": 2, "low": 1}
    merged: dict = {"sources": [r.source_url for r in records]}

    for field in CompanyRecord.model_fields:
        if field in ("source_url", "confidence"):
            continue

        values = [
            (getattr(r, field), rank[r.confidence])
            for r in records
            if getattr(r, field) is not None
        ]
        if not values:
            merged[field] = None
            continue

        counts = Counter(str(v) for v, _ in values)
        best = max(values, key=lambda vw: (counts[str(vw[0])], vw[1]))
        merged[field] = best[0]
        merged[f"{field}_agreement"] = counts[str(best[0])] / len(values)

    return merged

Sorting by agreement count first and confidence second means three medium-confidence sources beat one high-confidence outlier — which is usually right, because the outlier is often a stale directory listing.

Emitting an _agreement ratio per field lets downstream code apply its own bar. A CRM import might accept anything above 0.5; a published report might want 1.0.

Running it

def build_record(company: str) -> dict:
    results = search(f"{company} company headquarters founded", num=8)
    urls = [r["link"] for r in results[:5] if r.get("link")]

    with ThreadPoolExecutor(max_workers=5) as pool:
        pages = list(pool.map(read, urls))

    records = [
        rec for url, text in zip(urls, pages) if text
        for rec in [extract(text, url)] if rec
    ]
    return merge(records)

What to watch

Two metrics tell you whether this is healthy: the rejection rate from the validator, and the share of fields coming back None.

A rising rejection rate means the model’s output has drifted from the schema — usually after a prompt edit. A high None rate isn’t a bug; it’s the pipeline correctly declining to invent data, and it’s what you traded for trustworthy records. If it’s too high, the problem is upstream in your search queries, not in the extraction.