diff --git a/sdk/core/azure-core/CHANGELOG.md b/sdk/core/azure-core/CHANGELOG.md
index 90e3b12f5520..3641c7733f78 100644
--- a/sdk/core/azure-core/CHANGELOG.md
+++ b/sdk/core/azure-core/CHANGELOG.md
@@ -1,7 +1,10 @@
 # Release History
 
-## 1.11.1 (Unreleased)
+## 1.12.0 (Unreleased)
 
+### Features
+
+- Added `azure.core.messaging.CloudEvent` model that follows the cloud event spec.
 
 ## 1.11.0 (2021-02-08)
 
diff --git a/sdk/core/azure-core/azure/core/_version.py b/sdk/core/azure-core/azure/core/_version.py
index 14d127f747d9..7643b787eff9 100644
--- a/sdk/core/azure-core/azure/core/_version.py
+++ b/sdk/core/azure-core/azure/core/_version.py
@@ -9,4 +9,4 @@
 # regenerated.
 # --------------------------------------------------------------------------
 
-VERSION = "1.11.1"
+VERSION = "1.12.0"
diff --git a/sdk/core/azure-core/azure/core/messaging.py b/sdk/core/azure-core/azure/core/messaging.py
new file mode 100644
index 000000000000..b9b7cc3880b1
--- /dev/null
+++ b/sdk/core/azure-core/azure/core/messaging.py
@@ -0,0 +1,135 @@
+# coding=utf-8
+# --------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for
+# license information.
+# --------------------------------------------------------------------------
+import uuid
+from  base64 import b64decode
+from datetime import tzinfo, timedelta, datetime
+
+try:
+    from datetime import timezone
+    TZ_UTC = timezone.utc  # type: ignore
+except ImportError:
+    class UTC(tzinfo):
+        """Time Zone info for handling UTC in python2"""
+
+        def utcoffset(self, dt):
+            """UTF offset for UTC is 0."""
+            return timedelta(0)
+
+        def tzname(self, dt):
+            """Timestamp representation."""
+            return "Z"
+
+        def dst(self, dt):
+            """No daylight saving for UTC."""
+            return timedelta(hours=1)
+
+    TZ_UTC = UTC() # type: ignore
+
+try:
+    from typing import TYPE_CHECKING
+except ImportError:
+    TYPE_CHECKING = False
+
+if TYPE_CHECKING:
+    from typing import Any, Dict
+
+__all__ = ["CloudEvent"]
+
+
+class CloudEvent(object):   #pylint:disable=too-many-instance-attributes
+    """Properties of the CloudEvent 1.0 Schema.
+    All required parameters must be populated in order to send to Azure.
+    If data is of binary type, data_base64 can be used alternatively. Note that data and data_base64
+    cannot be present at the same time.
+    :param source: Required. Identifies the context in which an event happened. The combination of id and source must
+     be unique for each distinct event. If publishing to a domain topic, source must be the domain name.
+    :type source: str
+    :param type: Required. Type of event related to the originating occurrence.
+    :type type: str
+    :keyword data: Optional. Event data specific to the event type. If data is of bytes type, it will be sent
+     as data_base64 in the outgoing request.
+    :type data: object
+    :keyword time: Optional. The time (in UTC) the event was generated, in RFC3339 format.
+    :type time: ~datetime.datetime
+    :keyword dataschema: Optional. Identifies the schema that data adheres to.
+    :type dataschema: str
+    :keyword datacontenttype: Optional. Content type of data value.
+    :type datacontenttype: str
+    :keyword subject: Optional. This describes the subject of the event in the context of the event producer
+     (identified by source).
+    :type subject: str
+    :keyword specversion: Optional. The version of the CloudEvent spec. Defaults to "1.0"
+    :type specversion: str
+    :keyword id: Optional. An identifier for the event. The combination of id and source must be
+     unique for each distinct event. If not provided, a random UUID will be generated and used.
+    :type id: Optional[str]
+    :ivar source: Identifies the context in which an event happened. The combination of id and source must
+     be unique for each distinct event. If publishing to a domain topic, source must be the domain name.
+    :vartype source: str
+    :ivar data: Event data specific to the event type.
+    :vartype data: object
+    :ivar type: Type of event related to the originating occurrence.
+    :vartype type: str
+    :ivar time: The time (in UTC) the event was generated, in RFC3339 format.
+    :vartype time: ~datetime.datetime
+    :ivar dataschema: Identifies the schema that data adheres to.
+    :vartype dataschema: str
+    :ivar datacontenttype: Content type of data value.
+    :vartype datacontenttype: str
+    :ivar subject: This describes the subject of the event in the context of the event producer
+     (identified by source).
+    :vartype subject: str
+    :ivar specversion: Optional. The version of the CloudEvent spec. Defaults to "1.0"
+    :vartype specversion: str
+    :ivar id: An identifier for the event. The combination of id and source must be
+     unique for each distinct event. If not provided, a random UUID will be generated and used.
+    :vartype id: Optional[str]
+    """
+    def __init__(self, source, type, **kwargs): # pylint: disable=redefined-builtin
+        # type: (str, str, Any) -> None
+        self.source = source
+        self.type = type
+        self.specversion = kwargs.pop("specversion", "1.0")
+        self.id = kwargs.pop("id", str(uuid.uuid4()))
+        self.time = kwargs.pop("time", datetime.now(TZ_UTC).isoformat())
+        self.datacontenttype = kwargs.pop("datacontenttype", None)
+        self.dataschema = kwargs.pop("dataschema", None)
+        self.subject = kwargs.pop("subject", None)
+        self.extensions = {}
+        _extensions = dict(kwargs.pop('extensions', {}))
+        for key in _extensions.keys():
+            if not key.islower() or not key.isalnum():
+                raise ValueError("Extensions must be lower case and alphanumeric.")
+        self.extensions.update(_extensions)
+        self.data = kwargs.pop("data", None)
+
+    @classmethod
+    def from_dict(cls, event, **kwargs):
+        # type: (Dict, Any) -> CloudEvent
+        """
+        Returns the deserialized CloudEvent object when a dict is provided.
+        :param event: The dict representation of the event which needs to be deserialized.
+        :type event: dict
+        :rtype: CloudEvent
+        """
+        data = event.pop("data", None)
+        data_base64 = event.pop("data_base64", None)
+        if data and data_base64:
+            raise ValueError("Invalid input. Only one of data and data_base64 must be present.")
+        return cls(
+        id=event.pop("id", None),
+        source=event.pop("source", None),
+        type=event.pop("type", None),
+        specversion=event.pop("specversion", None),
+        data=data or b64decode(data_base64),
+        time=event.pop("time", None),
+        dataschema=event.pop("dataschema", None),
+        datacontenttype=event.pop("datacontenttype", None),
+        subject=event.pop("subject", None),
+        extensions=event,
+        **kwargs
+        )
diff --git a/sdk/core/azure-core/tests/test_messaging_cloud_event.py b/sdk/core/azure-core/tests/test_messaging_cloud_event.py
new file mode 100644
index 000000000000..bb94799bce26
--- /dev/null
+++ b/sdk/core/azure-core/tests/test_messaging_cloud_event.py
@@ -0,0 +1,95 @@
+import logging
+import sys
+import os
+import pytest
+import json
+
+from azure.core.messaging import CloudEvent
+
+# Cloud Event tests
+def test_cloud_storage_dict():
+    cloud_storage_dict = {
+        "id":"a0517898-9fa4-4e70-b4a3-afda1dd68672",
+        "source":"/subscriptions/{subscription-id}/resourceGroups/{resource-group}/providers/Microsoft.Storage/storageAccounts/{storage-account}",
+        "data":{
+            "api":"PutBlockList",
+            "client_request_id":"6d79dbfb-0e37-4fc4-981f-442c9ca65760",
+            "request_id":"831e1650-001e-001b-66ab-eeb76e000000",
+            "e_tag":"0x8D4BCC2E4835CD0",
+            "content_type":"application/octet-stream",
+            "content_length":524288,
+            "blob_type":"BlockBlob",
+            "url":"https://oc2d2817345i60006.blob.core.windows.net/oc2d2817345i200097container/oc2d2817345i20002296blob",
+            "sequencer":"00000000000004420000000000028963",
+            "storage_diagnostics":{"batchId":"b68529f3-68cd-4744-baa4-3c0498ec19f0"}
+        },
+        "type":"Microsoft.Storage.BlobCreated",
+        "time":"2020-08-07T01:11:49.765846Z",
+        "specversion":"1.0"
+    }
+
+    event = CloudEvent.from_dict(cloud_storage_dict)
+    assert event.data == {
+        "api":"PutBlockList",
+        "client_request_id":"6d79dbfb-0e37-4fc4-981f-442c9ca65760",
+        "request_id":"831e1650-001e-001b-66ab-eeb76e000000",
+        "e_tag":"0x8D4BCC2E4835CD0",
+        "content_type":"application/octet-stream",
+        "content_length":524288,
+        "blob_type":"BlockBlob",
+        "url":"https://oc2d2817345i60006.blob.core.windows.net/oc2d2817345i200097container/oc2d2817345i20002296blob",
+        "sequencer":"00000000000004420000000000028963",
+        "storage_diagnostics":{"batchId":"b68529f3-68cd-4744-baa4-3c0498ec19f0"}
+    }
+    assert event.specversion == "1.0"
+    assert event.__class__ == CloudEvent
+
+
+def test_cloud_custom_dict_with_extensions():
+    cloud_custom_dict_with_extensions = {
+        "id":"de0fd76c-4ef4-4dfb-ab3a-8f24a307e033",
+        "source":"https://egtest.dev/cloudcustomevent",
+        "data":{"team": "event grid squad"},
+        "type":"Azure.Sdk.Sample",
+        "time":"2020-08-07T02:06:08.11969Z",
+        "specversion":"1.0",
+        "ext1": "example",
+        "ext2": "example2"
+    }
+    event = CloudEvent.from_dict(cloud_custom_dict_with_extensions)
+    assert event.data == {"team": "event grid squad"}
+    assert event.__class__ == CloudEvent
+    assert event.extensions == {"ext1": "example", "ext2": "example2"}
+
+def test_cloud_custom_dict_base64():
+    cloud_custom_dict_base64 = {
+        "id":"de0fd76c-4ef4-4dfb-ab3a-8f24a307e033",
+        "source":"https://egtest.dev/cloudcustomevent",
+        "data_base64":'Y2xvdWRldmVudA==',
+        "type":"Azure.Sdk.Sample",
+        "time":"2020-08-07T02:06:08.11969Z",
+        "specversion":"1.0"
+    }
+    event = CloudEvent.from_dict(cloud_custom_dict_base64)
+    assert event.data == b'cloudevent'
+    assert event.specversion == "1.0"
+    assert event.__class__ == CloudEvent
+
+def test_extensions_upper_case_value_error():
+    with pytest.raises(ValueError):
+        event = CloudEvent(
+            source='sample',
+            type='type',
+            data='data',
+            extensions={"lowercase123": "accepted", "NOTlower123": "not allowed"}
+        )
+
+def test_data_and_base64_both_exist_raises():
+    with pytest.raises(ValueError):
+        CloudEvent.from_dict(
+            {"source":'sample',
+            "type":'type',
+            "data":'data',
+            "data_base64":'Y2kQ=='
+            }
+        )
diff --git a/sdk/eventgrid/azure-eventgrid/CHANGELOG.md b/sdk/eventgrid/azure-eventgrid/CHANGELOG.md
index 7826d7f3b0a6..42d1fc88db70 100644
--- a/sdk/eventgrid/azure-eventgrid/CHANGELOG.md
+++ b/sdk/eventgrid/azure-eventgrid/CHANGELOG.md
@@ -2,6 +2,8 @@
 
 ## 2.0.0b6 (Unreleased)
 
