KnowledgeSDK Python SDK

Official Python client for the KnowledgeSDK API — extract, classify, scrape, screenshot, and search web knowledge programmatically.

Installation

pip install knowledgesdk

Quick Start

from knowledgesdk import KnowledgeSDK

ks = KnowledgeSDK("sk_ks_your_key_here")

Usage

Extract

Run a full knowledge extraction on a website (synchronous):

result = ks.extract.run("https://stripe.com")

print(result.business.business_name)
print(result.business.industry_sector)
print(result.pages_scraped)

for item in result.knowledge_items:
    print(item.title, item.content)

Run an asynchronous extraction with a callback:

job = ks.extract.run_async(
    "https://stripe.com",
    max_pages=20,
    callback_url="https://myapp.com/webhook"
)

print(job.job_id)   # e.g. "job_abc123"
print(job.status)   # e.g. "PENDING"

Scrape

Scrape a single web page and get its Markdown content:

page = ks.scrape.run("https://docs.stripe.com/get-started")

print(page.title)
print(page.markdown)
print(page.links)

Classify

Classify a business from its website:

biz = ks.classify.run("https://stripe.com")

print(biz.business_name)
print(biz.business_type)
print(biz.industry_sector)
print(biz.target_audience)
print(biz.confidence_score)

Screenshot

Capture a screenshot of a web page:

shot = ks.screenshot.run("https://stripe.com")

# shot.screenshot is a base64-encoded PNG string
import base64
image_bytes = base64.b64decode(shot.screenshot)
with open("screenshot.png", "wb") as f:
    f.write(image_bytes)

Sitemap

Fetch the sitemap for a website:

site_map = ks.sitemap.run("https://stripe.com")

print(site_map.count)
for url in site_map.urls:
    print(url)

Search

Search the extracted knowledge base:

results = ks.search.run("pricing plans", limit=5)

print(f"Found {results.total} results")
for hit in results.hits:
    print(hit.title, hit.score)
    print(hit.content)

Webhooks

# Create a webhook
wh = ks.webhooks.create(
    url="https://myapp.com/hook",
    events=["EXTRACTION_COMPLETED", "JOB_FAILED"],
    display_name="My App Webhook"
)
print(wh.id)    # e.g. "weh_xxx"
print(wh.token) # signing token

# List all webhooks
all_webhooks = ks.webhooks.list()
for w in all_webhooks:
    print(w.id, w.url, w.status)

# Send a test event to a webhook
ks.webhooks.test("weh_xxx")

# Delete a webhook
ks.webhooks.delete("weh_xxx")

Jobs

Retrieve a job by ID:

job = ks.jobs.get("job_xxx")
print(job.status)   # PENDING | RUNNING | COMPLETED | FAILED
print(job.progress) # 0–100
print(job.result)

Poll until a job completes (blocking):

completed = ks.jobs.poll("job_xxx", interval_sec=5, timeout_sec=300)
print(completed.result)

Configuration

Environment Variables

export KNOWLEDGESDK_BASE_URL="https://api.knowledgesdk.com"

Debug Mode

ks = KnowledgeSDK("sk_ks_your_key", debug=True)

# Or toggle at runtime
ks.set_debug_mode(True)

Custom Headers

ks.set_header("X-Custom-Header", "value")
ks.set_headers({"X-Header-A": "a", "X-Header-B": "b"})

Error Handling

from knowledgesdk import (
    KnowledgeSDK,
    AuthenticationError,
    APIError,
    RateLimitError,
    NetworkError,
    TimeoutError,
)

ks = KnowledgeSDK("sk_ks_your_key")

try:
    result = ks.extract.run("https://stripe.com")
except AuthenticationError as e:
    print(f"Auth error: {e.message}")
except RateLimitError as e:
    print(f"Rate limited: {e.message}")
except APIError as e:
    print(f"API error {e.status_code}: {e.message}")
except NetworkError as e:
    print(f"Network error: {e.message}")
except TimeoutError as e:
    print(f"Request timed out: {e.message}")

Type Reference

All response objects are Pydantic models and are fully typed.

Requirements

• Python >= 3.8
• requests >= 2.31.0
• pydantic >= 2.0.0

License

MIT