webclaw 0.7.0

Python SDK for the Webclaw web extraction API

---

Note: The webclaw Cloud API is public. Create an API key at webclaw.io or use the open-source CLI/MCP for local extraction.

---

Installation

pip install webclaw

Requires Python 3.9+. The only dependency is httpx.

Quick Start

Sync

from webclaw import Webclaw

client = Webclaw("wc-YOUR_API_KEY")

result = client.scrape("https://example.com", formats=["markdown"])
print(result.markdown)

Async

from webclaw import AsyncWebclaw

async with AsyncWebclaw("wc-YOUR_API_KEY") as client:
    result = await client.scrape("https://example.com", formats=["markdown"])
    print(result.markdown)

Both clients support identical method signatures. Every sync method has an async equivalent. The examples below use the sync client for brevity.

Endpoints

Scrape

Extract content from a single URL. Supports multiple output formats: "markdown", "text", "llm", "json".

result = client.scrape(
    "https://example.com",
    formats=["markdown", "text", "llm"],
    include_selectors=["article", ".content"],
    exclude_selectors=["nav", "footer"],
    only_main_content=True,
    no_cache=True,
)

result.url        # str
result.markdown   # str | None
result.text       # str | None
result.llm        # str | None
result.json_data  # Any | None
result.metadata   # dict
result.cache      # CacheInfo | None  (.status: "hit" | "miss" | "bypass")
result.warning    # str | None

Vertical extractors

Site-specific extractors return structured JSON (GitHub, Reddit, Amazon, YouTube, PyPI, HuggingFace, Trustpilot, etc.) instead of generic markdown. See the catalog for the full list.

# Discover available extractors
catalog = client.list_extractors()
for e in catalog["extractors"]:
    print(e["name"], "-", e["label"])

# Run a specific extractor
pr = client.scrape_vertical(
    "github_pr",
    "https://github.com/rust-lang/rust/pull/123456",
)
print(pr["data"])  # {title, state, author, commits, reviews, ...}

# Amazon product as typed JSON
product = client.scrape_vertical(
    "amazon_product",
    "https://www.amazon.com/dp/B0C6KKQ7ND",
)
print(product["data"]["price"], product["data"]["rating"])

The data field is extractor-specific; call list_extractors() to discover what each returns. Both methods have async equivalents on AsyncWebclaw.

Search

Web search with strict source filters, provider freshness/date/locale hints, pagination, and optional scraping of result pages.

results = client.search(
    "website pain points",
    num_results=10,
    include_domains=["reddit.com"],
    include_url_prefixes=["https://www.reddit.com/r/webdesign/comments/"],
    freshness="month",
    page=1,
    location="Austin, Texas, United States",
    autocorrect=False,
    scrape=False,
    country="us",
    lang="en",
    no_cache=True,
)

for r in results["results"]:
    print(r["title"], r["url"])

print(results.get("filtered_out_count"))
print(results.get("applied_filters"))

Use published_after / published_before (YYYY-MM-DD) instead of freshness when you need explicit provider discovery hints; published_before is exclusive. These hints do not verify a result's actual publication date, so inspect the source timestamp when correctness matters. topic remains accepted for compatibility but is deprecated and ignored by the hosted API.

Map

Discover URLs via sitemap.

result = client.map("https://example.com")

print(result.count)
for url in result.urls:
    print(url)

Batch

Scrape multiple URLs in parallel.

result = client.batch(
    ["https://a.com", "https://b.com", "https://c.com"],
    formats=["markdown"],
    concurrency=5,
)

for item in result.results:
    print(item.url, item.markdown, item.error or "ok")

Parameters: urls (list[str]), formats (optional), concurrency (int, default 5).

Endpoints

Discover the API endpoints a page calls at runtime by scanning its inline JavaScript and external <script src> bundles. This surfaces the routes a single-page app hits that map (sitemap-based) can't see: relative paths, absolute URLs, GraphQL operations, and WebSocket endpoints.

result = client.endpoints(
    "https://app.example.com",
    include_third_party=False,  # default: skip analytics/CDN hosts
    max_bundles=20,             # default & server max: external scripts to scan
)

print(result.bundles_scanned, result.endpoint_count, result.truncated)
print(result.hosts)  # list[str] of hosts seen across endpoints

for e in result.endpoints:
    print(e.kind, e.value, "first-party" if e.first_party else "third-party", "via", e.source)

Each endpoint's kind is one of "relative_path", "absolute_url", "graph_ql", "web_socket" (the values in EndpointKind). truncated is True when more bundles existed than max_bundles allowed.

Parameters: url (str), include_third_party (bool, default False), max_bundles (int, default 20, capped at 20).

Extract

