Courier Python SDK

The Courier Python SDK gives you typed access to the . It includes sync and async clients, Pydantic response models, and TypedDict request params. Available on GitHub and PyPI.

Installation

pip install trycourier

poetry add trycourier

pipenv install trycourier

Requires Python 3.9+.

Quick start

from courier import Courier

client = Courier()

response = client.send.message(
    message={
        "to": {"user_id": "user_123"},
        "template": "nt_01kx4h2jdafq8bk9aftxak4b40",
        "data": {"name": "Sarah Bennett"},
    },
)

print(response.request_id)

The client reads COURIER_API_KEY from your environment automatically. You can also pass it explicitly: Courier(api_key='YOUR_COURIER_API_KEY').

Authentication

Get your API key from in the Courier dashboard. Set it as an environment variable:

export COURIER_API_KEY="YOUR_COURIER_API_KEY"

The SDK picks this up by default. To pass it explicitly:

client = Courier(api_key="YOUR_COURIER_API_KEY")

Use a .env file with python-dotenv to keep your API key out of source control.

Sending notifications

With a template

Design your notification in , then reference it by ID:

response = client.send.message(
    message={
        "to": {"user_id": "user_123"},
        "template": "nt_01kx4h2jdafq8bk9aftxak4b40",
        "data": {"order_id": "10042", "item_name": "Courier Hoodie"},
    },
)

To multiple recipients

Send to a list of users in a single call:

response = client.send.message(
    message={
        "to": [
            {"user_id": "user_123"},
            {"user_id": "user_123"},
            {"email": "sarah@acme-corp.com"},
        ],
        "template": "nt_01kx4h2jdafq8bk9aftxak4b40",
    },
)

Async usage

Basic async client

Import AsyncCourier instead of Courier and use await:

import asyncio
from courier import AsyncCourier

client = AsyncCourier()

async def main():
    response = await client.send.message(
        message={
            "to": {"user_id": "user_123"},
            "template": "nt_01kx4h2jdafq8bk9aftxak4b40",
            "data": {"name": "Sarah Bennett"},
        },
    )
    print(response.request_id)

asyncio.run(main())

The async client has the same API as the sync client. Every method returns an awaitable.

Using aiohttp instead of httpx

For better concurrency, swap the HTTP backend to aiohttp:

pip install trycourier[aiohttp]

import asyncio
from courier import AsyncCourier, DefaultAioHttpClient

async def main():
    async with AsyncCourier(http_client=DefaultAioHttpClient()) as client:
        response = await client.send.message(
            message={
                "to": {"user_id": "user_123"},
                "template": "nt_01kx4h2jdafq8bk9aftxak4b40",
            },
        )
        print(response.request_id)

asyncio.run(main())

Available resources

The SDK covers the full Courier API. Every method is typed and documented with docstrings.

Resource	Namespace	Description
Send	`client.send`	Send messages to one or more recipients
Messages	`client.messages`	Retrieve status, history, and content of sent messages
Profiles	`client.profiles`	Create, update, and retrieve user profiles
Users	`client.users`	Manage preferences, tenants, and push tokens per user
Auth	`client.auth`	Issue JWT tokens for client-side SDK authentication
Lists	`client.lists`	Manage subscription lists and their subscribers
Audiences	`client.audiences`	Define and query audience segments
Tenants	`client.tenants`	Manage tenants for multi-tenant setups
Journeys	`client.journeys`	Invoke and cancel runs. Create, list, publish, and archive
Brands	`client.brands`	Manage brand settings (logos, colors, templates)
Notifications	`client.notifications`	List and inspect notification templates
Translations	`client.translations`	Workspace `.po` strings for the `{{t}}` helper

Common operations

Checking message status

After sending, use the request_id to check delivery status or get the full event timeline:

message = client.messages.retrieve("your-message-id")
print(message.status)    # e.g. 'DELIVERED'
print(message.delivered)  # delivery timestamp (ms)

history = client.messages.history("your-message-id")
for event in history.results:
    print(event["type"], event["event"])

List recent messages with optional filters:

