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 crawlbruleequickstart
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 resultUsage (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
| option | default | description |
|---|---|---|
api_key | — | sent as Authorization: Bearer …. required, or use Crawlbrulee.from_env(). |
base_url | production host | override the api host — for local development and staging. |
timeout | None (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
| method | description |
|---|---|
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:
| class | when |
|---|---|
AuthenticationError | 401 — missing, invalid, or revoked key. Also a 403 whose error name is not recognized. |
AntibotBlockedError | 403 antibot_blocked — the target site’s anti-bot protection blocked the request (scrape or map). |
TooManyRedirectsError | 422 too_many_redirects — the target site redirected the request in a loop (scrape or map). not a bad request. |
PageTooLargeError | 422 page_too_large — the page’s html was too large to process (scrape). terminal; don’t retry it. |
RateLimitError | 429 — exposes retry_after_ms, limited_by. |
UsageAllocationError | plan limit — exposes reason, usage. |
ValidationError | bad request. |
NotFoundError | 404. |
TransportError | network 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.