FastAPI

The Spidra FastAPI integration (spidra-fastapi) brings the full Spidra API into FastAPI with idiomatic dependency injection, native async/await, a pre-built APIRouter that auto-documents itself in Swagger UI, HMAC-verified webhook dependencies, and a global proxy for cases where DI feels like overhead.

Installation

pip install spidra-fastapi uvicorn

Requires Python 3.9+ and FastAPI 0.100+.

Get your API key from app.spidra.io under API Keys. Store it as an environment variable. Never hardcode it.

Configuration

Set environment variables before starting your app:

export SPIDRA_API_KEY="spd_live_your_api_key"
export SPIDRA_WEBHOOK_SECRET="whsec_your_signing_secret"

# Optional
export SPIDRA_API_URL="https://api.spidra.io/api"
export SPIDRA_MAX_RETRIES=3
export SPIDRA_BACKOFF_FACTOR=1.0

Two integration styles

spidra-fastapi offers two approaches. Both work, and you can mix them freely.

Option A — Dependency injection (recommended)

get_async_spidra and get_spidra are ready-made FastAPI dependencies. They create (or reuse) a scoped client on each request. This is the idiomatic FastAPI approach and makes unit testing easy — just override the dependency.

from fastapi import Depends, FastAPI
from spidra import AsyncSpidra
from spidra_fastapi import get_async_spidra, to_dict

app = FastAPI()

@app.post("/scrape")
async def scrape(url: str, spidra: AsyncSpidra = Depends(get_async_spidra)):
    result = await spidra.scrape(url, output="markdown")
    return to_dict(result)

Option B — Global proxy

For scripts, background tasks, or simple routes where passing DI is cumbersome, import the global spidra proxy directly:

from spidra_fastapi import spidra

result = await spidra.async_client.scrape("https://example.com")

spidra.async_client returns the async Spidra client. spidra.scrape(...), spidra.search(...), etc. are also available as synchronous shortcuts on the proxy.

Scraping

from fastapi import Depends, FastAPI
from pydantic import BaseModel
from spidra import AsyncSpidra
from spidra_fastapi import get_async_spidra, to_dict

app = FastAPI()


class ScrapeRequest(BaseModel):
    url: str
    prompt: str = "Extract the main headline and key points"


@app.post("/scrape-article")
async def scrape_article(
    payload: ScrapeRequest,
    spidra: AsyncSpidra = Depends(get_async_spidra),
):
    result = await spidra.scrape(
        payload.url,
        prompt=payload.prompt,
        output="markdown",
        use_proxy=True,
    )
    return {"success": True, "result": to_dict(result)}

For structured extraction, add a schema:

@app.post("/extract-product")
async def extract_product(
    payload: ScrapeRequest,
    spidra: AsyncSpidra = Depends(get_async_spidra),
):
    result = await spidra.scrape(
        payload.url,
        prompt="Extract product details",
        schema={
            "type": "object",
            "properties": {
                "title":    {"type": "string"},
                "price":    {"type": "number"},
                "in_stock": {"type": "boolean"},
            },
            "required": ["title", "price", "in_stock"],
        },
        output="json",
    )
    return to_dict(result.content)

to_dict() recursively converts Spidra dataclasses and nested objects to plain dicts, ready to return from FastAPI routes.

Searching

@app.get("/search")
async def search_web(
    q: str,
    limit: int = 10,
    spidra: AsyncSpidra = Depends(get_async_spidra),
):
    results = await spidra.search(q, sources=["web", "news"], limit=limit)
    return {"query": q, "results": to_dict(results)}

Available sources: "web", "news", "images", "videos", "research", "developer". Defaults to ["web"].

Crawling

@app.post("/crawl-docs")
async def crawl_docs(
    start_url: str,
    max_pages: int = 30,
    spidra: AsyncSpidra = Depends(get_async_spidra),
):
    job = await spidra.crawl(
        start_url,
        crawl_instruction="Follow documentation pages only",
        transform_instruction="Extract title and summary",
        max_pages=max_pages,
    )
    return to_dict(job)

For long crawls, submit without waiting:

queued = await spidra.start_crawl(start_url, crawl_instruction="...", max_pages=100)
# later...
status = await spidra.get_crawl(queued.job_id)

Mapping

@app.get("/map")
async def map_site(
    url: str,
    limit: int = 100,
    spidra: AsyncSpidra = Depends(get_async_spidra),
):
    result = await spidra.map(url, limit=limit)
    return [{"url": l.url, "title": l.title} for l in result.links]

Batch scraping

@app.post("/batch")
async def batch_scrape(
    urls: list[str],
    prompt: str = "Extract the page title and description",
    spidra: AsyncSpidra = Depends(get_async_spidra),
):
    batch = await spidra.batch_scrape(urls, prompt=prompt, output="json")
    return to_dict(batch)

Parsing documents

@app.post("/parse")
async def parse_doc(
    url: str,
    spidra: AsyncSpidra = Depends(get_async_spidra),
):
    doc = await spidra.parse(url, prompt="Extract key figures and dates")
    return {"markdown": doc.markdown}

Webhooks

Spidra sends signed webhook events to your endpoint. spidra-fastapi ships two webhook dependencies.

`verify_spidra_webhook_json` — validated JSON dict

The simplest approach. Validates the HMAC-SHA256 signature from X-Spidra-Signature and returns the parsed payload:

from fastapi import Depends, FastAPI
from spidra_fastapi import spidra, verify_spidra_webhook_json

app = FastAPI()

@app.post("/webhooks/spidra")
async def receive_webhook(payload: dict = Depends(verify_spidra_webhook_json)):
    # Signature verified, payload is ready to use
    spidra.dispatch_event(payload)  # fires @spidra.on(...) listeners
    return {"status": "received", "event": payload.get("event")}

`verify_spidra_webhook` — raw bytes

When you need the raw body (e.g. to re-verify yourself or pass it elsewhere):

from fastapi import Depends
from spidra_fastapi import verify_spidra_webhook
import json

@app.post("/webhooks/spidra-raw")
async def receive_raw(raw: bytes = Depends(verify_spidra_webhook)):
    event = json.loads(raw)
    return {"event": event.get("event")}

Both dependencies return HTTPException(401) automatically if the signature is missing or invalid.

Event listeners

from spidra_fastapi import spidra

@spidra.on("crawl.completed")
def on_crawl_done(data):
    print(f"Crawl finished: {data.get('jobId')}")

@spidra.on("crawl.page")
def on_page(data):
    print(f"New page: {data['page']['url']}")

Pydantic webhook model

SpidraWebhookEvent is a Pydantic model for strongly typed webhook payloads:

from spidra_fastapi import SpidraWebhookEvent

@app.post("/webhooks/typed")
async def typed_webhook(event: SpidraWebhookEvent):
    print(event.event, event.job_id)
    return {"ok": True}

Advanced dependency injection

Use SpidraDependency to configure a scoped client with specific credentials, useful when handling multiple Spidra accounts in the same app:

from spidra_fastapi import SpidraDependency
from fastapi import Depends

dep = SpidraDependency(api_key="spd_account_b_key")

@app.get("/account-b/scrape")
async def scrape_b(url: str, spidra = Depends(dep.get_async)):
    return to_dict(await spidra.scrape(url))

dep.get_async() returns an AsyncSpidra client; dep.get_sync() returns a synchronous Spidra client.

Pre-built APIRouter

Mount all Spidra REST endpoints into your app in one line, auto-documented in OpenAPI/Swagger:

from fastapi import FastAPI
from spidra_fastapi import create_spidra_router

app = FastAPI(title="My Scraper API")
app.include_router(create_spidra_router(prefix="/spidra"))

Visit http://localhost:8000/docs to see the full interactive Swagger documentation. Or use the default pre-built router:

from spidra_fastapi import spidra_router

app.include_router(spidra_router)

The router exposes:

Method	Endpoint	Description
`POST`	`/spidra/scrape`	Scrape a page (`url`, `prompt`, `schema`, `renderJs`, `useProxy`)
`GET`/`POST`	`/spidra/search`	Web search (`q`, `limit`, `sources`)
`POST`	`/spidra/map`	Map domain links
`POST`	`/spidra/crawl`	Start or execute crawl
`GET`	`/spidra/crawl/{id}`	Crawl status and results
`GET`	`/spidra/crawl/{id}/report`	Crawl skip report
`DELETE`	`/spidra/crawl/{id}`	Cancel crawl
`POST`	`/spidra/crawl/{id}/retry/{page_id}`	Retry one page’s extraction
`POST`	`/spidra/batch`	Batch scrape
`GET`	`/spidra/batch/{id}`	Batch status
`DELETE`	`/spidra/batch/{id}`	Cancel batch
`POST`	`/spidra/batch/{id}/retry`	Retry failed items
`POST`	`/spidra/parse`	Parse document
`POST`	`/spidra/parse/estimate`	Estimate parse credits
`GET`	`/spidra/parse/{id}`	Parse result
`DELETE`	`/spidra/parse/files/{id}`	Delete uploaded file
`GET`	`/spidra/templates`	List templates
`GET`	`/spidra/templates/{id}`	Template details
`POST`	`/spidra/templates/{id}/run`	Run template
`GET`/`POST`	`/spidra/presets`	List or create presets
`GET`/`POST`	`/spidra/monitors`	List or create monitors
`POST`	`/spidra/monitors/{id}/check`	Trigger monitor check
`GET`	`/spidra/account`	Account and credit info
`GET`	`/spidra/usage`	Usage history
`POST`	`/spidra/webhooks`	HMAC-verified webhook receiver

Error handling

All Spidra exceptions inherit from SpidraError and are re-exported from spidra_fastapi. Register them as exception handlers:

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from spidra_fastapi import (
    SpidraAuthenticationError,
    SpidraPaymentRequiredError,
    SpidraRateLimitError,
    SpidraError,
)

app = FastAPI()

@app.exception_handler(SpidraAuthenticationError)
async def auth_handler(request: Request, exc: SpidraAuthenticationError):
    return JSONResponse(status_code=401, content={"error": "Invalid API key"})

@app.exception_handler(SpidraPaymentRequiredError)
async def payment_handler(request: Request, exc: SpidraPaymentRequiredError):
    return JSONResponse(status_code=402, content={"error": "Spidra credits exhausted"})

@app.exception_handler(SpidraRateLimitError)
async def rate_limit_handler(request: Request, exc: SpidraRateLimitError):
    return JSONResponse(
        status_code=429,
        content={"error": "Rate limited", "retry_after": getattr(exc, "retry_after", 5)},
    )

@app.exception_handler(SpidraError)
async def general_handler(request: Request, exc: SpidraError):
    return JSONResponse(status_code=500, content={"error": str(exc)})

See the Python SDK for the full exception hierarchy.

Flask

Flask extension with Blueprint routing, flask spidra CLI, and HMAC webhook decorator.

Django

Django integration with management commands, signals, and built-in URL routing.

SDKs

AI tools

Installation

Configuration

Two integration styles

Option A — Dependency injection (recommended)

Option B — Global proxy

Scraping

Searching

Crawling

Mapping

Batch scraping

Parsing documents

Webhooks

`verify_spidra_webhook_json` — validated JSON dict

`verify_spidra_webhook` — raw bytes

Event listeners

Pydantic webhook model

Advanced dependency injection

Pre-built APIRouter

Error handling

Flask

Django

​Installation

​Configuration

​Two integration styles

​Option A — Dependency injection (recommended)

​Option B — Global proxy

​Scraping

​Searching

​Crawling

​Mapping

​Batch scraping

​Parsing documents

​Webhooks

​verify_spidra_webhook_json — validated JSON dict

​verify_spidra_webhook — raw bytes

​Event listeners

​Pydantic webhook model

​Advanced dependency injection

​Pre-built APIRouter

​Error handling

Flask

Django

Installation

Configuration

Two integration styles

Option A — Dependency injection (recommended)

Option B — Global proxy

Scraping

Searching

Crawling

Mapping

Batch scraping

Parsing documents

Webhooks

`verify_spidra_webhook_json` — validated JSON dict

`verify_spidra_webhook` — raw bytes

Event listeners

Pydantic webhook model

Advanced dependency injection

Pre-built APIRouter

Error handling