recent = client.messages.list(status=["DELIVERED"])

Managing user profiles

Profiles store recipient data (email, phone, custom fields) that Courier uses for delivery. You need a profile before you can send to a user_id.

# Create or merge into a profile
client.profiles.create("user_123", profile={
    "email": "sarah@acme-corp.com",
    "phone_number": "+15551234567",
    "name": "Sarah Bennett",
})

# Retrieve a profile
profile = client.profiles.retrieve("user_123")
print(profile.profile["email"])

# Partial update (merge)
client.profiles.update("user_123", patch=[{"op": "replace", "path": "/name", "value": "Jane Smith"}])

create merges with any existing profile. Use replace for a full overwrite (any fields not included will be removed).

Issuing JWT tokens

Courier’s (React, JavaScript, mobile) authenticate users with a JWT your backend issues. covers the token flow, the scope strings, and token refresh. Use auth.issue_token:

result = client.auth.issue_token(
    scope="user_id:user_123 inbox:read:messages inbox:write:events read:preferences write:preferences",
    expires_in="2 days",
)

# Return result.token to your frontend
print(result.token)

The scope string controls what the token can access. Common scopes:

Scope	Permission
`user_id:<id>`	Which user the token is for (required)
`inbox:read:messages`	Read inbox messages
`inbox:write:events`	Mark messages as read/archived
`read:preferences`	Read notification preferences
`write:preferences`	Update notification preferences
`write:user-tokens`	Register push notification tokens

Type safety

Request parameters are TypedDicts, and responses are Pydantic models. You get autocomplete and inline docs in your editor.

response = client.send.message(
    message={
        "to": {"user_id": "user_123"},
        "template": "nt_01kx4h2jdafq8bk9aftxak4b40",
        "data": {"foo": "bar"},
    },
)

# Response is a Pydantic model
print(response.request_id)
print(response.to_dict())
print(response.to_json())

For VS Code type checking, set python.analysis.typeCheckingMode to basic in your settings.

Configuration

Error handling

The SDK throws typed errors for API failures. All errors extend courier.APIError:

import courier
from courier import Courier

client = Courier()

try:
    client.send.message(
        message={
            "to": {"user_id": "user_123"},
            "template": "nt_01kx4h2jdafq8bk9aftxak4b40",
        },
    )
except courier.APIConnectionError as e:
    print("The server could not be reached")
    print(e.__cause__)
except courier.RateLimitError as e:
    print("Rate limited; back off a bit.")
except courier.APIStatusError as e:
    print(f"API error: {e.status_code}")
    print(e.response)

Status Code	Error Type
400	`BadRequestError`
401	`AuthenticationError`
403	`PermissionDeniedError`
404	`NotFoundError`
422	`UnprocessableEntityError`
429	`RateLimitError`
>=500	`InternalServerError`
N/A	`APIConnectionError`

Retries

The SDK retries failed requests up to 2 times with exponential backoff. Retried errors include connection failures, timeouts, 408, 409, 429, and 5xx responses. Only network errors are retried. Any other exception raised while a request is in flight reaches your code unchanged. That includes a Celery or RQ time limit that fires while the SDK waits for Courier, so your task’s cleanup runs and the SDK doesn’t send the request again. A retry after a timeout can still repeat a send that Courier already received. To stop a retried send from going out twice, pass an idempotency key.

# Disable retries
client = Courier(max_retries=0)

# Override per-request
client.with_options(max_retries=5).send.message(
    message={
        "to": {"user_id": "user_123"},
        "template": "nt_01kx4h2jdafq8bk9aftxak4b40",
    },
)

Timeouts

Requests time out after 60 seconds by default. Configure globally or per-request:

import httpx
from courier import Courier

# Simple timeout
client = Courier(timeout=20.0)  # 20 seconds

# Granular control
client = Courier(
    timeout=httpx.Timeout(60.0, read=5.0, write=10.0, connect=2.0),
)

# Override per-request
client.with_options(timeout=5.0).send.message(
    message={
        "to": {"user_id": "user_123"},
        "template": "nt_01kx4h2jdafq8bk9aftxak4b40",
    },
)

