Python

The forculus package verifies the identity token the edge sends your app and gives you the caller. It works with any framework, and ships integrations for Flask, FastAPI, Django, Streamlit and Gradio.

pip install forculus

It needs Python 3.10 or newer, and depends only on PyJWT with its crypto extra. Each framework integration has an extra that installs the framework it needs:

pip install "forculus[flask]"      # or [fastapi], [django], [streamlit], [gradio]

Any framework

require_user(headers) takes the request’s headers and returns the caller, or raises Unauthorized.

from forculus import Unauthorized, require_user

@app.get("/")
def index():
    try:
        user = require_user(request.headers)
    except Unauthorized:
        return "unauthorized", 401
    return f"hello {user.email or user.sub}"

The headers can be any mapping with .get: Flask’s and Starlette’s request.headers, Django’s request.META, or a plain dict. The header name is matched in any case.

It returns a User:

@dataclass(frozen=True)
class User:
    kind: str              # "user", or "service" for a bypass token or an agent
    sub: str               # "usr_…", "service:<token id>", or "agent:<agent id>"
    email: str | None      # None for a bypass token or an agent
    role: str              # "viewer", "editor" or "owner"
    workspace_id: str
    agent_name: str | None # the agent's name when an agent called, None for everyone else

user.is_agent says whether one of the workspace’s agents made the request. It and agent_name are new in version 0.2.0: with an earlier version, pip install --upgrade forculus.

by = f"{user.agent_name} (agent)" if user.is_agent else (user.email or user.sub)

See the SDK overview for what each field means and what the SDK checks.

Roles

from forculus import Forbidden, has_role, require_role

require_role(user, "editor")   # raises Forbidden unless the caller is an editor or an owner

if has_role(user, "owner"):
    ...

Forbidden has status = 403 and needed, the role that was asked for.

Flask

from forculus.flask import current_user, get_user, login_required

@app.get("/")
@login_required
def index():
    return f"hello {current_user.email}"

@app.post("/orders")
@login_required(role="editor")
def place_order():
    ...

@login_required answers 401 unauthorized without a valid caller, and 403 forbidden: this needs editor when the caller has less than role. Neither reaches your view.

current_user is the caller @login_required verified. Using it in a view without the decorator raises RuntimeError. get_user() returns the caller or None, for a page that renders either way.

FastAPI

from typing import Annotated

from fastapi import Depends
from forculus import User
from forculus.fastapi import CurrentUser, optional_user, requires

@app.get("/")
async def index(user: CurrentUser):
    return {"hello": user.email}

@app.post("/orders")
async def place_order(user: Annotated[User, Depends(requires("editor"))]):
    ...

@app.get("/welcome")
async def welcome(user: Annotated[User | None, Depends(optional_user)]):
    ...

The dependencies raise HTTPException, so FastAPI answers 401 ("detail": "unauthorized") or 403 ("detail": "forbidden: this needs editor") and your handler never runs. optional_user gives None instead of refusing.

They verify without blocking the event loop. The request is verified once, however many dependencies ask for the caller.

Django

Add the middleware:

MIDDLEWARE = [
    "forculus.django.ForculusMiddleware",
    # ...
]

Then guard the views that need a caller:

from django.http import HttpResponse
from forculus.django import login_required, role_required

def index(request):
    user = request.forculus_user            # None when there is no valid caller
    return HttpResponse(f"hello {user.email if user else 'stranger'}")

@login_required
def account(request):
    return HttpResponse(f"hello {request.forculus_user.email}")

@role_required("editor")
def place_order(request):
    ...

The middleware sets request.forculus_user to the caller or None, and refuses nothing by itself, so one app can have public and private pages. login_required answers 401 unauthorized. role_required answers 401 without a caller and 403 forbidden: this needs editor with too little role.

Caution

Import login_required from forculus.django, not django.contrib.auth.decorators. Django’s own checks Django’s sessions and redirects to LOGIN_URL.

Streamlit

import streamlit as st
from forculus.streamlit import get_user

user = get_user()
if user is None:
    st.error("sign in to use this")
    st.stop()
st.write(f"hello {user.email}")

require_user() raises Unauthorized instead of returning None.

The caller is verified once per session and remembered. Streamlit only sees the headers of the request that opened the session, and the token in them lasts 60 seconds, so checking again on a rerun would refuse someone still using the app. On Forculus, the edge closes the connection within a minute of someone’s access being removed, and the session ends with it.

Gradio

import os

import gradio as gr
from forculus.gradio import require_user

def answer(question, request: gr.Request):
    user = require_user(request)
    return f"{user.email} asked: {question}"

gr.Interface(answer, "textbox", "textbox").launch(
    server_name="0.0.0.0", server_port=int(os.environ.get("PORT", "7860"))
)

On Forculus the app has to listen on 0.0.0.0 at the port in PORT. Gradio’s defaults, 127.0.0.1 and 7860, can’t be reached there.

Gradio passes the request to any handler that asks for one by type. It passes None when the handler is reached some other way, such as the API or a cached example, and require_user refuses those calls. get_user(request) returns None instead of raising.

async

require_user fetches our signing keys when it has none cached, about once every five minutes, and that fetch can block for up to three seconds. In an async app, use forculus.aio. It does the fetch on a thread, one at a time, and everything else stays on the loop.