+  **Breaking Changes**
+  - `azure.eventgrid.CloudEvent` is now removed. `azure.core.messaging.CloudEvent` must be used instead.
 
 ## 2.0.0b5 (2021-02-10)
 
diff --git a/sdk/eventgrid/azure-eventgrid/azure/eventgrid/__init__.py b/sdk/eventgrid/azure-eventgrid/azure/eventgrid/__init__.py
index 8c945af545a1..0b34a09f9889 100644
--- a/sdk/eventgrid/azure-eventgrid/azure/eventgrid/__init__.py
+++ b/sdk/eventgrid/azure-eventgrid/azure/eventgrid/__init__.py
@@ -7,10 +7,8 @@
 from ._publisher_client import EventGridPublisherClient
 from ._event_mappings import SystemEventNames
 from ._helpers import generate_sas
-from ._models import CloudEvent, EventGridEvent
+from ._models import EventGridEvent
 from ._version import VERSION
 
-__all__ = ['EventGridPublisherClient', 'CloudEvent',
-            'EventGridEvent', 'generate_sas', 'SystemEventNames'
-            ]
+__all__ = ['EventGridPublisherClient', 'EventGridEvent', 'generate_sas', 'SystemEventNames']
 __version__ = VERSION
diff --git a/sdk/eventgrid/azure-eventgrid/azure/eventgrid/_helpers.py b/sdk/eventgrid/azure-eventgrid/azure/eventgrid/_helpers.py
index a09f0ed7b0ee..ffaec8ad94d3 100644
--- a/sdk/eventgrid/azure-eventgrid/azure/eventgrid/_helpers.py
+++ b/sdk/eventgrid/azure-eventgrid/azure/eventgrid/_helpers.py
@@ -16,6 +16,7 @@
 from azure.core.credentials import AzureKeyCredential, AzureSasCredential
 from ._signature_credential_policy import EventGridSasCredentialPolicy
 from . import _constants as constants
