Python

pip install spidra

from spidra import Spidra

spidra = Spidra(api_key="spd_YOUR_API_KEY")

job = spidra.search("best espresso machine 2026", sources=["web", "news"])

for result in job.data.web or []:
    print(result.title, result.url)

job = spidra.search("transformer attention mechanisms", sources=["research"])

for paper in job.data.research or []:
    print(paper.title, paper.authors, paper.doi)

job = spidra.search("memory leak is:issue repo:vercel/next.js", sources=["developer"])

page1 = spidra.search("espresso machine reviews", limit=10)
page2 = spidra.search("espresso machine reviews", page_tokens=page1.data.next_page_tokens)

job = spidra.search("product launch", sources=["news"], time_range="week")

job = spidra.search("espresso machine reviews", include_domains=["reddit.com"])

from spidra.types.search import SearchScrapeOptions

job = spidra.search(
    "posthog before_send config",
    scrape_options=SearchScrapeOptions(formats=["markdown"], max_results=5),
)

job.data.web[0].markdown  # the scraped page's content, right there on the result

queued = spidra.start_search("electric cars")
status = spidra.get_search(queued.job_id)

from spidra import Spidra

spidra = Spidra(api_key="spd_YOUR_API_KEY")

result = spidra.scrape(
    "https://spidra.io/pricing",
    prompt="Extract all pricing plans with name, price, and included features",
    output="json",
)

print(result.content)
# {"plans": [{"name": "Starter", "price": "$9/mo", ...}]}

queued = spidra.start_scrape(
    "https://spidra.io",
    prompt="Extract the main headline",
)

# Later...
status = spidra.get_scrape(queued.job_id)

if status.status == "completed":
    print(status.result.content)

from spidra import ScrapeParams, ScrapeUrl

result = spidra.scrape(ScrapeParams(
    urls=[ScrapeUrl(url="https://github.careers")],
    prompt="Extract the job listing details",
    output="json",
    schema={
        "type": "object",
        "required": ["title", "company", "remote"],
        "properties": {
            "title":      {"type": "string"},
            "company":    {"type": "string"},
            "remote":     {"type": ["boolean", "null"]},
            "salary_min": {"type": ["number", "null"]},
            "skills":     {"type": "array", "items": {"type": "string"}},
        },
    },
))

from pydantic import BaseModel

class JobListing(BaseModel):
    title: str
    company: str
    remote: bool | None = None
    skills: list[str] = []

result = spidra.scrape(
    "https://github.careers",
    prompt="Extract the job listing details",
    output="json",
    schema=JobListing,
)

listing = JobListing.model_validate(result.content)  # validated, typed access

result = spidra.scrape(
    "https://www.amazon.de/gp/bestsellers",
    prompt="List the top 10 products with name and price",
    use_proxy=True,
    proxy_country="de",
)

result = spidra.scrape(
    "https://app.spidra.io/dashboard",
    prompt="Extract the monthly revenue and active user count",
    cookies="session=abc123; auth_token=xyz789",
)

from spidra import ScrapeUrl, BrowserAction

result = spidra.scrape(
    ScrapeUrl(
        url="https://spidra.io/products",
        actions=[
            BrowserAction(type="click", selector="#accept-cookies"),
            BrowserAction(type="wait", duration=1000),
            BrowserAction(type="scroll", to="80%"),
        ],
    ),
    prompt="Extract all product names and prices visible on the page",
)

result = spidra.scrape(
    "https://spidra.io",
    prompt="...",
    poll_interval=5,
    timeout=300,
)

batch = spidra.batch_scrape(
    ["https://books.toscrape.com/catalogue/a-light-in-the-attic_1000/index.html", "https://books.toscrape.com/catalogue/tipping-the-velvet_999/index.html"],
    prompt="Extract product name, price, and whether it is in stock",
    output="json",
)

print(f"{batch.completed_count}/{batch.total_urls} completed")

for item in batch.items:
    if item.status == "completed":
        print(item.url, item.result)
    else:
        print(f"Failed: {item.url}: {item.error}")

queued = spidra.start_batch_scrape(
    ["https://spidra.io/blog", "https://spidra.io/pricing"],
    prompt="Extract the page title and meta description",
)

# Come back later
result = spidra.get_batch_scrape(queued.batch_id)
print(f"{result.completed_count} of {result.total_urls} done")

