DataBlue is a self-hosted web scraping and structured data API platform. It provides a Firecrawl-compatible REST API for scraping, crawling, searching, and extracting content from any website, plus 22+ Data APIs for Google, YouTube, Twitter, and Reddit.

Make Your First Request

curl -X POST "https://api.datablue.dev/v1/scrape" \
  -H "Authorization: Bearer wh_your_api_key" \
  -H "Content-Type: application/json" \
  -d '{
    "url": "https://news.ycombinator.com",
    "formats": ["markdown", "links"]
  }'

DataBlue supports two authentication methods. Both are sent as Authorization: Bearer <token> in the request header.

JWT Authentication

Obtain a JWT token by authenticating with your email and password:

curl -X POST "https://api.datablue.dev/v1/auth/login" \
  -H "Content-Type: application/json" \
  -d '{"email": "you@example.com", "password": "your_password"}'

# Response: {"access_token": "eyJ...", "token_type": "bearer"}

API Key Authentication (Recommended)

API keys are persistent and do not expire. Generate them from the Dashboard under the API Keys panel. All API keys use the wh_ prefix.

# Use your API key in every request
curl -X POST "https://api.datablue.dev/v1/scrape" \
  -H "Authorization: Bearer wh_abc123def456" \
  -H "Content-Type: application/json" \
  -d '{"url": "https://example.com"}'

Step 1: Sign Up

Create an account at datablue.dev or register via the API:

curl -X POST "https://api.datablue.dev/v1/auth/register" \
  -H "Content-Type: application/json" \
  -d '{"email": "you@example.com", "password": "secure_password", "name": "Your Name"}'

Step 2: Generate an API Key

Log into the Dashboard and navigate to API Keys. Click Create New Key to generate a persistent API key with the wh_ prefix.

Step 3: Make Your First Request

cURL

curl -X POST "https://api.datablue.dev/v1/scrape" \
  -H "Authorization: Bearer wh_your_api_key" \
  -H "Content-Type: application/json" \
  -d '{
    "url": "https://news.ycombinator.com",
    "formats": ["markdown"]
  }'

Python

import requests

response = requests.post(
    "https://api.datablue.dev/v1/scrape",
    headers={"Authorization": "Bearer wh_your_api_key"},
    json={
        "url": "https://news.ycombinator.com",
        "formats": ["markdown"]
    }
)
data = response.json()
print(data["data"]["markdown"][:200])

JavaScript

const response = await fetch("https://api.datablue.dev/v1/scrape", {
  method: "POST",
  headers: {
    "Authorization": "Bearer wh_your_api_key",
    "Content-Type": "application/json"
  },
  body: JSON.stringify({
    url: "https://news.ycombinator.com",
    formats: ["markdown"]
  })
});
const data = await response.json();
console.log(data.data.markdown.slice(0, 200));

{
  "success": true,
  "data": {
    "markdown": "# Hacker News\n\n1. Show HN: I built an open-source web scraper...",
    "metadata": {
      "title": "Hacker News",
      "language": "en",
      "source_url": "https://news.ycombinator.com",
      "status_code": 200,
      "word_count": 1847
    }
  }
}

The official datablue Python SDK provides both synchronous and asynchronous clients for the DataBlue API. Built on httpx and Pydantic v2, it offers full type safety, automatic retries with exponential backoff, and strongly-typed response models.

Requirements

Install from PyPI

pip install datablue

Or with a specific version:

pip install datablue==2.0.0

Install with Poetry / uv

# Poetry
poetry add datablue

# uv
uv add datablue

Verify Installation

python -c "import datablue; print(datablue.__version__)"
# 2.0.0

Quick Start

from datablue import DataBlue

with DataBlue(api_key="wh_your_api_key") as client:
    result = client.scrape("https://example.com")
    print(result.data.markdown)

Using with AI Assistants

The v2.0.0 SDK is designed to be AI-first. It ships with two machine-readable reference files that AI coding assistants (Claude Code, Cursor, GitHub Copilot, etc.) can read for accurate code generation:

• CLAUDE.md — A structured quick-reference file at the SDK root (sdk/CLAUDE.md) containing all method signatures, response models, error types, and common patterns. AI assistants automatically read this file for context.
• llms.txt — A standardized machine-readable documentation file following the llms.txt convention. Provides a condensed API surface for LLM consumption.

SDK Features

