Python SDK

Full Python SDK reference with service namespaces

Python SDK

The Python SDK mirrors every method in the TypeScript SDK with Pythonic naming conventions. It uses httpx for HTTP with retry, timeout, and circuit breaker support.

pip install platformxe

from platformxe import PlatformXeClient

client = PlatformXeClient(api_key="pxk_live_your_key_here")

Service namespaces

The client exposes 10 service namespaces with 103 methods total.

---

client.permissions (36 methods)

Permission checks, role management, policies, relationships, federation, and audit.

Check & resolve

# Check a single permission
result = client.permissions.check(
    admin_id="usr_123",
    path="chat/session",
    action="READ",
)
# result["allowed"] -> True/False

# Batch check multiple permissions
results = client.permissions.batch_check(
    admin_id="usr_123",
    checks=[
        {"path": "chat/session", "action": "READ"},
        {"path": "chat/session", "action": "WRITE"},
    ],
)

# Resolve all permissions for a user
resolved = client.permissions.resolve(admin_id="usr_123")

Roles

# List roles
roles = client.permissions.list_roles()

# Create a role
role = client.permissions.create_role(
    name="Support Agent",
    description="Can view and manage support tickets",
    model="SIMPLE",
)

# Get a role
role = client.permissions.get_role(role_id="role_abc123")

# Update a role
client.permissions.update_role(
    role_id="role_abc123",
    name="Senior Agent",
)

# Delete a role
client.permissions.delete_role(role_id="role_abc123")

# Assign role to user
client.permissions.assign_role(admin_id="usr_123", role_id="role_abc123")

# Unassign role from user
client.permissions.unassign_role(admin_id="usr_123", role_id="role_abc123")

Capabilities (SIMPLE model)

# Set capabilities on a role
client.permissions.set_capabilities(
    role_id="role_abc123",
    capabilities=["chat:read", "chat:write", "tickets:read"],
)

# Get capabilities for a role
caps = client.permissions.get_capabilities(role_id="role_abc123")

Module permissions (FULL model)

# Set module permissions on a role
client.permissions.set_module_permissions(
    role_id="role_abc123",
    module_id="mod_xyz",
    permissions={"read": True, "write": True, "delete": False},
)

# Get module permissions for a role
perms = client.permissions.get_module_permissions(
    role_id="role_abc123",
    module_id="mod_xyz",
)

Resource policies (ABAC)

# Create a resource policy
policy = client.permissions.create_resource_policy(
    name="EU Data Access",
    resource_path="data/*",
    conditions={
        "all": [
            {"attribute": "user.region", "operator": "eq", "value": "EU"},
            {"attribute": "resource.classification", "operator": "in", "value": ["public", "internal"]},
        ]
    },
)

# List resource policies
policies = client.permissions.list_resource_policies()

# Update a resource policy
client.permissions.update_resource_policy(policy_id="pol_abc", name="Updated Policy")

# Delete a resource policy
client.permissions.delete_resource_policy(policy_id="pol_abc")

Relationships (ReBAC)

# Create a relationship
client.permissions.create_relationship(
    subject_type="user",
    subject_id="usr_123",
    relation="viewer",
    object_type="document",
    object_id="doc_456",
)

# List relationships
rels = client.permissions.list_relationships(
    subject_type="user",
    subject_id="usr_123",
)

# Delete a relationship
client.permissions.delete_relationship(relationship_id="rel_abc")

# Check relationship
result = client.permissions.check_relationship(
    subject_type="user",
    subject_id="usr_123",
    relation="viewer",
    object_type="document",
    object_id="doc_456",
)

Overrides

# Create an override
client.permissions.create_override(
    admin_id="usr_123",
    path="admin/settings",
    action="WRITE",
    effect="DENY",
)

# List overrides
overrides = client.permissions.list_overrides(admin_id="usr_123")

# Delete an override
client.permissions.delete_override(override_id="ovr_abc")