A timeout raises APITimeoutError. Timed-out requests are retried by default.

Logging

Enable debug logging with the COURIER_LOG environment variable:

export COURIER_LOG=info   # or 'debug' for full request/response logging

The SDK uses Python’s standard logging module, so it works with your existing setup.

Raw response access

Access HTTP headers or stream the response body:

from courier import Courier

client = Courier()

# Get headers + parsed data
response = client.send.with_raw_response.message(
    message={
        "to": {"user_id": "user_123"},
        "template": "nt_01kx4h2jdafq8bk9aftxak4b40",
    },
)
print(response.headers.get("x-request-id"))
send_result = response.parse()

# Stream the response
with client.send.with_streaming_response.message(
    message={
        "to": {"user_id": "user_123"},
        "template": "nt_01kx4h2jdafq8bk9aftxak4b40",
    },
) as response:
    for line in response.iter_lines():
        print(line)

Custom HTTP client

Override the default httpx client for proxies or custom transports:

import httpx
from courier import Courier, DefaultHttpxClient

client = Courier(
    base_url="http://my.test.server.example.com:8083",
    http_client=DefaultHttpxClient(
        proxy="http://my.proxy.example.com",
        transport=httpx.HTTPTransport(local_address="0.0.0.0"),
    ),
)

Journeys

are multi-step workflows (send, delay, branch, throttle, digest, and more). Invoke one by ID or alias to start a run, then cancel runs by ID or a shared cancelation token.

# Start a run (by journey ID or alias); the response includes run_id
response = client.journeys.invoke(
    "your-journey-id",
    user_id="user_123",
    data={"order_id": "A-1042"},
)

# Cancel one run, or every run sharing a cancelation token
client.journeys.cancel(run_id=response.run_id)
# client.journeys.cancel(cancelation_token="order-A-1042")

cancel takes exactly one of run_id or cancelation_token (spelled with one “l”). You can also build and manage journeys from code with client.journeys.create, list, publish, and archive. See .

More operations

A few more resources from the full :

Resource	Method	Use case
User preferences	`client.users.preferences.retrieve(user_id)`	Fetch a user’s notification preferences for your
Cancel a message	`client.messages.cancel(message_id)`	Cancel a delayed or queued message before delivery (returns 409 if already delivered)
Push tokens	`client.users.tokens.add_single(token, user_id=user_id, provider_key="firebase-fcm")`	Register a device push token for , , , or

API Reference

Full REST API docs with request/response examples.

Send API

Learn about the Send endpoint, routing, and message options.

Quickstart

Send your first notification in under two minutes.

GitHub

Source code, issues, and changelog.

Idempotent sends

Pass an idempotency key and Courier replays the first response for that key instead of sending again. The SDK sets the Idempotency-Key header for you. covers the scoping rule and the replay window.

response = client.send.message(
    message={"to": {"user_id": "user_123"}, "template": "nt_01kx4h2jdafq8bk9aftxak4b40"},
    idempotency_key="6f3b8c2a-9d4e-4b7a-a1c5-2e8f0d5b7a91",
)

​Installation

​Quick start

​Authentication

​Sending notifications

​With a template

​To multiple recipients

​Async usage

​Basic async client

​Using aiohttp instead of httpx

​Available resources

​Common operations

​Checking message status

​Managing user profiles

​Issuing JWT tokens

​Type safety

​Configuration

​Error handling

​Retries

​Timeouts

​Logging

​Raw response access

​Custom HTTP client

​Journeys

​More operations

API Reference

Send API

Quickstart

GitHub

​Idempotent sends

Installation

Quick start

Authentication

Sending notifications

With a template

To multiple recipients

Async usage

Basic async client

Using aiohttp instead of httpx

Available resources

Common operations

Checking message status

Managing user profiles

Issuing JWT tokens

Type safety

Configuration

Error handling

Retries

Timeouts

Logging

Raw response access

Custom HTTP client

Journeys

More operations

Idempotent sends