Python SDK

Sync and async clients for the CrawlKit API. Build data pipelines, deploy spiders, enrich datasets, export to pandas, and ship code that Google actually indexes.

Coming soon: pip install crawlkit. The examples below use requests (sync) and httpx (async). You can use these patterns today.

Installation

# Future official package
pip install crawlkit

# For now, use requests (sync) or httpx (async)
pip install requests
pip install httpx  # for async support
pip install pandas  # for dataset export

Client Setup (Sync)

import requests
from typing import Any, Optional

class CrawlKit:
    def __init__(self, api_key: str, base_url: str = "https://api.crawlkit.app/api/v1"):
        self.base_url = base_url
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
        })
        self.session.timeout = 30

    def _request(self, method: str, path: str, json: Any = None, params: dict = None) -> dict:
        url = f"{self.base_url}{path}"
        response = self.session.request(method, url, json=json, params=params)

        if not response.ok:
            error = response.json()
            raise CrawlKitError(
                message=error.get("message", f"HTTP {response.status_code}"),
                status=response.status_code,
                retryable=error.get("retryable", False),
                retry_after=error.get("retry_after"),
            )

        return response.json()

    # ── Data Platform ──

    def create_dataset(self, name: str, columns: list[dict], description: str = None) -> dict:
        body = {"name": name, "columns": columns}
        if description:
            body["description"] = description
        return self._request("POST", "/datasets", json=body)

    def list_datasets(self) -> list[dict]:
        return self._request("GET", "/datasets")

    def add_rows(self, dataset_id: str, rows: list[dict]) -> dict:
        return self._request("POST", f"/datasets/{dataset_id}/rows", json={
            "format": "json",
            "data": rows,
        })

    def list_rows(self, dataset_id: str, limit: int = 100, offset: int = 0) -> dict:
        return self._request("GET", f"/datasets/{dataset_id}/rows", params={
            "limit": limit,
            "offset": offset,
        })

    def export_dataset(self, dataset_id: str, format: str = "csv") -> dict:
        return self._request("GET", f"/datasets/{dataset_id}/export", params={
            "format": format,
        })

    def create_spider(self, config: dict) -> dict:
        return self._request("POST", "/spiders", json=config)

    def list_spiders(self) -> list[dict]:
        return self._request("GET", "/spiders")

    def enrich(self, dataset_id: str, source_name: str,
               input_mapping: dict, output_mapping: dict) -> dict:
        return self._request("POST", "/enrichment/enrich", json={
            "dataset_id": dataset_id,
            "source_name": source_name,
            "input_mapping": input_mapping,
            "output_mapping": output_mapping,
        })

    def enrichment_sources(self) -> list[dict]:
        return self._request("GET", "/enrichment/sources")

    def enrichment_status(self, job_id: str) -> dict:
        return self._request("GET", f"/enrichment/jobs/{job_id}")

    def create_pipeline(self, name: str, steps: list[dict]) -> dict:
        return self._request("POST", "/pipelines", json={
            "name": name,
            "steps": steps,
        })

    def run_pipeline(self, pipeline_id: str) -> dict:
        return self._request("POST", f"/pipelines/{pipeline_id}/run")

    # ── Crawl Jobs ──

    def start_crawl(self, site_url: str, spider_id: str, max_pages: int = 100) -> dict:
        return self._request("POST", "/crawl/jobs", json={
            "site_url": site_url,
            "spider_id": spider_id,
            "budget": {"max_pages": max_pages},
        })

    def crawl_status(self, job_id: str) -> dict:
        return self._request("GET", f"/crawl/jobs/{job_id}")

    # ── SEO Analysis ──

    def check_url(self, url: str, js_rendering: bool = False) -> dict:
        return self._request("POST", "/check-url", json={
            "url": url,
            "js_rendering": js_rendering,
        })

    def audit_site(self, site_url: str, max_pages: int = 50, js_rendering: bool = False) -> dict:
        return self._request("POST", "/audit/run", json={
            "site_url": site_url,
            "max_pages": max_pages,
            "js_rendering": js_rendering,
        })

    def dns_audit(self, domain: str) -> dict:
        return self._request("POST", "/dns/audit", json={"domain": domain})