Federation (permissions — v1.x.x admin permissions cross-app sync)

# Create federation group
group = client.permissions.create_federation_group(
    name="Caldera Suite",
    description="Cross-app permission sharing",
)

# Add member to federation group
client.permissions.add_federation_member(
    group_id="fg_abc",
    app_id="app_xyz",
)

# Sync federation permissions
client.permissions.sync_federation(group_id="fg_abc")

Note: This is the v1.x.x Permission Federation surface — it shares admin permission modules + role grants across Caldera apps. For sharing live tenant event traffic with peer Enterprise orgs (Phase 9D), use client.events.custom.federation instead. Different concept, different access path.

Audit

# Query audit logs
logs = client.permissions.query_audit_logs(
    admin_id="usr_123",
    action="READ",
    start_date="2026-01-01T00:00:00Z",
    end_date="2026-03-31T23:59:59Z",
    limit=50,
)

# Export audit logs
export = client.permissions.export_audit_logs(
    start_date="2026-01-01T00:00:00Z",
    end_date="2026-03-31T23:59:59Z",
    format="csv",
)

---

client.identity

Country-pluggable identity resolution + per-kind verifications across NG / KE / GH / ZA. The engine dispatches through the right per-country provider chain based on the identifier kind.

# Resolve identity (BVN, NIN, phone, etc.)
profile = client.identity.resolve(
    type="BVN",
    value="22012345678",
    resolve_linked=True,
    consent_reference="ref_123",
)

# Generic verify (cross-country)
result = client.identity.verify(
    type="NIN",
    value="12345678901",
    first_name="Adaeze",
    last_name="Okafor",
)

# Per-kind verifications (Phase 6F)
bvn = client.identity.verify_bvn(
    subject_id="usr_001", bvn="22012345678",
    first_name="Adaeze", last_name="Okafor",
)
nin = client.identity.verify_nin(
    subject_id="usr_001", nin="12345678901",
    first_name="Adaeze", last_name="Okafor",
)
acct = client.identity.verify_account(
    subject_id="usr_001",
    account_number="0123456789", bank_code="058",
    first_name="Adaeze", last_name="Okafor",
)
live = client.identity.liveness(
    subject_id="usr_001",
    image_url="https://example.com/selfie.jpg",
)
face = client.identity.face_match(
    subject_id="usr_001",
    selfie_url="https://example.com/selfie.jpg",
    reference_url="https://example.com/id-photo.jpg",
)

# Provider health rollup (Phase 6F.5) — per-(country, provider)
# circuit-breaker state + latency p50/p95/p99
health = client.identity.providers_health()

# Dead-letter queue (Phase 6F.5b)
dlq = client.identity.dlq.list(unreplayed_only=True)
stats = client.identity.dlq.stats()
row = client.identity.dlq.get("idlq_123")
replay = client.identity.dlq.replay("idlq_123")

---

client.messaging (7 methods)

Transactional email and (coming soon) SMS/WhatsApp.

# Send email
result = client.messaging.send_email(
    to="user@example.com",
    subject="Order Confirmation",
    html="<h1>Order #1234 confirmed</h1>",
    reply_to="support@example.com",
)

# Send email to multiple recipients
result = client.messaging.send_email(
    to=["alice@example.com", "bob@example.com"],
    subject="Team Update",
    html="<p>New release deployed.</p>",
)

# Get email status
status = client.messaging.get_email_status(message_id="msg_abc123")

# List email queue
queue = client.messaging.list_email_queue(status="pending", limit=20)

# Retry a failed email
client.messaging.retry_email(message_id="msg_abc123")

# Send SMS (coming soon)
# result = client.messaging.send_sms(to="+2348012345678", body="Your OTP is 123456")

# Send WhatsApp (coming soon)
# result = client.messaging.send_whatsapp(to="+2348012345678", template="order_update", params={"order_id": "1234"})

---