• Sync + Async clients — DataBlue for synchronous code, AsyncDataBlue for asyncio/FastAPI/Django
• Pydantic v2 response models — every response is a typed dataclass with autocomplete and validation
• Automatic retries — exponential backoff on 429 and 5xx errors, configurable max retries
• Context manager support — with / async with for clean resource management
• Job polling built-in — crawl/search blocking methods poll automatically with configurable timeout
• Batch scraping — concurrent scraping with semaphore-based concurrency control and streaming results
• Typed error hierarchy — catch specific errors like RateLimitError, AuthenticationError, etc.
• Environment variable config — zero-config setup with DataBlue.from_env()

The SDK supports three authentication methods: API key (recommended), environment variables, and email/password login.

API Key (Recommended)

Pass your API key directly to the constructor. API keys use the wh_ prefix and never expire.

from datablue import DataBlue

# Sync client
client = DataBlue(api_key="wh_your_api_key")
result = client.scrape("https://example.com")
client.close()

# Or use context manager (recommended)
with DataBlue(api_key="wh_your_api_key") as client:
    result = client.scrape("https://example.com")
    print(result.data.markdown)

from datablue import AsyncDataBlue

# Async client
async with AsyncDataBlue(api_key="wh_your_api_key") as client:
    result = await client.scrape("https://example.com")
    print(result.data.markdown)

Environment Variables

Set DATABLUE_API_KEY and optionally DATABLUE_API_URL in your environment, then use from_env():

# Shell — set environment variables
export DATABLUE_API_KEY=wh_your_api_key
export DATABLUE_API_URL=https://api.datablue.dev    # optional, default: http://localhost:8000
export DATABLUE_TIMEOUT=120                          # optional, seconds (default: 60)
export DATABLUE_MAX_RETRIES=5                        # optional (default: 3)

from datablue import DataBlue

# Reads all DATABLUE_* environment variables automatically
with DataBlue.from_env() as client:
    result = client.scrape("https://example.com")
    print(result.data.markdown)

Email/Password Login

Use login() to authenticate with email and password. This obtains a JWT token (7-day TTL) and stores it internally on the client instance.

from datablue import DataBlue

with DataBlue(api_url="https://api.datablue.dev") as client:
    # Authenticate — JWT is stored automatically
    auth = client.login("you@example.com", "your_password")
    print(f"Token: {auth['access_token'][:20]}...")

    # All subsequent requests use the JWT
    result = client.scrape("https://example.com")
    print(result.data.markdown)

# Async variant
from datablue import AsyncDataBlue

async with AsyncDataBlue(api_url="https://api.datablue.dev") as client:
    await client.login("you@example.com", "your_password")
    result = await client.scrape("https://example.com")

Scrape a single URL and return structured content. The scrape() method is synchronous and returns immediately with the result. The scraping engine uses a 5-tier parallel race: HTTP, stealth browser, Playwright, headless Chrome, and anti-bot bypass — first valid result wins.

Basic Usage

from datablue import DataBlue

with DataBlue(api_key="wh_your_api_key") as client:
    result = client.scrape("https://example.com")
    print(result.success)          # True
    print(result.data.markdown)    # "# Example Domain\n\nThis domain is..."
    print(result.data.metadata)    # PageMetadata(title="Example Domain", ...)

With All Options

result = client.scrape(
    "https://news.ycombinator.com",
    formats=["markdown", "html", "links", "screenshot"],
    only_main_content=True,
    wait_for=2000,                          # wait 2s after page load
    timeout=45000,                          # 45s timeout
    include_tags=["article", "main"],       # only these HTML tags
    exclude_tags=["nav", "footer"],         # remove these tags
    headers={"Accept-Language": "en-US"},
    cookies={"session": "abc123"},
    mobile=True,
    mobile_device="iphone_14",
    css_selector="main.content",            # target specific element
    use_proxy=True,
    capture_network=True,
)

print(result.data.markdown)
print(result.data.html)
print(result.data.links)
print(result.data.screenshot)       # base64-encoded PNG
print(result.data.metadata.title)
print(result.data.metadata.word_count)

Saving Screenshots

Screenshots are returned as base64-encoded PNG strings. Save to a file, embed in HTML, or pass directly to an LLM:

import base64

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

# Save to file
with open("screenshot.png", "wb") as f:
    f.write(base64.b64decode(result.data.screenshot))

# Embed in HTML
html_img = f'<img src="data:image/png;base64,{result.data.screenshot}" />'

# Store in database as a string field
db.save(url=result.data.metadata.source_url, screenshot=result.data.screenshot)

Async

from datablue import AsyncDataBlue

async with AsyncDataBlue(api_key="wh_your_api_key") as client:
    result = await client.scrape(
        "https://example.com",
        formats=["markdown", "links"],
        only_main_content=True,
    )
    print(result.data.markdown)

