CDP Python SDK

A Python client library for Codematic's Customer Data Platform (CDP) with optional Customer.io integration.

Installation

pip install cdp-python-sdk

Or via requirements.txt:

cdp-python-sdk

Features

• Async Support: Built on httpx and asyncio for high-performance non-blocking I/O.
• Dual-Write: Optional integration to send data to both CDP and Customer.io simultaneously.
• Type Safety: Uses Pydantic models for request validation.
• Transactional Messaging: Support for Email, Push, and SMS.
• Device Registration: Register devices for push notification targeting.

Examples

A complete runnable example is available in example/:

example/
├── main.py       # Full flow: init → identify → track → register_device → email → push → sms
└── README.md     # How to run

See example/README.md for run instructions.

Quick Start

import asyncio
from cdp_client import CDPClient, CDPConfig

async def main():
    config = CDPConfig(cdp_api_key="your-cdp-api-key")
    client = CDPClient(config)

    await client.identify("user-123", {"email": "user@example.com", "name": "Jane"})
    await client.track("user-123", "signed_up", {"plan": "pro"})

    await client.close()

asyncio.run(main())

Usage

Initialization

import asyncio
from cdp_client import CDPClient, CDPConfig, CustomerIoConfig

async def main():
    config = CDPConfig(
        cdp_api_key="your-cdp-api-key",
        debug=True,
        # Optional: enable Customer.io dual-write
        send_to_customer_io=True,
        customer_io=CustomerIoConfig(
            site_id="your-cio-site-id",
            api_key="your-cio-api-key",
            region="us"   # "us" or "eu"
        )
    )
    client = CDPClient(config)

    # ... use client ...

    await client.close()

asyncio.run(main())

Identify a User

Associate a user ID with a set of traits (name, email, plan, etc.). Call this on sign-up, login, or whenever user attributes change.

await client.identify("user-123", {
    "email": "user@example.com",
    "name": "Jane Doe",
    "plan": "premium"
})

Track an Event

Record a user action or behaviour.

await client.track("user-123", "purchase_completed", {
    "amount": 99.99,
    "currency": "USD",
    "item_id": "prod-456"
})

Register a Device

Register a device token for push notification targeting. Call this after you receive a push token from your mobile platform.

await client.register_device(
    user_id="user-123",
    device_id="device-abc",      # Unique device identifier
    platform="ios",              # "ios", "android", or "web"
    token="apns-or-fcm-token",
    attributes={                  # Optional device metadata
        "app_version": "2.1.0",
        "os_version": "17.2"
    }
)

Send Transactional Email

from cdp_client import EmailPayload, Identifiers

await client.send_email(EmailPayload(
    to="user@example.com",
    identifiers=Identifiers(id="user-123"),
    transactional_message_id="WELCOME_EMAIL",
    subject="Welcome!",
    body="<h1>Thanks for joining!</h1>",
    body_plain="Thanks for joining!"
))

Attachments

Attach up to 5 files (2 MB decoded in total). Bytes are base64-encoded for you:

payload = EmailPayload(
    to="user@example.com",
    identifiers=Identifiers(id="user-123"),
    transactional_message_id="INVOICE_EMAIL",
)
payload.attach_file("./invoice.pdf")                          # named "invoice.pdf"
payload.attach("notes.txt", "Plain text content")             # strings are base64-encoded by default
payload.attach("report.csv", existing_base64, encode=False)   # already base64, sent as-is

await client.send_email(payload)

You can also pass attachments={"invoice.pdf": "<base64>"} directly. send_email validates attachments before sending: at most 5 files, at most 2 MB decoded in total, filenames without /, \ or .., and non-empty base64 content. Invalid attachments raise CDPValidationError when fail_on_exception=True; otherwise the error is logged and nothing is sent. The content type is inferred from the file extension.

Send Push Notification

from cdp_client import PushPayload, Identifiers

await client.send_push(PushPayload(
    identifiers=Identifiers(id="user-123"),
    transactional_message_id="PROMO_PUSH",
    title="Flash Sale 🔥",
    body="50% off for the next hour."
))

Send SMS

from cdp_client import SmsPayload, Identifiers

await client.send_sms(SmsPayload(
    identifiers=Identifiers(id="user-123"),
    to="+14155551234",            # Optional: raw phone number
    transactional_message_id="OTP_MSG",
    body="Your one-time code is 881234."
))

Send WhatsApp

WhatsApp sends use a saved WhatsApp transactional, so transactional_message_id is required.

from cdp_client import WhatsAppPayload, Identifiers

result = await client.send_whatsapp(WhatsAppPayload(
    identifiers=Identifiers(id="user-123"),
    transactional_message_id="ORDER_WHATSAPP",
    to="+14155551234",                          # Optional: overrides the profile phone number
    message_data={"order_number": "12345"},     # Available in the template as {{trigger.order_number}}
))

• A successful return means the message was queued, not delivered. Delivery runs asynchronously, so a missing WhatsApp provider, no phone number, or a template rejected by Meta does not fail this call. result holds the transactional execution record; keep its id to trace the send.
• template_variables ({"header": {...}, "body": {...}, "button": {...}}) sets the template slots from code. Keys must be slot numbers ("1", "2", ...), and values may use Liquid such as {{customer.first_name}}. Passing it replaces all variables saved on the transactional, so include every section the template needs.
• Sends are not retried on another gateway host after a read, write, or pool timeout or an HTTP error, because the message may already have been queued. The exceptions are connection failures (including connection timeouts) and the Cloudflare errors 521, 523, 525 and 526, which mean the gateway never received the request.

Clear Identity / Logout

To reset the client's user context (e.g., on logout), close the current client and re-initialize without a user session:

# On user logout: flush pending work and release the HTTP client
await client.close()

# Re-initialize for anonymous or new user session
client = CDPClient(config)

Error Handling

By default (fail_on_exception=False) the SDK logs errors and does not raise, so a failed call never crashes your app. Set fail_on_exception=True to raise instead:

from cdp_client import CDPError, CDPValidationError

config = CDPConfig(cdp_api_key="...", fail_on_exception=True)
client = CDPClient(config)

try:
    await client.identify("user-123", {"email": "user@example.com"})
except CDPValidationError as e:
    # Invalid input, caught before any request is sent
    print(f"Invalid {e.field}: {e}")
except CDPError as e:
    # status_code is the HTTP status, or None when the gateway could not be reached
    print(f"API error {e.status_code}: {e}")

This applies to all methods (identify, track, send_email, send_push, send_sms, send_whatsapp, register_device). Dual-write Customer.io errors are non-fatal and only emit a warning log.

Configuration Options

CustomerIoConfig Options

Dual-Write to Customer.io

When send_to_customer_io=True, all identify, track, and register_device calls are mirrored to Customer.io automatically. Customer.io failures are non-blocking — the CDP call succeeds even if the Customer.io call fails.

config = CDPConfig(
    cdp_api_key="your-cdp-api-key",
    send_to_customer_io=True,
    customer_io=CustomerIoConfig(
        site_id="cio-site-id",
        api_key="cio-api-key",
        region="eu"
    )
)

Development

Setup

python3 -m venv venv
source venv/bin/activate
python -m pip install --upgrade pip setuptools wheel build
python -m pip install -e ".[test]"

Run Tests

pytest -v

Building the Package

python -m build

This creates files in the dist/ directory:

• cdp_python_sdk-{version}-py3-none-any.whl
• cdp_python_sdk-{version}.tar.gz

Versioning

Follow Semantic Versioning: