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!"
))

---

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."
))

---

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, the Python SDK raises exceptions on HTTP errors or network failures. Wrap calls in try/except to handle them gracefully:

import httpx

try:
    await client.identify("user-123", {"email": "user@example.com"})
except httpx.HTTPStatusError as e:
    # Server returned 4xx or 5xx
    print(f"API error {e.response.status_code}: {e.response.text}")
except Exception as e:
    # Network error, timeout, etc.
    print(f"Unexpected error: {e}")

All methods (identify, track, send_email, send_push, send_sms, register_device) raise on failure. Dual-write Customer.io errors are non-fatal and only emit a warning log.

---

Configuration Options

Option	Type	Default	Description
cdp_api_key	str	Required	Your CDP API Key
cdp_endpoint	str	Production URL	Custom CDP Gateway URL
debug	bool	False	Enable verbose debug logging
send_to_customer_io	bool	False	Enable dual-write to Customer.io
customer_io	CustomerIoConfig	None	Customer.io integration config (see below)

CustomerIoConfig Options

Option	Type	Description
site_id	str	Customer.io Site ID
api_key	str	Customer.io API Key
region	str	"us" (default) or "eu"

---

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:

Bump	When
PATCH 1.0.0 → 1.0.1	Bug fixes
MINOR 1.0.0 → 1.1.0	New features, backward compatible
MAJOR 1.0.0 → 2.0.0	Breaking API changes