Browser Actions

Execute browser actions before content extraction. Useful for pages that require interaction (clicking buttons, filling forms, scrolling to load content).

result = client.scrape(
    "https://example.com/infinite-scroll",
    actions=[
        {"type": "wait", "milliseconds": 1000},
        {"type": "click", "selector": "button.load-more"},
        {"type": "wait", "milliseconds": 2000},
        {"type": "scroll", "direction": "down", "amount": 3},
        {"type": "screenshot"},
    ],
)

Action Types Reference

LLM Extraction

Extract structured data from a page using an LLM. Pass a prompt and optional schema (JSON Schema) in the extract parameter.

result = client.scrape(
    "https://openai.com/pricing",
    extract={
        "prompt": "Extract all pricing tiers with name and price",
        "schema": {
            "type": "object",
            "properties": {
                "tiers": {
                    "type": "array",
                    "items": {
                        "type": "object",
                        "properties": {
                            "name": {"type": "string"},
                            "price": {"type": "string"},
                        },
                    },
                },
            },
        },
    },
)
print(result.data.extract)  # {"tiers": [{"name": "GPT-4o", "price": "$2.50/1M"}, ...]}

Parameters

Response Model

Returns ScrapeResult — a Pydantic model with the following fields:

class ScrapeResult:
    success: bool                    # Whether the scrape request succeeded
    data: PageData | None            # Scraped page content including markdown, html, links, and metadata
    error: str | None                # Human-readable error message if the scrape failed
    error_code: str | None           # Machine-readable error code (BLOCKED_BY_WAF, CAPTCHA_REQUIRED, TIMEOUT, JS_REQUIRED, NETWORK_ERROR)
    job_id: str | None               # Unique job identifier for async scrape requests

class PageData:
    url: str | None                  # The URL of the scraped page
    markdown: str | None             # Clean Markdown conversion of the page content
    fit_markdown: str | None         # Markdown trimmed to fit LLM context windows
    html: str | None                 # Cleaned HTML content with boilerplate removed
    raw_html: str | None             # Original unmodified HTML source of the page
    links: list[str] | None          # List of URLs found on the page
    links_detail: list[dict] | None  # Detailed link info including anchor text and attributes
    screenshot: str | None           # Base64-encoded PNG screenshot of the page
    structured_data: dict | None     # JSON-LD and microdata extracted from the page
    headings: list[dict] | None      # List of headings with level and text (e.g. [{level: 1, text: "Title"}])
    images: list[dict] | None        # List of images with src, alt, and dimensions
    extract: dict | None             # LLM-extracted structured data matching the provided schema
    citations: list[dict] | None     # Source citations for extracted content
    markdown_with_citations: str | None  # Markdown content with inline citation references
    content_hash: str | None         # SHA-256 hash of the content for change detection
    metadata: PageMetadata | None    # Page metadata including title, status_code, word_count, and SEO tags

class PageMetadata:
    title: str | None                # Page title from <title> tag
    description: str | None          # Meta description from <meta name='description'>
    language: str | None             # Page language code (e.g. 'en', 'fr', 'de')
    source_url: str | None           # The URL that was actually scraped (after redirects)
    status_code: int | None          # HTTP response status code (200, 404, 500, etc.)
    word_count: int                  # Number of words in the main content
    reading_time_seconds: int        # Estimated reading time in seconds
    content_length: int              # Response body size in bytes
    og_image: str | None             # OpenGraph image URL for social sharing previews
    canonical_url: str | None        # Canonical URL from <link rel='canonical'>
    favicon: str | None              # Favicon URL
    robots: str | None               # Robots meta tag content (e.g. 'noindex, nofollow')
    response_headers: dict | None    # HTTP response headers as key-value pairs

Crawl a website starting from a seed URL, discovering and scraping pages via BFS, DFS, or best-first strategy. The SDK provides two methods: crawl() blocks until all pages are scraped, while start_crawl() returns immediately for manual polling.

Blocking Crawl

The simplest approach. crawl() starts the job and polls until completion, then returns all results at once.

from datablue import DataBlue

with DataBlue(api_key="wh_your_api_key") as client:
    status = client.crawl(
        "https://docs.python.org/3/",
        max_pages=50,
        max_depth=2,
        include_paths=["/3/library/*"],
        poll_interval=2.0,     # check every 2 seconds
        timeout=300.0,         # give up after 5 minutes
    )

    print(f"Status: {status.status}")                # "completed"
    print(f"Pages: {status.completed_pages}/{status.total_pages}")
    print(f"Progress: {status.progress:.0%}")         # "100%"

    for page in status.data:
        print(f"  {page.url} — {page.metadata.word_count} words")

