Python SDK integration guide.

§ 01 · Install

The paphwey package is published on PyPI. It targets Python 3.10+ and ships type stubs — mypy --strict passes out of the box.

pip install paphwey

# Or with the optional telemetry extra (OpenTelemetry spans on every call):
pip install "paphwey[otel]"

The package bundles:

• paphwey.client — PaphweyClient and AsyncPaphweyClient.
• paphwey.models — Pydantic models for every request and response shape.
• paphwey.errors — exception hierarchy rooted at PaphweyError.
• paphwey.verify — offline JWS verification against the gateway JWKS.

§ 02 · Configure the client

Build the client once per process. It is thread-safe and holds a pooled httpx.Client under the hood. Supply credentials and the base URL from environment variables — never inline.

import os
from paphwey import PaphweyClient

paphwey = PaphweyClient(
    base_url=os.environ["PAPHWEY_BASE_URL"],
    api_key=os.environ["PAPHWEY_API_KEY"],
    timeout=10.0,               # seconds, total
    max_retries=3,              # 429 / 5xx only
    user_agent="merchant-api/2.1 (+https://merchant.example)",
)

§ 03 · Step 1 — Create a delegation

create_delegation() registers a scoped authority for the agent. It does not prompt the user. Persist the returned delegation_id — you will need it at every later step.

from datetime import datetime, timezone, timedelta
from paphwey.models import CreateDelegationRequest

delegation = paphwey.create_delegation(CreateDelegationRequest(
    principal_email="alice@example.com",
    provider="openai",
    agent_id="shopping-agent",
    allowed_scopes=["purchase"],
    allowed_challenge_types=["HIGH_VALUE_PURCHASE_REQUIRED"],
    max_amount_minor=2500,
    currency="GBP",
    valid_until=datetime.now(timezone.utc) + timedelta(days=14),
    metadata={"channel": "web-chat"},
))

assert delegation.status == "active"
print(delegation.delegation_id)

import uuid
from paphwey.models import CreateDelegationRequest

key = str(uuid.uuid4())
delegation = paphwey.create_delegation(
    CreateDelegationRequest(...),
    idempotency_key=key,  # safe to retry — the gateway dedupes for 24h.
)

§ 04 · Step 2 — Present the delegation

present_delegation() opens a challenge tied to a specific action and returns an approval_url. Surface that URL to the user's phone — by SMS, email, push, or a QR code.

from paphwey.models import PresentDelegationRequest, ActionContext

challenge = paphwey.present_delegation(
    delegation_id=delegation.delegation_id,
    request=PresentDelegationRequest(
        challenge_type="HIGH_VALUE_PURCHASE_REQUIRED",
        audience="merchant.example",
        payload={
            "action_context": ActionContext(
                scope="purchase",
                amount_minor=1500,
                currency="GBP",
                merchant="acme.example",
            ),
            "minimum_assurance": 10,
        },
        callback_url="https://merchant.example/hooks/paphwey",
        ttl_seconds=900,
    ),
)

# Hand this to the user's device. Do NOT give it to the agent.
sms.send(user.phone, f"Approve the purchase: {challenge.approval_url}")

§ 05 · Step 3 — Verify the outcome

verify_outcome() validates the attestation JWS, checks the audience and delegation, and returns a VerifiedOutcome with a pinned confirmation JWK.

from paphwey.errors import PaphweyVerificationError
from paphwey.models import VerifyOutcomeRequest

try:
    outcome = paphwey.verify_outcome(VerifyOutcomeRequest(
        attestation_token=token_from_agent,
        audience="merchant.example",
        expected_delegation_id=delegation.delegation_id,
        expected_action={"scope": "purchase",
                         "amount_minor": 1500,
                         "currency": "GBP"},
    ))
except PaphweyVerificationError as exc:
    logger.warning("attestation rejected", extra={"code": exc.code})
    raise HTTPException(status_code=403, detail="attestation invalid")

assert outcome.valid
# `claims` is the verified JWT body; `delegation` is the resolved
# delegation snapshot; `actor_chain` is the principal/agent envelope.
session["receipt_id"] = outcome.receipt_id
session["agent_did"] = outcome.delegation["agent_did"]
session["principal_id"] = outcome.actor_chain["principal_id"]

Offline verification

For hot paths, skip the round-trip and verify against the cached JWKS locally:

from paphwey.verify import OfflineVerifier

verifier = OfflineVerifier.from_base_url("https://www.paphwey.com")
# The verifier caches JWKS for 5 minutes and auto-refreshes on unknown kid.

outcome = verifier.verify(
    attestation_token=token_from_agent,
    audience="merchant.example",
    expected_delegation_id=delegation.delegation_id,
)

§ 06 · Async client

AsyncPaphweyClient mirrors the sync API one-for-one. Use it in FastAPI, aiohttp, or any asyncio runtime. Create one instance per event loop.

import asyncio
from paphwey import AsyncPaphweyClient

async def main() -> None:
    async with AsyncPaphweyClient(
        base_url=os.environ["PAPHWEY_BASE_URL"],
        api_key=os.environ["PAPHWEY_API_KEY"],
    ) as paphwey:
        delegation = await paphwey.create_delegation(...)
        challenge = await paphwey.present_delegation(...)
        outcome = await paphwey.verify_outcome(...)

