Spidra Python SDK

This is the official Python SDK for Spidra. It searches the web, scrapes pages, runs browser actions, batch-processes URLs, and crawls entire sites.

Every result comes back as structured data, clean markdown ready to feed into an LLM pipeline, your own systems and research, or just store directly.

Installation

pip install spidra

Get your API key at app.spidra.io under API Keys.

Quick start

Synchronous (simplest)

If you're not already inside an event loop, this is the version to reach for:

from spidra import Spidra, ScrapeParams, ScrapeUrl

client = Spidra(api_key="spd_YOUR_API_KEY")

result = client.scrape(ScrapeParams(
    urls=[ScrapeUrl(url="https://news.ycombinator.com")],
    prompt="List the top 5 stories with title, points, and comment count",
    output="json",
))

print(result.content)

In an async function

Already inside asyncio, FastAPI, or similar? Use AsyncSpidra and await it directly:

import asyncio
from spidra import AsyncSpidra, ScrapeParams, ScrapeUrl

async def main():
    client = AsyncSpidra(api_key="spd_YOUR_API_KEY")

    result = await client.scrape(ScrapeParams(
        urls=[ScrapeUrl(url="https://news.ycombinator.com")],
        prompt="List the top 5 stories with title, points, and comment count",
        output="json",
    ))

    print(result.content)

asyncio.run(main())

In a Jupyter Notebook

Jupyter already runs its own event loop, so await directly in a cell:

from spidra import AsyncSpidra, ScrapeParams, ScrapeUrl

client = AsyncSpidra(api_key="spd_YOUR_API_KEY")

result = await client.scrape(ScrapeParams(
    urls=[ScrapeUrl(url="https://news.ycombinator.com")],
    prompt="List the top 5 stories with title, points, and comment count",
    output="json",
))

print(result.content)

Why? Jupyter's event loop is already running. Using bare await in a cell works because Jupyter patches its loop to accept top-level awaits. Calling asyncio.run() would start a second loop, which Python does not allow.

Two clients

Client	When to use
Spidra	Scripts, Django, Flask, CLI tools, anywhere you don't have an event loop
AsyncSpidra	FastAPI, async libraries, Jupyter notebooks

Both have identical method signatures. All code examples below use AsyncSpidra with await. For sync usage, swap AsyncSpidra → Spidra and drop the await.

Table of contents

• Spidra Python SDK
  • Installation
  • Quick start
    • Synchronous (simplest)
    • In an async function
    • In a Jupyter Notebook
  • Two clients
  • Table of contents
  • Searching
  • Scraping
    • Basic scrape
    • Structured output with JSON schema
    • Structured output with Pydantic
    • Geo-targeted scraping
    • Authenticated pages
    • Browser actions
    • forEach: process every element on a page
      • inline mode
      • navigate mode
      • click mode
      • Pagination
      • Per-element actions
      • item_prompt vs top-level prompt
    • Manual job control
    • Poll options
  • Batch scraping
  • Parsing documents
  • Crawling
    • Why a page was skipped
  • Mapping a site
  • Watching jobs (streaming results)
  • Logs
  • Usage statistics
  • Retries and reliability
  • Error handling
  • Verifying webhooks
  • Debugging
  • Context manager
  • Requirements
  • License

Searching

Search runs a real query and returns structured results: titles, links, descriptions, thumbnails. It's the same data you'd get scraping a search engine yourself, minus the scraping. Unlike scrape/batch/crawl, a plain search usually resolves in a few seconds, often within the single search() call below with no polling at all.

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

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

By default only web results come back. Request more with sources:

Source	Description
web	Standard web results (default)
news	News articles
images	Image results
videos	Video results
research	Academic papers from arXiv, PubMed, bioRxiv, and medRxiv
developer	GitHub issues and PRs.

Each source is independent. If one comes back empty or is temporarily unavailable, the others are unaffected.

All SearchParams options:

Parameter	Type	Description
query	str	Required. What to search for.
sources	list[str] | None	Which result types to request. Defaults to ["web"].
limit	int | None	Results per source, 1–20. Defaults to 10.
page	int | None	1-indexed page number. Prefer page_tokens once you have one, it's a more reliable way to continue to the next page than a bare page number.
page_tokens	dict[str, str] | None	Opaque per-source token from a prior response's next_page_tokens, for reliably continuing to that source's next page.
time_range	str | None	"hour" | "day" | "week" | "month" | "year". Restricts results to a recency window. Support can vary by source and query, and a window that can't be honored is simply ignored.
country	str | None	Two-letter country code, or "global" / "eu" / "asia", for localized results.
include_domains	list[str] | None	Only return web results from these domains. Mutually exclusive with exclude_domains. Doesn't apply to research/developer.
exclude_domains	list[str] | None	Keep web results from these domains out. Mutually exclusive with include_domains. Doesn't apply to research/developer.
filetype	str | None	"pdf". Restricts results to PDFs.
scrape_options	SearchScrapeOptions | None	Opt-in, also scrapes each web result's page content (see below).

Research and developer sources

research searches academic papers, and returns authors and doi alongside the usual fields:

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

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

developer searches GitHub issues and PRs. Write the query the way you'd type it into GitHub's own search, qualifiers and all:

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

include_domains, exclude_domains, and filetype only apply to web/news/images/videos. They're not supported on research or developer.

Pagination and time range

Pass the previous response's next_page_tokens back as page_tokens to get the next page of a specific source:

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

time_range narrows results to a recency window, useful for sources like news where older results matter less:

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

Domain filtering

Restrict web results to specific domains, or keep specific domains out. Pass one or the other, never both.

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

Scrape content from results

Add scrape_options to also fetch each web result's actual page content in the same call. No second request, no separate job to poll.

from spidra.types.search import SearchScrapeOptions

job = await client.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

This makes the search take as long as its slowest scraped page, not the usual few seconds. Real scraping isn't instant, and Spidra would rather you wait once on one job than build your own polling loop around N separate scrape jobs. max_results caps how many of the top-ranked web results get scraped; omit it to scrape all of them, up to a cap of 10. A result that fails to scrape or falls outside the time budget is simply left without markdown, not treated as an error.

Need AI extraction, a schema, or a screenshot instead of plain markdown? Scrape that specific URL directly with client.scrape().

Manual job control

If you'd rather not wait on the call at all, submit the search and check on it yourself whenever you're ready. Unlike search(), this never tries to resolve inline, it always returns a job_id immediately:

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

Scraping

All scrape jobs run asynchronously on the server side. The scrape() method submits a job, polls until it finishes, and returns the result directly. If you need more control, use start_scrape() and get_scrape().

Up to 3 URLs can be passed per request and they are processed in parallel.

Basic scrape

Give it a URL and a prompt describing what to pull out, and it comes back as structured JSON:

from spidra import AsyncSpidra, ScrapeParams, ScrapeUrl

client = AsyncSpidra(api_key="spd_YOUR_API_KEY")

result = await client.scrape(ScrapeParams(
    urls=[ScrapeUrl(url="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", "features": [...] }, ...] }

Structured output with JSON schema

When you need a guaranteed shape, pass a schema. The API will enforce the structure and return None for any missing fields rather than hallucinating values.

Define every field you want extracted. An untyped object with no properties gives the AI nothing to fill in, so those members come back empty.

result = await client.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"] },
            "salary_max": { "type": ["number", "null"] },
            "skills":     { "type": "array", "items": { "type": "string" } },
        },
    },
))

If your schema uses a keyword the API doesn't support (e.g. anyOf, $ref), the response includes schema_warnings listing what was ignored. The rest of the schema still applies.

Structured output with Pydantic

You can pass a Pydantic model (class or instance) directly instead of hand-writing JSON Schema. The SDK converts it automatically via model_json_schema().

from pydantic import BaseModel

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

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

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

The same works for batch_scrape() and crawl() (applied per page). Pydantic is an optional dependency, install it only if you use this (pip install spidra[pydantic]). Both Pydantic v2 (model_json_schema()) and v1 (schema()) are supported.

