Python SDK

The xdive package wraps the API in two clients: XDive (synchronous) and AsyncXDive (asyncio). Each method makes one API call, or polls until a job finishes, and returns the parsed JSON. The clients discover your organization from the API key, retry rate-limited requests, and raise typed exceptions.

Install

pip install xdive

Python 3.9 or later. The only runtime dependency is httpx. The package ships type annotations and a py.typed marker, so type checkers use them.

Optional extras install the framework integrations:

pip install "xdive[langchain]"
pip install "xdive[llama-index]"

Configure the client

from xdive import XDive
 
xdive = XDive()  # reads XDIVE_API_KEY and, if set, XDIVE_BASE_URL
 
xdive = XDive(
    api_key="ck_live_…",
    base_url="https://xdive.dev/api",
    org_id="org_3f9a1c2e7b5d4a6f8e0c1b2a3d4e5f60",
    timeout=60.0,
    max_retries=5,
)

api_key is the only positional argument; the others are keyword-only. Every request sends the key in the X-API-Key header.

An API key belongs to one organization, and organization endpoints live under /orgs/{id}/. The first call that needs the organization id calls GET /whoami and caches the orgId it returns for the life of the client. crawler.org_id is a property that triggers the same lookup.

The key's scopes decide which methods work: scrape for scraping, crawl for crawls and maps, search for search and research. A call outside the key's scopes raises XDiveError with status 403. See Authentication.

A first request

from xdive import XDive, page_markdown
 
with XDive() as xdive:
    page = xdive.scrape("https://example.com")
    print(page["title"])
    print(page_markdown(page))

scrape returns the page object from the response's data field. Each requested format is an entry under artifacts, keyed by the format's name; Markdown is the default format. page_markdown(page) returns the Markdown whether it is in artifacts["markdown"]["content"] (a scrape, or a crawl with formats) or in markdown (a crawl without formats), and an empty string when there is none.

Methods

Keyword options are sent as the API's JSON fields, unchanged, so every field in the API reference is available even when the SDK does not name it. Options set to None are left out of the request.

formats takes the API's format names: markdown, text, html, rawHtml, links, images, json, summary, question, highlights, branding, audio, screenshot, product and menu (audio and screenshot on scrape only). A name is sent as {"type": "markdown"}; pass a dict, such as {"type": "json", "schema": {...}} or {"type": "question", "question": "..."}, to set a format's options.

whoami

xdive.whoami()

{
  "auth": "api_key",
  "keyId": "key_9b2e4f71c0a35d8e6f1a2b3c4d5e6f70",
  "orgId": "org_3f9a1c2e7b5d4a6f8e0c1b2a3d4e5f60",
  "scopes": ["crawl", "scrape", "search"],
  "plan": "free",
  "suspended": false
}

scrape

page = xdive.scrape(
    "https://example.com/pricing",
    formats=["markdown", "links"],
    max_age=3_600_000,  # accept a cached copy up to an hour old (milliseconds)
)
markdown = page_markdown(page)
print(sorted(page["artifacts"]))  # ['links', 'markdown']

Scrapes one page while you wait and returns the data object: url, title, description, status_code, the requested artifacts and more. Other options (headers, respect_robots, redact_pii, actions, …) are the scrape parameters.

crawl

record = xdive.crawl("https://docs.example.com", max_pages=50, max_depth=2)
for page in record["results"]:
    print(page["url"], page.get("error") or len(page_markdown(page)))

Starts a crawl, then polls until it reaches success, error or cancelled, and returns the final record: id, url, status, pages, durationMs, error, createdAt, results and fetch statistics. A failed page stays in results with an error field. Without formats, each page carries its Markdown in markdown; with formats, in artifacts. Crawl options (max_pages, max_depth, include_paths, exclude_paths, webhook_url, …) are the crawl parameters.

timeout (seconds) bounds the wait, not the crawl. When it runs out, CrawlTimeout is raised and the crawl keeps running on the server; call cancel_crawl to stop it.

To start a crawl and come back to it later, use the parts crawl is built from:

crawl_id = xdive.start_crawl("https://docs.example.com", max_pages=500)
status = xdive.get_crawl(crawl_id)["status"]       # queued, running, success, error or cancelled
record = xdive.wait_for_crawl(crawl_id, timeout=1800)

cancel_crawl

xdive.cancel_crawl(crawl_id)

Cancels a queued or running crawl or batch scrape; the pages fetched so far are kept. A crawl that already finished raises XDiveError with status 409.

crawl_errors

report = xdive.crawl_errors(crawl_id)

{
  "id": "crl_5c8d2e1f4a7b4c3d9e0f1a2b3c4d5e6f",
  "status": "success",
  "error": "",
  "errors": [{ "url": "https://docs.example.com/old-page", "error": "HTTP 404" }],
  "robots_blocked": ["https://docs.example.com/private/"]
}

Lists the pages of a crawl or batch scrape that failed, without downloading the results. Pages skipped because of robots.txt are listed separately in robots_blocked.

batch_scrape

record = xdive.batch_scrape(["https://example.com/a", "https://example.com/b"], formats=["markdown"])
 
job_id = xdive.batch_scrape(urls, wait=False)  # returns immediately
record = xdive.wait_for_crawl(job_id)

Scrapes up to 100 URLs as one job. It runs through the crawl pipeline, so the result is a crawl record and the crawl methods (get_crawl, cancel_crawl, crawl_errors) work on its id. See Batch scrape.

map

urls = xdive.map("https://example.com", search="pricing", limit=100)

