// field guide 003
Build a Google SERP API with Python, FastAPI, and Playwright
Run a complete Google search API that drives Chromium, paginates organic results, detects robot checks, caps browser concurrency, and returns typed JSON.
This is a complete Google SERP API built around a real Chromium browser. FastAPI exposes a REST API; Playwright loads the search page and reads organic headings from the live DOM. The code runs as one file.
01 / outcome
What you will build
The finished service exposes GET /health and GET /search. A search request controls query, page count, requested results per page, country, language, and the delay between pages. The response reports how many pages were scraped and returns ordered organic rows.
Each row contains its global position, source page, title, destination URL, displayed URL, and snippet when the rendered result exposes one. The parser leaves missing snippets as empty strings.
The browser makes this easy to understand but is expensive to operate. One small API call can execute megabytes of JavaScript, create renderer processes, allocate hundreds of megabytes of memory, and spend seconds waiting for a page lifecycle.
02 / browser boundary
A browser solves rendering, not reliability
Playwright can read the result page after JavaScript runs, but Google can still vary the page by region, browser state, experiments, network reputation, and automation signals. Google may challenge either headed or headless Chromium, so neither mode guarantees reliable access.
The sample removes two obvious automation flags. It includes no CAPTCHA solver, fingerprint farm, proxy rotator, or challenge bypass. When Google serves /sorry/, a reCAPTCHA frame, or unusual-traffic text, the request stops and returns upstream_blocked.
03 / setup
Install the API and browser runtime
Use Python 3.11 or newer. The Python package controls Chromium over Playwright's driver, while the separate install command downloads the compatible browser. CI and minimal Linux images may need playwright install --with-deps chromium so the required system libraries are present.
python3 -m venv .venvsource .venv/bin/activatepython -m pip install --upgrade pip fastapi uvicorn httpx playwrightplaywright install chromium04 / HTTP contract
Validate every control before opening a page
The route bounds inputs with FastAPI's typed query parameters. A malformed country, negative delay, zero pages, or out-of-range result count receives HTTP 422 before BrowserRuntime.search() runs. A whitespace-only query receives a stable invalid_query error.
| Parameter | Default | Accepted values |
|---|---|---|
q | Required | 1–200 characters after trimming |
pages | 1 | 1–10 |
results_per_page | 10 | 10–100 |
country | US | Two ASCII letters |
language | en | Language or language-region |
delay_ms | 1,000 | 0–10,000 milliseconds |
Google may return fewer rows than results_per_page. Search features, omitted duplicates, query interpretation, and result availability all affect the rendered page. The value caps the request; the rendered page may return fewer rows.
05 / lifecycle
Reuse Chromium without sharing caller state
FastAPI's lifespan starts the Playwright driver and Chromium once. The browser process is the expensive shared resource. Every search still creates a fresh context with its own cookies, storage, locale, and viewport, then closes that context in a finally block.
Pagination happens sequentially in one page by changing the start query parameter. Sequential navigation retains the same request context and makes the inter-page delay meaningful. Running all pages in parallel would multiply renderer load and send a much sharper traffic burst.
/health exposes browser_ready. Browser startup failure leaves the web process alive but degraded; search calls return HTTP 503 instead of hanging on a missing browser object.
06 / extraction
Use semantic headings, then narrow to result cards
The parser starts from clickable organic headings with a h3. It walks from each heading to the anchor and nearest known result-card container, then reads the displayed source and snippet. One browser evaluation returns all candidate rows for the page.
clean_rows() unwraps google.com/url?q=... links, accepts only absolute HTTP destinations, filters internal Google pages, removes fragments, and deduplicates across every requested page. Positions are assigned only after filtering, so they stay contiguous in the public response.
07 / complete file
The complete Google SERP browser API
Save this as google_serp_api.py. It includes the typed models, URL builder, redirect normalizer, Chromium lifecycle, concurrency limit, DOM extractor, block detection, stable errors, routes, and local Uvicorn entry point.
from __future__ import annotations import asyncioimport osimport sysimport timefrom contextlib import asynccontextmanagerfrom typing import Annotated, Any, AsyncIteratorfrom urllib.parse import parse_qs, urlencode, urlsplit, urlunsplit from fastapi import FastAPI, Query, Requestfrom fastapi.responses import JSONResponsefrom pydantic import BaseModel, Field ENGINE = "google"RESULT_SELECTOR = "a h3"MAX_CONCURRENT_REQUESTS = 2BLOCKED_MARKERS = ( "our systems have detected unusual traffic", "unusual traffic from your computer network", "verify you're not a robot",) class OrganicResult(BaseModel): position: int = Field(ge=1) page: int = Field(ge=1) title: str url: str displayed_url: str = "" snippet: str = "" class SearchResponse(BaseModel): engine: str query: str requested_pages: int pages_scraped: int result_count: int duration_ms: int results: list[OrganicResult] class SerpError(Exception): def __init__( self, code: str, message: str, status_code: int, retryable: bool, ) -> None: super().__init__(message) self.code = code self.message = message self.status_code = status_code self.retryable = retryable def build_search_url( query: str, page: int, results_per_page: int, country: str, language: str,) -> str: params = { "q": query, "num": results_per_page, "start": (page - 1) * results_per_page, "gl": country.lower(), "hl": language, "filter": 0, "pws": 0, } return f"https://www.google.com/search?{urlencode(params)}" def unwrap_google_url(value: str) -> str: try: parsed = urlsplit(value) except ValueError: return "" host = (parsed.hostname or "").casefold() if (host == "google.com" or host.endswith(".google.com")) and parsed.path == "/url": return parse_qs(parsed.query).get("q", [value])[0] return value def normalize_url(value: str) -> str: try: parsed = urlsplit(unwrap_google_url(value.strip())) except ValueError: return "" if parsed.scheme not in {"http", "https"} or not parsed.netloc: return "" path = parsed.path.rstrip("/") or "/" return urlunsplit( ( parsed.scheme.casefold(), parsed.netloc.casefold(), path, parsed.query, "", ) ) def clean_rows( rows: list[dict[str, Any]], page: int, starting_position: int, seen: set[str],) -> list[OrganicResult]: results: list[OrganicResult] = [] for row in rows: title = str(row.get("title") or "").strip() url = normalize_url(str(row.get("url") or "")) if not title or not url or url in seen: continue host = (urlsplit(url).hostname or "").casefold() if host == "google.com" or host.endswith(".google.com"): continue seen.add(url) results.append( OrganicResult( position=starting_position + len(results), page=page, title=title, url=url, displayed_url=str(row.get("displayed_url") or "").strip(), snippet=str(row.get("snippet") or "").strip(), ) ) return results async def dismiss_consent(page: Any) -> None: for label in ("Reject all", "Accept all", "I agree"): button = page.get_by_role("button", name=label, exact=True) try: if await button.count() and await button.first.is_visible(): await button.first.click(timeout=2_000) await page.wait_for_timeout(300) return except Exception: continue async def raise_if_blocked(page: Any) -> None: lowered_url = page.url.casefold() if "/sorry/" in lowered_url or "captcha" in lowered_url: raise SerpError( "upstream_blocked", "Google returned a robot-check page. Stop and retry later or use a headed browser.", 429, True, ) if await page.locator('form[action*="/sorry/"], iframe[src*="recaptcha"]').count(): raise SerpError( "upstream_blocked", "Google returned a robot-check page. Stop and retry later or use a headed browser.", 429, True, ) try: body = (await page.locator("body").inner_text(timeout=3_000)).casefold() except Exception: return if any(marker in body for marker in BLOCKED_MARKERS): raise SerpError( "upstream_blocked", "Google returned a robot-check page. Stop and retry later or use a headed browser.", 429, True, ) async def rows_from_page(page: Any) -> list[dict[str, str]]: headings = page.locator("a h3") return await headings.evaluate_all( """ headings => headings.map(heading => { const anchor = heading.closest('a'); const card = heading.closest( 'div.N54PNb, div.MjjYud, div.tF2Cxc, div.Gx5Zad, div[data-snhf]' ) || anchor?.parentElement?.parentElement; const displayed = card?.querySelector('cite, .VuuXrf'); const snippet = card?.querySelector( '.VwiC3b, [data-sncf], .IsZvec, [data-content-feature="1"]' ); const cardText = card?.innerText?.trim() || ''; const anchorText = anchor?.innerText?.trim() || ''; const fallbackSnippet = cardText.startsWith(anchorText) ? cardText.slice(anchorText.length).trim() : ''; return { title: heading.textContent?.trim() || '', url: anchor?.href || '', displayed_url: displayed?.textContent?.trim() || '', snippet: snippet?.textContent?.trim() || fallbackSnippet }; }) """ ) class BrowserRuntime: def __init__(self) -> None: self.playwright: Any = None self.browser: Any = None self.semaphore = asyncio.Semaphore(MAX_CONCURRENT_REQUESTS) @property def ready(self) -> bool: return self.browser is not None and self.browser.is_connected() async def start(self) -> None: from playwright.async_api import async_playwright self.playwright = await async_playwright().start() headless = os.getenv("SERP_HEADLESS", "true").casefold() not in {"0", "false", "no"} self.browser = await self.playwright.chromium.launch( headless=headless, args=[ "--disable-dev-shm-usage", "--disable-blink-features=AutomationControlled", ], ) async def stop(self) -> None: if self.browser is not None: if self.browser.is_connected(): try: await self.browser.close() except Exception: pass self.browser = None if self.playwright is not None: try: await self.playwright.stop() except Exception: pass self.playwright = None async def search( self, query: str, pages: int, results_per_page: int, country: str, language: str, delay_ms: int, ) -> tuple[list[OrganicResult], int]: if not self.ready: raise SerpError( "browser_unavailable", "Chromium is not ready. Check the server startup logs.", 503, True, ) from playwright.async_api import TimeoutError as PlaywrightTimeoutError async with self.semaphore: context = await self.browser.new_context( locale=language, viewport={"width": 1440, "height": 1000}, ) await context.add_init_script( "Object.defineProperty(navigator, 'webdriver', {get: () => undefined})" ) page = await context.new_page() results: list[OrganicResult] = [] seen: set[str] = set() pages_scraped = 0 try: for page_number in range(1, pages + 1): url = build_search_url( query, page_number, results_per_page, country, language, ) try: await page.goto(url, wait_until="domcontentloaded", timeout=45_000) await dismiss_consent(page) await raise_if_blocked(page) await page.locator(RESULT_SELECTOR).first.wait_for( state="visible", timeout=15_000, ) await page.wait_for_timeout(350) except SerpError: raise except PlaywrightTimeoutError as error: await raise_if_blocked(page) no_results = await page.get_by_text( "did not match any documents", exact=False, ).count() if no_results: break raise SerpError( "upstream_layout_changed", "Google loaded, but organic result headings were not found.", 502, False, ) from error rows = await rows_from_page(page) cleaned = clean_rows( rows, page_number, len(results) + 1, seen, ) results.extend(cleaned) pages_scraped += 1 if not cleaned: break if page_number < pages: await page.wait_for_timeout(delay_ms) finally: await context.close() return results, pages_scraped runtime = BrowserRuntime() @asynccontextmanagerasync def lifespan(_: FastAPI) -> AsyncIterator[None]: try: await runtime.start() except Exception as error: print(f"Chromium startup failed: {type(error).__name__}: {error}", file=sys.stderr) yield await runtime.stop() app = FastAPI( title="Google SERP Browser API", version="1.0.0", lifespan=lifespan,) @app.exception_handler(SerpError)async def serp_error_handler(_: Request, error: SerpError) -> JSONResponse: return JSONResponse( { "error": error.message, "error_code": error.code, "status_code": error.status_code, "retryable": error.retryable, }, status_code=error.status_code, ) @app.get("/health")async def health() -> dict[str, str | bool]: return { "status": "ok" if runtime.ready else "degraded", "engine": ENGINE, "browser_ready": runtime.ready, } @app.get("/search", response_model=SearchResponse)async def search( q: Annotated[str, Query(min_length=1, max_length=200)], pages: Annotated[int, Query(ge=1, le=10)] = 1, results_per_page: Annotated[int, Query(ge=10, le=100)] = 10, country: Annotated[str, Query(pattern=r"^[A-Za-z]{2}$")] = "US", language: Annotated[str, Query(pattern=r"^[a-z]{2}(?:-[A-Z]{2})?$")] = "en", delay_ms: Annotated[int, Query(ge=0, le=10_000)] = 1_000,) -> SearchResponse: query = q.strip() if not query: raise SerpError("invalid_query", "q cannot be blank.", 422, False) started = time.perf_counter() try: results, pages_scraped = await runtime.search( query, pages, results_per_page, country.upper(), language, delay_ms, ) except SerpError: raise except Exception as error: print(f"Unhandled search error: {type(error).__name__}: {error}", file=sys.stderr) raise SerpError( "internal_error", "The browser request failed unexpectedly.", 500, False, ) from error return SearchResponse( engine=ENGINE, query=query, requested_pages=pages, pages_scraped=pages_scraped, result_count=len(results), duration_ms=round((time.perf_counter() - started) * 1_000), results=results, ) if __name__ == "__main__": import uvicorn uvicorn.run(app, host="127.0.0.1", port=int(os.getenv("PORT", "8000")))08 / run it
Start locally and make the first request with httpx
The file binds to localhost on port 8000. Start with a visible browser when debugging on a desktop. Linux servers without a display need a virtual display for headed mode; headless mode is the default.
python google_serp_api.py # Watch a visible browser while debugging:SERP_HEADLESS=false python google_serp_api.pyThe client uses a 60-second timeout because browser startup, semaphore queueing, and page navigation can all exceed the default timeout of a typical HTTP client. Keep page counts small until you have measured the host and upstream behavior.
import httpx response = httpx.get( "http://127.0.0.1:8000/search", params={ "q": "San Francisco Coffee Shops", "pages": 1, "results_per_page": 10, "country": "US", "language": "en", }, timeout=60,)response.raise_for_status() for result in response.json()["results"]: print(result["position"], result["title"], result["url"])09 / measured run
Verified with the exact file and a real Google result page
On July 17, 2026, we started the published file under a virtual display and requested one page for San Francisco Coffee Shops. Google returned a robot-check page, and the API responded with HTTP 429 and error_code: upstream_blocked.
import httpx response = httpx.get( "http://127.0.0.1:8000/search", params={ "q": "San Francisco Coffee Shops", "pages": 1, "results_per_page": 10, "country": "US", "language": "en", }, timeout=60,)response.raise_for_status() for result in response.json()["results"]: print(result["position"], result["title"], result["url"])The 429 response confirms block detection and the public error envelope. Run the same request from your own network to test organic extraction; challenge rates depend on IP reputation and browser state.
10 / operations
Return explicit failures instead of empty datasets
| Signal | Meaning | Response |
|---|---|---|
422 | Invalid parameter or blank query | Correct the request; browser is untouched |
429 upstream_blocked | Robot check or unusual-traffic page | Stop, lower frequency, and retry later |
502 upstream_layout_changed | No organic headings were found | Inspect a headed run and update the parser |
503 browser_unavailable | Chromium startup failed | Verify binaries, libraries, memory, and logs |
500 internal_error | Unexpected application failure | Read server logs; response hides exception text |
A real deployment also needs bounded request queues, worker recycling, crash metrics, structured logs, alerting on result completeness, and per-host concurrency tuning. Do not interpret a blank page as zero results. The code differentiates a known no-results message from a changed layout or challenge so callers can detect silent data loss.