LLM-powered structured data extraction. Use either a JSON schema or a natural language prompt.

# Schema-based extraction
result = client.extract(
    "https://example.com/pricing",
    schema={
        "type": "object",
        "properties": {
            "plans": {
                "type": "array",
                "items": {
                    "type": "object",
                    "properties": {
                        "name": {"type": "string"},
                        "price": {"type": "string"},
                    },
                },
            }
        },
    },
)
print(result.data)  # dict matching your schema

# Prompt-based extraction
result = client.extract(
    "https://example.com/pricing",
    prompt="Extract all pricing tiers with names and monthly prices",
)
print(result.data)

Summarize

Summarize page content with an optional sentence limit.

result = client.summarize("https://example.com", max_sentences=3)
print(result.summary)

Diff

Compare a page with your most recent cached extraction. Use the same API account for both calls; a missing or expired baseline returns an error.

# Establish the cached baseline once, then check for changes later.
client.scrape("https://example.com", formats=["json"])
result = client.diff("https://example.com")
print(result["status"])    # "Same", "Changed", or "New"
print(result["text_diff"]) # unified diff, or None
print(result["metadata_changes"])

To compare against a saved baseline instead, pass its complete extraction as previous, including metadata and content. An arbitrary title/body object is not accepted.

Brand

Extract brand identity (colors, fonts, logos) from a URL.

result = client.brand("https://example.com")
print(result.data)  # dict with brand identity fields

Research

Deep research that searches, reads, and synthesizes information from multiple sources. This is an async job: the SDK starts it and polls until completion.

# Blocks until research completes (up to 1200s)
result = client.research(
    "How do modern web crawlers handle JavaScript rendering?",
    max_sources=15,
    topic="tech",
)

print(result.report)
print(result.iterations)
print(result.elapsed_ms)

for source in result.sources:
    print(source["url"], source["title"])

To check status without blocking:

status = client.get_research_status("job-id-here")
print(status.status)  # "running" | "completed" | "failed"

Parameters: query (str), max_sources (int, optional), max_iterations (int, optional), topic (str, optional), deep (bool, deprecated — every research job runs in deep mode; the flag is ignored and passing deep=True emits a DeprecationWarning).

Crawl

Start an async crawl that follows links from a seed URL.

job = client.crawl(
    "https://example.com",
    max_depth=3,
    max_pages=100,
    use_sitemap=True,
)

# Poll until complete (default timeout 300s)
status = job.wait(interval=2.0, timeout=300.0)

print(status.total, status.completed, status.errors)
for page in status.pages:
    print(page.url, len(page.markdown or ""))

Check status without waiting:

status = job.get_status()
print(status.status)  # "running" | "completed" | "failed"

Async variant:

job = await client.crawl("https://example.com", max_depth=2)
status = await job.wait()

Watch

Monitor URLs for content changes with automatic periodic checks.

Create a watch:

watch = client.watch_create(
    "https://example.com/pricing",
    name="Pricing page monitor",
    interval_minutes=60,
    webhook_url="https://hooks.example.com/webclaw",
)
print(watch.id, watch.active)

List all watches:

result = client.watch_list(limit=50, offset=0)
for w in result.watches:
    print(w.id, w.url, w.name, w.last_checked_at)

Get a single watch:

watch = client.watch_get("watch-id-here")
print(watch.url, watch.interval_minutes)

Delete a watch:

client.watch_delete("watch-id-here")

Trigger an immediate check:

check = client.watch_check("watch-id-here")
print(check.status)  # "checking"; the snapshot is produced asynchronously.
# Fetch watch_get later to inspect its snapshots.

X (Twitter) monitoring

Monitor X for new tweets matching a profile, search, list, or reply thread, and fire a webhook on new matches — plus export an account's followers or following. These are paid-only features (a free/lapsed account gets ScopeError for 403). ScopeError remains an AuthenticationError subclass for backward-compatible catches. Monitors and audience export are billed per X request at your plan rate (Starter 5, Growth 3, Pro 2, Scale 1 credits). Max 50 monitors per user.

Create a monitor:

monitor = client.create_x_monitor(
    "search",                 # "profile" | "search" | "list" | "replies"
    "web scraping",           # handle / search query / list id / tweet id (per kind)
    name="Scraping mentions",
    interval_minutes=15,      # default 15, clamped 2..10080
    webhook_url="https://hooks.example.com/x",
    include_retweets=True,    # defaults: retweets/replies/quotes all True
    include_replies=False,
    include_quotes=True,
    min_faves=50,             # minimum likes to match (default 0)
    keyword="rust",           # only match tweets containing this
    lang="en",                # only match this language code
)
print(monitor.id, monitor.kind, monitor.active)