Geo-targeted scraping

Pass use_proxy=True and a proxy_country code to route the request through a specific country. Useful for geo-restricted content or localized pricing.

result = await client.scrape(ScrapeParams(
    urls=[ScrapeUrl(url="https://www.amazon.de/gp/bestsellers")],
    prompt="List the top 10 products with name and price",
    use_proxy=True,
    proxy_country="de",
))

Supported country codes include: us, gb, de, fr, jp, au, ca, br, in, nl, sg, es, it, mx, and 40+ more. Use "global" or "eu" for regional routing.

Authenticated pages

Pass cookies as a string to scrape pages that require a login session.

result = await client.scrape(ScrapeParams(
    urls=[ScrapeUrl(url="https://app.spidra.io/dashboard")],
    prompt="Extract the monthly revenue and active user count",
    cookies="session=abc123; auth_token=xyz789",
))

Browser actions

Actions let you interact with the page before the scrape runs. They execute in order, and the scrape happens after all actions complete.

from spidra import BrowserAction

result = await client.scrape(ScrapeParams(
    urls=[
        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",
))

Available actions:

Action	Required fields	Description
click	selector or value	Click a button, link, or any element
type	selector, value	Type text into an input or textarea
check	selector or value	Check a checkbox
uncheck	selector or value	Uncheck a checkbox
wait	duration (ms)	Pause execution for a set number of milliseconds
scroll	to (0–100%)	Scroll the page to a percentage of its height
forEach	observe	Loop over every matched element and process each one

For selector, use a CSS selector or XPath. For value, use a plain English description and Spidra will locate the element using AI.

# CSS selector
BrowserAction(type="click", selector="button[data-testid='submit']")

# Plain English
BrowserAction(type="click", value="Accept all cookies button")

# Type into a field
BrowserAction(type="type", selector="input[name='q']", value="wireless headphones")

# Wait for content to load
BrowserAction(type="wait", duration=2000)

# Scroll to bottom
BrowserAction(type="scroll", to="100%")

forEach: process every element on a page

forEach finds a set of elements on the page and processes each one individually. It is the right tool when you need to collect data from a list of items, paginate through multiple pages, or click into each item's detail page.

You don't need forEach if the data fits on a single page and is short. A plain prompt is simpler and works just as well.

Use forEach when:

• The list spans multiple pages and you need pagination
• You need to click into each item's detail page (navigate mode)
• You have 20+ items and want per-item AI extraction to stay consistent (item_prompt)

inline mode

Read each element's content directly without navigating. Best for product cards, search results, table rows.

from spidra import BrowserAction

result = await client.scrape(ScrapeParams(
    urls=[
        ScrapeUrl(
            url="https://books.toscrape.com/catalogue/category/books/mystery_3/index.html",
            actions=[
                BrowserAction(
                    type="forEach",
                    observe="Find all book cards in the product grid",
                    mode="inline",
                    capture_selector="article.product_pod",
                    max_items=20,
                    item_prompt="Extract title, price, and star rating. Return as JSON: {title, price, star_rating}",
                ),
            ],
        ),
    ],
    prompt="Return a clean JSON array of all books",
    output="json",
))

navigate mode

Follow each element's link to its destination page and capture content there. Best for product listings where the full detail is only on the individual page.

BrowserAction(
    type="forEach",
    observe="Find all book title links in the product grid",
    mode="navigate",
    capture_selector="article.product_page",
    max_items=10,
    wait_after_click=800,
    item_prompt="Extract title, price, star rating, and availability. Return as JSON.",
)

click mode

Click each element, capture the content that appears (a modal, drawer, or expanded section), then move on. Best for hotel room cards, FAQ accordions, or any UI where clicking reveals hidden content.

BrowserAction(
    type="forEach",
    observe="Find all room type cards",
    mode="click",
    capture_selector="[role='dialog']",
    max_items=8,
    wait_after_click=1200,
    item_prompt="Extract room name, bed type, price per night, and amenities. Return as JSON.",
)

Pagination

After processing all elements on the current page, follow the next-page link and continue collecting.

BrowserAction(
    type="forEach",
    observe="Find all book title links",
    mode="navigate",
    max_items=40,
    pagination={
        "next_selector": "li.next > a",
        "max_pages": 3,  # 3 additional pages beyond the first
    },
)

max_items applies across all pages combined. The loop stops when you hit max_items, run out of pages, or reach max_pages.

Per-element actions

Run additional browser actions on each item after navigating or clicking into it, before the content is captured.

BrowserAction(
    type="forEach",
    observe="Find all book title links",
    mode="navigate",
    capture_selector="article.product_page",
    max_items=5,
    wait_after_click=1000,
    actions=[
        BrowserAction(type="scroll", to="50%"),
    ],
    item_prompt="Extract title, price, and full description. Return as JSON.",
)

item_prompt vs top-level prompt

Both are optional and serve different purposes.

	item_prompt	prompt
When it runs	During scraping, once per item	After all items are collected
What it sees	One item's content	All items combined
Output location	Feeds into the top-level prompt	result.content

Manual job control

Use start_scrape() and get_scrape() when you want to manage polling yourself, or fire-and-forget and check back later.

# Submit a job and get the job_id immediately
queued = await client.start_scrape(ScrapeParams(
    urls=[ScrapeUrl(url="https://spidra.io")],
    prompt="Extract the main headline",
))

# Check status at any point
status = await client.get_scrape(queued.job_id)

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

Job statuses: queued, waiting, active, completed, failed.

Poll options

scrape(), batch_scrape(), and crawl() accept poll_interval and timeout keyword arguments to control polling behaviour.

result = await client.scrape(
    params,
    poll_interval=3.0,   # seconds between status checks (default: 3)
    timeout=600.0,       # max seconds to wait (default: None, waits until the job finishes)
)

By default there is no timeout. The call waits until the job reaches a terminal state, so long crawls just work. If you set a timeout and it fires, SpidraTimeoutError is raised (a TimeoutError subclass); the job keeps running server-side, so you can keep checking it with get_scrape()/get_crawl() or cancel it. Transient errors during polling (a 502 blip, a dropped connection, a rate limit) don't kill the wait, polling continues unless several happen in a row.

Batch scraping

Submit up to 50 URLs in a single request. All URLs are processed in parallel. Each URL is a plain string.

from spidra import BatchScrapeParams

batch = await client.batch_scrape(BatchScrapeParams(
    urls=[
        "https://books.toscrape.com/catalogue/a-light-in-the-attic_1000/index.html",
        "https://books.toscrape.com/catalogue/tipping-the-velvet_999/index.html",
        "https://books.toscrape.com/catalogue/soumission_998/index.html",
    ],
    prompt="Extract product name, price, and availability",
    output="json",
    use_proxy=True,
))

for item in batch.items:
    if item.status == "completed":
        print(item.url, item.result)
    elif item.status == "failed":
        print(item.url, item.error)

Item statuses: pending, running, completed, failed.

If some URLs fail (a site was down, a page timed out), you don't have to resubmit the whole batch. Re-queue just the failed items:

queued = await client.start_batch_scrape(BatchScrapeParams(
    urls=["https://spidra.io/blog", "https://spidra.io/pricing"],
    prompt="Extract the page title",
))

# Later, after checking status
result = await client.get_batch_scrape(queued.batch_id)
if result.failed_count > 0:
    await client.retry_batch_scrape(queued.batch_id)

Cancelling stops everything still pending and refunds the credits for it:

response = await client.cancel_batch_scrape(batch_id)
print(f"Cancelled {response.cancelled_items} items, refunded {response.credits_refunded} credits")

To look back through batches you've already run:

response = await client.list_batch_scrapes(page=1, limit=20)

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

Parsing documents

parse() turns a document into Markdown, page by page: PDF, Word, Excel, PowerPoint, OpenDocument, RTF, EPUB, CSV and HTML. Pass a file path, a link, bytes (with file_name), an open file, or file_id= from an earlier upload. Scanned pages are read with AI; pages with their own text aren't. It costs 1 credit per page parsed, plus AI tokens when you extract with a prompt or schema. Failed pages are free.

doc = client.parse("report.pdf", pages="1-10", include_pages=True)
print(doc.markdown)
for page in doc.pages or []:
    print(page.page_number, page.method)  # "text" or "ocr"

invoice = client.parse("https://example.com/invoice.pdf", prompt="Invoice number, total and due date")
print(invoice.json)

Options: mode ("auto", "fast" or "ocr"), pages (like "1-5, 9" or "10-"), max_pages, include_pages, page_markers, prompt and schema. Short documents answer in the same request and longer ones are polled, so parse() always returns the result; a document that can't be parsed raises SpidraJobFailedError.

• estimate_parse(...) returns the pages and credits without parsing or charging, and a file_id you can parse for 24 hours.
• parse_batch([...], file_ids=[...]) parses several paths, links and (file_name, bytes) tuples together.
• upload_for_parse(path) uploads files up to 500 MB (direct uploads take up to 50 MB) and returns a file_id.
• start_parse() / get_parse() submit and check a job yourself; delete_parse_file(file_id) deletes an upload straight away.

Crawling

Given a starting URL, Spidra discovers pages automatically according to your instruction and extracts structured data from each one.

from spidra import CrawlParams

job = await client.crawl(CrawlParams(
    base_url="https://posthog.com/blog",
    crawl_instruction="Find all blog posts published in 2024",
    transform_instruction="Extract the title, author, publish date, and a one-sentence summary",
    max_pages=30,
    use_proxy=True,
))

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

transform_instruction is optional. When omitted (and no schema is set), each page's data field contains the raw page markdown. No AI extraction is called and no token credits are charged for extraction.

All CrawlParams options:

Parameter	Type	Description
base_url	str	Required. Starting URL for the crawl.
crawl_instruction	str	Which pages to discover. Defaults to "Find all pages on the website".
transform_instruction	str | None	What to extract from each page. Omit to get raw markdown with no AI charges.
schema	dict | None	JSON Schema for structured per-page output. Root must be type: object.
max_pages	int | None	Most pages returned. Defaults to 5. Up to 100 on Free, 1,000 on Starter, 5,000 on Builder and 10,000 on Pro.
max_depth	int | None	Max link depth from the base URL. 0 = base URL only. Omit for no limit.
include_paths	list[str] | None	Only crawl pages whose path matches one of these patterns.
exclude_paths	list[str] | None	Skip pages whose path matches any of these patterns.
allow_subdomains	bool | None	Follow links to subdomains of the base domain.
crawl_entire_domain	bool | None	Follow any link on the same root domain regardless of path.
ignore_query_params	bool | None	Treat URLs differing only by query string as the same page.
webhook_url	str | None	URL that receives POST notifications as the job progresses.
use_proxy	bool | None	Route requests through a residential proxy.
proxy_country	str | None	Two-letter country code for geo-targeted proxy routing.
cookies	str | None	Cookie string for authenticated crawls.
scrape_mode	"default" | "fast" | None	"fast" reads pages over HTTP and opens a browser only for pages that need one. "default" uses a browser for every page.
use_sitemap	bool | None	Read the site's sitemap to find pages. Defaults to True.
respect_robots	bool | None	Follow the site's robots.txt rules and its Crawl-delay. Defaults to False.

Spidra reads category, listing and section pages to find the pages you asked for. Those pages are not returned and you are not charged for them. Each returned page has depth (clicks from base_url) and parent_url (the page it was found on).

If you don't want to hold a connection open for a long crawl, submit it and check back whenever you like:

queued = await client.start_crawl(CrawlParams(
    base_url="https://docs.spidra.io",
    crawl_instruction="Find all documentation pages",
    max_pages=50,
))

# Check status later
status = await client.get_crawl(queued.job_id)

max_depth, include_paths, and exclude_paths keep a crawl from wandering off into pages you don't care about:

job = await client.crawl(CrawlParams(
    base_url="https://spidra.io/blog",
    crawl_instruction="Find all blog posts",
    max_depth=2,
    include_paths=["/blog/"],
    exclude_paths=["/blog/tag/", "/blog/author/"],
    ignore_query_params=True,
))

Each crawled page includes html and markdown fields with S3-signed URLs that expire after 1 hour:

response = await client.crawl_pages(job_id)

for page in response.pages:
    print(page.url, page.status)
    # Download raw HTML:     page.html
    # Download markdown:     page.markdown

Crawled a site and want different information out of it? You don't have to re-crawl. This runs a new AI transformation over the already-crawled content and only charges for the transformation:

queued = await client.crawl_extract(source_job_id, "Extract only the product SKUs and prices as a CSV")

# Poll the new job manually
result = await client.get_crawl(queued.job_id)

Look back through crawls you've already run, or check your total count for the account:

response = await client.crawl_history(page=1, limit=10)
stats = await client.crawl_stats()
print(f"Total crawls: {stats.total}")

get_crawl_job_details() gives you a flat snapshot of one crawl job's config and counters, the same data crawl_history() returns per row:

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

If a single page's extraction failed or you want to re-run it, retry just that page instead of the whole crawl. It only charges credits for that page's transformation:

retry = await client.retry_crawl_page(job_id, page_id)
print(retry.data)

To get the raw files instead of working through the API, download everything a crawl produced as one zip:

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

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

include defaults to all three (html, markdown, data) if you omit it.

Large crawls

crawl() returns every page, however large the crawl. The status endpoint keeps only the first 100 pages inline, so for bigger crawls crawl() reads the rest for you. When you poll yourself, read them in batches with crawl_pages(job_id, offset=0, limit=1000), or all at once with crawl_all_pages(job_id).

Why a page was skipped

When a crawl finishes, its report says how Spidra understood the instruction, how many pages it read only to find others, and every link it chose not to follow, with the reason and the page the link was on:

res = await client.crawl_report(job_id)

if res.report:
    print(res.report.understood_as)      # "Individual blog posts"
    print(res.report.navigation_pages)   # pages read to find them (free)
    for s in res.report.skipped:
        print(s.reason, s.count, s.description)

# List the URLs for one reason, 200 at a time
res = await client.crawl_report(job_id, reason="not_relevant", limit=200, offset=0)
for u in res.report.skipped_urls if res.report else []:
    print(u.url, "found on", u.found_on)

Reasons: not_relevant, outside_domain, include_pattern, exclude_pattern, file_type, other_version_or_language, duplicate_page, redirected_off_site, robots_txt and action_link (links like add to cart or log out). report is None while the crawl is still running and for crawls that ran before reports existed.

Mapping a site

map() lists a site's page URLs without scraping them. It reads the sitemap and the links on the site's main pages, and answers in the same request (usually a few seconds). It costs 1 credit, and nothing when no URLs are found.

result = await client.map("https://docs.python.org/3/", search="asyncio", limit=100)

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

Parameter	Type	Description
url	str	Required. The website to map.
search	str | None	Only return URLs related to these words, best matches first.
limit	int | None	Most URLs returned, 1 to 5000. Defaults to 500.
sitemap	"include" | "only" | "skip" | None	"include" reads the sitemap and page links, "only" the sitemap, "skip" the links. Defaults to "include".
include_subdomains	bool | None	Include URLs on subdomains.
ignore_query_params	bool | None	Treat URLs that differ only by query string as one. Defaults to True.
respect_robots	bool | None	Leave out URLs the site's robots.txt disallows.
cookies	str | None	Cookie string for pages behind a login.

result.total is how many URLs matched, result.scanned how many were found before the search filter. Map is a good first step before a batch scrape: pick the URLs you want, then scrape exactly those.

Watching jobs (streaming results)

For long-running crawls and batches, watch_crawl() / watch_batch() yield each result as it lands instead of one snapshot at the end. They poll under the hood, but page content is only re-fetched when progress actually changes.

job = await client.start_crawl(
    "https://posthog.com/blog",
    crawl_instruction="Find all blog posts",
    transform_instruction="Extract title, author, and publish date",
    max_pages=50,
)

async for page in client.watch_crawl(job.job_id):
    print(page.url, page.data)  # fires once per crawled page, as soon as it is available

Batch works the same way, yielding each item as it finishes (completed or failed):

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

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

Both are plain for loops on the sync client:

job = client.start_crawl("https://spidra.io", max_pages=20)
for page in client.watch_crawl(job.job_id):
    print(page.url)

Every page/item is yielded exactly once, including ones that already existed when you started watching. The generator ends when the job completes or is cancelled, raises SpidraJobFailedError if the job fails, and SpidraTimeoutError if you pass a timeout and it is exceeded. Watching is read-only: breaking out of the loop early does not cancel the job, use cancel_crawl() / cancel_batch_scrape() for that.

Logs

Scrape logs are stored for every job that runs through the API.

# List logs with optional filters
response = await client.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)

