Skip to Content

python sdk

the official python sdk — crawlbrulee on pypi. fully typed (ships py.typed), sync and async clients, one runtime dependency (httpx). Python 3.10+.

install

pip install crawlbrulee # or: uv add crawlbrulee

quickstart

from crawlbrulee import Crawlbrulee, ScrapeExtract # pass api_key=..., or use from_env() to read CRAWLBRULEE_API_KEY client = Crawlbrulee.from_env() page = client.scrape( url="https://example.com", extract=ScrapeExtract(markdown=True, links=True), ) print(page.markdown) print(len(page.links or []), "links found")

usage object

every scrape/map response, terminal async status, and scrape.complete webhook payload carries billing + routing info on response_meta.usage — no separate usage() call is needed to attribute a single request’s cost.

page = client.scrape(url="https://example.com") print(page.response_meta.usage.credits) # 0 on a fully cached result

Usage (credits, engine, proxy, screenshot_slices) covers scrape, async, and webhook responses. MapUsage is the map-specific (credits, engine, proxy) dataclass; map cannot generate screenshot slices, and its engine is http | cache. access both via attributes, not dict keys.

client options

optiondefaultdescription
api_keysent as Authorization: Bearer …. required, or use Crawlbrulee.from_env().
base_urlproduction hostoverride the api host — for local development and staging.
timeoutNone (no timeout)per-request timeout in seconds. a per-call timeout= overrides it.

Crawlbrulee.from_env(**overrides) reads CRAWLBRULEE_API_KEY and forwards any other keyword argument.

async

AsyncCrawlbrulee mirrors the sync client method-for-method:

import asyncio from crawlbrulee import AsyncCrawlbrulee async def main() -> None: async with AsyncCrawlbrulee.from_env() as client: page = await client.scrape(url="https://example.com") print(page.markdown) asyncio.run(main())

both clients are context managers (with / async with) and expose close() / aclose().

request inputs

top-level fields are keyword args; nested structures are typed dataclasses (or plain dicts):

from crawlbrulee import ScrapeCleanup, ScrapeExtract, ScreenshotRequest client.scrape( url="https://news.example.com/article-1", extract=ScrapeExtract( markdown=True, metadata=True, links=True, screenshot=ScreenshotRequest(type="full_page", device_mode="desktop"), ), require_js=True, proxy="advanced", cleanup=ScrapeCleanup(ads_and_popups=True, exclude_selectors=["nav", "footer"]), cache={"max_age": 3600}, location={"country": "US"}, )

methods

methoddescription
scrape(url, **opts)scrape synchronously; blocks until done.
scrape_async(url, **opts)submit a background job; returns { job_id }.
get_scrape_status(job_id)pending / running / done / failed.
get_scrape_result(job_id)result of a finished job.
wait_for_scrape(job_id, interval=2.0, timeout=300.0)poll until terminal, then return the result.
map(url, **opts)build (or return a cached) link map for a site.
usage()current billing-cycle snapshot.
whoami()organization + token identity.
fetch_scrape_result_from_webhook(body)fetch the result of a verified scrape.complete delivery; raises on a failed/cancelled job.
verify_webhook_signature(payload=, headers=, secret=)standalone helper (no client needed); returns a WebhookVerificationResult instead of raising.

map() takes the same keyword-args pattern as scrape():

result = client.map( url="https://example.com", proxy="advanced", sitemap_only=True, max_urls=500, )

every method accepts a per-call timeout= (seconds) — except wait_for_scrape, whose timeout is the overall wait budget across all polls (default 300.0, 0 = wait forever), not a per-request http timeout.

the two webhook helpers in the table have extra options worth knowing: verify_webhook_signature(payload=, headers=, secret=, tolerance_seconds=300) takes a tolerance_seconds (pass 0 to disable the replay-protection check) and returns a WebhookVerificationResult rather than raising — .verified (bool), and either .signed_with ("primary" / "rotated") on success or .reason on failure. both are covered in detail on webhook verification.

errors

every failure subclasses CrawlbruleeError:

classwhen
AuthenticationError401 — missing, invalid, or revoked key. Also a 403 whose error name is not recognized.
AntibotBlockedError403 antibot_blocked — the target site’s anti-bot protection blocked the request (scrape or map).
TooManyRedirectsError422 too_many_redirects — the target site redirected the request in a loop (scrape or map). not a bad request.
PageTooLargeError422 page_too_large — the page’s html was too large to process (scrape). terminal; don’t retry it.
RateLimitError429 — exposes retry_after_ms, limited_by.
UsageAllocationErrorplan limit — exposes reason, usage.
ValidationErrorbad request.
NotFoundError404.
TransportErrornetwork failure / timeout / non-json.
import time from crawlbrulee import Crawlbrulee, RateLimitError, UsageAllocationError client = Crawlbrulee.from_env() try: client.scrape(url="https://example.com") except RateLimitError as err: time.sleep((err.retry_after_ms or 1000) / 1000) except UsageAllocationError as err: print("plan limit:", err.reason)

for exhaustive branching, switch on err.error_name. to check whether an arbitrary caught exception is a crawlbrulee error, use is_crawlbrulee_error():

from crawlbrulee import is_crawlbrulee_error if is_crawlbrulee_error(err): print(err.status, err.error_name)

see credits & pricing for operation costs and errors for the full http error catalog.