Keelson Python SDK

Python SDK for building apps on the Keelson platform. Provides four modules:

Note: This repository is a read-only release mirror. Development happens in the private Keelson monorepo; issues are welcome here, but pull requests are not accepted — changes land through the next release.

Cross-language parity across Node, Python, and Go is defined in ./PARITY.md. APIs below are labelled as guaranteed (same capability in all 3 languages) or Python-specific (convenience helpers unique to this SDK).

Installation

pip install keelson-sdk

The PyPI distribution is named keelson-sdk; Python imports continue to use keelson and the domain modules shown below.

Media SDK (keelson_media)

Upload immutable media (images, PDFs, generated assets) and serve it by ID. On Keelson it uses the managed media storage (internal media API); for local development it uses the local filesystem. The runtime-mode contract is fail-closed: it never silently writes to ephemeral local storage when platform Media configuration is missing or incomplete (see Modes below).

from keelson import media

# Store a file — returns a generated ULID file ID
file_id = media.put(b"Hello, world!", content_type="text/plain")

# Store with filename (MIME type auto-detected)
with open("report.pdf", "rb") as f:
    pdf_id = media.put(f.read(), filename="report.pdf")

# Retrieve
data = media.get(file_id)

# Check existence and metadata
if media.exists(file_id):
    info = media.stat(file_id)
    print(info.content_type, info.content_length)

# Public URL (for embedding in HTML)
src = media.url(file_id)  # "/media/01HXYZ..."

# Delete
media.delete(file_id)

Cross-language guaranteed API

Python-specific helpers

Data class: MediaStat (frozen dataclass with content_type: str, content_length: int, status: int).

Exception: MediaError (subclass of RuntimeError).

Modes (fail-closed runtime-mode contract)

The SDK never silently falls back to ephemeral local storage on Keelson: set KEELSON_MODE=local explicitly for local development.

Environment variables

Files (data) SDK (keelson_files)

Durable file storage for your app's own files — state, settings, caches. Reads and writes are always whole-file, and write() is write-through: once it returns, the data is persisted. There is no background sync and nothing is stored on ephemeral local disk. Overwriting an existing key is the normal case; updates to the same key are limited to about once per second. For user-uploaded or generated media referenced by ID and served over HTTP, use keelson_media; for data read/written on every request, use the database.

from keelson import files

files.write("seen_urls.json", json.dumps(seen))
seen = json.loads(files.read("seen_urls.json") or "[]")  # read() -> bytes | None
keys = files.list()                                       # sorted list[str]
files.delete("seen_urls.json")                            # idempotent

Cross-language guaranteed API

Key grammar: /-separated relative path, well-formed UTF-8 ≤ 512 bytes total and ≤ 255 bytes per segment, no leading/trailing /, no empty / . / .. segments, no control characters. One-object soft limit 10 MiB. There is no exists() — read() returning None covers it.

Environment variables

Fail-closed: KEELSON_MODE=keelson requires bucket + prefix + platform identity; missing config raises FilesError. See ./PARITY.md for the full contract.

Identity SDK (keelson_identity)

User identity and tenant directory lookup. In production, the Keelson auth gateway injects trusted X-Keelson-User-* headers before requests reach the app. Use get_current_user when the basic user profile is enough; use get_current_identity when the app needs tenant role, app permissions, app roles, or group attributes.

import os

from keelson_identity import (
    get_current_user,
    get_current_identity,
    list_members,
    get_user,
    list_groups,
)

user = get_current_user(headers=request.headers)
print(user.email)

identity = get_current_identity(
    headers=request.headers,
    app_token=os.environ["KEELSON_DIRECTORY_TOKEN"],
)
print(identity.tenant.role)
print(identity.app.permissions)   # ["manage", "view"]
if identity.attributes:
    print(identity.attributes.groups)  # ["developers", "everyone"]

# Directory lookup as the app actor
page = list_members(
    app_token=os.environ["KEELSON_DIRECTORY_TOKEN"],
    limit=25, offset=0, q="alice",
)
for member in page.items:
    print(member.id, member.email, member.name)

member = get_user("user-id-here", app_token=os.environ["KEELSON_DIRECTORY_TOKEN"])
groups = list_groups(app_token=os.environ["KEELSON_DIRECTORY_TOKEN"])

Cross-language guaranteed API

get_current_user and get_current_identity accept common request header mappings. The required header is x-keelson-user-id; x-keelson-user-email and x-keelson-user-name are optional.

Directory functions also support app-as-actor access with app_token or the KEELSON_DIRECTORY_TOKEN env fallback. Keep app tokens on the server.

Python-specific helpers

Data classes: CurrentIdentity, UserIdentity, TenantIdentity, AppIdentity, AttributesIdentity, MemberItem, PaginatedMembers, GroupItem.

Exception: IdentityError.

Filtering members by group

list_members narrows results to a single group via either group_id or group_key:

• group_key — the group's code-facing identifier. Always present, stable, and immutable. Prefer this for code references. Keys may be non-ASCII (e.g. a Japanese 経理).
• group_id — a stable UUID for machine integration / internal wiring.

Passing both raises IdentityError. On GroupItem, key is str | None kept nullable for backward compatibility, but the server always populates it; id is the UUID.

attributes.groups vs list_groups()

• attributes.groups (from get_current_identity()): the group keys the current user belongs to, scoped to the current app — the system groups the caller holds by role (a subset of owners/admins/developers/everyone, not all four: an OWNER gets owners/developers/everyone, an app user gets only everyone; exposed regardless of app binding) plus custom groups bound to this app (via a view/manage permission binding or an app-role binding). Custom groups not bound to the app are excluded, and a system group the caller does not hold never appears. Keys are stable and immutable, so they are safe for authorization checks; bind a group to the app if you need to branch on it.
• list_groups(): all groups in the tenant (each with its stable id and key). Use for building UI pickers and admin views.

Modes

Environment variables

Email SDK (keelson_email)

Inbound/outbound email.

from keelson import email

# Send
email.send(
    to="user@example.com",
    subject="Hello",
    text="Plain text body",
    html="<p>HTML body</p>",
)

# Receive inbound emails via decorator
@email.on_receive
def handle(msg: email.InboundMessage):
    print(msg.subject, msg.text)
    email.send(
        to=msg.reply_to or msg.from_,
        subject=f"Re: {msg.subject}",
        text="Got it!",
        in_reply_to=msg.provider_message_id,
    )

# Handle bounce/complaint events
@email.on_event
def handle_event(event: email.EmailEventPayload):
    if event.event_type == "bounce":
        print("Bounced:", event.email_address)

Cross-language guaranteed API

Attachment download is also guaranteed; in Python it is an instance method on InboundAttachment:

for att in msg.attachments:
    data = att.download()  # -> bytes

Python-specific helpers

When KEELSON_EMAIL_WEBHOOK_SECRET is set, serve() and the decorators automatically verify inbound webhooks.

Data classes: Address, Attachment, InboundMessage, InboundAttachment, EmailEventPayload, SpamAssessment, AuthenticationResult.

Exception: EmailError.

Environment variables

Testing

Run all SDK tests:

uv sync --group dev
uv run pytest

Build the source distribution and wheel from the repository root:

uv build

Run tests for individual modules:

uv run pytest keelson_media/tests/
uv run pytest keelson_identity/tests/
uv run pytest keelson_email/tests/

Lint:

uv run ruff check .