Need the full detail on one job, not just the summary from the list? Fetch it by UUID:

log = await client.get_scrape_log("log-uuid")
print(log.result_data)  # the full AI output for that job

Usage statistics

Returns credit and request usage broken down by day or week.

# Range options: "7d" | "30d" | "weekly"
rows = await client.usage("30d")

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

Retries and reliability

Transient failures (network blips, 502/503/504 gateway errors, QUEUE_UNAVAILABLE) are retried automatically with exponential backoff, so a single hiccup never fails your call. Both knobs are configurable on either client:

client = AsyncSpidra(
    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), capped at 5s
)

Safety rules the SDK follows so retries never double-charge you:

• 4xx client errors are never retried.
• Job submissions (POSTs) are only retried when the server explicitly rejected them (502/503). Never on network errors or 504s, where the job may already have been queued.
• When the server sends a Retry-After hint (e.g. a 503 SERVICE_BUSY), the SDK honors it instead of its own backoff.

Error handling

Every API error raises a typed exception. Catch the specific class you care about or fall back to the base SpidraError.

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

try:
    result = await client.scrape("https://spidra.io", prompt="...")
except SpidraAuthenticationError:
    # 401: Missing or invalid Authorization header
    print("Check your API key")
except SpidraForbiddenError:
    # 403: Monthly credit limit reached
    print("Out of credits")
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:
    # 5xx: Something went wrong on Spidra's side (already retried automatically)
    print("Server error")
