authpi-admin

Official Python Admin SDK for the AuthPI Core API.

Requirements: Python 3.11+, async-only (httpx + asyncio)

Stability

authpi-admin follows semantic versioning. Public imports, constructor options, generated request and response models, resource accessors, and documented method behavior are stable across 1.x; incompatible changes will ship in a new major version.

Installation

pip install authpi-admin

Quick Start

from authpi_admin import AuthPIAdmin
from authpi_admin.types import WebhookEventType

async with AuthPIAdmin(api_key=("key_xxx", "your_key_secret"), account_id="acc_xxx") as admin:
    # List issuers
    page = await admin.issuers.list(limit=10)
    print(page.data)

    # Scope into an issuer and manage users
    users = await admin.issuer("iss_xxx").users.list()

    # Auto-paginate
    async for user in admin.issuer("iss_xxx").users.list_all():
        print(user)

    # Create a user
    user = await admin.issuer("iss_xxx").users.create({
        "email": "alice@example.com",
        "display_name": "Alice",
    })

    # Create a webhook with typed event subscriptions
    webhook = await admin.webhooks.create({
        "name": "Lifecycle events",
        "url": "https://example.com/webhooks/authpi",
        "auth": {"type": "signature"},
        "events": [
            WebhookEventType.ORGANIZATION_CREATED,
            WebhookEventType.USER_CREATED,
        ],
    })

Generated response models, request TypedDicts, and enums are exported from authpi_admin.types. Most calls can use dict literals directly; import request types only when you want to annotate a reusable payload:

from authpi_admin.types import CreateWebhookInput, WebhookEventType

payload: CreateWebhookInput = {
    "name": "Lifecycle events",
    "url": "https://example.com/webhooks/authpi",
    "auth": {"type": "signature"},
    "events": [WebhookEventType.USER_CREATED],
}
await admin.webhooks.create(payload)

Authentication

API Key (default)

API keys are issued as an id + secret pair — both parts are shown once when you create the key in the dashboard. The SDK sends them as HTTP Basic credentials (key_id:key_secret):

admin = AuthPIAdmin(api_key=("key_xxx", "your_key_secret"), account_id="acc_xxx")

Bearer Token

For server-side applications authenticating on behalf of a user session:

admin = AuthPIAdmin(
    access_token="tok_xxx",
    account_id="acc_xxx",
)

With optional token refresh callback:

async def refresh():
    new_tokens = await my_refresh_logic()
    return {"access_token": new_tokens.access_token}

admin = AuthPIAdmin(
    access_token="tok_xxx",
    account_id="acc_xxx",
    on_token_expired=refresh,
)

When on_token_expired is provided, the SDK calls it on 401 responses and retries the request with the new token. Concurrent 401s are deduplicated — only one refresh runs at a time.

Account resolution

account_id is optional. When omitted, the SDK resolves it once via GET /v1/me on the first request and caches it — an API key always maps to exactly one account:

async with AuthPIAdmin(api_key=("key_xxx", "your_key_secret")) as admin:
    issuers = await admin.issuers.list()  # resolves the account transparently

If the credential can act on zero or multiple accounts (possible with user bearer tokens), the SDK raises a ConfigurationError naming the choices — pass account_id explicitly in that case.

You can also ask directly who the API considers you to be:

me = await admin.whoami()
# {"type": "api_key", "key_id": "key_...", "issuer_id": "i_...", "accounts": [{"account_id": ..., "org_id": ..., "scopes": [...]}]}

Scoped Client Pattern

The SDK mirrors the API's resource hierarchy. Navigate with chained accessors:

# Account-level resources
await admin.issuers.list()
await admin.webhooks.create({
    "name": "Lifecycle events",
    "url": "https://...",
    "auth": {"type": "signature"},
    "events": [WebhookEventType.USER_CREATED],
})
await admin.events.list(limit=50)

# Issuer scope
iss = admin.issuer("iss_xxx")
await iss.users.list()
await iss.agents.create({"name": "bot"})
await iss.clients.list()
await iss.organizations.list()