+from ._generated.models import CloudEvent as InternalCloudEvent
 
 if TYPE_CHECKING:
     from datetime import datetime
@@ -110,3 +111,25 @@ def _eventgrid_data_typecheck(event):
     if isinstance(data, six.binary_type):
         raise TypeError("Data in EventGridEvent cannot be bytes. Please refer to"\
             "https://docs.microsoft.com/en-us/azure/event-grid/event-schema")
+
+def _cloud_event_to_generated(cloud_event, **kwargs):
+    if isinstance(cloud_event.data, six.binary_type):
+        data_base64 = cloud_event.data
+        data = None
+    else:
+        data = cloud_event.data
+        data_base64 = None
+    return InternalCloudEvent(
+        id=cloud_event.id,
+        source=cloud_event.source,
+        type=cloud_event.type,
+        specversion=cloud_event.specversion,
+        data=data,
+        data_base64=data_base64,
+        time=cloud_event.time,
+        dataschema=cloud_event.dataschema,
+        datacontenttype=cloud_event.datacontenttype,
+        subject=cloud_event.subject,
+        additional_properties=cloud_event.extensions,
+        **kwargs
+    )
diff --git a/sdk/eventgrid/azure-eventgrid/azure/eventgrid/_models.py b/sdk/eventgrid/azure-eventgrid/azure/eventgrid/_models.py
index c886e78553a6..ef325bf8cff5 100644
--- a/sdk/eventgrid/azure-eventgrid/azure/eventgrid/_models.py
+++ b/sdk/eventgrid/azure-eventgrid/azure/eventgrid/_models.py
@@ -3,153 +3,14 @@
 # Licensed under the MIT License. See License.txt in the project root for license information.
 # --------------------------------------------------------------------------------------------
 # pylint:disable=protected-access
-from typing import Union, Any, Dict
+from typing import Any
 import datetime as dt
 import uuid
-import json
-import six
 from msrest.serialization import UTC