class CrawlKitError(Exception):
    def __init__(self, message: str, status: int, retryable: bool = False,
                 retry_after: Optional[int] = None):
        super().__init__(message)
        self.status = status
        self.retryable = retryable
        self.retry_after = retry_after

Data Platform Examples

Create a Dataset and Add Rows

ck = CrawlKit("ck_live_YOUR_API_KEY")

# Create a dataset for market research
dataset = ck.create_dataset(
    name="saas_market_research",
    description="SaaS companies with revenue and tech stack data",
    columns=[
        {"name": "domain", "data_type": "text"},
        {"name": "company_name", "data_type": "text"},
        {"name": "monthly_traffic", "data_type": "integer"},
        {"name": "tech_stack", "data_type": "jsonb"},
        {"name": "funding_stage", "data_type": "text"},
        {"name": "employee_count", "data_type": "integer"},
    ],
)

print(f"Dataset created: {dataset['id']}")

# Add rows
result = ck.add_rows(dataset["id"], [
    {
        "domain": "notion.so",
        "company_name": "Notion",
        "monthly_traffic": 30_000_000,
        "tech_stack": {"frameworks": ["React"], "hosting": "AWS"},
        "funding_stage": "Series C",
        "employee_count": 800,
    },
    {
        "domain": "figma.com",
        "company_name": "Figma",
        "monthly_traffic": 25_000_000,
        "tech_stack": {"frameworks": ["React", "C++/WebAssembly"], "hosting": "AWS"},
        "funding_stage": "Acquired",
        "employee_count": 1500,
    },
])

print(f"Inserted {result['inserted']} rows")

Deploy a Spider

# Create a spider to scrape a news site
spider = ck.create_spider({
    "name": "tech_news_scraper",
    "scope": {
        "start_urls": ["https://news.ycombinator.com/news"],
    },
    "definition": {
        "handlers": [
            {
                "url_pattern": "/article/*",
                "fields": [
                    {"name": "headline", "selector": "h1.article-title", "type": "text"},
                    {"name": "author", "selector": ".author-name", "type": "text"},
                    {"name": "published_at", "selector": "time", "type": "attribute", "attribute": "datetime"},
                    {"name": "body", "selector": ".article-body", "type": "html"},
                    {"name": "tags", "selector": ".tag-list a", "type": "text"},
                ],
            },
        ],
        "navigation": {
            "follow_links": ".pagination a",
            "max_depth": 5,
        },
    },
    "dataset_id": dataset["id"],
})

print(f"Spider {spider['id']}: {spider['status']}")

Run Enrichment

import time

# Start enrichment with tech stack detection
job = ck.enrich(
    dataset_id=dataset["id"],
    source_name="tech_stack_detector",
    input_mapping={"url": "domain"},
    output_mapping={"technologies": "tech_stack"},
)

print(f"Enrichment job {job['job_id']} started")

# Poll until complete
while True:
    status = ck.enrichment_status(job["job_id"])
    print(f"Progress: {status['processed_rows']}/{status['total_rows']}")
    if status["status"] in ("completed", "failed"):
        break
    time.sleep(5)

print(f"Enrichment {status['status']}")

Build and Run a Pipeline

# Create a full data pipeline: crawl → enrich → filter → export
pipeline = ck.create_pipeline(
    name="competitor_intelligence",
    steps=[
        {
            "type": "crawl",
            "config": {"spider_id": spider["id"], "max_pages": 500},
        },
        {
            "type": "enrich",
            "config": {
                "source_name": "company_data",
                "input_mapping": {"url": "domain"},
                "output_mapping": {"revenue": "estimated_revenue"},
            },
        },
        {
            "type": "filter",
            "config": {
                "conditions": [
                    {"field": "employee_count", "operator": "gte", "value": 100},
                ],
            },
        },
        {
            "type": "export",
            "config": {
                "format": "csv",
                "destination": "webhook",
                "webhook_url": "https://your-app.com/api/pipeline-results",
            },
        },
    ],
)

# Run the pipeline
run = ck.run_pipeline(pipeline["id"])
print(f"Pipeline run {run['run_id']}: {run['status']}")

SEO Analysis Examples

Check a URL

# Ship code that Google actually indexes
result = ck.check_url("https://crawlkit.app")

print(f"SEO Score: {result['score']}/100")
print(f"Issues: {len(result['issues'])}")