from fastapi import HTTPException, Request
from forculus import Unauthorized
from forculus.aio import require_user

@app.get("/")
async def index(request: Request):
    try:
        user = await require_user(request.headers)
    except Unauthorized as err:
        raise HTTPException(401, str(err))
    return {"hello": user.email}

An uncaught Unauthorized becomes a 500. In FastAPI you rarely need this, because the CurrentUser dependency uses forculus.aio and answers 401 for you.

Webhooks

A delivery on a verified public path arrives with a token of kind webhook. require_webhook checks it and says which provider signed the delivery. require_user refuses a webhook’s token, and require_webhook refuses a person’s.

from forculus import Unauthorized, require_webhook

@app.post("/hooks/stripe")
def stripe_hook():
    try:
        hook = require_webhook(request.headers, provider="stripe")
    except Unauthorized:
        return "unauthorized", 401
    event = request.get_json()
    # hook.delivery is Stripe's event id. Skip one you've handled.
    return "", 204

provider is required: stripe, github, slack, shopify, standard_webhooks or hmac_sha256, or a list of them, like provider=["github", "stripe"]. A delivery from any other provider is refused, and so is every delivery when the list is empty. It returns a Webhook:

@dataclass(frozen=True)
class Webhook:
    provider: str             # "stripe", "github", "slack", "shopify", "standard_webhooks" or "hmac_sha256"
    pattern: str              # the public path it came in on, like "/hooks/stripe"
    delivery: str | None      # the provider's signed id, None where it signs none
    signed_at: int | None     # unix seconds, None where the provider signs no time
    workspace_id: str

The body arrives byte for byte as the provider sent it, so parse it as usual. In an async app, forculus.aio.require_webhook is the same check as a coroutine. verify_webhook(token, provider="stripe") checks a webhook’s token you already have.

Each framework integration checks per route, so webhook routes and routes for people can be in the same app:

# Flask
from forculus.flask import get_webhook, webhook_required

@app.post("/hooks/stripe")
@webhook_required(provider="stripe")
def stripe_hook():
    hook = get_webhook()
    ...

# FastAPI
from typing import Annotated

from fastapi import Depends, Request
from forculus import Webhook
from forculus.fastapi import webhook

@app.post("/hooks/stripe")
async def stripe_hook(hook: Annotated[Webhook, Depends(webhook("stripe"))], request: Request):
    event = await request.json()

# Django
from forculus.django import webhook_required

@webhook_required(provider="stripe")
def stripe_hook(request):
    hook = request.forculus_webhook
    event = json.loads(request.body)
    ...

Each one answers 401 before the view runs unless the request is a verified delivery from a provider it names. A signed-in person gets 401 too. Flask’s get_webhook() raises if @webhook_required didn’t run on that view.

A provider has no CSRF token to send. webhook_required exempts its Django view from the CSRF check. With Flask-WTF’s CSRFProtect, add @csrf.exempt to the route yourself.

Configuration

The SDK reads FORCULUS_APP_ID, FORCULUS_IDENTITY_ISSUER and FORCULUS_JWKS_URL from os.environ on each call. We set them. To pass them yourself, use a Config:

from forculus import Config, require_user

config = Config(app_id="…", issuer="https://edge.forculuslabs.tech", jwks_url="https://edge.forculuslabs.tech/.well-known/jwks.json")
user = require_user(request.headers, config)

Config also takes dev, whether FORCULUS_DEV is 1. Config.from_env() reads it. Only forculus dev sets it, and without it a Config with an http issuer or key set raises Unauthorized.

verify_identity(token) checks a token you already have, the same way.

Errors

A refusal raises Unauthorized (status = 401) or Forbidden (status = 403). The message says why, for your logs:

Message	Why
no identity token	The request has no X-Forculus-Identity header
invalid identity token	The token failed a check
unknown signing key	No key in the key set matches the token’s kid
couldn't fetch the identity keys	The key set couldn’t be fetched, and there’s no fresh copy
FORCULUS_APP_ID, FORCULUS_IDENTITY_ISSUER and FORCULUS_JWKS_URL must be set	The variables aren’t in the environment
FORCULUS_IDENTITY_ISSUER and FORCULUS_JWKS_URL must use https, except under forculus dev (FORCULUS_DEV=1, on this machine)	The issuer or key set is http, and FORCULUS_DEV isn’t 1. Only forculus dev uses http
no request to read an identity from	Gradio passed None for the request
this needs editor	A Forbidden from require_role(user, "editor")
not a webhook this route takes	From require_webhook: a provider the route doesn’t name signed the delivery
invalid webhook token	From require_webhook: the token’s webhook claims failed a check
no provider named	From require_webhook: the call named no provider, so it refuses everything

A person’s token given to require_webhook, or a webhook’s given to require_user, fails as invalid identity token, because its audience is wrong for that call.

Run it locally

forculus dev runs your app on your machine behind a copy of our front door, with real tokens. It can start Flask, FastAPI, Django, Streamlit and Gradio apps without being told how. See Local development.

forculus dev

Next steps

• SDK overview: every claim in the token, and each check.
• Local development.
• Webhooks and public paths.
• Deploy a Streamlit, Gradio or FastAPI app that only your team can open, start to finish.