Non-blocking Crawl (Manual Polling)

Use start_crawl() to get a job ID, then poll with get_crawl_status() at your own pace.

import time
from datablue import DataBlue

with DataBlue(api_key="wh_your_api_key") as client:
    # Start the crawl — returns immediately
    job = client.start_crawl(
        "https://docs.python.org/3/",
        max_pages=100,
        max_depth=3,
        concurrency=5,
        crawl_strategy="bfs",
    )
    print(f"Job ID: {job.job_id}")
    print(f"Status: {job.status}")    # "started"

    # Poll until done
    while True:
        status = client.get_crawl_status(job.job_id)
        print(f"  Progress: {status.completed_pages}/{status.total_pages}")

        if status.is_complete:
            break
        time.sleep(3)

    # Process results
    for page in status.data:
        if page.success:
            print(f"  {page.url}: {len(page.markdown or '')} chars")
        else:
            print(f"  {page.url}: FAILED — {page.error}")

Paginating Large Crawls

Results are paginated (20 per page by default). For crawls with more than 20 pages, you must paginate through the results to retrieve all scraped data:

# Get page 2 of results
status = client.get_crawl_status(job.job_id)  # page 1 (default)
# Access pagination info:
# status.total_results — total pages crawled
# status.page — current page number
# status.per_page — results per page (default 20)

Use the total_results, page, and per_page fields on the CrawlStatus response to determine how many pages of results exist and iterate through them. If your crawl returns fewer than 20 pages, all results will be in the first response.

Cancel a Crawl

result = client.cancel_crawl(job.job_id)
print(result)  # {"success": True, "message": "Crawl cancelled"}

Async

from datablue import AsyncDataBlue

async with AsyncDataBlue(api_key="wh_your_api_key") as client:
    # Blocking async crawl
    status = await client.crawl(
        "https://docs.python.org/3/",
        max_pages=50,
        max_depth=2,
    )
    for page in status.data:
        print(page.url)

    # Non-blocking: start + poll
    job = await client.start_crawl("https://example.com", max_pages=20)
    status = await client.get_crawl_status(job.job_id)
    await client.cancel_crawl(job.job_id)

Parameters

The crawl() blocking method accepts two additional parameters:

Streaming Crawl

Use crawl_stream() to receive pages in real time via NDJSON streaming — no polling required. Pages are yielded as they are discovered and scraped.

from datablue import DataBlue

with DataBlue(api_key="wh_your_api_key") as client:
    for page in client.crawl_stream("https://docs.example.com", max_pages=50):
        print(f"{page.url} — {len(page.markdown or '')} chars")

Streaming with Callbacks

Use crawl_stream_with_callback() for event-driven architectures. Provide callback functions instead of iterating.

from datablue import DataBlue

pages = []

with DataBlue(api_key="wh_your_api_key") as client:
    client.crawl_stream_with_callback(
        "https://example.com",
        max_pages=100,
        on_document=lambda page: pages.append(page),
        on_complete=lambda: print(f"Done! {len(pages)} pages"),
        on_error=lambda e: print(f"Error: {e}"),
    )

Response Model

class CrawlJob:                         # from start_crawl()
    success: bool                        # Whether the crawl job was accepted
    job_id: str                          # Unique job identifier for polling crawl status
    status: str                          # Current job status (typically "started")
    message: str | None                  # Human-readable status message or error description

class CrawlStatus:                       # from crawl() or get_crawl_status()
    success: bool                        # Whether the status request succeeded
    job_id: str                          # Unique job identifier
    status: str                          # "pending" | "running" | "completed" | "failed" | "cancelled" | "started"
    total_pages: int                     # Total number of pages discovered for crawling
    completed_pages: int                 # Number of pages successfully scraped so far
    data: list[CrawlPageData]            # List of scraped page results (CrawlPageData objects)
    total_results: int                   # Total number of results available (for pagination)
    page: int                            # Current page number in paginated results (default: 1)
    per_page: int                        # Number of results per page (default: 20)
    error: str | None                    # Human-readable error message if the crawl failed

    # Properties
    is_complete: bool                    # True when status in {"completed", "failed", "cancelled"}
    progress: float                      # completed_pages / total_pages (0.0 to 1.0)