# User scope (nested under issuer)
org = admin.issuer("iss_xxx").organization("org_xxx")
await org.members.list()
await org.sso.add_domain(domain="acme.com")

usr = admin.issuer("iss_xxx").user("usr_xxx")
await usr.get()
await usr.sessions.list()
await usr.tokens.list()
await usr.trusted_devices.list()
await usr.verifiers.list()

# Webhook scope
wh = admin.webhook("wh_xxx")
await wh.get()
await wh.deliveries.list()

Pagination

List endpoints return a Page with cursor-based pagination:

# Manual pagination
page = await admin.issuer("iss_xxx").users.list(limit=25)
print(page.data)        # list[dict]
print(page.has_more)    # bool
print(page.next_cursor) # str | None

# Fetch next page
if page.has_more:
    next_page = await admin.issuer("iss_xxx").users.list(
        limit=25, cursor=page.next_cursor
    )

# Auto-pagination (yields individual items across all pages)
async for user in admin.issuer("iss_xxx").users.list_all():
    print(user)

Retries

Read-only requests (GET, HEAD, OPTIONS) are automatically retried on 429, 502, 503, and 504 responses with exponential backoff. The Retry-After header is respected when present.

# Default: retries enabled (3 attempts, 1s base delay, exponential backoff)
admin = AuthPIAdmin(api_key=("key_xxx", "your_key_secret"), account_id="acc_xxx")

# Disable retries
admin = AuthPIAdmin(api_key=("key_xxx", "your_key_secret"), retries=False)

# Custom retry config
admin = AuthPIAdmin(
    api_key=("key_xxx", "your_key_secret"),
    account_id="acc_xxx",
    retries={"limit": 5, "delay": 0.5, "backoff": "linear"},
)

Mutations (POST, PATCH, DELETE) are never retried automatically. Use idempotency keys and handle retries explicitly for writes.

ETags & Optimistic Concurrency

GET responses include an _etag field. Pass it back on updates to prevent overwriting concurrent changes:

user = await admin.issuer("iss_xxx").user("usr_xxx").get()

# Conditional update — fails with PreconditionFailedError if modified
await admin.issuer("iss_xxx").users.update(
    "usr_xxx",
    {"display_name": "Bob"},
    if_match=user.get("_etag"),
)

Error Handling

The SDK maps HTTP status codes to specific error classes:

from authpi_admin import (
    NotFoundError,
    ValidationError,
    AuthenticationError,
    RateLimitError,
    PreconditionFailedError,
)

try:
    await admin.issuer("iss_xxx").user("usr_xxx").get()
except NotFoundError:
    print("User not found")
except ValidationError as err:
    print("Validation failed:", err.fields)
except RateLimitError as err:
    print(f"Retry after {err.retry_after} seconds")
except AuthenticationError:
    print("Invalid credentials")

Error Hierarchy

Configuration

from authpi_admin import AuthPIAdmin

admin = AuthPIAdmin(
    api_key=("key_xxx", "your_key_secret"),     # or access_token instead
    account_id="acc_xxx",                       # optional — resolved via GET /v1/me when omitted
    base_url="https://api.authpi.com",          # default
    timeout=30.0,                               # default, in seconds
    default_headers={"X-Custom": "value"},      # optional extra headers
    retries=True,                               # default (or False, or dict)
)

Custom httpx Client

Inject a pre-configured httpx.AsyncClient for advanced use cases (proxies, certificates, connection pooling):

import httpx
from authpi_admin import AuthPIAdmin

async with httpx.AsyncClient(proxies="http://proxy:8080") as http:
    admin = AuthPIAdmin(api_key=("key_xxx", "your_key_secret"), http_client=http)
    await admin.issuers.list()

Context Manager

The SDK supports async context managers for clean resource cleanup:

async with AuthPIAdmin(api_key=("key_xxx", "your_key_secret"), account_id="acc_xxx") as admin:
    await admin.issuers.list()
# httpx client is closed automatically

Or close manually:

admin = AuthPIAdmin(api_key=("key_xxx", "your_key_secret"), account_id="acc_xxx")
try:
    await admin.issuers.list()
finally:
    await admin.close()

License

MIT