except SpidraError as e:
    # Any other API error
    print(f"{e.status}: {e.message}")

Every error class exposes e.status (the HTTP status code, or 0 for non-HTTP errors like job failures and timeouts) and e.message. API errors also carry e.code (a machine-readable identifier like SERVICE_BUSY or TOO_MANY_PENDING_JOBS) and e.details (the raw error body). SpidraRateLimitError carries limit, remaining, reset_at, and retry_after parsed from the response headers. Other classes: SpidraPaymentRequiredError (402) and SpidraNotFoundError (404).

Verifying webhooks

Crawl jobs can push crawl.page, crawl.completed, and crawl.failed events to your webhook_url. Spidra signs each delivery with HMAC-SHA256 in the X-Spidra-Signature header, and the SDK ships a verification helper (stdlib-only, constant-time compare):

import json
from spidra import verify_webhook

# FastAPI example, use 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}

Always pass the raw request body. Re-serialising parsed JSON produces different bytes and fails verification.

Debugging

Enable debug logging to see every HTTP request, response, and retry attempt:

import logging
logging.getLogger("spidra").setLevel(logging.DEBUG)

Sample output:

DEBUG:spidra:POST /scrape (attempt 1/4)
DEBUG:spidra:Response 200 in 1.23s

Context manager

Use AsyncSpidra as an async context manager to ensure the HTTP connection pool is properly closed.

async with AsyncSpidra(api_key="spd_YOUR_API_KEY") as client:
    result = await client.scrape(ScrapeParams(
        urls=[ScrapeUrl(url="https://spidra.io")],
        prompt="Extract the page title",
    ))
    print(result.content)

Requirements

• Python 3.9 or later
• A Spidra API key (sign up free)

License

MIT