class CrawlPageData:
    id: str | None                       # Unique identifier for this page within the crawl job
    url: str | None                      # The URL of the crawled page
    markdown: str | None                 # Clean Markdown conversion of the page content
    fit_markdown: str | None             # Markdown trimmed to fit LLM context windows
    html: str | None                     # Cleaned HTML content with boilerplate removed
    raw_html: str | None                 # Original unmodified HTML source of the page
    links: list[str] | None              # List of URLs found on the page
    links_detail: dict | list | None     # Detailed link info including anchor text and attributes
    screenshot: str | None               # Base64-encoded PNG screenshot of the page
    structured_data: dict | None         # JSON-LD and microdata extracted from the page
    headings: list[dict] | None          # List of headings with level and text
    images: list[dict] | None            # List of images with src, alt, and dimensions
    extract: dict | None                 # LLM-extracted structured data matching the provided schema
    citations: list[dict] | None         # Source citations for extracted content
    markdown_with_citations: str | None  # Markdown content with inline citation references
    content_hash: str | None             # SHA-256 hash of the content for change detection
    metadata: PageMetadata | None        # Page metadata including title, status_code, word_count, and SEO tags
    error: str | None                    # Error message if scraping this page failed
    success: bool                        # Whether this individual page was scraped successfully

Search the web using Google (via SearXNG), DuckDuckGo, or Brave, then scrape each result page and return structured content. Like crawl, search offers a blocking search() method and a non-blocking start_search() method.

Blocking Search

from datablue import DataBlue

with DataBlue(api_key="wh_your_api_key") as client:
    status = client.search(
        "best python web scraping libraries 2026",
        num_results=5,
        engine="google",
        formats=["markdown"],
        only_main_content=True,
    )

    print(f"Query: {status.query}")
    print(f"Results: {status.completed_results}/{status.total_results}")

    for item in status.data:
        print(f"\n--- {item.title} ---")
        print(f"URL: {item.url}")
        print(f"Snippet: {item.snippet}")
        if item.markdown:
            print(f"Content: {item.markdown[:200]}...")

Non-blocking Search

import time
from datablue import DataBlue

with DataBlue(api_key="wh_your_api_key") as client:
    job = client.start_search(
        "machine learning frameworks comparison",
        num_results=10,
        engine="google",
    )
    print(f"Job ID: {job.job_id}")

    while True:
        status = client.get_search_status(job.job_id)
        print(f"  Progress: {status.completed_results}/{status.total_results}")
        if status.is_complete:
            break
        time.sleep(2)

    for item in status.data:
        print(f"  {item.title}: {item.url}")

Async

from datablue import AsyncDataBlue

async with AsyncDataBlue(api_key="wh_your_api_key") as client:
    status = await client.search(
        "latest AI research papers",
        num_results=5,
        engine="google",
    )
    for item in status.data:
        print(item.title, item.url)

With LLM Extraction

Apply LLM extraction to each search result for structured output:

status = client.search(
    "python web frameworks comparison",
    num_results=3,
    extract={
        "prompt": "Extract the framework name, pros, and cons",
        "schema": {
            "type": "object",
            "properties": {
                "framework": {"type": "string"},
                "pros": {"type": "array", "items": {"type": "string"}},
                "cons": {"type": "array", "items": {"type": "string"}},
            },
        },
    },
)

for item in status.data:
    if item.extract:
        print(f"{item.extract['framework']}: {item.extract['pros']}")

Parameters

The search() blocking method accepts two additional parameters:

Response Model

class SearchJob:                         # from start_search()
    success: bool                        # Whether the search job was accepted
    job_id: str                          # Unique job identifier for polling search status
    status: str                          # Current job status (typically "started")
    message: str | None                  # Human-readable status message or error description

class SearchStatus:                      # from search() or get_search_status()
    job_id: str                          # Unique job identifier
    status: str                          # "pending" | "running" | "completed" | "failed" | "cancelled" | "started"
    query: str | None                    # The search query that was executed
    total_results: int                   # Total number of search results found
    completed_results: int               # Number of results successfully scraped so far
    data: list[SearchResultItem]         # List of search result items with optional scraped content
    error: str | None                    # Human-readable error message if the search failed

    # Properties
    is_complete: bool                    # True when status in {"completed", "failed", "cancelled"}
    progress: float                      # completed_results / total_results (0.0 to 1.0)

