CloudEvents Python SDK

v2 uses the Format protocol instead of marshaller callbacks. To support a new format (like YAML), implement the Format class with the following methods:

• read(event_factory, data): Parse data into attributes and return a BaseCloudEvent.
• write(event): Serialize the entire event to bytes.
• write_data(data, datacontenttype): Serialize just the data payload.
• read_data(body, datacontenttype): Deserialize just the data payload.
• get_content_type(): Return the MIME type.

Example usage:

from cloudevents.core.bindings.http import to_binary

# Use your custom format implementation
message = to_binary(event, event_format=YAMLFormat())

This sample demonstrates how to run a local server and a client to exchange structured and binary CloudEvents over HTTP using JSON.

1. Install dependencies: Ensure you have the necessary packages installed using the provided requirements file.
2. Start the server: Run the server script to listen for incoming CloudEvents.
3. Run the client: Execute the client script, passing the server's URL as an argument, to send both structured and binary CloudEvents to the server.

# 1. Install dependencies
pip install -r requirements.txt

# 2. Start the server
python json_sample_server.py

# 3. In a new shell, run the client
python client.py http://localhost:3000/

This guide demonstrates how to run a local server and a client to send CloudEvents containing image payloads using both structured and binary formats.

1. Setup the Server

First, install the necessary dependencies and start the sample server:

pip install -r requirements.txt
python image_sample_server.py

2. Run the Client

In a separate terminal, execute the client script to send a CloudEvent (containing an image) to the running server at http://localhost:3000/:

python client.py http://localhost:3000/

# Install dependencies
pip install -r requirements.txt

# Start the server
python image_sample_server.py

# Run the client (in a new shell)
python client.py http://localhost:3000/

This guide outlines the breaking changes and new patterns introduced in v2 of the CloudEvents Python SDK.

Requirements

• Python: v2 requires 3.10+ (v1 supported 3.7+).
• Dependencies: v2 requires python-dateutil>=2.8.2 only. The optional pydantic extra from v1 is no longer required.

Architectural Shifts in v2

1. Protocol-based design: BaseCloudEvent is now a Protocol, not a base class. Events use explicit getter methods instead of dict-like access.
2. Explicit serialization: Implicit JSON handling is replaced by a Format protocol (e.g., JSONFormat).
3. Strict validation: Events are validated at construction time. Extension attribute names must be 1-20 lowercase alphanumeric characters, and time must be a timezone-aware datetime.

Serializing

to_binary_event returns a frozen KafkaMessage dataclass with .headers, .key, and .value.

from cloudevents.core.bindings.kafka import to_binary_event, KafkaMessage

kafka_msg = to_binary_event(event)
# Supports custom key mapping
kafka_msg = to_binary_event(
    event,
    key_mapper=lambda e: e.get_extension("partitionkey"),
)

Deserializing

Wrap the raw components in a KafkaMessage and use from_kafka_event to auto-detect the mode and spec version.

from cloudevents.core.bindings.kafka import from_kafka_event, KafkaMessage

msg = KafkaMessage(headers=headers, key=key, value=value)
event = from_kafka_event(msg)

If you are not ready to migrate to the v2 core API, the cloudevents.v1 package provides a drop-in compatibility layer. This preserves the v1 API under a new namespace, allowing you to unpin from old top-level imports without rewriting logic.

Note: The compat layer behaves identically to the old v1 SDK (dict-like, mutable, accepts marshaller callables) but does not enforce strict mypy or v2 validation rules.

# Example mapping for common imports
# Old:
# from cloudevents.http import CloudEvent
# New (Compat):
from cloudevents.v1.http import CloudEvent

# Old:
# from cloudevents.conversion import to_dict
# New (Compat):
from cloudevents.v1.conversion import to_dict

To support a new wire format (like Avro or Protobuf), implement the Format protocol. A Format implementation is responsible for serializing and deserializing CloudEvents between their object representation and their byte representation.

Key responsibilities include:

• read: Deserializing a full CloudEvent from bytes/string into a BaseCloudEvent instance.
• write: Serializing a BaseCloudEvent instance into bytes.
• write_data: Serializing only the data payload (useful for protocol bindings like HTTP binary mode).
• read_data: Deserializing only the data payload from a body.
• get_content_type: Returning the appropriate Content-Type header for structured mode.