asyncio.run(main())

§ 07 · FastAPI pattern

Wire the client as a singleton dependency and let FastAPI inject it per request. Store the pinned cnf_jwk on your session once verified.

from fastapi import Depends, FastAPI, HTTPException, Request
from paphwey import AsyncPaphweyClient

app = FastAPI()

async def get_paphwey(request: Request) -> AsyncPaphweyClient:
    return request.app.state.paphwey

@app.on_event("startup")
async def _startup() -> None:
    app.state.paphwey = AsyncPaphweyClient(
        base_url=os.environ["PAPHWEY_BASE_URL"],
        api_key=os.environ["PAPHWEY_API_KEY"],
    )

@app.on_event("shutdown")
async def _shutdown() -> None:
    await app.state.paphwey.aclose()

@app.post("/actions/purchase")
async def purchase(
    body: PurchaseBody,
    paphwey: AsyncPaphweyClient = Depends(get_paphwey),
):
    outcome = await paphwey.verify_outcome(
        attestation_token=body.attestation_token,
        audience="merchant.example",
        expected_delegation_id=body.delegation_id,
    )
    if not outcome.valid:
        raise HTTPException(403, "attestation invalid")
    # ... fulfil the purchase
    return {"status": "accepted"}

§ 08 · Django pattern

In Django, the sync client is usually the right choice — most views are WSGI. Build it once in AppConfig.ready() and reuse.

# app/apps.py
from django.apps import AppConfig
from paphwey import PaphweyClient

class MerchantConfig(AppConfig):
    name = "merchant"
    paphwey: PaphweyClient

    def ready(self) -> None:
        from django.conf import settings
        MerchantConfig.paphwey = PaphweyClient(
            base_url=settings.PAPHWEY_BASE_URL,
            api_key=settings.PAPHWEY_API_KEY,
        )

# app/views.py
from django.apps import apps
from django.http import JsonResponse

def accept_outcome(request):
    paphwey = apps.get_app_config("merchant").paphwey
    outcome = paphwey.verify_outcome(
        attestation_token=request.POST["token"],
        audience="merchant.example",
        expected_delegation_id=request.POST["delegation_id"],
    )
    if not outcome.valid:
        return JsonResponse({"error": "invalid"}, status=403)
    return JsonResponse({"receipt_id": outcome.receipt_id})

§ 09 · Celery / background tasks

Verification happens on the hot path — but post-verification fulfilment (PDF generation, ledger writes, shipping notices) belongs in a task queue. Pass the receipt_id along so your workers can reconcile against the audit log.

from celery import shared_task

@shared_task(bind=True, max_retries=5, autoretry_for=(TransientError,))
def fulfil_purchase(self, *, delegation_id: str, receipt_id: str, order: dict):
    # The outcome was already verified on the request path.
    # Here we only trust the receipt_id — never the raw token.
    record = ledger.open(receipt_id=receipt_id,
                         delegation_id=delegation_id)
    shipping.schedule(order, record_id=record.id)
    return record.id

§ 10 · Error handling

All SDK exceptions inherit from PaphweyError. The hierarchy is stable across minor releases.

from paphwey.errors import (
    PaphweyPolicyError, PaphweyVerificationError, PaphweyRateLimitError,
)

try:
    delegation = paphwey.create_delegation(...)
except PaphweyPolicyError as exc:
    # Surface a user-safe message; log exc.code + exc.trace_id internally.
    return Response({"error": "ceiling_exceeded"}, status=422)
except PaphweyRateLimitError as exc:
    response = Response(status=503)
    response.headers["Retry-After"] = str(exc.retry_after_seconds)
    return response

§ 11 · Testing & mocks

The SDK ships a paphwey.testing module for test fixtures. Prefer recording real fixtures over hand-rolling dict stubs — the Pydantic models are strict and evolve between releases.

from paphwey.testing import FakePaphweyClient, make_outcome

def test_accepts_valid_outcome():
    fake = FakePaphweyClient()
    fake.enqueue_outcome(make_outcome(
        delegation_id="dlg_test",
        audience="merchant.example",
        claims={"sub": "principal-uuid", "challenge_id": "chl_test"},
        actor_chain={"principal_id": "principal-uuid",
                     "agent": {"provider": "openai", "agent_id": "shopping-agent"}},
    ))

    outcome = fake.verify_outcome(
        attestation_token="anything",
        audience="merchant.example",
        expected_delegation_id="dlg_test",
    )
    assert outcome.valid

§ 12 · Pre-production checklist

• Client is built once per process, reused across requests.
• API key loaded from a secrets manager; never in source, logs, or stack traces.
• Async client is used in asyncio runtimes; sync client everywhere else.
• Offline verifier's JWKS cache refresh is wired to the process lifecycle.
• All PaphweyError subclasses have handlers; no bare except Exception.
• Integration tests run against a real Paphwey tenant (your test tenant), not FakePaphweyClient.
• Pinned confirmation JWK is stored on the session after first verify.
• OpenTelemetry spans enabled in production (paphwey[otel]).