class SearchResultItem:
    id: str | None                       # Unique identifier for this search result
    url: str                             # URL of the search result
    title: str | None                    # Title of the search result from the search engine
    snippet: str | None                  # Search result snippet/description text
    success: bool                        # Whether this result was successfully scraped
    markdown: str | None                 # Clean Markdown conversion of the scraped page content
    html: str | None                     # Cleaned HTML content of the scraped page
    links: list[str] | None              # List of URLs found on the scraped page
    links_detail: list[dict] | None      # Detailed link info including anchor text and attributes
    screenshot: str | None               # Base64-encoded PNG screenshot of the page
    structured_data: dict | None         # JSON-LD and microdata extracted from the page
    headings: list[dict] | None          # List of headings with level and text
    images: list[dict] | None            # List of images with src, alt, and dimensions
    extract: dict | None                 # LLM-extracted structured data matching the provided schema
    metadata: PageMetadata | None        # Page metadata including title, status_code, word_count, and SEO tags
    error: str | None                    # Error message if scraping this result failed

Discover all URLs on a website by combining sitemap.xml parsing, robots.txt discovery, and link crawling. The map() method returns a flat list of discovered URLs with metadata. This is useful for understanding site structure before launching a targeted crawl.

Basic Usage

from datablue import DataBlue

with DataBlue(api_key="wh_your_api_key") as client:
    result = client.map("https://docs.python.org")

    print(f"Total URLs: {result.total}")
    for link in result.links:
        print(f"  {link.url}")
        if link.title:
            print(f"    Title: {link.title}")
        if link.lastmod:
            print(f"    Last modified: {link.lastmod}")

With Search Filter

# Only find URLs containing "tutorial"
result = client.map(
    "https://docs.python.org",
    search="tutorial",
    limit=50,
)

print(f"Found {result.total} tutorial URLs")
for link in result.links:
    print(f"  {link.url}")

URL Shorthand

Use the urls property to get a flat list of URL strings:

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

# Get just the URLs as a plain list
url_list = result.urls    # ["https://example.com/", "https://example.com/about", ...]
print(f"Found {len(url_list)} URLs")

# Feed into a crawl or batch scrape
crawl = client.crawl(url_list[0], max_pages=len(url_list))

Async

from datablue import AsyncDataBlue

async with AsyncDataBlue(api_key="wh_your_api_key") as client:
    result = await client.map(
        "https://example.com",
        limit=500,
        include_subdomains=True,
    )
    print(f"Found {result.total} URLs")
    for url in result.urls:
        print(url)

Parameters

Response Model

class MapResult:
    success: bool                        # Whether the map request succeeded
    total: int                           # Total number of links discovered on the site
    links: list[LinkResult]              # List of discovered links with URL and optional metadata
    error: str | None                    # Human-readable error message if the map failed
    job_id: str | None                   # Unique job identifier for async map requests

    # Properties
    urls: list[str]                      # Convenience: [link.url for link in links]

class LinkResult:
    url: str                             # The discovered URL
    title: str | None                    # Page title (from sitemap or page metadata)
    description: str | None              # Page description (from sitemap or meta tags)
    lastmod: str | None                  # Last modification date in ISO 8601 format (from sitemap)
    priority: float | None               # Sitemap priority value between 0.0 and 1.0

Scrape multiple URLs in a single call. The sync client runs sequentially, while the async client runs concurrently with configurable parallelism. For the async client, batch_scrape_iter() yields results as they complete for streaming processing.

Sync Batch Scrape

from datablue import DataBlue

urls = [
    "https://example.com",
    "https://news.ycombinator.com",
    "https://github.com",
    "https://stackoverflow.com",
]

with DataBlue(api_key="wh_your_api_key") as client:
    results = client.batch_scrape(urls, concurrency=5)

    for result in results:
        if result.success:
            print(f"{result.data.url}: {result.data.metadata.word_count} words")
        else:
            print(f"FAILED: {result.error}")

Sync with Scrape Options

results = client.batch_scrape(
    urls,
    concurrency=3,
    scrape_options={
        "formats": ["markdown", "links"],
        "only_main_content": True,
        "timeout": 45000,
    },
)

Async Batch (Collect All)

from datablue import AsyncDataBlue

async with AsyncDataBlue(api_key="wh_your_api_key") as client:
    results = await client.batch_scrape(
        urls,
        concurrency=10,
        scrape_options={"formats": ["markdown"]},
    )
    print(f"Scraped {len(results)} pages")
    for r in results:
        print(f"  {r.data.url}: {r.success}")

Async Streaming (Recommended for Large Batches)

Use batch_scrape_iter() to process results as they arrive — no need to wait for all pages to finish before starting processing.

from datablue import AsyncDataBlue

urls = ["https://example.com/page/" + str(i) for i in range(100)]

