# 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](/docs/sdk/django), [Flask](/docs/sdk/flask) and [FastAPI](/docs/sdk/fastapi). To protect an API that receives bearer tokens, see [Protecting your API](/docs/sdk/protect-api).

## Install

```sh
pip install keenauth-python
```

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

```sh
pip install "keenauth-python[m2m]"
```

## Call the management API with an API key

```python
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:

```python
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)
```

- The assertion has `iss` and `sub` set to `client_id`, `aud` set to the issuer, a random `jti` and a lifetime of 60 seconds. Its `kid` is the RFC 7638 thumbprint of the public key.
- Supported keys: EC P-256 (`ES256`), EC P-384 (`ES384`), Ed25519 (`EdDSA`) and RSA of at least 2048 bits (`PS256` by default, or `algorithm="RS256"`).
- Register the public key on the management application. `keenauth.m2m.load_private_key(pem).public_jwk()` returns it as a JWK.
- The token is cached. When several threads need a new token at the same time, only one request goes to the token endpoint.
- A refused request raises `keenauth.m2m.TokenError` with `status`, `error` and `error_description`. The message never contains the assertion or the token.

## 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.

- If the response has a `Retry-After` header (seconds or an HTTP date), the SDK waits that long. If `Retry-After` is longer than 60 seconds, the SDK does not wait and returns the response to you.
- Without `Retry-After`, it waits 0.5 seconds and then 1 second, each with ±25% random jitter.
- The SDK adds a random UUIDv4 `Idempotency-Key` to every `POST` operation that accepts one, unless you pass your own `idempotency_key=`. A retry sends the same key, so the server runs the operation only once.

## 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:

```python
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
```

- Other arguments, such as filters and `order`, are sent with every page request.
- Pass `before=` to walk backwards from a cursor. `after` and `before` cannot be used together.
- `pages(...)` takes the same arguments and returns whole pages instead of items.

## 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.

```python
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:

```python
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
```

- Pass the raw request body, before any JSON parsing; a changed byte fails the check.
- The timestamp must be within 5 minutes of your server's clock, in either direction.
- While you rotate the secret, pass both: `WebhookVerifier([new_secret, old_secret])`. A request is accepted when any signature in the header matches any secret.
- Errors are typed: `WebhookHeadersError`, `WebhookTimestampError`, `WebhookSignatureError` and `WebhookSecretError` (a malformed secret, raised when the verifier is created). Each has a `reason` field.
- The comparison runs in constant time, and the error message never contains the secret or the signature.

## 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`:

```python
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
```

- `parse_event` accepts bytes, a string or an already-decoded mapping. It checks the envelope and the members the type's schema declares. A missing or wrong-typed member raises `EventParseError` with its path, such as `data.object.reason`.
- An unknown type, or a known type with a `schema_version` this SDK does not model, is an `UnknownEvent`. It is not an error.
- Members that the schema does not declare are kept in `raw` (on the event and on `data.object`).
- Arrays are tuples. Values of enumerated fields are not checked against the known values.
- The models pass `mypy --strict`.

`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:

```python
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.

```python
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.