-from ._generated.models import EventGridEvent as InternalEventGridEvent, CloudEvent as InternalCloudEvent
+from ._generated.models import EventGridEvent as InternalEventGridEvent
 
 
-class EventMixin(object):
-    """
-    Mixin for the event models comprising of some helper methods.
-    """
-    @staticmethod
-    def _from_json(event, encode):
-        """
-        Load the event into the json
-        :param dict eventgrid_event: The event to be deserialized.
-        :type eventgrid_event: Union[str, dict, bytes]
-        :param str encode: The encoding to be used. Defaults to 'utf-8'
-        """
-        if isinstance(event, six.binary_type):
-            event = json.loads(event.decode(encode))
-        elif isinstance(event, six.string_types):
-            event = json.loads(event)
-        return event
-
-
-class CloudEvent(EventMixin):   #pylint:disable=too-many-instance-attributes
-    """Properties of an event published to an Event Grid topic using the CloudEvent 1.0 Schema.
-
-    All required parameters must be populated in order to send to Azure.
-    If data is of binary type, data_base64 can be used alternatively. Note that data and data_base64
-    cannot be present at the same time.
-
-    :param source: Required. Identifies the context in which an event happened. The combination of id and source must
-     be unique for each distinct event. If publishing to a domain topic, source must be the domain name.
-    :type source: str
-    :param type: Required. Type of event related to the originating occurrence.
-    :type type: str
-    :keyword data: Optional. Event data specific to the event type. Only one of the `data` or `data_base64`
-     argument must be present. If data is of bytes type, it will be sent as data_base64 in the outgoing request.
-    :type data: object
-    :keyword time: Optional. The time (in UTC) the event was generated, in RFC3339 format.
-    :type time: ~datetime.datetime
-    :keyword dataschema: Optional. Identifies the schema that data adheres to.
-    :type dataschema: str
-    :keyword datacontenttype: Optional. Content type of data value.
-    :type datacontenttype: str
-    :keyword subject: Optional. This describes the subject of the event in the context of the event producer
-     (identified by source).
-    :type subject: str
-    :keyword specversion: Optional. The version of the CloudEvent spec. Defaults to "1.0"
-    :type specversion: str
-    :keyword id: Optional. An identifier for the event. The combination of id and source must be
-     unique for each distinct event. If not provided, a random UUID will be generated and used.
-    :type id: Optional[str]
-    :keyword data_base64: Optional. Event data specific to the event type if the data is of bytes type.
-     Only data of bytes type is accepted by `data-base64` and only one of the `data` or `data_base64` argument
-     must be present.
-    :type data_base64: bytes
-    :ivar source: Identifies the context in which an event happened. The combination of id and source must
-     be unique for each distinct event. If publishing to a domain topic, source must be the domain name.
-    :vartype source: str
-    :ivar data: Event data specific to the event type.
-    :vartype data: object
-    :ivar data_base64: Event data specific to the event type if the data is of bytes type.
-    :vartype data_base64: bytes
-    :ivar type: Type of event related to the originating occurrence.
-    :vartype type: str
-    :ivar time: The time (in UTC) the event was generated, in RFC3339 format.
-    :vartype time: ~datetime.datetime
-    :ivar dataschema: Identifies the schema that data adheres to.
-    :vartype dataschema: str
-    :ivar datacontenttype: Content type of data value.
-    :vartype datacontenttype: str
-    :ivar subject: This describes the subject of the event in the context of the event producer
-     (identified by source).
-    :vartype subject: str
-    :ivar specversion: Optional. The version of the CloudEvent spec. Defaults to "1.0"
-    :vartype specversion: str
-    :ivar id: An identifier for the event. The combination of id and source must be
-     unique for each distinct event. If not provided, a random UUID will be generated and used.
-    :vartype id: Optional[str]
-    """
-    def __init__(self, source, type, **kwargs): # pylint: disable=redefined-builtin
-        # type: (str, str, Any) -> None
-        self.source = source
-        self.type = type
-        self.specversion = kwargs.pop("specversion", "1.0")
-        self.id = kwargs.pop("id", str(uuid.uuid4()))
-        self.time = kwargs.pop("time", dt.datetime.now(UTC()).isoformat())
-        self.data = kwargs.pop("data", None)
-        self.datacontenttype = kwargs.pop("datacontenttype", None)
-        self.dataschema = kwargs.pop("dataschema", None)
-        self.subject = kwargs.pop("subject", None)
-        self.data_base64 = kwargs.pop("data_base64", None)
-        self.extensions = {}
-        self.extensions.update(dict(kwargs.pop('extensions', {})))
-        if self.data is not None and self.data_base64 is not None:
-            raise ValueError("data and data_base64 cannot be provided at the same time.\
-                Use data_base64 only if you are sending bytes, and use data otherwise.")
-
-    @classmethod
-    def _from_generated(cls, cloud_event, **kwargs):
-        # type: (Union[str, Dict, bytes], Any) -> CloudEvent
-        generated = InternalCloudEvent.deserialize(cloud_event)
-        if generated.additional_properties:
-            extensions = dict(generated.additional_properties)
-            kwargs.setdefault('extensions', extensions)
-        return cls(
-            id=generated.id,
-            source=generated.source,
-            type=generated.type,
-            specversion=generated.specversion,
-            data=generated.data or generated.data_base64,
-            time=generated.time,
-            dataschema=generated.dataschema,
-            datacontenttype=generated.datacontenttype,
-            subject=generated.subject,
-            **kwargs
-        )
-
-    def _to_generated(self, **kwargs):
-        if isinstance(self.data, six.binary_type):
-            data_base64 = self.data
-            data = None
-        else:
-            data = self.data
-            data_base64 = None
-        return InternalCloudEvent(
-            id=self.id,
-            source=self.source,
-            type=self.type,
-            specversion=self.specversion,
-            data=data,
-            data_base64=self.data_base64 or data_base64,
-            time=self.time,
-            dataschema=self.dataschema,
-            datacontenttype=self.datacontenttype,
-            subject=self.subject,
-            additional_properties=self.extensions,
-            **kwargs
-        )
-
-
-class EventGridEvent(InternalEventGridEvent, EventMixin):
+class EventGridEvent(InternalEventGridEvent):
     """Properties of an event published to an Event Grid topic using the EventGrid Schema.
 
     Variables are only populated by the server, and will be ignored when sending a request.
diff --git a/sdk/eventgrid/azure-eventgrid/azure/eventgrid/_publisher_client.py b/sdk/eventgrid/azure-eventgrid/azure/eventgrid/_publisher_client.py
index 02fd405b1aa6..96e5e155cb34 100644
--- a/sdk/eventgrid/azure-eventgrid/azure/eventgrid/_publisher_client.py
+++ b/sdk/eventgrid/azure-eventgrid/azure/eventgrid/_publisher_client.py
@@ -22,16 +22,18 @@
     UserAgentPolicy
 )
 
-from ._models import CloudEvent, EventGridEvent
+from azure.core.messaging import CloudEvent
+from ._policies import CloudEventDistributedTracingPolicy
+from ._models import EventGridEvent
 from ._helpers import (
     _get_endpoint_only_fqdn,
     _get_authentication_policy,
     _is_cloud_event,
     _is_eventgrid_event,
-    _eventgrid_data_typecheck
+    _eventgrid_data_typecheck,
+    _cloud_event_to_generated
 )
 from ._generated._event_grid_publisher_client import EventGridPublisherClient as EventGridPublisherClientImpl
-from ._policies import CloudEventDistributedTracingPolicy
 from ._version import VERSION
 from ._generated.models import CloudEvent as InternalCloudEvent
 
@@ -116,7 +118,7 @@ def send(self, events, **kwargs):
 
         if isinstance(events[0], CloudEvent) or _is_cloud_event(events[0]):
             try:
-                events = [cast(CloudEvent, e)._to_generated(**kwargs) for e in events] # pylint: disable=protected-access
+                events = [_cloud_event_to_generated(e, **kwargs) for e in events] # pylint: disable=protected-access
             except AttributeError:
                 pass # means it's a dictionary
             kwargs.setdefault("content_type", "application/cloudevents-batch+json; charset=utf-8")
diff --git a/sdk/eventgrid/azure-eventgrid/azure/eventgrid/aio/_publisher_client_async.py b/sdk/eventgrid/azure-eventgrid/azure/eventgrid/aio/_publisher_client_async.py
index 670b68a4ebf0..6e62f323492e 100644
--- a/sdk/eventgrid/azure-eventgrid/azure/eventgrid/aio/_publisher_client_async.py
+++ b/sdk/eventgrid/azure-eventgrid/azure/eventgrid/aio/_publisher_client_async.py
@@ -22,14 +22,16 @@
     HttpLoggingPolicy,
     UserAgentPolicy
 )
+from azure.core.messaging import CloudEvent
 from .._policies import CloudEventDistributedTracingPolicy
-from .._models import CloudEvent, EventGridEvent
+from .._models import EventGridEvent
 from .._helpers import (
     _get_endpoint_only_fqdn,
     _get_authentication_policy,
     _is_cloud_event,
     _is_eventgrid_event,
-    _eventgrid_data_typecheck
+    _eventgrid_data_typecheck,
+    _cloud_event_to_generated
 )
 from .._generated.aio import EventGridPublisherClient as EventGridPublisherClientAsync
 from .._generated.models import CloudEvent as InternalCloudEvent
@@ -119,7 +121,7 @@ async def send(
 
         if isinstance(events[0], CloudEvent) or _is_cloud_event(events[0]):
             try:
-                events = [cast(CloudEvent, e)._to_generated(**kwargs) for e in events] # pylint: disable=protected-access
+                events = [_cloud_event_to_generated(e, **kwargs) for e in events] # pylint: disable=protected-access
             except AttributeError:
                 pass # means it's a dictionary
             kwargs.setdefault("content_type", "application/cloudevents-batch+json; charset=utf-8")
diff --git a/sdk/eventgrid/azure-eventgrid/dev_requirements.txt b/sdk/eventgrid/azure-eventgrid/dev_requirements.txt
index 2d79b6ebd683..f9efb595635a 100644
--- a/sdk/eventgrid/azure-eventgrid/dev_requirements.txt
+++ b/sdk/eventgrid/azure-eventgrid/dev_requirements.txt
@@ -3,3 +3,4 @@
 -e ../../core/azure-core
 -e ../../identity/azure-identity
 -e ../azure-mgmt-eventgrid
+aiohttp>=3.0; python_version >= '3.5'
diff --git a/sdk/eventgrid/azure-eventgrid/samples/champion_scenarios/cs1_publish_custom_events_to_a_topic.py b/sdk/eventgrid/azure-eventgrid/samples/champion_scenarios/cs1_publish_custom_events_to_a_topic.py
index b91cb5a9f0cb..9101f7cc12d9 100644
--- a/sdk/eventgrid/azure-eventgrid/samples/champion_scenarios/cs1_publish_custom_events_to_a_topic.py
+++ b/sdk/eventgrid/azure-eventgrid/samples/champion_scenarios/cs1_publish_custom_events_to_a_topic.py
@@ -15,7 +15,7 @@
     "<YOUR-TOPIC-NAME>.<REGION-NAME>.eventgrid.azure.net".
 """
 import os
