cmp-consent

Python SDK for the CMP officer-assisted deferred consent flow. Runs in your backend (server-to-server) and holds the application API key — never ship the key to a browser.

Three jobs: raise consent requests, gate PII processing on verify(), and verify CMP's signed webhooks.

Install

pip install cmp-consent

Authentication

from promiz import CMPClient

cmp = CMPClient(
    api_base="https://cmp-api.yourdomain",
    api_key="cmpk_live_…",
    webhook_secret="whsec_…",   # only needed for verify_webhook
    timeout=10, retries=2,
)

1) Raise a request (officer submitted)

req = cmp.create_consent_request(
    email="user@example.com",
    notice="kyc-notice",           # the notice key for the application
    requested_by=officer.id,
    idempotency_key=submit_id,     # a retried submit won't double-send the email
)
save(applicant_id, consent_id=req.consent_id)   # store consent_id at t=0

CMP creates a PENDING request, emails the principal a single-use magic link, and returns consent_id immediately.

2) Gate every use of the PII

from cmp_consent import ConsentPending

res = cmp.verify(consent_id=req.consent_id, purpose="bureau_pull")
if not res.allowed:
    raise ConsentPending(res.reason)     # PENDING/DECLINED/EXPIRED → do not process
proceed(proof=res.integrity_hash)        # allowed → consent-backed

verify() is fail-closed: any network error or timeout returns allowed=False (it never raises), so PII is never processed on an ambiguous result. Prefer require_consent(...) when you want it to raise instead:

cmp.require_consent(consent_id=cid, purpose="bureau_pull")   # raises if not granted

3) Receive the callback

# in your webhook route — pass the RAW body bytes
event = cmp.verify_webhook(request.headers, request.body)
if event.type == "CONSENT_GRANTED" and is_new(event.event_id):   # dedupe by event_id
    unblock(event.consent_id)

Webhooks are at-least-once and unordered: dedupe by event.event_id, order by event.consent_seq, and treat verify() as the source of truth on any ambiguity.

Manage

cmp.get_request(req.request_id)   # poll status (fallback if a webhook is missed)
cmp.resend(req.request_id)        # re-mint an expired magic link
cmp.cancel(req.request_id)        # cancel-on-save-failure

Async

AsyncCMPClient mirrors the same surface with await:

from cmp_consent import AsyncCMPClient

async with AsyncCMPClient(api_base=…, api_key=…) as cmp:
    req = await cmp.create_consent_request(email=e, notice="kyc-notice", requested_by=o)
    res = await cmp.verify(consent_id=req.consent_id, purpose="bureau_pull")

Exceptions

ConsentPending · ConsentDeclined · ConsentExpired · SignatureError · CMPError (base). Each carries .status (HTTP) and .code (envelope code) when available.