async with AsyncDataBlue(api_key="wh_your_api_key") as client:
    completed = 0
    async for result in client.batch_scrape_iter(urls, concurrency=10):
        completed += 1
        if result.success:
            print(f"[{completed}/{len(urls)}] {result.data.url} — {result.data.metadata.word_count} words")
        else:
            print(f"[{completed}/{len(urls)}] FAILED: {result.error}")

Parameters

Methods Summary

The SDK raises typed exceptions for all API errors. Every exception inherits from DataBlueError, making it easy to catch all errors or handle specific types. The HTTP client automatically retries transient errors (429 and 5xx) with exponential backoff before raising.

Exception Hierarchy

DataBlueError                         # Base exception for all SDK errors
    AuthenticationError               # 401 — bad or missing API key / JWT
    NotFoundError                     # 404 — resource does not exist
    RateLimitError                    # 429 — rate limit exceeded (retryable)
    ServerError                       # 5xx — server error (retryable)
    JobFailedError                    # Polled job completed with "failed" status
    TimeoutError                      # Polling timeout exceeded

Basic Error Handling

from datablue import (
    DataBlue,
    DataBlueError,
    AuthenticationError,
    RateLimitError,
    NotFoundError,
    ServerError,
    JobFailedError,
    TimeoutError,
)

with DataBlue(api_key="wh_your_api_key") as client:
    try:
        result = client.scrape("https://example.com")
    except AuthenticationError as e:
        print(f"Auth failed: {e.message}")
        print(f"Status: {e.status_code}")        # 401
        print(f"Docs: {e.docs_url}")              # https://docs.datablue.dev/errors/authentication
    except RateLimitError as e:
        print(f"Rate limited: {e.message}")
        print(f"Retry after: {e.retry_after}s")   # seconds to wait
        print(f"Retryable: {e.is_retryable}")     # True
    except NotFoundError as e:
        print(f"Not found: {e.message}")           # 404
    except ServerError as e:
        print(f"Server error ({e.status_code}): {e.message}")
        print(f"Retryable: {e.is_retryable}")     # True
    except DataBlueError as e:
        print(f"API error: {e.message}")
        print(f"Status: {e.status_code}")
        print(f"Body: {e.response_body}")

Job Errors (Crawl / Search)

from datablue import DataBlue, JobFailedError, TimeoutError

with DataBlue(api_key="wh_your_api_key") as client:
    try:
        status = client.crawl(
            "https://example.com",
            max_pages=100,
            timeout=60.0,       # fail if not done in 60s
        )
    except TimeoutError as e:
        print(f"Timed out after {e.elapsed:.1f}s")
        print(f"Job ID: {e.job_id}")
        # Optionally cancel the still-running job
        client.cancel_crawl(e.job_id)
    except JobFailedError as e:
        print(f"Job failed: {e.message}")
        print(f"Job ID: {e.job_id}")
        print(f"Response: {e.response_body}")

Exception Attributes

AI-Friendly Error Messages

v2.0.0 errors include fix suggestions directly in the message, making them useful for both humans and AI coding assistants:

# AuthenticationError message includes fix instructions:
# "Authentication failed. Set DATABLUE_API_KEY environment variable
#  or pass api_key to DataBlue(api_key='wh_...')"

# RateLimitError includes wait time:
# "Rate limit exceeded. Wait 42s before retrying,
#  or reduce request frequency."

# TimeoutError includes fix suggestion:
# "Job crawl-abc123 did not complete within 300s.
#  Try increasing the timeout parameter."

# ServerError indicates auto-retry:
# "Server error (502). This request will be automatically retried."

Automatic Retries

The SDK uses an immutable ClientConfig dataclass for all configuration. You can pass parameters directly to the constructor, use environment variables, or build a config object manually.

Constructor Parameters

from datablue import DataBlue

client = DataBlue(
    api_url="https://api.datablue.dev",   # Base URL (default: http://localhost:8000)
    api_key="wh_your_api_key",             # API key (wh_ prefix)
    timeout=120.0,                          # Request timeout in seconds (default: 60)
    max_retries=5,                          # Max retry attempts (default: 3)
)

ClientConfig Object

For advanced control, build a ClientConfig and pass it to the constructor:

from datablue import DataBlue, ClientConfig

config = ClientConfig(
    api_url="https://api.datablue.dev",
    api_key="wh_your_api_key",
    timeout=120.0,
    max_retries=5,
    backoff_factor=1.0,                  # Multiplier for exponential backoff (default: 0.5)
)

client = DataBlue(config=config)

Config from Environment