-from azure.eventgrid import EventGridPublisherClient, EventGridEvent, CloudEvent
+from azure.eventgrid import EventGridPublisherClient, EventGridEvent
 from azure.core.credentials import AzureKeyCredential
 
 topic_key = os.environ["EG_ACCESS_KEY"]
diff --git a/sdk/eventgrid/azure-eventgrid/samples/champion_scenarios/cs5_publish_events_using_cloud_events_1.0_schema.py b/sdk/eventgrid/azure-eventgrid/samples/champion_scenarios/cs5_publish_events_using_cloud_events_1.0_schema.py
index 94649fe65502..ba0d9729dc96 100644
--- a/sdk/eventgrid/azure-eventgrid/samples/champion_scenarios/cs5_publish_events_using_cloud_events_1.0_schema.py
+++ b/sdk/eventgrid/azure-eventgrid/samples/champion_scenarios/cs5_publish_events_using_cloud_events_1.0_schema.py
@@ -15,7 +15,8 @@
     "<YOUR-TOPIC-NAME>.<REGION-NAME>.eventgrid.azure.net".
 """
 import os
-from azure.eventgrid import EventGridPublisherClient, CloudEvent
+from azure.eventgrid import EventGridPublisherClient
+from azure.core.messaging import CloudEvent
 from azure.core.credentials import AzureKeyCredential
 
 topic_key = os.environ["CLOUD_ACCESS_KEY"]
diff --git a/sdk/eventgrid/azure-eventgrid/samples/champion_scenarios/sample_consume_custom_payload.py b/sdk/eventgrid/azure-eventgrid/samples/champion_scenarios/sample_consume_custom_payload.py
index f321d95e8e67..1a8bc89056f1 100644
--- a/sdk/eventgrid/azure-eventgrid/samples/champion_scenarios/sample_consume_custom_payload.py
+++ b/sdk/eventgrid/azure-eventgrid/samples/champion_scenarios/sample_consume_custom_payload.py
@@ -1,4 +1,4 @@
-from azure.eventgrid import CloudEvent
+from azure.core.messaging import CloudEvent
 import json
 
 # all types of CloudEvents below produce same DeserializedEvent
diff --git a/sdk/eventgrid/azure-eventgrid/samples/publish_samples/publish_cloud_events_to_custom_topic_sample.py b/sdk/eventgrid/azure-eventgrid/samples/publish_samples/publish_cloud_events_to_custom_topic_sample.py
index 91ab801e9d5a..170fa03a36d8 100644
--- a/sdk/eventgrid/azure-eventgrid/samples/publish_samples/publish_cloud_events_to_custom_topic_sample.py
+++ b/sdk/eventgrid/azure-eventgrid/samples/publish_samples/publish_cloud_events_to_custom_topic_sample.py
@@ -20,7 +20,8 @@
 import time
 
 from azure.core.credentials import AzureKeyCredential
-from azure.eventgrid import EventGridPublisherClient, CloudEvent
+from azure.core.messaging import CloudEvent
+from azure.eventgrid import EventGridPublisherClient
 
 key = os.environ.get("CLOUD_ACCESS_KEY")
 endpoint = os.environ["CLOUD_TOPIC_HOSTNAME"]
diff --git a/sdk/eventgrid/azure-eventgrid/samples/publish_samples/publish_cloud_events_to_domain_topic_sample.py b/sdk/eventgrid/azure-eventgrid/samples/publish_samples/publish_cloud_events_to_domain_topic_sample.py
index a83ec797d273..d3bbec2e35a2 100644
--- a/sdk/eventgrid/azure-eventgrid/samples/publish_samples/publish_cloud_events_to_domain_topic_sample.py
+++ b/sdk/eventgrid/azure-eventgrid/samples/publish_samples/publish_cloud_events_to_domain_topic_sample.py
@@ -22,7 +22,8 @@
 import time
 
 from azure.core.credentials import AzureKeyCredential
-from azure.eventgrid import EventGridPublisherClient, CloudEvent
+from azure.core.messaging import CloudEvent
+from azure.eventgrid import EventGridPublisherClient
 
 domain_key = os.environ["DOMAIN_ACCESS_KEY"]
 domain_endpoint = os.environ["DOMAIN_TOPIC_HOSTNAME"]
diff --git a/sdk/eventgrid/azure-eventgrid/samples/publish_samples/publish_with_shared_access_signature_sample.py b/sdk/eventgrid/azure-eventgrid/samples/publish_samples/publish_with_shared_access_signature_sample.py
index f08a7d2345a4..8c0ee9cfc531 100644
--- a/sdk/eventgrid/azure-eventgrid/samples/publish_samples/publish_with_shared_access_signature_sample.py
+++ b/sdk/eventgrid/azure-eventgrid/samples/publish_samples/publish_with_shared_access_signature_sample.py
@@ -21,7 +21,8 @@
 
 from datetime import datetime, timedelta
 from azure.core.credentials import AzureSasCredential
-from azure.eventgrid import EventGridPublisherClient, CloudEvent, generate_sas
+from azure.core.messaging import CloudEvent
+from azure.eventgrid import EventGridPublisherClient, generate_sas
 
 key = os.environ["CLOUD_ACCESS_KEY"]
 endpoint = os.environ["CLOUD_TOPIC_HOSTNAME"]
diff --git a/sdk/eventgrid/azure-eventgrid/tests/_mocks.py b/sdk/eventgrid/azure-eventgrid/tests/_mocks.py
index 107d6d6175ed..08c9b5a63dca 100644
--- a/sdk/eventgrid/azure-eventgrid/tests/_mocks.py
+++ b/sdk/eventgrid/azure-eventgrid/tests/_mocks.py
@@ -1,6 +1,3 @@
-import json
-
-
 # storage cloud event
 cloud_storage_dict = {
     "id":"a0517898-9fa4-4e70-b4a3-afda1dd68672",
@@ -21,20 +18,27 @@
     "time":"2020-08-07T01:11:49.765846Z",
     "specversion":"1.0"
 }
-cloud_storage_string = json.dumps(cloud_storage_dict)
-cloud_storage_bytes = cloud_storage_string.encode("utf-8")
 
 # custom cloud event
-cloud_custom_dict = {
+cloud_custom_dict_base64 = {
     "id":"de0fd76c-4ef4-4dfb-ab3a-8f24a307e033",
     "source":"https://egtest.dev/cloudcustomevent",
-    "data":{"team": "event grid squad"},
+    "data_base64":'Y2xvdWRldmVudA==',
     "type":"Azure.Sdk.Sample",
     "time":"2020-08-07T02:06:08.11969Z",
     "specversion":"1.0"
 }
-cloud_custom_string = json.dumps(cloud_custom_dict)
-cloud_custom_bytes = cloud_custom_string.encode("utf-8")
+
+cloud_custom_dict_with_extensions = {
+    "id":"de0fd76c-4ef4-4dfb-ab3a-8f24a307e033",
+    "source":"https://egtest.dev/cloudcustomevent",
+    "data":{"team": "event grid squad"},
+    "type":"Azure.Sdk.Sample",
+    "time":"2020-08-07T02:06:08.11969Z",
+    "specversion":"1.0",
+    "ext1": "example",
+    "ext2": "example2"
+}
 
 # storage eg event
 eg_storage_dict = {
@@ -58,20 +62,3 @@
     "eventTime":"2020-08-07T02:28:23.867525Z",
     "topic":"/subscriptions/faa080af-c1d8-40ad-9cce-e1a450ca5b57/resourceGroups/t-swpill-test/providers/Microsoft.EventGrid/topics/eventgridegsub"
 }
-
-eg_storage_string = json.dumps(eg_storage_dict)
-eg_storage_bytes = eg_storage_string.encode("utf-8")
-
-# custom eg event
-eg_custom_dict = {
-    "id":"3a30afef-b604-4b67-973e-7dfff7e178a7",
-    "subject":"Test EG Custom Event",
-    "data":{"team":"event grid squad"},
-    "eventType":"Azure.Sdk.Sample",
-    "dataVersion":"2.0",
-    "metadataVersion":"1",
-    "eventTime":"2020-08-07T02:19:05.16916Z",
-    "topic":"/subscriptions/f8aa80ae-d1c8-60ad-9bce-e1a850ba5b67/resourceGroups/sample-resource-group-test/providers/Microsoft.EventGrid/topics/egtopicsamplesub"
-}
-eg_custom_string = json.dumps(eg_custom_dict)
-eg_custom_bytes = eg_custom_string.encode("utf-8")
diff --git a/sdk/eventgrid/azure-eventgrid/tests/test_cloud_event_tracing.py b/sdk/eventgrid/azure-eventgrid/tests/test_cloud_event_tracing.py
index 9032189d0658..a4a47e64b341 100644
--- a/sdk/eventgrid/azure-eventgrid/tests/test_cloud_event_tracing.py
+++ b/sdk/eventgrid/azure-eventgrid/tests/test_cloud_event_tracing.py
@@ -12,7 +12,7 @@
     PipelineContext
 )
 from azure.core.pipeline.transport import HttpRequest
-from azure.eventgrid import CloudEvent
+from azure.core.messaging import CloudEvent
 from azure.eventgrid._policies import CloudEventDistributedTracingPolicy
 from _mocks import (
     cloud_storage_dict
@@ -23,45 +23,43 @@
 _tracestate_value = "rojo=00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01,congo=lZWRzIHRoNhcm5hbCBwbGVhc3VyZS4"
 
 
-class EventGridSerializationTests(object):
+def test_cloud_event_policy_copies():
+    policy = CloudEventDistributedTracingPolicy()
 
-    def test_cloud_event_policy_copies(self):
-        policy = CloudEventDistributedTracingPolicy()
+    body = json.dumps([cloud_storage_dict])
+    universal_request = HttpRequest('POST', 'http://127.0.0.1/', data=body)
+    universal_request.headers['content-type'] = _content_type
+    universal_request.headers['traceparent'] = _traceparent_value
+    universal_request.headers['tracestate'] = _tracestate_value
 
-        body = json.dumps([cloud_storage_dict])
-        universal_request = HttpRequest('POST', 'http://127.0.0.1/', data=body)
-        universal_request.headers['content-type'] = _content_type
-        universal_request.headers['traceparent'] = _traceparent_value
-        universal_request.headers['tracestate'] = _tracestate_value
+    request = PipelineRequest(universal_request, PipelineContext(None))
 
-        request = PipelineRequest(universal_request, PipelineContext(None))
+    resp = policy.on_request(request)
 
-        resp = policy.on_request(request)
+    body = json.loads(request.http_request.body)
 
-        body = json.loads(request.http_request.body)
-    
-        for item in body:
-            assert 'traceparent' in item
-            assert 'tracestate' in item
+    for item in body:
+        assert 'traceparent' in item
+        assert 'tracestate' in item
 
-    def test_cloud_event_policy_no_copy_if_trace_exists(self):
-        policy = CloudEventDistributedTracingPolicy()
+def test_cloud_event_policy_no_copy_if_trace_exists():
+    policy = CloudEventDistributedTracingPolicy()
 
-        cloud_storage_dict.update({'traceparent': 'exists', 'tracestate': 'state_exists'})
-        body = json.dumps([cloud_storage_dict])
-        universal_request = HttpRequest('POST', 'http://127.0.0.1/', data=body)
-        universal_request.headers['content-type'] = _content_type
-        universal_request.headers['traceparent'] = _traceparent_value
-        universal_request.headers['tracestate'] = _tracestate_value
+    cloud_storage_dict.update({'traceparent': 'exists', 'tracestate': 'state_exists'})
+    body = json.dumps([cloud_storage_dict])
+    universal_request = HttpRequest('POST', 'http://127.0.0.1/', data=body)
+    universal_request.headers['content-type'] = _content_type
+    universal_request.headers['traceparent'] = _traceparent_value
+    universal_request.headers['tracestate'] = _tracestate_value
 
-        request = PipelineRequest(universal_request, PipelineContext(None))
+    request = PipelineRequest(universal_request, PipelineContext(None))
 
-        resp = policy.on_request(request)
+    resp = policy.on_request(request)
 
-        body = json.loads(request.http_request.body)
-    
-        for item in body:
-            assert 'traceparent' in item
-            assert 'tracestate' in item 
-            assert item['traceparent'] == 'exists'
-            assert item['tracestate'] == 'state_exists'
+    body = json.loads(request.http_request.body)
+
+    for item in body:
+        assert 'traceparent' in item
+        assert 'tracestate' in item 
+        assert item['traceparent'] == 'exists'
+        assert item['tracestate'] == 'state_exists'
diff --git a/sdk/eventgrid/azure-eventgrid/tests/test_eg_publisher_client.py b/sdk/eventgrid/azure-eventgrid/tests/test_eg_publisher_client.py
index efd2458cf3d7..0c4afab168b4 100644
--- a/sdk/eventgrid/azure-eventgrid/tests/test_eg_publisher_client.py
+++ b/sdk/eventgrid/azure-eventgrid/tests/test_eg_publisher_client.py
@@ -18,7 +18,9 @@
 
 from azure_devtools.scenario_tests import ReplayableTest
 from azure.core.credentials import AzureKeyCredential, AzureSasCredential
-from azure.eventgrid import EventGridPublisherClient, CloudEvent, EventGridEvent, generate_sas
+from azure.core.messaging import CloudEvent
+from azure.eventgrid import EventGridPublisherClient, EventGridEvent, generate_sas
+from azure.eventgrid._helpers import _cloud_event_to_generated
 
 from eventgrid_preparer import (
     CachedEventGridTopicPreparer
@@ -132,34 +134,6 @@ def callback(request):
 
         client.send(cloud_event, raw_response_hook=callback)
 
-    @CachedResourceGroupPreparer(name_prefix='eventgridtest')
-    @CachedEventGridTopicPreparer(name_prefix='cloudeventgridtest')
-    def test_send_cloud_event_bytes_using_data_base64(self, resource_group, eventgrid_topic, eventgrid_topic_primary_key, eventgrid_topic_endpoint):
-        akc_credential = AzureKeyCredential(eventgrid_topic_primary_key)
-        client = EventGridPublisherClient(eventgrid_topic_endpoint, akc_credential)
-        cloud_event = CloudEvent(
-                source = "http://samplesource.dev",
-                data_base64 = b'cloudevent',
-                type="Sample.Cloud.Event"
-                )
-
-        def callback(request):
-            req = json.loads(request.http_request.body)
-            assert req[0].get("data_base64") is not None
-            assert req[0].get("data") is None
-
-        client.send(cloud_event, raw_response_hook=callback)
-
-
-    def test_send_cloud_event_fails_on_providing_data_and_b64(self):
-        with pytest.raises(ValueError, match="data and data_base64 cannot be provided at the same time*"):
-            cloud_event = CloudEvent(
-                    source = "http://samplesource.dev",
-                    data_base64 = b'cloudevent',
-                    data = "random data",
-                    type="Sample.Cloud.Event"
-                    )
-
     @CachedResourceGroupPreparer(name_prefix='eventgridtest')
     @CachedEventGridTopicPreparer(name_prefix='cloudeventgridtest')
     def test_send_cloud_event_data_none(self, resource_group, eventgrid_topic, eventgrid_topic_primary_key, eventgrid_topic_endpoint):
@@ -223,7 +197,7 @@ def test_send_cloud_event_data_with_extensions(self, resource_group, eventgrid_t
                     }
                 )
         client.send([cloud_event])
-        internal = cloud_event._to_generated().serialize()
+        internal = _cloud_event_to_generated(cloud_event).serialize()
         assert 'reason_code' in internal
         assert 'extension' in internal
         assert internal['reason_code'] == 204
diff --git a/sdk/eventgrid/azure-eventgrid/tests/test_eg_publisher_client_async.py b/sdk/eventgrid/azure-eventgrid/tests/test_eg_publisher_client_async.py
index 5a9152432c05..a7b50199c4cc 100644
--- a/sdk/eventgrid/azure-eventgrid/tests/test_eg_publisher_client_async.py
+++ b/sdk/eventgrid/azure-eventgrid/tests/test_eg_publisher_client_async.py
@@ -16,10 +16,11 @@
 from devtools_testutils import AzureMgmtTestCase, CachedResourceGroupPreparer
 
 from azure_devtools.scenario_tests import ReplayableTest
+from azure.core.messaging import CloudEvent
 from azure.core.credentials import AzureKeyCredential, AzureSasCredential
-from azure.eventgrid import CloudEvent, EventGridEvent, generate_sas
+from azure.eventgrid import EventGridEvent, generate_sas
 from azure.eventgrid.aio import EventGridPublisherClient
-
+from azure.eventgrid._helpers import _cloud_event_to_generated
 from eventgrid_preparer import (
     CachedEventGridTopicPreparer
 )
@@ -179,7 +180,7 @@ async def test_send_cloud_event_data_with_extensions(self, resource_group, event
                     }
                 )
         await client.send([cloud_event])
-        internal = cloud_event._to_generated().serialize()
+        internal = _cloud_event_to_generated(cloud_event).serialize()
         assert 'reason_code' in internal
         assert 'extension' in internal
         assert internal['reason_code'] == 204
diff --git a/sdk/eventgrid/azure-eventgrid/tests/test_from_dict.py b/sdk/eventgrid/azure-eventgrid/tests/test_from_dict.py
new file mode 100644
index 000000000000..55b851d543b8
--- /dev/null
+++ b/sdk/eventgrid/azure-eventgrid/tests/test_from_dict.py
@@ -0,0 +1,20 @@
+import logging
+import sys
+import os
+import pytest
+import json
+
+from azure.eventgrid import EventGridEvent
+from devtools_testutils import AzureMgmtTestCase
+from _mocks import (
+    eg_storage_dict
+)
+
+class EventGridDeserializerTests(AzureMgmtTestCase):
+    
+    # EG Event tests
+
+    def test_eg_storage_from_dict(self, **kwargs):
+        event = EventGridEvent.from_dict(eg_storage_dict)
+        assert event.__class__ == EventGridEvent
+        assert event.event_type == "Microsoft.Storage.BlobCreated"
diff --git a/sdk/eventgrid/azure-eventgrid/tests/test_serialization.py b/sdk/eventgrid/azure-eventgrid/tests/test_serialization.py
index 5bed9c1c9212..f639571d640c 100644
--- a/sdk/eventgrid/azure-eventgrid/tests/test_serialization.py
+++ b/sdk/eventgrid/azure-eventgrid/tests/test_serialization.py
@@ -14,14 +14,11 @@
 
 from devtools_testutils import AzureMgmtTestCase
 from msrest.serialization import UTC
-from azure.eventgrid import CloudEvent, EventGridEvent
+from azure.core.messaging import CloudEvent
+from azure.eventgrid import EventGridEvent
 from azure.eventgrid._generated import models as internal_models
+from azure.eventgrid._helpers import _cloud_event_to_generated
 from azure.eventgrid import SystemEventNames
-from _mocks import (
-    cloud_storage_dict,
-    cloud_storage_string,
-    cloud_storage_bytes,
-    )
 
 class EventGridSerializationTests(AzureMgmtTestCase):
 
@@ -45,7 +42,7 @@ def test_cloud_event_serialization_extension_bytes(self, **kwargs):
         
         cloud_event.subject = "subject" # to test explicit setting of prop
         encoded = base64.b64encode(data).decode('utf-8')
-        internal = cloud_event._to_generated()
+        internal = _cloud_event_to_generated(cloud_event)
 
         assert internal.additional_properties is not None
         assert 'foo' not in internal.additional_properties
@@ -77,7 +74,7 @@ def test_cloud_event_serialization_extension_string(self, **kwargs):
                 )
         
         cloud_event.subject = "subject" # to test explicit setting of prop
-        internal = cloud_event._to_generated()
+        internal = _cloud_event_to_generated(cloud_event)
 
         assert internal.additional_properties is not None
         assert 'foo' not in internal.additional_properties