Python SDK Reference

from sarvcrawl import KBClient

kb = KBClient(
    api_key='YOUR_API_KEY',        # or set KB_API_KEY env var
    api_url='https://crawl.sarv.com',  # the default; set your own host to self-host
    timeout=30,
    max_retries=3,
)

status = kb.health()
print(status)

user = kb.me()
print(user['id'], user['email'])

new_kb = kb.create_kb(
    'Research KB',
    description='My research knowledge base',
)
print(new_kb.id)  # 'kb_abc123'

kbs = kb.list_kbs()
for k in kbs:
    print(k.id, k.name)

my_kb = kb.get_kb('kb_abc123')
print(my_kb.name)

stats = kb.get_kb_stats('kb_abc123')
print(stats.total_pages, stats.total_mb)

kb.delete_kb('kb_abc123')

job = kb.scrape(
    'kb_abc123',
    url='https://example.com/article',
    formats=['markdown', 'html'],  # default: ['markdown', 'json']
    include_docs=True,             # auto-download linked PDFs/DOCX
)
print(job.job_id, job.status)  # 'queued'

job = kb.crawl(
    'kb_abc123',
    url='https://docs.example.com',
    formats=['markdown', 'json'],
    max_depth=5,
    limit=500,
    exclude_paths=['/admin', '/api'],
    include_docs=True,
)
print(job.job_id)

job = kb.search_job(
    'kb_abc123',
    query='knowledge infrastructure 2024',
    limit=10,    # number of results to scrape, default: 10
    lang='en',   # result language filter, default: 'en'
    formats=['markdown', 'json'],
)
print(job.job_id)

job = kb.map_job(
    'kb_abc123',
    url='https://example.com',
    limit=5000,  # default: 1000
)
print(job.job_id)

job = kb.upload(
    'kb_abc123',
    file_path='/path/to/report.pdf',
    formats=['markdown', 'json'],
    ocr_language='eng',  # default: 'eng'. Also supports 'hin'
)
print(job.job_id)

job = kb.monitor(
    'kb_abc123',
    monitor_type='page',
    urls=['https://example.com/pricing'],
    schedule='daily',
    notify_url='https://your-endpoint.example.com/webhook',
)
print(job.job_id, job.status)  # "queued"

# Website mode — crawl and watch every discovered page:
kb.monitor(
    'kb_abc123',
    monitor_type='website',
    url='https://docs.example.com',
    limit=100,
    include_paths=['/docs'],
    schedule='every 6 hours',
    notify_url='https://your-endpoint.example.com/webhook',
)

job = kb.get_job('dacf0540-f611-47eb-8e47-d5c72acea679')
print(job.status, job.urls_processed)

jobs = kb.list_jobs(
    kb_id='kb_abc123',
    status='completed',  # queued | running | completed | failed | cancelled
    type='scrape',       # scrape | crawl | search | map | upload
    limit=20,
    offset=0,
)

kb.cancel_job('dacf0540-f611-47eb-8e47-d5c72acea679')

kb.delete_job('dacf0540-f611-47eb-8e47-d5c72acea679')

logs = kb.get_job_logs('dacf0540-f611-47eb-8e47-d5c72acea679')
for log in logs:
    print(log.level, log.message)

audit = kb.get_job_audit('dacf0540-f611-47eb-8e47-d5c72acea679')
print(audit.similarity_score)

pdf_bytes = kb.get_job_source_pdf('dacf0540-f611-47eb-8e47-d5c72acea679')
with open('source.pdf', 'wb') as f:
    f.write(pdf_bytes)

md = kb.get_job_source_md('dacf0540-f611-47eb-8e47-d5c72acea679')
print(md[:500])

pages = kb.list_pages('kb_abc123', limit=50, offset=0)
for p in pages:
    print(p.url, p.title)

page = kb.get_page('kb_abc123', 'page_abc', format='markdown')
print(page.content)

results = kb.search_kb(
    'kb_abc123',
    q='knowledge infrastructure',
    page=1,
    size=10,
    type='scrape',     # optional: filter by job type
    job_id='job_xyz',  # optional: filter by specific job
)
for hit in results.hits:
    print(hit.title, hit.score)

results = kb.embed_search_kb(
    'kb_abc123',
    q='how do refunds work?',
    size=10,
    type='crawl',      # optional: filter by job type
    job_id='job_xyz',  # optional: filter by specific job
)
for hit in results.hits:
    print(hit.rrf_score, hit.title, hit.text)

jobs = kb.list_kb_jobs('kb_abc123', limit=20, offset=0)

files = kb.list_job_files('kb_abc123', 'dacf0540-f611-47eb-8e47-d5c72acea679')
for f in files.files:
    print(f.path, f.size)

data = kb.download_file(
    'kb_abc123',
    'dacf0540-f611-47eb-8e47-d5c72acea679',
    'pages/page_abc/content.md',
)
with open('content.md', 'wb') as f:
    f.write(data)

zip_bytes = kb.export_job_zip('kb_abc123', 'dacf0540-f611-47eb-8e47-d5c72acea679')
with open('job-export.zip', 'wb') as f:
    f.write(zip_bytes)

zip_bytes = kb.export_kb_zip('kb_abc123')
with open('kb-export.zip', 'wb') as f:
    f.write(zip_bytes)

with open('kb-export.jsonl', 'wb') as f:
    for chunk in kb.export_kb('kb_abc123', format='jsonl'):
        f.write(chunk)

import asyncio
from sarvcrawl import AsyncKBClient

async def main():
    kb = AsyncKBClient(api_key='YOUR_API_KEY')

    job = await kb.scrape('kb_abc123', 'https://example.com')
    print(job.job_id)

    # All methods mirror the sync client
    jobs = await kb.list_jobs(kb_id='kb_abc123', status='completed')
    pages = await kb.list_pages('kb_abc123')
    results = await kb.search_kb('kb_abc123', 'knowledge infrastructure')
    semantic = await kb.embed_search_kb('kb_abc123', 'how do refunds work?')

    # Streaming export
    with open('export.jsonl', 'wb') as f:
        async for chunk in kb.export_kb('kb_abc123'):
            f.write(chunk)

asyncio.run(main())