Python SDK¶

from archivus import ArchivusClient

client = ArchivusClient(
    api_key=os.environ['ARCHIVUS_API_KEY'],
    tenant='your-tenant'
)

# Upload document
with open('contract.pdf', 'rb') as f:
    document = client.documents.upload(file=f, enable_ai=True)

# Search
results = client.search.query('contract terms', mode='semantic')

# Chat
session = client.chat.create_session(document_id=document.id)
answer = client.chat.ask(session.id, 'What are the key terms?')
print(answer.content)

from archivus import ArchivusClient

client = ArchivusClient(
    api_key='your-api-key',        # API key or JWT token
    tenant='your-tenant',          # Your tenant subdomain
    base_url='https://api.archivus.app/api/v1',  # Optional
    max_retries=3,                 # Optional
    timeout=30,                    # Optional (seconds)
)

# List documents
documents = client.documents.list(page=1, limit=20, folder_id='folder_abc123')

# Get document
document = client.documents.get('doc_abc123')

# Upload document
with open('file.pdf', 'rb') as f:
    document = client.documents.upload(
        file=f,
        folder_id='folder_abc123',
        enable_ai=True,
        tags=['contract', 'legal']
    )

# Update document
updated = client.documents.update(
    'doc_abc123',
    filename='new-name.pdf',
    tags=['updated']
)

# Delete document
client.documents.delete('doc_abc123')

# Download document
with client.documents.download('doc_abc123') as response:
    with open('output.pdf', 'wb') as f:
        for chunk in response.iter_content(chunk_size=8192):
            f.write(chunk)

# Get content
content = client.documents.get_content('doc_abc123')

# Basic search
results = client.search.query(
    'contract terms',
    mode='semantic',
    limit=10,
    folder_id='folder_abc123'
)

# Enhanced search (Pro+)
results = client.search.enhanced(
    query='Find contracts expiring in Q4',
    filters={
        'tags': ['contract'],
        'date_range': {
            'start': '2026-10-01',
            'end': '2026-12-31'
        }
    }
)

# Create session
session = client.chat.create_session(
    document_id='doc_abc123',
    name='Analysis'
)

# Ask question
answer = client.chat.ask(session.id, 'What are the key terms?')

# RAG query (Pro+)
answer = client.chat.ask_rag(session.id, 'Find all contracts')

# Get messages
messages = client.chat.get_messages(session.id)

# List sessions
sessions = client.chat.list_sessions()

# Delete session
client.chat.delete_session(session.id)

# Create webhook
webhook = client.webhooks.create(
    url='https://yourapp.com/webhooks',
    events=['document.processed', 'document.failed'],
    secret='your-secret'
)

# List webhooks
webhooks = client.webhooks.list()

# Update webhook
client.webhooks.update(
    'webhook_abc123',
    events=['document.processed']
)

# Delete webhook
client.webhooks.delete('webhook_abc123')

from archivus import ArchivusClient, Document, SearchResult, ChatSession

client: ArchivusClient = ArchivusClient(
    api_key=os.environ['ARCHIVUS_API_KEY'],
    tenant='your-tenant'
)

document: Document = client.documents.get('doc_abc123')
results: list[SearchResult] = client.search.query('query')
session: ChatSession = client.chat.create_session(document_id=document.id)

from archivus.exceptions import (
    ArchivusError,
    NotFoundError,
    RateLimitError,
    ValidationError
)

try:
    document = client.documents.get('doc_abc123')
except NotFoundError:
    print('Document not found')
except RateLimitError as e:
    print(f'Rate limit exceeded, retry after: {e.retry_after}')
except ValidationError as e:
    print(f'Validation error: {e.details}')
except ArchivusError as e:
    print(f'API error: {e.code} - {e.message}')

from archivus import AsyncArchivusClient
import asyncio

async def main():
    client = AsyncArchivusClient(
        api_key=os.environ['ARCHIVUS_API_KEY'],
        tenant='your-tenant'
    )

    # Upload document
    async with aiofiles.open('contract.pdf', 'rb') as f:
        document = await client.documents.upload(file=f, enable_ai=True)

    # Search
    results = await client.search.query('contract terms')

    # Chat
    session = await client.chat.create_session(document_id=document.id)
    answer = await client.chat.ask(session.id, 'What are the key terms?')

    print(answer.content)

asyncio.run(main())

from flask import Flask, request, jsonify
from archivus.webhooks import verify_webhook_signature

app = Flask(__name__)
WEBHOOK_SECRET = os.environ['WEBHOOK_SECRET']

@app.route('/webhooks/archivus', methods=['POST'])
def archivus_webhook():
    signature = request.headers.get('X-Archivus-Signature')
    payload = request.get_data(as_text=True)

    if not verify_webhook_signature(payload, signature, WEBHOOK_SECRET):
        return jsonify({'error': 'Invalid signature'}), 401

    event = request.json

    if event['type'] == 'document.processed':
        handle_document_processed(event['data'])

    return jsonify({'status': 'ok'}), 200

# Automatic resource cleanup
with ArchivusClient(api_key='...', tenant='...') as client:
    documents = client.documents.list()
    # Client automatically cleaned up after with block

# Iterate through all pages
for document in client.documents.iter_all():
    print(document.filename)

# Manual pagination
page = 1
while True:
    documents = client.documents.list(page=page, limit=100)
    if not documents.data:
        break
    for doc in documents.data:
        print(doc.filename)
    page += 1