client.webhooks (7 methods)

Manage outbound webhook endpoints.

# Create webhook
webhook = client.webhooks.create(
    url="https://example.com/webhooks",
    events=["email.sent", "email.bounced"],
    secret="whsec_your_signing_secret",
)

# List webhooks
webhooks = client.webhooks.list()

# Get webhook
webhook = client.webhooks.get(webhook_id="wh_abc123")

# Update webhook
client.webhooks.update(webhook_id="wh_abc123", url="https://example.com/webhooks/v2")

# Delete webhook
client.webhooks.delete(webhook_id="wh_abc123")

# Test webhook
client.webhooks.test(webhook_id="wh_abc123")

# Get webhook deliveries
deliveries = client.webhooks.list_deliveries(webhook_id="wh_abc123", limit=20)

---

client.templates (7 methods)

Manage reusable content templates.

# Create template
template = client.templates.create(
    name="welcome-email",
    subject="Welcome to {{company}}",
    html="<h1>Welcome, {{name}}!</h1>",
    variables=["company", "name"],
)

# List templates
templates = client.templates.list()

# Get template
template = client.templates.get(template_id="tmpl_abc123")

# Update template
client.templates.update(template_id="tmpl_abc123", subject="Updated Subject")

# Delete template
client.templates.delete(template_id="tmpl_abc123")

# Render template (preview)
rendered = client.templates.render(
    template_id="tmpl_abc123",
    variables={"company": "Acme", "name": "Jane"},
)

# List template versions
versions = client.templates.list_versions(template_id="tmpl_abc123")

---

client.storage (10 methods)

File upload, retrieval, and management via multi-provider storage.

# Upload file
file = client.storage.upload(
    file=open("invoice.pdf", "rb"),
    filename="invoice.pdf",
    folder="invoices",
    access="private",
)

# Upload from URL
file = client.storage.upload_from_url(
    url="https://example.com/image.png",
    filename="image.png",
    folder="images",
)

# Get file metadata
meta = client.storage.get(file_id="file_abc123")

# List files
files = client.storage.list(folder="invoices", limit=20)

# Delete file
client.storage.delete(file_id="file_abc123")

# Get signed URL (temporary access)
url = client.storage.get_signed_url(file_id="file_abc123", expires_in=3600)

# Update file metadata
client.storage.update(file_id="file_abc123", folder="archive")

# Copy file
copy = client.storage.copy(file_id="file_abc123", folder="backup")

# Move file
client.storage.move(file_id="file_abc123", folder="processed")

# Get storage usage
usage = client.storage.get_usage()

---

client.workflows (6 methods)

Event-driven workflow automation.

# Create workflow
workflow = client.workflows.create(
    name="New User Onboarding",
    trigger={"event": "user.created"},
    actions=[
        {"type": "send_email", "template_id": "tmpl_welcome"},
        {"type": "webhook", "url": "https://example.com/onboard"},
    ],
)

# List workflows
workflows = client.workflows.list()

# Get workflow
workflow = client.workflows.get(workflow_id="wf_abc123")

# Update workflow
client.workflows.update(workflow_id="wf_abc123", name="Updated Onboarding")

# Delete workflow
client.workflows.delete(workflow_id="wf_abc123")

# Trigger workflow manually
client.workflows.trigger(workflow_id="wf_abc123", payload={"user_id": "usr_123"})

---

client.domains (5 methods)

Manage sending domains for email delivery.

# Add sending domain
domain = client.domains.create(domain="mail.example.com")

# List domains
domains = client.domains.list()

# Get domain (includes DNS records)
domain = client.domains.get(domain_id="dom_abc123")

# Verify domain
result = client.domains.verify(domain_id="dom_abc123")

# Delete domain
client.domains.delete(domain_id="dom_abc123")

---

client.subscriptions (8 methods)

Manage event subscriptions for real-time notifications.

# Create subscription
sub = client.subscriptions.create(
    event="email.sent",
    url="https://example.com/events",
)