Only kind and target are required; every other argument is omitted from the request when left unset, so the server applies its own default.

List monitors (each is a full monitor object):

result = client.list_x_monitors(limit=50, offset=0)
for m in result.monitors:
    print(m.id, m.kind, m.target, m.last_matched_at, m.active)

Get, update, delete, and check:

m = client.get_x_monitor("monitor-id")

# PATCH — only the fields you pass are changed
client.update_x_monitor("monitor-id", active=False, interval_minutes=60)

client.delete_x_monitor("monitor-id")

# Trigger an immediate check (runs in the background; billed at your plan rate)
client.check_x_monitor("monitor-id")

Webhook payload posted to webhook_url on a match (Discord/Slack URLs get native formatting instead):

{
  "event": "x.monitor.matched",
  "monitor_id": "xm-1", "kind": "search", "target": "web scraping",
  "new_count": 1,
  "tweets": [
    {
      "id": "...", "screen_name": "...", "text": "...", "url": "...",
      "created_at": "...", "favorite_count": 12, "retweet_count": 3,
      "reply_count": 1, "lang": "en", "is_retweet": false,
      "is_reply": false, "is_quote": false
    }
  ],
  "checked_at": "..."
}

Export an audience (followers or following), cursor-paginated and metered per page at your plan rate (Starter 5, Growth 3, Pro 2, Scale 1 credits):

# Provide handle OR user_id. A pre-resolved user_id skips the (unbilled)
# re-resolve on later pages.
page = client.export_x_audience(
    handle="@jack",
    direction="followers",   # "followers" (default) | "following"
    max_pages=2,             # default 2, clamped 1..10 (~1-2k users/page)
)

for u in page.users:
    print(u.screen_name, u.name, u.followers, u.description)

print(page.pages_fetched, page.credits_charged)

# Walk the full audience: keep paging until next_cursor is None.
user_id, cursor = page.user_id, page.next_cursor
while cursor is not None:
    page = client.export_x_audience(user_id=user_id, cursor=cursor)
    # ... process page.users ...
    cursor = page.next_cursor

Every X method has an async equivalent on AsyncWebclaw with identical parameters.

Error Handling

All errors inherit from WebclawError, which carries the HTTP status code when available.

from webclaw import (
    WebclawError,
    AuthenticationError,
    ScopeError,
    NotFoundError,
    RateLimitError,
    TimeoutError,
)

try:
    result = client.scrape("https://example.com")
except ScopeError:
    print("API key is missing the required plan or scope")
except AuthenticationError:
    print("Invalid or missing API key")
except RateLimitError:
    print("Too many requests, slow down")
except NotFoundError:
    print("Resource not found")
except TimeoutError as e:
    print(f"Operation timed out: {e}")
except WebclawError as e:
    print(f"API error (status {e.status_code}): {e}")

Exception	HTTP Status	When
AuthenticationError	401	Invalid or missing API key
ScopeError	403	Authenticated but missing the required plan or scope
NotFoundError	404	Resource does not exist
RateLimitError	429	Too many requests
TimeoutError	--	Crawl/research polling exceeded timeout
WebclawError	Any	Base class for all other API errors

Configuration

import os
from webclaw import Webclaw

client = Webclaw(
    os.environ["WEBCLAW_API_KEY"],
    base_url="https://api.webclaw.io",  # default
    timeout=60.0,                        # seconds, default 30
)

Both Webclaw and AsyncWebclaw support context managers for automatic cleanup:

# Sync
with Webclaw("wc-YOUR_API_KEY") as client:
    result = client.scrape("https://example.com")

# Async
async with AsyncWebclaw("wc-YOUR_API_KEY") as client:
    result = await client.scrape("https://example.com")

Async Usage

Every endpoint is available on AsyncWebclaw with identical parameters. Use await on all method calls and async with for the context manager.

import asyncio
from webclaw import AsyncWebclaw

async def main():
    async with AsyncWebclaw("wc-YOUR_API_KEY") as client:
        # Run multiple scrapes concurrently
        results = await asyncio.gather(
            client.scrape("https://a.com", formats=["markdown"]),
            client.scrape("https://b.com", formats=["markdown"]),
            client.scrape("https://c.com", formats=["markdown"]),
        )
        for r in results:
            print(r.url, len(r.markdown or ""))

asyncio.run(main())

Type Support

This package ships with a py.typed marker (PEP 561). Type checkers like mypy and pyright will pick up all type annotations automatically. All response types are dataclasses importable from the top-level package:

from webclaw import (
    ScrapeResponse, CrawlStatus, MapResponse, ExtractResponse, EndpointsResponse,
    XMonitor, XMonitorListResponse, XAudienceResponse, XAudienceUser,
)

License

MIT