zaniiid

Zanii ID — Python reference

zanii-id on PyPI, Python 3.10+. Apache-2.0. Extras: [fastapi], [subject], [dev].

Configuration

ZaniiConfig is a pydantic-settings model over ZANII_* env vars, validated eagerly. Explicit None kwargs are dropped rather than overriding the environment, so ZaniiClient() falls through to env vars cleanly.

from zanii_id import ZaniiClient
zanii = ZaniiClient()                     # or ZaniiClient(issuer=..., client_id=..., ...)

The issuer must be https:// (localhost and testserver are allowed for development). The default scopes are openid profile email offline_access. offline_access is what asks for a refresh token — it does not guarantee one. Consent is per permission, so a user of a third-party client can untick it and still sign in: read tokens.scope and treat tokens.refresh_token as possibly None. It is typed str | None for exactly this reason; a session without it is bound to the IdP session and cannot be refreshed.

Public clients (no client_secret): the SDK sends client_id in the token body, as RFC 6749 requires; confidential clients use HTTP Basic. revoke() authenticates the same way. (Both were broken before 0.2.0 - if an old integration gets invalid_client, upgrade.)

FastAPI

from zanii_id import ZaniiClient
from zanii_id.integrations.fastapi import build_auth_router, install_zanii, require_zanii_auth

zanii = ZaniiClient()
install_zanii(app, zanii, session_secret=SECRET)      # app.state + a 401 handler
app.include_router(build_auth_router(
    zanii,
    session_secret=SECRET,
    on_user=provision_local_user,                     # awaited with IdTokenClaims
    default_post_login_redirect="/dashboard",
    post_logout_redirect_uri="https://yourapp.com/",
))

@app.get("/dashboard")
async def dashboard(user = Depends(require_zanii_auth)):
    return {"zanii_user_id": user.zanii_user_id, "email": user.email, "did": user.did}

install_zanii must run before the router is mounted — it puts the client and the session codec on app.state and registers the ZaniiAuthRequired → 401 handler.

require_zanii_auth calls /userinfo; on failure it refreshes once, rotates the session cookie, and retries. If the refresh fails the family is gone and the user must sign in again. It also backfills did from the session when userinfo lags a fresh issuance.

Manual flow

req = zanii.get_authorization_url()       # AuthorizationRequest(url, state, nonce, verifier)
# persist req.state / req.nonce / req.verifier in YOUR session, then redirect to req.url

# step-up: require a second-factor session, force a fresh login, or both
from zanii_id import ACR_MFA
req = zanii.get_authorization_url(acr_values=ACR_MFA, prompt="login", max_age=0)

tokens = await zanii.exchange_code(code, req, received_state)   # StateMismatchError if wrong
claims = zanii.verify_id_token(tokens.id_token, nonce=req.nonce)
user   = await zanii.get_user(tokens.access_token)

exchange_code raises StateMismatchError before any network call. verify_id_token is synchronous: it checks the signature via a cached PyJWKClient, plus iss, aud, exp and nonce, with alg pinned to RS256 — never read from the token header.

For resource servers validating tokens they were handed:

claims = zanii.verify_access_token(access_token)   # requires exp, sub, scope, sid

IdTokenClaims has typed acr and amr. Read claims.acr == "urn:zanii:mfa" for "this session has a second factor"; the FastAPI router accepts /auth/login?acr=mfa and ?prompt=login to request it, and /auth/logout now passes id_token_hint so the IdP signs the user out without a confirmation page.

A machine-to-machine or agent token is not an SDK flow: POST /token yourself with grant_type=client_credentials (Basic auth) or the token-exchange grant (see api.md), then zanii.verify_access_token() the result - act is present on exchanged tokens and there is no sid on client-credentials tokens.

Sync frameworks (Flask, Django)

from zanii_id import SyncZaniiClient
zanii = SyncZaniiClient()                 # same API, blocking; owns a private event loop
tokens = zanii.exchange_code(code, req, received_state)

Do not use SyncZaniiClient inside an already-running event loop — its private loop will raise. In async code use ZaniiClient directly.

Models

  • Tokens — access_token, id_token, refresh_token, token_type, expires_in, scope. refresh_token is None whenever offline_access was not granted, which the user can decide; scope is what was actually granted, not what was requested.
  • UserInfo — sub, did, email, email_verified, name, picture, plus a given_name, family_name, org, and a zanii_user_id property aliasing sub. Since 0.5.1 it also allows extra fields, so a claim the issuer adds later arrives untyped instead of being dropped — which is exactly how given_name went missing before 0.5.1.
  • IdTokenClaims — the OIDC set plus did; extra="allow", so new claims do not break

JIT provisioning

sub is the immutable zanii_user_id. Upsert on it, never on email:

INSERT INTO users (zanii_user_id, email) VALUES ($1, $2)
ON CONFLICT (zanii_user_id) DO NOTHING;

Then re-select. Two products can provision the same user at the same moment; the unique constraint is what makes that safe.

Agent activity

from zanii_id.activity import subject_tag, fetch_activity   # pip install 'zanii-id[subject]'

tag = subject_tag(user.did, CLIENT_ID)          # pass to your agent's record(..., subject_tag=tag)
entries = await fetch_activity(tag)             # each verified offline

ActivityEntry carries verified and, when false, a flag_reason. Failures are returned, never dropped, and a truncated slice appends its own flagged entry. Without the zanii peer installed both functions raise ConfigError naming the install command.

Lifecycle webhooks

verify_lifecycle_event(raw_body, signature, webhook_secret) (0.1.3+) checks the HMAC in constant time and returns a LifecycleEvent(event, sub, occurred_at, data). Pass the raw request bytes, never re-serialised JSON. sign_lifecycle_body exists for your tests.