# List subscriptions
subs = client.subscriptions.list()

# Get subscription
sub = client.subscriptions.get(subscription_id="sub_abc123")

# Update subscription
client.subscriptions.update(subscription_id="sub_abc123", url="https://example.com/events/v2")

# Delete subscription
client.subscriptions.delete(subscription_id="sub_abc123")

# Pause subscription
client.subscriptions.pause(subscription_id="sub_abc123")

# Resume subscription
client.subscriptions.resume(subscription_id="sub_abc123")

# List subscription deliveries
deliveries = client.subscriptions.list_deliveries(subscription_id="sub_abc123")

---

client.misc (7 methods)

Utility and account-level operations.

# Health check
health = client.misc.health()

# Get API key info
info = client.misc.get_api_key_info()

# Get organization details
org = client.misc.get_organization()

# Get usage summary
usage = client.misc.get_usage(period="2026-03")

# Generate PDF from HTML
pdf = client.misc.generate_pdf(
    html="<h1>Invoice #1234</h1><p>Total: ₦50,000</p>",
    options={"format": "A4", "margin": "20mm"},
)

# OCR - extract text from image
text = client.misc.ocr(file_id="file_abc123")

# Verify identity document
result = client.misc.verify_document(
    file_id="file_abc123",
    document_type="national_id",
)

---

client.fraud (29 methods)

Full Phase 6 Detection Engine surface — decisions, rules, screening, devices, cases, and T&Cs. See the SDK Fraud reference for the full method matrix.

# Render a production verdict
verdict = client.fraud.decisions.decide(
    subject={"id": "user_123", "kind": "user"},
    action="withdraw",
    resource={"kind": "transaction"},
    context={"amount": {"value": 75000, "currency": "NGN"}},
)

Sub-namespaces: client.fraud.decisions, client.fraud.rules, client.fraud.screening, client.fraud.devices, client.fraud.cases, client.fraud.terms.

---

client.audit (1 method)

Federated audit log dispatch. See the Audit reference.

client.audit.log(
    app="lettings",
    actor_id="user_123",
    entity_type="BOOKING",
    entity_id="bk_abc",
    action="CONFIRMED",
    metadata={"propertyId": "prop_xyz"},
)

---

client.issues (2 methods)

Federated bug / support-ticket workflow. See the Issues reference.

issues = client.issues.list(app="lettings", status="OPEN")

created = client.issues.create(
    app="lettings",
    reporter_id="user_123",
    title="Booking confirmation email shows wrong nights",
    description="Booked 3 nights, email says 2.",
    priority="HIGH",
)

---

client.search (2 methods)

Federated search index + query. See the Search reference.

client.search.index(
    app="lettings",
    entity_type="PROPERTY",
    entity_id="prop_abc",
    title="Lekki Penthouse",
    keywords="lekki lagos penthouse 3 bedroom",
)

results = client.search.query(q="lekki", app="lettings")

---

client.whoami (1 method)

API key verification + boot-time self-check. See the Whoami reference.

me = client.whoami.get()
print(f"Authenticated as {me['callerService']} ({me['organizationId']})")
if me.get("gracePeriodWarning"):
    print(me["gracePeriodWarning"])

---

platformxe.register (module-level)

Unauthenticated bootstrap for new developers. Sits at the package root because it runs before you have an API key.

from platformxe import register, PlatformXeClient

r = register(name="Acme", email="dev@acme.test")
client = PlatformXeClient(api_key=r["apiKey"])
# Save r["apiKey"] — it cannot be retrieved again.

---

Error handling

from platformxe import PlatformXeError, RateLimitError

try:
    client.messaging.send_email(to="user@example.com", subject="Test", html="<p>Hi</p>")
except RateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after}s")
except PlatformXeError as e:
    print(f"Error {e.code}: {e.message}")

See Error Handling for the full list of error codes.