if batch.failed_count > 0:
    spidra.retry_batch_scrape(queued.batch_id)

# Cancel a running batch and refund unprocessed credits
response = spidra.cancel_batch_scrape(batch_id)
print(f"Cancelled {response.cancelled_items} items, refunded {response.credits_refunded} credits")

page = spidra.list_batch_scrapes(page=1, limit=20)

for job in page.jobs:
    print(job.uuid, job.status, f"{job.completed_count}/{job.total_urls}")

template = "https://www.indeed.com/jobs?q={title}&l={location}&fromage=7&sort=date"
variables = {
    "title": ["merchandiser", "category manager"],
    "location": ["Columbus, OH", "Dallas, TX", "Atlanta, GA"],
}

estimate = spidra.estimate_bulk(template, variables)
# BulkEstimate(pages=6, batches=1, credits=6, ai_tokens_extra=False)

result = spidra.bulk_scrape(
    template,
    variables,
    items_path="jobs",  # where the list of items sits in each page's result
    dedupe_by="id",  # the field that identifies an item
    target=500,  # optional: stop once this many unique items are collected
    output="json",
)
print(result.stats)
# BulkStats(pages=6, failed_pages=0, raw_items=230, unique_items=209, duplicates_removed=21, credits_used=6)
for job in result.items:
    print(job["title"], job["company"])

result = spidra.bulk_scrape(
    "https://example-shop.com/search?q={keyword}",
    {"keyword": ["desk lamp", "standing desk", "monitor arm"]},
    schema={
        "type": "object",
        "properties": {
            "products": {
                "type": "array",
                "items": {
                    "type": "object",
                    "properties": {"sku": {"type": "string"}, "name": {"type": "string"}, "price": {"type": "number"}},
                },
            }
        },
    },
    items_path="products",
    dedupe_by="sku",
)

job = spidra.crawl(
    "https://posthog.com/blog",
    crawl_instruction="Follow links to blog posts only, skip tag pages and the homepage",
    transform_instruction="Extract the post title, author name, publish date, and a one-sentence summary",
    max_pages=30,
)

for page in job.result:
    print(page.url, page.data)

job = spidra.crawl(
    "https://docs.spidra.io",
    crawl_instruction="Crawl all documentation pages",
    max_pages=50,
)

for page in job.result:
    # page.data contains the raw markdown of each page
    print(page.url, page.data[:200])

from spidra import CrawlParams

job = spidra.crawl(CrawlParams(
    base_url="https://github.careers",
    crawl_instruction="Crawl all job listing pages",
    schema={
        "type": "object",
        "properties": {
            "title":    {"type": "string"},
            "location": {"type": "string"},
            "salary":   {"type": "string"},
            "remote":   {"type": "boolean"},
        },
    },
    max_pages=20,
))

for page in job.result:
    print(page.data)  # {"title": "...", "location": "...", ...}

job = spidra.crawl(CrawlParams(
    base_url="https://spidra.io",
    crawl_instruction="Crawl all documentation pages",
    transform_instruction="Extract the page title and main content",
    include_paths=["/docs/*"],
    exclude_paths=["/docs/changelog/*", "/docs/legacy/*"],
    max_pages=30,
))

queued = spidra.start_crawl(
    "https://docs.spidra.io",
    crawl_instruction="Follow all documentation pages",
    transform_instruction="Extract the page title and a short summary",
    max_pages=50,
)

# Poll manually
status = spidra.get_crawl(queued.job_id)
if status.status == "completed":
    for page in status.result:
        print(page.url, page.data)

response = spidra.cancel_crawl(job_id)
print(response.status)  # "cancelled"

# Retrieve whatever was processed before cancellation
pages = spidra.crawl_pages(job_id)
for page in pages.pages:
    if page.status == "success":
        print(page.url, page.data)

response = spidra.crawl_pages(job_id)

for page in response.pages:
    print(page.url, page.status)
    # page.html: signed URL for the raw HTML snapshot
    # page.markdown: signed URL for the Markdown version

queued = spidra.crawl_extract(
    completed_job_id,
    "Extract only product SKUs and prices as structured JSON",
)

result = spidra.get_crawl(queued.job_id)

response = spidra.crawl_history(page=1, limit=10)
print(f"Total crawl jobs: {response.total}")