from zanii_id import LifecycleSignatureError, verify_lifecycle_event

@app.post("/webhooks/zanii-id")
async def lifecycle(request: Request):
    try:
        ev = verify_lifecycle_event(await request.body(), request.headers.get("X-Zanii-Signature"), WEBHOOK_SECRET)
    except LifecycleSignatureError:
        return Response(status_code=401)
    if ev.event == "user.deleted":
        await users.erase(ev.sub)
    return Response(status_code=204)

sub is your subject for the user (pairwise for org clients). See api.md for the event table and the retry policy.

Machines, agents and logout (0.3.0)

# A service acting as itself, or an agent that will act for users.
cc = await client.client_credentials(scope="agent", audience="cli_other")

# RFC 8693: act for a user towards another Zanii service; name the agent in `act`.
ex = await client.exchange_token(user_access_token, actor_token=cc.access_token, audience="cli_other")
# invalid_grant here means the audience's organization refuses this actor (receipted either way).

# RFC 8628 for CLIs, TVs and headless agents: show the code, then poll at the server's pace.
d = await client.device_authorize(scope="openid")
print(d.user_code, d.verification_uri_complete)
tokens = await client.poll_device_token(d)  # honours interval, slow_down, expires_in, Retry-After

# DPoP (RFC 9449): tokens bound to a key; a stolen token is useless.
from zanii_id import generate_dpop_key
client = ZaniiClient(dpop_key=generate_dpop_key())  # every token, userinfo and revoke call carries a proof

# private_key_jwt: no shared secret; register the public JWK Set on the client.
client = ZaniiClient(client_assertion_key=pem_private_key, client_assertion_kid="k1")

# PAR (RFC 9126): nothing but client_id + request_uri reaches the browser.
req = await client.get_authorization_url_par(acr_values=ACR_MFA)

# Logout: RP-initiated URL, and the two signals the IdP sends you.
client.logout_url(id_token, "https://app.test/bye", state="s1")
token = client.verify_logout_token(logout_token)      # back-channel POST -> end sessions for token.sid
sid = client.parse_frontchannel_logout(request.query)  # front-channel iframe ?iss=&sid=

The FastAPI router mounts POST /auth/backchannel-logout and GET /auth/frontchannel-logout when you pass on_backchannel_logout=async def (sid): ...; the hook ends every local session tied to that IdP session id. Register the two URIs on the client.

Server-to-server: post_receipt_event(issuer, platform_client_id=..., webhook_secret=..., event=...) relays a ledger receipt event to the user's activity page, and OrgClient(issuer, api_key) wraps /orgs/* (settings and the agent deny list, clients, lifecycle URLs, domain and workload binding, screen_agent).

Claims: IdTokenClaims and UserInfo type act, cnf and the age scope (age_over + age_method + age_provider, or age_declared; .age_verified).

Cheap for the issuer: one connection pool per event loop (10 connections), a timeout on every call, JWKS cached, reads retried once only when the server answered 429/502/503/504 (honouring Retry-After), token grants never retried, device polling never faster than told.

Prove it (0.4.0)

info = await client.introspect(access_token)        # RFC 7662 + Zanii: act, cnf, cst, kya, consent
if info.active and info.consent:                   # the receipt the ledger holds for this consent
    print(info.consent.commitment, info.consent.scope)

pip install 'zanii-id[verify]'
zanii-id verify did:key:z6Mk...        # bundle verified offline, governance per constitution version,
zanii-id verify did:key:z6Mk... --json # bilingual evidence pack; exit 1 on any failure

Exceptions

ZaniiError is the base: ConfigError (also a ValueError), AuthorizationError (.error, .description, .status), StateMismatchError, TokenValidationError, HTTPError.

The SDK never retries anything. Token POSTs must not be retried — the grant is single-use and a retry burns it; wrap idempotent GETs (get_user) in your own retry if you need one.

The agent surfaces (0.6.0)

from zanii_id import TaskGrant, ZaniiClient

z = ZaniiClient()

# A token for one tool server, and a job the user approves rather than a blank cheque.
req = z.get_authorization_url(
    login_hint="alice@example.com",
    resource="https://tools.example.com/mcp",          # registered first, see OrgClient
    task=TaskGrant(purpose="Book one flight to Dubai", max_uses=3, grant_ttl=3600,
                   ceiling={"AED": 4000}),             # declared; your API enforces it
)

tokens = await z.exchange_code(code, req, state)
print(tokens.scope)                                     # read it: consent is per permission

# Stop and ask the person.
approval = await z.create_approval(tokens.access_token, action="Pay invoice 4417",
                                   detail={"amount": "500 AED"}, expires_in=600)
final = await z.await_approval(approval)
if not final.approved:                                  # "expired" lands here too, on purpose
    raise RuntimeError(f"the human said no ({final.status})")

# Hear about revocation instead of waiting out a token's ten minutes.
await z.configure_stream(delivery="poll")
batch = await z.poll_signals()
for jti, raw in batch.sets.items():
    event = z.verify_security_event(raw)                # verify before acting on it
    handle(event.event_type, event.sub_id)
await z.poll_signals(ack=list(batch.sets))              # acknowledge only what you handled

OrgClient covers the rest: register_resource, list_resources, delete_resource, invite_member, list_members, remove_member, set_agent_model, clear_agent_model, update(require_member_mfa=True) and register_dynamic_client (RFC 7591).

Still not in the SDK, deliberately: the /manage/v1 admin API. It is authenticated by the single global admin token, and an application SDK is the wrong place to invite that token.