Returns the site's URLs, found from its sitemaps and links, without fetching their content. See Map.

search

response = xdive.search("python packaging guide", limit=5, search_depth="advanced")
for result in response["results"]:
    print(result["rank"], result["url"])
    for passage in result.get("passages", []):
        print("  ", passage["text"][:120])

Returns the search response: query, provider and results. With search_depth="advanced", the top results also carry passages (the excerpts most relevant to the query) and a relevance_score; results stay in the search engine's order. Other options (sources, country, freshness, include_domains, auto_parameters, include_answer, scrape, …) are the search parameters.

research

result = xdive.research("What does PEP 639 change about license metadata?", depth="quick")
print(result["answer"])
for citation in result["citations"]:
    print(citation["number"], citation["url"])

Answers a question from several web sources with verified citations. depth is "quick" (seconds) or "deep" (one to two minutes, planned sub-queries). output is "answer" or "report". Pass schema (a JSON Schema) to get structured data instead of prose, and no_cache=True to bypass the short-lived result cache (cached is true on a cache hit). The result has answer or data, citations, sources, queries, confidence and usage. In usage, plan (when present) says how the question was planned: "single" (searched directly), "multi_hop" (answered through an intermediate search) or "unavailable" (the planning step was too slow or failed). See Research.

extract

result = xdive.extract(
    ["https://example.com/pricing", "https://example.com/plans"],
    prompt="List each plan with its monthly price",
)

Extracts structured data from up to 10 URLs with a prompt, a JSON Schema in schema, or both. See Extract.

llms_txt

files = xdive.llms_txt("https://docs.example.com", max_urls=20, full_text=True)
print(files["llms_txt"])

Generates an llms.txt index for a site, and with full_text=True also llms_full_txt. See llms.txt.

Async client

AsyncXDive has the same constructor and the same methods, as coroutines. Use it as an async context manager so its connection pool is closed:

import asyncio
 
from xdive import AsyncXDive
 
async def main() -> None:
    async with AsyncXDive() as client:
        urls = ["https://example.com/a", "https://example.com/b", "https://example.com/c"]
        pages = await asyncio.gather(*(client.scrape(url) for url in urls))
        for page in pages:
            print(page["url"], page["title"])
 
asyncio.run(main())

One difference: on AsyncXDive, org_id is a coroutine method, await client.org_id(), not a property.

Concurrent requests count against your plan's concurrency limit; requests over it are answered with 429 and retried as described below.

Errors

All exceptions derive from XDiveError, importable from xdive.

message is the API's error field. code is the API's machine-readable code when it sends one, for example CONCURRENCY_LIMIT or SERVER_BUSY. str(error) reads 429 CONCURRENCY_LIMIT: too many concurrent requests…, or 402: … without a code.

from xdive import XDive, XDiveError, QuotaExceeded, RateLimited
 
xdive = XDive()
try:
    page = xdive.scrape("https://example.com")
except RateLimited as error:
    print(f"still throttled, retry in {error.retry_after:.0f}s")
except QuotaExceeded:
    print("monthly quota used up")
except XDiveError as error:
    print(error.status, error.code, error.message)

Network failures and request timeouts are not wrapped: they raise httpx exceptions (httpx.TimeoutException, httpx.ConnectError, …). See Errors and limits for the API's status codes.

Retries and rate limits

The client retries only overload, the API's way of saying "slow down":

• 429: the organization exceeded its plan's request rate or concurrency.
• 503 with the code SERVER_BUSY in the JSON body: the server is at capacity.

Both carry a Retry-After header. The client sleeps for that many seconds (1 second when the header is missing) and sends the request again, up to max_retries times, so a request is attempted at most max_retries + 1 times. When the last attempt is still throttled it raises RateLimited with the final Retry-After in retry_after.

Every other error is raised at once, including any other 503, such as a feature the server has not configured or a job queue that is unavailable: you get a plain XDiveError with the API's status, message and code. max_retries=0 turns retries off. Retries are synchronous sleeps on XDive and asyncio.sleep on AsyncXDive.

Crawl polling

crawl, wait_for_crawl and batch_scrape (with wait=True) poll GET /orgs/{id}/crawls/{crawlId}:

• Each poll sends the ETag of the last response in If-None-Match. While nothing changed, the server answers 304 without a body.
• Between polls the client waits the number of seconds in the server's Retry-After header, which is 2 while a crawl runs, and falls back to poll_interval when the header is absent. The wait never runs past the deadline.
• Polling stops at success, error or cancelled and returns that record. A crawl that ends in error is returned, not raised; check record["status"].
• After timeout seconds it raises CrawlTimeout. The crawl itself is not cancelled.

Closing the client

Each client holds an httpx connection pool. Close it when you are done, either with a context manager or explicitly:

with XDive() as xdive:
    xdive.scrape("https://example.com")
 
xdive = XDive()
try:
    xdive.scrape("https://example.com")
finally:
    xdive.close()

For AsyncXDive, use async with AsyncXDive() as client: or await client.aclose(). A long-running service can create one client at startup, reuse it for every call, and close it at shutdown; reusing it also reuses the cached organization id and open connections.

Self-hosted

Point the client at your own server with base_url, including the /api prefix, or set XDIVE_BASE_URL:

xdive = XDive(api_key="ck_live_…", base_url="https://your-host/api")

export XDIVE_BASE_URL="https://your-host/api"

Keys are issued by the server they are used against. See Self-hosting.

Related

• JavaScript SDK
• LangChain and LlamaIndex integrations
• Errors and limits