# Show issues by severity
for issue in sorted(result["issues"], key=lambda i: {
    "critical": 0, "high": 1, "medium": 2, "low": 3, "info": 4
}[i["severity"]]):
    print(f"  [{issue['severity'].upper()}] {issue['message']}")

Run a Full Site Audit

audit = ck.audit_site("https://crawlkit.app", max_pages=100)

print(f"Site score: {audit['score']}/100")
print(f"Pages crawled: {audit['crawl_stats']['pages_crawled']}")
print(f"Pages with issues: {audit['crawl_stats']['pages_with_issues']}")

# Group issues by category
from collections import Counter
categories = Counter(i["category"] for i in audit["issues"])
for category, count in categories.most_common():
    print(f"  {category}: {count}")

Pandas Integration

import pandas as pd

# Export a dataset and load into pandas
rows = ck.list_rows(dataset["id"], limit=1000)
df = pd.DataFrame([row["data"] for row in rows["rows"]])

# Analyze the data
print(df.describe())
print(f"\nTop companies by traffic:")
print(df.nlargest(10, "monthly_traffic")[["company_name", "monthly_traffic", "employee_count"]])

# Export to CSV
df.to_csv("saas_companies.csv", index=False)

# Direct dataset export via API
export = ck.export_dataset(dataset["id"], format="csv")
print(f"Export URL: {export['download_url']}")

Async Client (httpx)

import httpx
import asyncio

class AsyncCrawlKit:
    def __init__(self, api_key: str, base_url: str = "https://api.crawlkit.app/api/v1"):
        self.base_url = base_url
        self.client = httpx.AsyncClient(
            headers={
                "Authorization": f"Bearer {api_key}",
                "Content-Type": "application/json",
            },
            timeout=30.0,
        )

    async def _request(self, method: str, path: str, json=None, params=None) -> dict:
        response = await self.client.request(method, f"{self.base_url}{path}",
                                             json=json, params=params)
        if response.status_code >= 400:
            error = response.json()
            raise CrawlKitError(
                message=error.get("message", f"HTTP {response.status_code}"),
                status=response.status_code,
                retryable=error.get("retryable", False),
                retry_after=error.get("retry_after"),
            )
        return response.json()

    async def create_dataset(self, name: str, columns: list[dict], description: str = None) -> dict:
        body = {"name": name, "columns": columns}
        if description:
            body["description"] = description
        return await self._request("POST", "/datasets", json=body)

    async def check_url(self, url: str, js_rendering: bool = False) -> dict:
        return await self._request("POST", "/check-url", json={
            "url": url,
            "js_rendering": js_rendering,
        })

    async def enrich(self, dataset_id: str, source_name: str,
                     input_mapping: dict, output_mapping: dict) -> dict:
        return await self._request("POST", "/enrichment/enrich", json={
            "dataset_id": dataset_id,
            "source_name": source_name,
            "input_mapping": input_mapping,
            "output_mapping": output_mapping,
        })

    async def close(self):
        await self.client.aclose()

# Usage with asyncio
async def main():
    ck = AsyncCrawlKit("ck_live_YOUR_API_KEY")

    # Run multiple operations concurrently
    dataset, check = await asyncio.gather(
        ck.create_dataset("async_demo", columns=[
            {"name": "url", "data_type": "text"},
            {"name": "score", "data_type": "integer"},
        ]),
        ck.check_url("https://crawlkit.app"),
    )

    print(f"Dataset: {dataset['id']}")
    print(f"SEO Score: {check['score']}/100")

    await ck.close()

asyncio.run(main())

Error Handling with Retry

import time
import random

def with_retry(fn, max_retries=3, base_delay=1.0):
    """Call fn() with exponential backoff on retryable errors."""
    for attempt in range(max_retries + 1):
        try:
            return fn()
        except CrawlKitError as e:
            if not e.retryable or attempt == max_retries:
                raise

            delay = (e.retry_after if e.retry_after
                     else base_delay * (2 ** attempt) + random.uniform(0, 0.5))
            print(f"Retry {attempt + 1}/{max_retries} in {delay:.1f}s: {e}")
            time.sleep(delay)

# Usage
result = with_retry(lambda: ck.check_url("https://crawlkit.app"))
print(f"Score: {result['score']}")