Cloud event publishing

Author: jackgerrits

protos/agent_worker.proto

@@ -117,12 +117,11 @@ message Message {
     oneof message {
         RpcRequest request = 1;
         RpcResponse response = 2;
-        Event event = 3;
+        cloudevent.CloudEvent cloudEvent = 3;
         RegisterAgentTypeRequest registerAgentTypeRequest = 4;
         RegisterAgentTypeResponse registerAgentTypeResponse = 5;
         AddSubscriptionRequest addSubscriptionRequest = 6;
         AddSubscriptionResponse addSubscriptionResponse = 7;
-        cloudevent.CloudEvent cloudEvent = 8;
     }
 }
 

python/packages/autogen-core/src/autogen_core/application/_worker_runtime.py

@@ -28,10 +28,12 @@
     cast,
 )
 
+from autogen_core.application.protos import cloudevent_pb2
 from opentelemetry.trace import TracerProvider
 from typing_extensions import Self, deprecated
 
 from ..base import (
+    PROTOBUF_DATA_CONTENT_TYPE,
     JSON_DATA_CONTENT_TYPE,
     Agent,
     AgentId,
@@ -178,6 +180,7 @@ def __init__(
         host_address: str,
         tracer_provider: TracerProvider | None = None,
         extra_grpc_config: ChannelArgumentType | None = None,
+        payload_serialization_format: str = JSON_DATA_CONTENT_TYPE,
     ) -> None:
         self._host_address = host_address
         self._trace_helper = TraceHelper(tracer_provider, MessageRuntimeTracingConfig("Worker Runtime"))
@@ -198,6 +201,11 @@ def __init__(
         self._serialization_registry = SerializationRegistry()
         self._extra_grpc_config = extra_grpc_config or []
 
+        if payload_serialization_format not in {JSON_DATA_CONTENT_TYPE, PROTOBUF_DATA_CONTENT_TYPE}:
+            raise ValueError(f"Unsupported payload serialization format: {payload_serialization_format}")
+
+        self._payload_serialization_format = payload_serialization_format
+
     def start(self) -> None:
         """Start the runtime in a background task."""
         if self._running:
@@ -381,27 +389,33 @@ async def publish_message(
         if message_id is None:
             message_id = str(uuid.uuid4())
 
-        # TODO: consume message_id
-
         message_type = self._serialization_registry.type_name(message)
         with self._trace_helper.trace_block(
             "create", topic_id, parent=None, extraAttributes={"message_type": message_type}
         ):
             serialized_message = self._serialization_registry.serialize(
-                message, type_name=message_type, data_content_type=JSON_DATA_CONTENT_TYPE
+                message, type_name=message_type, data_content_type=self._payload_serialization_format
             )
+
+            sender_id = sender or AgentId("unknown", "unknown")
+            attributes = {
+                "datacontenttype": cloudevent_pb2.CloudEvent.CloudEventAttributeValue(ce_string=self._payload_serialization_format),
+                "dataschema": cloudevent_pb2.CloudEvent.CloudEventAttributeValue(ce_string=message_type),
+                "agagentsendertype": cloudevent_pb2.CloudEvent.CloudEventAttributeValue(ce_string=sender_id.type),
+                "agagentsenderkey": cloudevent_pb2.CloudEvent.CloudEventAttributeValue(ce_string=sender_id.key),
+                "agmsgkind": cloudevent_pb2.CloudEvent.CloudEventAttributeValue(ce_string="publish"),
+            }
+
             telemetry_metadata = get_telemetry_grpc_metadata()
             runtime_message = agent_worker_pb2.Message(
-                event=agent_worker_pb2.Event(
-                    topic_type=topic_id.type,
-                    topic_source=topic_id.source,
-                    source=agent_worker_pb2.AgentId(type=sender.type, key=sender.key) if sender is not None else None,
-                    metadata=telemetry_metadata,
-                    payload=agent_worker_pb2.Payload(
-                        data_type=message_type,
-                        data=serialized_message,
-                        data_content_type=JSON_DATA_CONTENT_TYPE,
-                    ),
+                cloudEvent=cloudevent_pb2.CloudEvent(
+                    id=message_id,
+                    spec_version="1.0",
+                    type=topic_id.type,
+                    source=topic_id.source,
+                    attributes=attributes,
+                    # TODO: use text, or proto fields appropriately
+                    binary_data=serialized_message,
                 )
             )
 

python/packages/autogen-core/src/autogen_core/application/protos/agent_worker_pb2.py

@@ -437,18 +437,17 @@ class Message(google.protobuf.message.Message):
 
     REQUEST_FIELD_NUMBER: builtins.int
     RESPONSE_FIELD_NUMBER: builtins.int
-    EVENT_FIELD_NUMBER: builtins.int
+    CLOUDEVENT_FIELD_NUMBER: builtins.int
     REGISTERAGENTTYPEREQUEST_FIELD_NUMBER: builtins.int
     REGISTERAGENTTYPERESPONSE_FIELD_NUMBER: builtins.int
     ADDSUBSCRIPTIONREQUEST_FIELD_NUMBER: builtins.int
     ADDSUBSCRIPTIONRESPONSE_FIELD_NUMBER: builtins.int
-    CLOUDEVENT_FIELD_NUMBER: builtins.int
     @property
     def request(self) -> global___RpcRequest: ...
     @property
     def response(self) -> global___RpcResponse: ...
     @property
-    def event(self) -> global___Event: ...
+    def cloudEvent(self) -> cloudevent_pb2.CloudEvent: ...
     @property
     def registerAgentTypeRequest(self) -> global___RegisterAgentTypeRequest: ...
     @property
@@ -457,22 +456,19 @@ class Message(google.protobuf.message.Message):
     def addSubscriptionRequest(self) -> global___AddSubscriptionRequest: ...
     @property
     def addSubscriptionResponse(self) -> global___AddSubscriptionResponse: ...
-    @property
-    def cloudEvent(self) -> cloudevent_pb2.CloudEvent: ...
     def __init__(
         self,
         *,
         request: global___RpcRequest | None = ...,
         response: global___RpcResponse | None = ...,
-        event: global___Event | None = ...,
+        cloudEvent: cloudevent_pb2.CloudEvent | None = ...,
         registerAgentTypeRequest: global___RegisterAgentTypeRequest | None = ...,
         registerAgentTypeResponse: global___RegisterAgentTypeResponse | None = ...,
         addSubscriptionRequest: global___AddSubscriptionRequest | None = ...,
         addSubscriptionResponse: global___AddSubscriptionResponse | None = ...,
-        cloudEvent: cloudevent_pb2.CloudEvent | None = ...,
     ) -> None: ...
-    def HasField(self, field_name: typing.Literal["addSubscriptionRequest", b"addSubscriptionRequest", "addSubscriptionResponse", b"addSubscriptionResponse", "cloudEvent", b"cloudEvent", "event", b"event", "message", b"message", "registerAgentTypeRequest", b"registerAgentTypeRequest", "registerAgentTypeResponse", b"registerAgentTypeResponse", "request", b"request", "response", b"response"]) -> builtins.bool: ...
-    def ClearField(self, field_name: typing.Literal["addSubscriptionRequest", b"addSubscriptionRequest", "addSubscriptionResponse", b"addSubscriptionResponse", "cloudEvent", b"cloudEvent", "event", b"event", "message", b"message", "registerAgentTypeRequest", b"registerAgentTypeRequest", "registerAgentTypeResponse", b"registerAgentTypeResponse", "request", b"request", "response", b"response"]) -> None: ...
-    def WhichOneof(self, oneof_group: typing.Literal["message", b"message"]) -> typing.Literal["request", "response", "event", "registerAgentTypeRequest", "registerAgentTypeResponse", "addSubscriptionRequest", "addSubscriptionResponse", "cloudEvent"] | None: ...
+    def HasField(self, field_name: typing.Literal["addSubscriptionRequest", b"addSubscriptionRequest", "addSubscriptionResponse", b"addSubscriptionResponse", "cloudEvent", b"cloudEvent", "message", b"message", "registerAgentTypeRequest", b"registerAgentTypeRequest", "registerAgentTypeResponse", b"registerAgentTypeResponse", "request", b"request", "response", b"response"]) -> builtins.bool: ...
+    def ClearField(self, field_name: typing.Literal["addSubscriptionRequest", b"addSubscriptionRequest", "addSubscriptionResponse", b"addSubscriptionResponse", "cloudEvent", b"cloudEvent", "message", b"message", "registerAgentTypeRequest", b"registerAgentTypeRequest", "registerAgentTypeResponse", b"registerAgentTypeResponse", "request", b"request", "response", b"response"]) -> None: ...
+    def WhichOneof(self, oneof_group: typing.Literal["message", b"message"]) -> typing.Literal["request", "response", "cloudEvent", "registerAgentTypeRequest", "registerAgentTypeResponse", "addSubscriptionRequest", "addSubscriptionResponse"] | None: ...
 
 global___Message = Message

python/packages/autogen-core/src/autogen_core/application/protos/agent_worker_pb2.pyi

@@ -15,6 +15,7 @@
 from ._message_context import MessageContext
 from ._message_handler_context import MessageHandlerContext
 from ._serialization import (
+    PROTOBUF_DATA_CONTENT_TYPE,
     JSON_DATA_CONTENT_TYPE,
     MessageSerializer,
     SerializationRegistry,
@@ -43,6 +44,7 @@
     "SubscriptionInstantiationContext",
     "MessageHandlerContext",
     "JSON_DATA_CONTENT_TYPE",
+    "PROTOBUF_DATA_CONTENT_TYPE",
     "MessageSerializer",
     "try_get_known_serializers_for_type",
     "UnknownPayload",