Python server SDK

keenauth-python is the server SDK for Python 3.10 or later. It calls the management API with typed methods and adds the behaviour every KeenAuth server SDK shares: authentication with an API key or a private_key_jwt machine-to-machine token, retries, cursor pagination, typed errors, webhook signature checks and deprecation warnings. The package is not published to PyPI yet.

For sign-in in a web app, see Django, Flask and FastAPI. To protect an API that receives bearer tokens, see Protecting your API.

Install

pip install keenauth-python

The private_key_jwt signer needs the cryptography package. Install it with the m2m extra:

pip install "keenauth-python[m2m]"

Call the management API with an API key

import os
from keenauth import KeenAuth

client = KeenAuth("https://api.eu.keenauth.com/v1", os.environ["KEENAUTH_API_KEY"])

page = client.users.list_users(limit=20)
for user in page.data:
    print(user.id)

Every generated API class is available as an attribute: client.users, client.organizations, client.api_keys and so on. The base URL must use https; plain http is accepted only for localhost. The SDK never follows a redirect, so your credential is never sent to another host.

Use a private_key_jwt management token

A management application authenticates with its private key instead of an API key. The SDK signs a client assertion (RFC 7523), gets a management API token from your environment's token endpoint and renews it 60 seconds before it expires:

import os
from keenauth import KeenAuth
from keenauth.m2m import ManagementTokenSource

source = ManagementTokenSource(
    issuer="https://acme.eu.keenauth.app",          # your environment's default host
    client_id="app_01j9zzzzzzzzzzzzzzzzzzzz",       # a management application
    private_key=os.environ["KEENAUTH_PRIVATE_KEY"],  # unencrypted PEM
    resource="https://api.eu.keenauth.com",          # the management API of your region
    scopes=["users:read", "users:write"],            # optional; default: the grant's default scopes
)
client = KeenAuth("https://api.eu.keenauth.com/v1", token_source=source)

Retries

The SDK retries a request that failed with 429, 502, 503 or 504, or with a connection error or timeout. It retries only GET, PUT, DELETE, and POST requests that carry an Idempotency-Key. A PATCH that carries If-Match is retried on 429 only: the rate limiter refuses a request before it runs, and If-Match keeps a change from being applied twice. Any other PATCH is never retried. It makes at most 3 attempts in total.

Pagination

List methods return one page: data and list_metadata with the after and before cursors. paginate walks through every page and fetches the next page only when you reach it:

from keenauth import paginate

for user in paginate(client.users.list_users, limit=100, status="active"):
    if user.email == "ada@example.com":
        break  # no further page is requested

Errors

An error response in application/problem+json (RFC 9457) raises keenauth.ProblemError. Use its code, status, detail, request_id and errors fields. A code that this SDK version does not know is still returned as a string, so new server error codes do not break your code.

from keenauth import ProblemError

try:
    client.users.get_user("usr_01j9zzzzzzzzzzzzzzzzzzzz")
except ProblemError as e:
    if e.code == "not_found":
        ...
    print(e.request_id)

Verify webhooks

KeenAuth signs webhooks with the Standard Webhooks scheme. WebhookVerifier checks the webhook-id, webhook-timestamp and webhook-signature headers against your endpoint's idp_whsec_ secret:

import os
from flask import Flask, request
from keenauth.webhooks import WebhookVerificationError, WebhookVerifier

app = Flask(__name__)
verifier = WebhookVerifier(os.environ["KEENAUTH_WEBHOOK_SECRET"])

@app.post("/webhooks/keenauth")
def keenauth_webhook():
    try:
        verifier.verify(request.get_data(), request.headers)
    except WebhookVerificationError as e:
        return {"error": e.reason}, 400
    event = request.get_json()
    ...
    return "", 204

Typed events and claims

keenauth.events has a frozen dataclass for every customer event type, generated from the public event schemas. parse_event checks a verified body and returns the class of its type. keenauth.webhooks.verify_event verifies and parses in one call. Narrow with match or isinstance:

from keenauth.events import EmailSendingSuspendedEvent, UnknownEvent, UserCreatedEvent
from keenauth.webhooks import verify_event

event = verify_event(verifier, request.get_data(), request.headers)
match event:
    case UserCreatedEvent():
        print(event.data.object.id)
    case EmailSendingSuspendedEvent():
        print(event.data.object.reason)
    case UnknownEvent():
        pass  # a type this SDK version does not know yet; event.type and event.raw are still available

keenauth.claims types the claims of a verified token: parse_id_token_claims (OpenID Connect Core) and parse_access_token_claims (RFC 9068). They check claim types but verify nothing, so call them on the claims that the verifier returns:

from keenauth.claims import parse_access_token_claims

at = verifier.verify_access_token(token)
claims = parse_access_token_claims(at.claims)
claims.client_id     # str
claims.org_roles     # tuple[str, ...] | None
claims.custom        # your own claims

aud is always a tuple. A claim that does not fit raises ClaimsParseError with its name.

Deprecation warnings

When a response has a Deprecation (RFC 9745) or Sunset (RFC 8594) header, the SDK writes one warning per process to the keenauth logger with the standard logging module. The warning names the method, the resource and the header values; it never contains IDs, query strings or credentials. The request itself is not affected.

import logging

logging.getLogger("keenauth").setLevel(logging.WARNING)

Telemetry header

Every request carries Idp-Client: keenauth-python/<version> (python/<version>). It contains no personal data and cannot be turned off; KeenAuth uses it for support and deprecation notices.