Flamecloak for Python

Standard library only (urllib), Python 3.9+, typed. Installed as flamecloak-sdk, imported as flamecloak_sdk.

pip install flamecloak-sdk

The developer guide, every status, and what to check before shipping: the Flamecloak developer guide.

import os
from flamecloak_sdk import Flamecloak

flamecloak = Flamecloak(os.environ["FLAMECLOAK_KEY"])

decision = flamecloak.authorize(
    "refund",
    resource=f"payment:{payment_id}",
    agent="support-bot",
    parameters={"amount": 1200},
)

if decision.allowed:
    refund = stripe.Refund.create(payment_intent=payment_id, amount=1200, idempotency_key=decision.decision_id)
    decision.done("succeeded", reference=refund.id)  # optional

Code that cannot wait (a request handler, a queue worker) asks without waiting and keeps the id:

asked = flamecloak.authorize("refund", parameters={"amount": 1200}, wait=False)
if asked.status == "pending":
    save_job(decision_id=asked.decision_id)
# later
decision = flamecloak.collect(decision_id)  # or flamecloak.wait(decision_id, max_wait_seconds=60)

Resuming the job when a person decides

Ask without waiting, with your job's own id as the idempotency key:

asked = flamecloak.authorize("refund", parameters={"amount": 1200}, idempotency_key=job.id, wait=False)

and resume it from the notice (see Notices). verify_webhook needs the RAW body (request.get_data() in Flask, await request.body() in Starlette):

from flask import request
from flamecloak_sdk import FlamecloakWebhookError, verify_webhook

@app.post("/flamecloak/notices")
def notice():
    try:
        event = verify_webhook(
            request.get_data(),
            request.headers.get("flamecloak-signature"),
            os.environ["FLAMECLOAK_WEBHOOK_SECRET"],
        )
    except FlamecloakWebhookError:  # .code says why
        return "", 400
    if event.decision_id is not None and event.idempotency_key is not None:
        job = jobs.find(event.idempotency_key)
        if job is not None:
            # THE NOTICE IS NOT THE PERMISSION; collect is, once.
            decision = flamecloak.collect(event.decision_id)
            if decision.allowed:
                refund(job, idempotency_key=event.decision_id)
            elif decision.status != "used":  # denied, expired, stale
                jobs.close(job, decision.status)
            # "used": this notice came again after the job ran. Nothing to do.
    return "", 200

A retried notice is harmless: its second collect answers used.

Flamecloak(key, base_url=None, timeout=10.0). base_url only for a key with no address or a self-hosted gateway; timeout is one request's, in seconds (a long-poll gets it on top of its wait). One client per process is enough, and it is safe to share between threads.

authorize() also takes idempotency_key, max_wait_seconds, and public_info (who asked and what changes, once your organisation has turned on Say who asked; see the guide). Errors are FlamecloakError, with .status set to the gateway's HTTP status, or None for a key that is not one.

When Flamecloak sits in front of an endpoint, call() makes the call and waits: a 423 pending (403 from a gateway older than October 2026) is sent again byte for byte with the decision, and a 202 (held) is waited on until the gateway has made the call itself.

result = flamecloak.call("https://api.example.com/api/orders/1")
# result.outcome: answered, executed, attempted, refused, pending or unavailable

See calls through your gateway.

What every status means, what raises, and what to check before shipping: the developer guide.

Apache License 2.0.