stats = spidra.crawl_stats()
print(f"All-time: {stats.total}")

details = spidra.get_crawl_job_details(job_id)
print(details.status, details.pages_crawled, details.credits_used)

retry = spidra.retry_crawl_page(job_id, page_id)
print(retry.data)

data = spidra.download_crawl_results(job_id, include=["markdown", "data"])

with open("crawl-results.zip", "wb") as f:
    f.write(data)

res = spidra.crawl_report(job_id)
if res.report:
    print(res.report.understood_as, res.report.navigation_pages)
    for s in res.report.skipped:
        print(s.reason, s.count)

# The links for one reason, 200 at a time
res = spidra.crawl_report(job_id, reason="not_relevant", limit=200, offset=0)

doc = spidra.parse("report.pdf", pages="1-10", include_pages=True, prompt="Invoice number, total and due date")
print(doc.markdown, doc.json)

paper = spidra.parse("https://arxiv.org/pdf/1706.03762", max_pages=3)

result = spidra.map("https://docs.python.org/3/", search="asyncio", limit=20)

for link in result.links:
    print(link.url, link.title)

queued = spidra.start_crawl(
    "https://posthog.com/blog",
    crawl_instruction="Follow blog post links only",
    transform_instruction="Extract title, author, and publish date",
    max_pages=50,
)

for page in spidra.watch_crawl(queued.job_id):
    print(page.url, page.data)  # fires once per crawled page

queued = spidra.start_batch_scrape(urls, prompt="Extract product data")

for item in spidra.watch_batch(queued.batch_id):
    print(item.url, item.status, item.result)

response = spidra.scrape_logs(
    status="failed",
    search_term="amazon.com",
    start_date="2024-01-01",
    end_date="2024-12-31",
    page=1,
    limit=20,
)

for log in response.logs:
    print(log.urls[0].get("url"), log.status, log.credits_used)

log = spidra.get_scrape_log(log_uuid)
print(log.result_data)

rows = spidra.usage("30d")  # "7d" | "30d" | "weekly"

for row in rows:
    print(row.date, row.requests, row.credits)

spidra = Spidra(
    api_key="spd_YOUR_API_KEY",
    max_retries=3,        # retry attempts for transient failures (default: 3, 0 disables)
    backoff_factor=1.0,   # base seconds, delay is backoff_factor * 2**(attempt-1)
)

from spidra import (
    SpidraError,
    SpidraAuthenticationError,
    SpidraForbiddenError,
    SpidraValidationError,
    SpidraRateLimitError,
    SpidraServerError,
    SpidraJobFailedError,
    SpidraTimeoutError,
)

try:
    result = spidra.scrape("https://spidra.io", prompt="Extract the main headline")
except SpidraAuthenticationError:
    # 401: Missing or invalid API key
    pass
except SpidraForbiddenError:
    # 403: Monthly credit limit reached
    pass
except SpidraValidationError as e:
    # 422: Bad request body, e.errors lists each problem
    print(e.errors)
except SpidraRateLimitError as e:
    # 429: Too many requests, metadata tells you exactly how long to wait
    print(f"Rate limited. {e.remaining}/{e.limit} left, retry in {e.retry_after}s")
except SpidraJobFailedError as e:
    # The job itself failed or was cancelled (not a transport error)
    print(f"Job {e.job_id} {e.job_status}: {e.message}")
except SpidraTimeoutError as e:
    # Your poll timeout elapsed, the job is still running server-side
    print(f"Still running after {e.timeout_seconds}s, check {e.job_id} later")
except SpidraServerError as e:
    # 5xx: Something went wrong on Spidra's side (already retried automatically)
    print(f"Server error ({e.status}): {e.message}")
except SpidraError as e:
    # Catch-all for anything else
    print(f"API error {e.status}: {e.message}")

import json
from spidra import verify_webhook

# FastAPI example, verify against the RAW body, not the parsed JSON
@app.post("/webhooks/spidra")
async def spidra_webhook(request: Request):
    raw = await request.body()
    if not verify_webhook(raw, request.headers.get("x-spidra-signature"), WEBHOOK_SECRET):
        raise HTTPException(status_code=401)

    event = json.loads(raw)
    if event["event"] == "crawl.page":
        print("New page:", event["page"]["url"])
    return {"ok": True}