kraken-captia 0.1.4

Official Python SDK for Kraken document processing API

Kraken SDK for Python

Official Python SDK for the public Kraken API exposed through PCS/Stratios.

Requirements

• Python 3.9 or higher
• httpx for HTTP requests
• pydantic for data validation

Installation

pip install kraken-sdk

Or install from source:

git clone https://github.com/Capt-IA/Kraken-SDK.git
cd Kraken-SDK/kraken-sdk-python
pip install -e .

Public architecture

The supported public flow is:

1. client application → PCS/Stratios
2. PCS verifies the API key and injects signed platform context
3. PCS proxies the request to Kraken

By default, the SDK therefore prefixes requests with /services/kraken.

Quick Start

from kraken_sdk import KrakenClient
from kraken_sdk.types import ExtractionTaskRequest

client = KrakenClient(
    api_key="your-api-key",
    base_url="https://pcs.example.com",
)

source = client.sources.upload("invoice.pdf")

# Read service metadata through PCS
info = client.info.info()
print(info.service, info.version)

source = client.sources.upload("path/to/document.pdf")
print(f"Uploaded: {source.id}")

job = client.jobs.create_single(
    ExtractionTaskRequest(
        task_type="extraction",
        source_id=source.id,
        ai_provider="openai",
        provider_model="gpt-4o",
    )
)

result = client.jobs.wait(job.job_id)
tasks = client.jobs.tasks(job.job_id)
print(result.status)
print(tasks.tasks[0].status)

Async Support

The SDK provides full async support:

import asyncio
from kraken_sdk import AsyncKrakenClient
from kraken_sdk.types import ExtractionTaskRequest

async def main():
    async with AsyncKrakenClient(
        api_key="your-api-key",
        base_url="https://pcs.example.com",
    ) as client:
        source = await client.sources.upload("document.pdf")
        job = await client.jobs.create_single(
            ExtractionTaskRequest(
                task_type="extraction",
                source_id=source.id,
                ai_provider="openai",
                provider_model="gpt-4o",
            )
        )
        result = await client.jobs.wait(job.job_id)
        print(result)

asyncio.run(main())

Direct internal access

Direct Kraken URLs should stay limited to internal tooling and trusted service-to-service use. If you really need it, disable PCS path prefixing explicitly:

client = KrakenClient(
    api_key="internal-key",
    base_url="https://kraken.internal.example.com",
    use_pcs_proxy=False,
)

Structured Annotation

Extract structured data using custom schemas:

from kraken_sdk import KrakenClient
from kraken_sdk.types import AnnotationTaskRequest

client = KrakenClient(
    api_key="your-api-key",
    base_url="https://pcs.example.com",
)

annotation_schema = {
    "Invoice": {
        "general_prompt": "Invoice information",
        "is_list": False,
        "is_optional": False,
        "table_fields": [
            {"class_name": "invoice_number", "type": "str", "prompt": "The invoice number"},
            {"class_name": "total_amount", "type": "str", "prompt": "The total amount"},
            {"class_name": "date", "type": "str", "prompt": "The invoice date"}
        ]
    }
}

job = client.jobs.create_single(
    AnnotationTaskRequest(
        task_type="annotation",
        source_id=source.id,
        ai_provider="openai",
        provider_model="gpt-4o",
        annotation_config={
            "annotation_schema": annotation_schema,
            "output_format": "json",
        },
    )
)

result = client.jobs.wait(job.job_id)

Bulk Processing

Process multiple documents with the same configuration:

from kraken_sdk.types import ExtractionTaskConfig

job = client.jobs.create_bulk(
    task_config=ExtractionTaskConfig(
        task_type="extraction",
        ai_provider="openai",
        provider_model="gpt-4o",
    ),
    source_ids=["source-1", "source-2", "source-3"],
)

Error Handling

from kraken_sdk.exceptions import KrakenAuthError, KrakenError, KrakenNotFoundError

try:
    result = client.jobs.get("invalid-id")
except KrakenNotFoundError:
    print("Job not found")
except KrakenAuthError:
    print("Invalid API key")
except KrakenError as e:
    print(f"API error: {e}")

API Reference

KrakenClient

Resource	Methods
client.info	info(), health(), tasks()
client.jobs	create(), create_single(), create_bulk(), list(), get(), tasks(), wait()
client.sources	upload()
client.benchmarks	create(), list(), get(), report(), create_expected_result(), upload_expected_result(), get_expected_result(), list_expected_results()

Development

# Install dev dependencies
pip install -e ".[dev]"

# Run tests
pytest

# Lint
ruff check src/