from datablue import DataBlue, ClientConfig

# Build config from DATABLUE_* env vars
config = ClientConfig.from_env()

# Use with either client type
sync_client = DataBlue(config=config)

from datablue import AsyncDataBlue, ClientConfig

config = ClientConfig.from_env()
async_client = AsyncDataBlue(config=config)

Cloning Configs

Configs are immutable (frozen dataclass). Use clone() to create modified copies for different environments:

from datablue import DataBlue, ClientConfig

# Base config
prod = ClientConfig(
    api_url="https://api.datablue.dev",
    api_key="wh_prod_key",
    timeout=60.0,
    max_retries=3,
)

# Derive staging config (inherits everything except overrides)
staging = prod.clone(
    api_url="https://staging.datablue.dev",
    api_key="wh_staging_key",
)

# Derive a fast config for time-sensitive operations
fast = prod.clone(timeout=10.0, max_retries=1)

# Use each
with DataBlue(config=prod) as client:
    result = client.scrape("https://example.com")

ClientConfig Fields

Self-Hosted Setup

Point the SDK at your self-hosted DataBlue instance by setting the api_url:

# Direct constructor
with DataBlue(
    api_url="https://scraper.internal.company.com",
    api_key="wh_internal_key",
) as client:
    result = client.scrape("https://example.com")

# Or via environment variables
export DATABLUE_API_URL=https://scraper.internal.company.com
export DATABLUE_API_KEY=wh_internal_key

from datablue import DataBlue

with DataBlue.from_env() as client:
    result = client.scrape("https://example.com")
    print(result.data.markdown)

Complete API Reference (v2.0.0)

AI-First Documentation Files

v2.0.0 ships with machine-readable reference files for AI coding assistants:

Rate limits are enforced per API key or JWT token. Limits vary by plan tier. When exceeded, the API returns 429 Too Many Requests.

Rate Limit Headers

Every API response includes these headers:

429 Response Example

{
  "error": "Rate limit exceeded",
  "message": "You have exceeded the rate limit of 10 requests per minute. Please wait and try again.",
  "retry_after": 42
}

DataBlue uses standard HTTP status codes. All error responses include a JSON body with error and message fields.

Error Response Format

// 400 Bad Request
{
  "success": false,
  "error": "Bad Request",
  "message": "Field 'url' is required"
}

// 401 Unauthorized
{
  "success": false,
  "error": "Unauthorized",
  "message": "Invalid or expired authentication token"
}

// 429 Rate Limited
{
  "success": false,
  "error": "Rate limit exceeded",
  "message": "You have exceeded the rate limit of 60 requests per minute",
  "retry_after": 23
}

// 500 Internal Error
{
  "success": false,
  "error": "Internal Server Error",
  "message": "An unexpected error occurred. Request ID: req_a1b2c3d4"
}

Webhooks allow you to receive real-time notifications when jobs complete instead of polling. Pass webhook_url and optionally webhook_secret when starting any async job (crawl, search, extract).

Webhook Events

Payload Format

POST https://your-server.com/webhook

Headers:
  Content-Type: application/json
  X-Webhook-Signature: sha256=a1b2c3d4e5f6...
  X-Webhook-Event: crawl.completed

Body:
{
  "event": "crawl.completed",
  "job_id": "550e8400-e29b-41d4-a716-446655440000",
  "status": "completed",
  "timestamp": "2026-04-03T12:00:00Z",
  "data": {
    "total_pages": 47,
    "completed_pages": 47,
    "url": "https://docs.python.org"
  }
}

Signature Verification

If you provide a webhook_secret, DataBlue signs each payload with HMAC-SHA256. The signature is sent in the X-Webhook-Signature header.

import hmac
import hashlib

def verify_webhook(payload: bytes, signature: str, secret: str) -> bool:
    """Verify the HMAC-SHA256 signature from DataBlue webhooks."""
    expected = "sha256=" + hmac.new(
        secret.encode(),
        payload,
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(expected, signature)

# Usage in a Flask/FastAPI handler:
@app.post("/webhook")
async def handle_webhook(request: Request):
    body = await request.body()
    signature = request.headers.get("X-Webhook-Signature", "")
    if not verify_webhook(body, signature, "your_webhook_secret"):
        raise HTTPException(status_code=401, detail="Invalid signature")
    data = await request.json()
    print(f"Event: {data['event']}, Job: {data['job_id']}")

Retry Policy

Every API call consumes credits. The cost varies by endpoint complexity. Credits reset monthly based on your plan.

Credit Costs by Endpoint

Plan Comparison