kubbi

Official Python SDK for the Kubbi API — send and receive ephemeral, encrypted payloads between decoupled actors.

Install

pip install kubbi

Quick Start

Create an API key at dashboard.kubbi.ai, then use it below.

Send a kubbi

from kubbi import KubbiClient

client = KubbiClient(api_key="kb_...")

result = client.send(
    content={"order_id": "123", "status": "approved"},
    content_type="application/json",
    ttl_seconds=3600,
    max_retrievals=1,
)

# Pass claim_url to the next actor (agent, service, webhook, etc.)
print(result.claim_url)

Receive a kubbi

from kubbi import claim

result = claim("https://api.kubbi.ai/r/abc123")

print(result.content)       # {"order_id": "123", "status": "approved"}
print(result.content_type)  # "application/json"

API Reference

KubbiClient

The main client for producing kubbis.

Constructor

client = KubbiClient(
    api_key="kb_...",                    # required
    base_url="https://api.kubbi.ai",     # optional (default)
    max_retries=2,                       # optional (default: 2)
    timeout=30.0,                        # optional (default: 30.0)
    debug=False,                         # optional (default: False)
)

Supports context manager usage:

with KubbiClient(api_key="kb_...") as client:
    result = client.send(...)

client.send(...) -> SendResult

Create a kubbi and get a claim URL.

result = client.send(
    content={"key": "value"},            # required — any JSON-serializable value or string
    content_type="application/json",     # required — "text/plain" | "application/json"
    ttl_seconds=3600,                    # required — 60 to 86400
    max_retrievals=1,                    # optional — burn after N retrievals
    metadata={"source": "billing"},      # optional — up to 1KB
)

# result.id, result.claim_url, result.claim_token, result.status,
# result.content_type, result.max_retrievals, result.metadata,
# result.created_at, result.expires_at

client.send_files(files, ttl_seconds, ...) -> SendFilesResult

Create a multi-file kubbi package and get a claim URL.

result = client.send_files(
    files=[
        {
            "name": "config.json",
            "content": '{"model": "gpt-4"}',
            "content_type": "application/json",
            "role": "config",
        },
        {
            "name": "prompt.txt",
            "content": "Summarize the attached data.",
            "content_type": "text/plain",
            "role": "instructions",
        },
    ],
    ttl_seconds=3600,
    max_retrievals=1,
    metadata={"source": "pipeline"},
)

# result.id, result.claim_url, result.claim_token, result.status,
# result.file_count, result.total_size_bytes, result.max_retrievals,
# result.metadata, result.created_at, result.expires_at

Each file entry accepts:

• name — filename (e.g. "config.json")
• content — file content as a string, or base64-encoded for binary
• content_type — MIME type (e.g. "application/json", "image/png")
• role (optional) — "instructions" | "data" | "context" | "config" | "attachment"
• encoding (optional) — set to "base64" for binary content

client.get(kubbi_id) -> KubbiDetail

Get details about a kubbi you created.

detail = client.get("kubbi-id-here")

# detail.id, detail.status, detail.content_type, detail.max_retrievals,
# detail.retrieval_count, detail.first_retrieved_at, detail.last_retrieved_at,
# detail.metadata, detail.created_at, detail.expires_at

client.list(page=None, limit=None) -> ListKubbisResult

List kubbis created with this API key.

result = client.list(page=1, limit=20)
# result.kubbis, result.pagination

client.delete(kubbi_id) -> dict

Delete a kubbi you created. Wipes the encrypted payload.

client.delete("kubbi-id-here")

Consumer Functions

Module-level functions for consuming kubbis. No API key required.

inspect(claim_url_or_token, ...) -> InspectResult

Inspect a kubbi without consuming it (does not count as a retrieval).

from kubbi import inspect

# With a full claim URL:
info = inspect("https://api.kubbi.ai/r/abc123")

# With just a claim token:
info = inspect("abc123", base_url="https://api.kubbi.ai")

# info.status, info.content_type, info.max_retrievals, info.retrieval_count,
# info.metadata, info.created_at, info.expires_at

claim(claim_url_or_token, ...) -> ClaimResult

Retrieve a kubbi's content. Counts as a retrieval.

from kubbi import claim

result = claim("https://api.kubbi.ai/r/abc123")

# result.content, result.content_type, result.metadata, result.created_at, result.expires_at

AsyncKubbiClient

Async version of KubbiClient with the same interface.

from kubbi import AsyncKubbiClient

async with AsyncKubbiClient(api_key="kb_...") as client:
    result = await client.send(
        content="hello",
        content_type="text/plain",
        ttl_seconds=300,
    )

Multi-file sending works the same way:

result = await client.send_files(
    files=[
        {"name": "data.csv", "content": "a,b\n1,2", "content_type": "text/csv", "role": "data"},
    ],
    ttl_seconds=3600,
)

Async consumer functions are also available:

from kubbi import aclaim, ainspect

info = await ainspect("https://api.kubbi.ai/r/abc123")
result = await aclaim("https://api.kubbi.ai/r/abc123")

Error Handling

All API errors raise typed exceptions that inherit from KubbiApiError:

from kubbi import (
    KubbiApiError,
    KubbiNotFoundError,
    KubbiGoneError,
    KubbiValidationError,
    KubbiQuotaExceededError,
    KubbiNetworkError,
    claim,
)

try:
    result = claim(claim_url)
except KubbiGoneError:
    print("Kubbi already consumed or expired")
except KubbiNotFoundError:
    print("Kubbi not found")
except KubbiNetworkError as e:
    print(f"Network issue: {e}")

Every KubbiApiError has:

• status — HTTP status code (e.g. 404, 410, 429)
• code — API error code (e.g. "not_found", "gone", "quota_exceeded")
• args[0] — Human-readable error message
• messages — List of validation error messages (for KubbiValidationError)

Retries

The SDK automatically retries on transient errors (502, 503, 504) and network failures with exponential backoff. Configure with max_retries (default: 2).

Security Notes

Claim URL origin handling: When a full URL is passed to inspect() or claim(), the SDK extracts the origin from the URL and directs HTTP requests to that host. If your application accepts claim URLs from untrusted input (e.g. user-supplied or agent-provided), validate the URL origin before passing it to the SDK, or use bare claim tokens with an explicit base_url instead.

# Safe: use bare tokens with a known base URL
result = claim("abc123", base_url="https://api.kubbi.ai")

Debug mode: When debug=True is set, request URLs including claim tokens are logged. Avoid enabling debug mode in production environments where logs may be persisted or forwarded.

Requirements

• Python 3.9+
• httpx >= 0.24.0
• pydantic >= 2.0.0

License

MIT