# scrapingpros Python SDK — Complete Reference

> Python SDK for the [Scraping Pros](https://scrapingpros.com) API. Web scraping with browser rendering, proxy rotation, and structured data extraction.

## Installation

**Requires Python 3.10 or higher.**

```bash
pip install scrapingpros
```

## Quick Start

```python
from scrapingpros import ScrapingPros

client = ScrapingPros("demo_6x595maoA6GdOdVb")
result = client.scrape("https://example.com")
print(result.content)
```

No signup needed. Demo token: 5,000 credits/month, 30 req/min.

## Credit System

- 1 simple request = 1 credit
- 1 browser request = 5 credits
- Anti-bot protection and proxy rotation included at no extra cost
- Credits are NOT consumed for requests that fail due to infrastructure errors
- `retry_on_block=True` only charges for the successful attempt
- Track credits: `client.credits_charged`, `client.quota_remaining`
- View plans: `client.plans()`, check usage: `client.billing()`

## Plans

| Plan | Price/mo | Credits/mo | Rate Limit |
|------|----------|------------|------------|
| Free | $0 | 1,000 | 30/min |
| Starter | $29 | 25,000 | 30/min |
| Growth | $69 | 100,000 | 60/min |
| Pro | $199 | 500,000 | 120/min |
| Scale | $499 | 2,500,000 | 200/min |
| Enterprise | Custom | Unlimited | 2,000/min |

See https://scrapingpros.com for details.

## Client Initialization

```python
from scrapingpros import ScrapingPros, AsyncScrapingPros

# Sync client
client = ScrapingPros(
    "your_token",          # or omit and set SP_TOKEN env var
    base_url="https://api.scrapingpros.com",  # default
    timeout=120.0,         # request timeout in seconds
    max_retries=3,         # auto-retry on 429 rate limits
)

# Async client (same API surface)
async with AsyncScrapingPros("your_token") as client:
    result = await client.scrape("https://example.com")
```

## Methods Reference

### scrape(url, ...) -> ScrapeResponse

Scrape a single URL. Returns HTML or clean markdown.

```python
result = client.scrape(
    "https://example.com",
    format="markdown",        # "html" (default) or "markdown"
    browser=False,            # True for JS-heavy sites
    browser_type="light",     # "light" (default, fast), "heavy" (anti-detection), or "stealth" (max anti-detection)
    use_proxy="any",          # "any" (default), None, country code "US", or ProxyConfig
    screenshot=False,         # True for full-page PNG (requires browser=True)
    actions=None,             # browser automation steps
    language="",              # "en-US", "es-AR", etc.
    http_method=None,         # MethodPOST(payload={...}) for POST requests
    headers=None,             # {"User-Agent": "..."}
    window_size="",           # "desktop", "mobile", "any"
    cookies=None,             # {"name": "value"}
    extract=None,             # CSS/XPath selectors (see below)
    network_capture=None,     # NetworkCaptureConfig(resource_types=["xhr"])
    retry_on_block=False,     # True: auto-retry on CAPTCHA/403, only charges on success
)
```

**ScrapeResponse fields:**
- `result.content` — convenience: returns markdown if available, else html
- `result.html` — raw HTML (when format="html")
- `result.markdown` — clean text (when format="markdown")
- `result.statusCode` — HTTP status from target page
- `result.extracted_data` — dict with extracted data
- `result.screenshot` — base64 PNG string
- `result.executionTime` — seconds
- `result.potentiallyBlockedByCaptcha` — bool
- `result.timings` — performance breakdown (always present, never null)
- `result.evaluate_results` — list of JS evaluation results
- `result.network_requests` — captured network activity
- `result.guidance` — `ScrapeGuidance` with error analysis and next steps (see below)

### Response Guidance (ScrapeGuidance)

Every scrape response includes a `guidance` object that explains the result and suggests what to do next.

```python
result = client.scrape("https://hard-site.com")

# Always present — check it after every scrape
g = result.guidance
print(g.success)          # True if content was retrieved
print(g.credits_charged)  # 1 (simple) or 5 (browser), 0 if refunded
```

When a scrape fails, guidance tells you why and what to try:

```python
if not result.guidance.success:
    print(result.guidance.error_type)         # "captcha", "ssl_error", "timeout", "login_wall", etc.
    print(result.guidance.error_provider)      # "cloudflare", "datadome", etc.
    print(result.guidance.next_steps)          # ["Try with browser=true", "Try retry_on_block=true"]
    print(result.guidance.suggested_request)   # ready-to-use params dict: {"url": "...", "browser": True}
    print(result.guidance.credits_refunded)    # True if credits were returned

    # If stop_reason is set, do NOT retry — it won't help
    if result.guidance.stop_reason:
        print("Stop:", result.guidance.stop_reason)  # "Site requires authentication"
    elif result.guidance.suggested_request:
        # Retry with suggested params
        result = client.scrape(**result.guidance.suggested_request)
```

**ScrapeGuidance fields:**
- `success` — `True` if usable content was retrieved
- `error_type` — category: `"captcha"`, `"ssl_error"`, `"timeout"`, `"login_wall"`, `"blocked"`, etc.
- `error_provider` — protection provider: `"cloudflare"`, `"datadome"`, `"recaptcha"`, etc.
- `credits_charged` — credits consumed (0 if refunded)
- `credits_refunded` — `True` if credits were returned due to infrastructure error
- `next_steps` — list of human-readable suggestions
- `suggested_request` — dict of params to retry with (pass directly to `client.scrape(**suggested_request)`)
- `stop_reason` — if set, retrying is futile (e.g. login wall, geo-blocked)

### retry_on_block — Server-Side Anti-Bot Retry

```python
# For protected sites: auto-retry with different IP/fingerprint
result = client.scrape(
    "https://hard-site.com",
    browser=True,
    retry_on_block=True,  # up to 3 retries, only charges on success
)
# Combined with early CAPTCHA detection: blocked pages return in ~5-10s
# instead of waiting for full page load (60-85s)
```

### Browser Types

Three browser modes, ordered by anti-detection strength:

| Mode | Anti-Detection | Speed | Best For |
|------|---------------|-------|----------|
| `"light"` (default) | Basic | Fast | Most sites, SPAs, high concurrency |
| `"heavy"` | Strong | Medium | Protected sites with soft blocks |
| `"stealth"` | Maximum | Slower | Hardest sites (Google, Amazon) |

```python
# Light (default) — fast rendering, best for concurrency
result = client.scrape("https://example.com", browser=True)

# Heavy — anti-detection for protected sites
result = client.scrape("https://protected-site.com", browser=True, browser_type="heavy")

# Stealth — maximum anti-detection for the hardest sites
result = client.scrape("https://www.google.com/shopping/...", browser=True, browser_type="stealth")
```

### Resource Blocking (block_resources, block_requests)

Block unnecessary resources to speed up page load and reduce detection surface. Requires `browser=True`.

**`block_resources`** — block by resource type:
```python
result = client.scrape(
    "https://heavy-site.com",
    browser=True,
    block_resources=["image", "font", "media", "stylesheet"],
)
# Valid types: "image", "stylesheet", "font", "media", "script", "document", "xhr", "fetch"
```

**`block_requests`** — block by URL substring (case-insensitive):
```python
result = client.scrape(
    "https://www.google.com/shopping/...",
    browser=True,
    block_requests=["google-analytics", "doubleclick", "googletagmanager", "googlesyndication"],
)
```

**Combined — optimal for hard sites:**
```python
result = client.scrape(
    "https://www.google.com/shopping/...",
    browser=True,
    browser_type="stealth",
    block_resources=["image", "font", "media"],
    block_requests=["google-analytics", "doubleclick", "googlesyndication"],
    retry_on_block=True,
)
```

### Data Extraction

```python
result = client.scrape(
    "https://quotes.toscrape.com/",
    extract={
        "title": "css:h1",
        "quotes": {"selector": "css:.text", "multiple": True},
        "links": {"selector": "css:a", "attribute": "href", "multiple": True},
        "xpath_example": "xpath://div[@class='quote']",
    },
)
for quote in result.extracted_data["quotes"]:
    print(quote)
```

### Browser Automation (Actions)

```python
from scrapingpros import ClickAction, InputAction, WaitForSelectorAction, EvaluateAction

result = client.scrape(
    "https://example.com/login",
    browser=True,
    actions=[
        InputAction(selector="css:#email", text="user@example.com"),
        InputAction(selector="css:#password", text="secret"),
        ClickAction(selector="css:button[type=submit]", wait_for_navigation=True),
        WaitForSelectorAction(selector="css:.dashboard", time=5000),
    ],
)
```

Available actions: `ClickAction`, `InputAction`, `SelectAction`, `KeyPressAction`, `WaitForSelectorAction`, `WaitForTimeoutAction`, `EvaluateAction`, `CollectAction`, `WhileControl`.

### JavaScript Execution (EvaluateAction)

Run arbitrary JavaScript in the browser page context. Use this to call internal APIs, extract data from JS variables, intercept XHR/fetch responses, or access data that isn't in the DOM.

```python
from scrapingpros import EvaluateAction, WaitForSelectorAction

# Example: call an internal API endpoint from the page context
result = client.scrape(
    "https://example.com/products",
    browser=True,
    actions=[
        WaitForSelectorAction(selector="css:.product-list", time=5000),
        EvaluateAction(script="fetch('/api/products').then(r => r.json())"),
    ],
)
# Results are in evaluate_results, one per EvaluateAction
api_data = result.evaluate_results[0]  # parsed JSON from the fetch

# Example: extract data from JavaScript variables
result = client.scrape(
    "https://example.com/dashboard",
    browser=True,
    actions=[
        EvaluateAction(script="window.__NEXT_DATA__"),        # Next.js page data
        EvaluateAction(script="document.title"),               # simple DOM access
        EvaluateAction(script="""
            // Multi-line scripts work too
            const items = document.querySelectorAll('.item');
            Array.from(items).map(el => ({
                name: el.textContent,
                href: el.querySelector('a')?.href
            }))
        """),
    ],
)
next_data = result.evaluate_results[0]
title = result.evaluate_results[1]
items = result.evaluate_results[2]
```

**EvaluateAction fields:**
- `script` — JavaScript code to execute in the page context. The return value is captured.
- `timeout` — max execution time in ms (default: 30000)

**When to use evaluate vs extract:**
- Use `extract` (CSS/XPath selectors) when the data is visible in the HTML DOM
- Use `EvaluateAction` when you need to: call internal APIs (`fetch`), read JS variables (`window.__DATA__`), execute complex DOM logic, or access data not rendered in HTML

### download(url, ...) -> DownloadResponse

Download a file as base64.

```python
result = client.download("https://example.com/report.pdf")
print(result.contentType)  # "application/pdf"
import base64
with open("report.pdf", "wb") as f:
    f.write(base64.b64decode(result.content))
```

### scrape_many(urls, ...) -> list[ScrapeResponse | Exception]

**Use `scrape_many()` whenever you have a list of URLs.** It handles concurrency automatically based on your plan limits. Much faster than a sequential loop.

```python
urls = ["https://example.com/1", "https://example.com/2", "https://example.com/3"]
results = client.scrape_many(urls, format="markdown", browser=True)
for url, result in zip(urls, results):
    if isinstance(result, Exception):
        print(f"FAILED: {url} — {result}")
    else:
        print(f"OK: {url} — {len(result.content or '')} chars")
```

Parameters: all `scrape()` params apply to every URL, plus:
- `max_concurrent` — max parallel requests. Auto-detected from your plan if omitted.
- `requests` — alternative: list of `ScrapeRequest` or dicts for per-URL parameters.

Returns a list in the same order as input. Failed requests return the `Exception` instead of raising.

```python
# Override concurrency
results = client.scrape_many(urls, max_concurrent=10, browser=True)

# With actions + stealth + blocking
results = client.scrape_many(urls, browser=True, browser_type="stealth",
    actions=[WaitForSelectorAction(selector="css:.item", time=5000),
             EvaluateAction(script="document.querySelectorAll('.item').length")],
    block_resources=["image", "font", "media"],
    block_requests=["google-analytics", "doubleclick"],
    headers={"Accept-Language": "en-US"},
)

# Heterogeneous: different params per URL
from scrapingpros import ScrapeRequest
results = client.scrape_many(requests=[
    ScrapeRequest(url="https://site1.com", browser=True, actions=[...]),
    ScrapeRequest(url="https://site2.com", format="markdown"),
    {"url": "https://site3.com", "browser": True, "extract": {"title": "css:h1"}},
])
```

### Batch Processing with Webhooks (collections)

```python
# Create collection with webhook
col = client.create_collection("my-batch", [
    {"url": "https://example.com/1"},
    {"url": "https://example.com/2", "browser": True},
], callback_url="https://your-server.com/webhook")

# Option A: Poll for completion
run = client.run_and_wait(col.id, timeout=120)

# Option B: Receive webhook POST when done
# Your server gets: {"event": "run.completed", "run_id": "...", "job_ids": [...]}
# Signed with HMAC-SHA256 in X-SP-Signature header

# Fetch individual results (HTML available 48h, metadata 90 days)
for job_id in run.job_ids:
    result = client.get_job_result(col.id, run.run_id, job_id)
    print(result.url, result.custom_id, result.content[:100])

# Check webhook delivery status
run = client.get_run(col.id, run.run_id)
print(run.callback_status)  # "sent", "pending", "failed", "retrying"

# Cleanup
client.delete_collection(col.id)
```

### URL traceability with custom_id

Each request can carry a `custom_id` echoed back in the job listing and result, so you can map results to your data without depending on order:

```python
col = client.create_collection("my-batch", [
    {"url": "https://example.com/tour/1", "custom_id": "tour_1"},
    {"url": "https://example.com/tour/2", "custom_id": "tour_2"},
])
run = client.run_and_wait(col.id)

# Iterate jobs with full traceability — each job has url + custom_id
for job in client.iter_run_jobs(col.id, run.run_id):
    if job.status == "completed":
        result = client.get_job_result(col.id, run.run_id, job.job_public_id)
        save_to_db(result, my_id=job.custom_id)
```

`scrape_many()` results carry `url` and `custom_id` too — same pattern applies for client-side concurrency.

### Pagination over jobs

`get_run_jobs()` paginates internally by default (returns all jobs). For very large runs (1000+) prefer `iter_run_jobs()` to stream lazily:

```python
# Default: fetch all jobs in memory
jobs = client.get_run_jobs(col.id, run.run_id)
print(f"{len(jobs.items)} jobs")

# Memory-efficient: stream lazily, optionally filter by status
for job in client.iter_run_jobs(col.id, run.run_id, status_filter="completed"):
    process(job)

# Manual pagination: pass cursor to control fetching
page = client.get_run_jobs(col.id, run.run_id, cursor=None, limit=500)
while page.has_more:
    page = client.get_run_jobs(col.id, run.run_id, cursor=page.cursor_next, limit=500)
```

`JobExecutionPublic` fields: `job_public_id`, `url`, `custom_id`, `status`, `status_code`, `queued_at`, `started_at`, `completed_at`, `execution_time_ms`, `retries_attempted`, `block_reason`, `url_truncated`, `protection_stack`, `rule_hits`.

### POST to a different URL than the navigation target (MethodPOST.url)

Some sites require navigating to one page (to set cookies, generate session) and POSTing to a different endpoint (an internal API or GraphQL):

```python
from scrapingpros import MethodPOST

result = client.scrape(
    "https://www.example.com/dashboard",     # navigation
    http_method=MethodPOST(
        url="https://api.example.com/graphql", # POST goes here
        payload={"query": "..."},
    ),
)
```

If `MethodPOST.url` is omitted, the POST goes to the same URL as the scrape (legacy behavior).

Other collection methods:
- `client.list_collections()` -> list[CollectionPublic]
- `client.get_collection(id)` -> CollectionPublic
- `client.update_collection(id, name, requests)` -> NewCollectionResponse
- `client.create_run(collection_id)` -> RunPublic
- `client.get_run(collection_id, run_id)` -> RunPublic
- `client.get_run_jobs(collection_id, run_id, *, cursor=None, limit=1000, status_filter=None)` -> JobExecutionListPublic
- `client.iter_run_jobs(collection_id, run_id, *, page_size=500, status_filter=None)` -> Iterator[JobExecutionPublic]

### Viability Test

Analyze sites before scraping. Tests each URL with multiple scraping modes to find what works.

```python
# depth controls how many modes are tested:
#   "quick"    — simple HTTP only (1 credit/URL)
#   "standard" — simple + proxy + browser (up to 7 credits/URL)
#   "full"     — all modes including browser+proxy (up to 12 credits/URL)
test = client.create_viability_test(
    ["https://example.com", "https://hard-site.com"],
    depth="full",  # default
)

import time
while True:
    result = client.get_viability_test(test.run_id)
    if result.status == "completed":
        break
    time.sleep(2)

for r in result.results:
    print(f"{r.url}: {r.recommended_strategy}")
    print(f"  Difficulty: {r.difficulty}")  # "easy", "medium", "hard"
    print(f"  Best mode: {r.best_mode}")    # cheapest mode that worked
    print(f"  CAPTCHA: {r.captcha_detected} {r.captcha_providers}")
    print(f"  Browser needed: {r.javascript_required}")
    print(f"  Credits used: {r.total_credits_used}")

    # modes_tested shows per-mode results
    if r.modes_tested:
        for mode, result_data in r.modes_tested.items():
            status = "OK" if result_data.success else "BLOCKED"
            print(f"    {mode}: {status} ({result_data.time_ms}ms, {result_data.credits} credits)")
```

**UrlViabilityResult key fields:**
- `difficulty` — `"easy"`, `"medium"`, or `"hard"`
- `best_mode` — cheapest successful mode (e.g. `"simple"`, `"browser_proxy"`)
- `modes_tested` — dict of mode name to `ViabilityModeResult`
- `total_credits_used` — total credits consumed across all mode tests
- `recommended_strategy` — suggested scraping approach
- `suggested_params` — dict of params to use with `client.scrape()`

**ViabilityModeResult fields:**
- `success` — whether this mode retrieved content
- `blocked` — whether the site blocked this mode
- `captcha_type` — CAPTCHA provider if detected
- `time_ms` — response time in milliseconds
- `credits` — credits consumed for this mode test
- `error` — error message if failed

### Proxy Management

```python
resp = client.list_proxy_countries()
print(resp.countries)  # ["AD", "AE", "AF", ...]

resp = client.request_proxy_country("US", reason="Need US pricing")
print(resp.status)  # "pending" or "already_approved"

status = client.proxy_status()
print(status.approved_countries)

result = client.scrape("https://example.com", use_proxy="US")
```

### Plans & Billing

```python
plans = client.plans()  # public, no auth needed
billing = client.billing(month="2026-04")
metrics = client.client_metrics(date="2026-04", detail="urls")

# After any call:
print(client.credits_charged)       # 1 or 5
print(client.quota_remaining)       # credits left this month
print(client.rate_limit_remaining)  # requests left this minute
```

### Health Check

```python
health = client.health()  # {"status": "healthy", ...}
```

## Async Client

Same API, all methods are `async def`:

```python
from scrapingpros import AsyncScrapingPros

async with AsyncScrapingPros("demo_6x595maoA6GdOdVb") as client:
    result = await client.scrape("https://example.com", format="markdown")
    print(result.content)

    col = await client.create_collection("batch", [{"url": "https://example.com"}],
                                          callback_url="https://my-server.com/hook")
    run = await client.run_and_wait(col.id)
```

## Error Handling

```python
from scrapingpros import (
    ScrapingPros,
    ScrapingProsError,
    AuthenticationError,
    RateLimitError,
    QuotaExceededError,
    URLBlockedError,
    ConnectionError,
)

try:
    result = client.scrape("https://example.com")
except AuthenticationError:
    # Invalid token — get one at https://scrapingpros.com
    pass
except RateLimitError as e:
    # Rate limited — SDK auto-retries, this means all retries exhausted
    print(f"Retry after {e.retry_after}s")
except QuotaExceededError:
    # Monthly credits exhausted — upgrade at https://scrapingpros.com
    pass
except URLBlockedError:
    # URL blocked by SSRF protection — only public URLs allowed
    pass
except ConnectionError:
    # Network error — DNS, connection refused, etc.
    pass
except ScrapingProsError:
    # Catch-all for any SDK error
    pass
```

Exception hierarchy:
- `ScrapingProsError` — base for all SDK errors
  - `APIError` — HTTP error (has `.status_code`, `.detail`)
    - `AuthenticationError` — 401
    - `CountryProxyNotApproved` — 403
    - `URLBlockedError` — 422 (SSRF)
    - `ValidationError` — 422 (params)
    - `RateLimitError` — 429 (has `.retry_after`)
    - `QuotaExceededError` — 429 (quota, no retry)
  - `ConnectionError` — network error
  - `TimeoutError` — polling timeout

## LangChain Integration

```bash
pip install langchain-scrapingpros
```

```python
from langchain_scrapingpros import ScrapingProsScraper, ScrapingProsExtractor

# Scrape for RAG
scraper = ScrapingProsScraper(api_key="demo_6x595maoA6GdOdVb")
text = scraper.invoke({"url": "https://example.com"})

# Extract structured data
extractor = ScrapingProsExtractor(api_key="demo_6x595maoA6GdOdVb")
data = extractor.invoke({"url": "https://quotes.toscrape.com/", "fields": {"quotes": "css:.text"}})
```

## Key Models

All importable from `from scrapingpros import ...`:

**Request/Response:** `ScrapeRequest`, `ScrapeResponse`, `ScrapeGuidance`, `DownloadRequest`, `DownloadResponse`
**Actions:** `ClickAction`, `InputAction`, `SelectAction`, `KeyPressAction`, `WaitForSelectorAction`, `WaitForTimeoutAction`, `EvaluateAction`, `CollectAction`, `WhileControl`
**Conditions:** `SelectorVisibleCondition`, `SelectorInvisibleCondition`
**Collections:** `CollectionPublic`, `NewCollectionResponse`, `RunPublic`, `FailedJobPublic`, `JobExecutionPublic`
**Viability:** `ViabilityTestRequest`, `ViabilityRunPublic`, `UrlViabilityResult`, `ViabilityModeResult`
**Proxy:** `ProxyConfig`, `ProxyCountriesResponse`, `ProxyRequestCountryResponse`, `ProxyStatusResponse`
**HTTP:** `MethodGET`, `MethodPOST`
**Constants:** `DEMO_TOKEN`

## Links

- Website: https://scrapingpros.com
- API Docs: https://api.scrapingpros.com/docs
- API Reference: https://api.scrapingpros.com/llms-full.txt
- PyPI: https://pypi.org/project/scrapingpros/
- LangChain: https://pypi.org/project/langchain-scrapingpros/
