diff --git a/CHANGELOG.md b/CHANGELOG.md
index 3fe0914d..2db0c5ff 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -10,6 +10,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - Updated dev and build-time dependencies.
 
+### Fixed
+
+- Fixed an issue where setting a non-JSON datacontenttype (e.g., application/octet-stream) with dict data produced non-JSON-decodable output; now best-effort json-encoding is applied regardless of datacontenttype. ([#291])
+- Updated behavior when datacontenttype is unset: now treats events as application/json in line with spec recommendation. ([#291])
+
 ## [2.1.0]
 
 ### Added
diff --git a/src/cloudevents/core/formats/json.py b/src/cloudevents/core/formats/json.py
index 9ac0e44a..f20e0538 100644
--- a/src/cloudevents/core/formats/json.py
+++ b/src/cloudevents/core/formats/json.py
@@ -44,6 +44,7 @@ def default(self, obj: Any) -> Any:
 
 class JSONFormat(Format):
     CONTENT_TYPE: Final[str] = "application/cloudevents+json"
+    DEFAULT_CONTENT_TYPE: Final[str] = "application/json"
     JSON_CONTENT_TYPE_PATTERN: Pattern[str] = re.compile(
         r"^(application|text)/([a-zA-Z0-9\-\.]+\+)?json(;.*)?$"
     )
@@ -135,7 +136,9 @@ def write(self, event: BaseCloudEvent) -> bytes:
                         "utf-8"
                     )
             else:
-                datacontenttype = event_dict.get("datacontenttype", "application/json")
+                datacontenttype = event_dict.get(
+                    "datacontenttype", self.DEFAULT_CONTENT_TYPE
+                )
                 if re.match(JSONFormat.JSON_CONTENT_TYPE_PATTERN, datacontenttype):
                     event_dict["data"] = event_data
                 else:
@@ -171,12 +174,19 @@ def write_data(
 
         # If data is a dict and content type is JSON, serialize as JSON
         if isinstance(data, dict):
-            if datacontenttype and re.match(
-                JSONFormat.JSON_CONTENT_TYPE_PATTERN, datacontenttype
-            ):
+            content_type = datacontenttype or self.DEFAULT_CONTENT_TYPE
+            if re.match(JSONFormat.JSON_CONTENT_TYPE_PATTERN, content_type):
                 return dumps(data, cls=_JSONEncoderWithDatetime).encode("utf-8")
 
-        # Default: convert to string and encode
+            # for other contenttypes we still try to generate a json-decodable string
+            # if not possible, a string representing
+            try:
+                return dumps(data, cls=_JSONEncoderWithDatetime).encode("utf-8")
+            except TypeError:
+                pass
+
+        # according to the spec, we return an encoded string per default
+        # careful: the result is not json-decodable as the dict keys are single-quoted
         return str(data).encode("utf-8")
 
     def read_data(
@@ -196,9 +206,8 @@ def read_data(
             return None
 
         # If content type indicates JSON, try to parse as JSON
-        if datacontenttype and re.match(
-            JSONFormat.JSON_CONTENT_TYPE_PATTERN, datacontenttype
-        ):
+        content_type = datacontenttype or self.DEFAULT_CONTENT_TYPE
+        if re.match(JSONFormat.JSON_CONTENT_TYPE_PATTERN, content_type):
             try:
                 decoded = body.decode("utf-8")
                 parsed: dict[str, Any] = loads(decoded)
diff --git a/tests/test_core/test_format/test_json.py b/tests/test_core/test_format/test_json.py
index 12f75435..9971779b 100644
--- a/tests/test_core/test_format/test_json.py
+++ b/tests/test_core/test_format/test_json.py
@@ -12,8 +12,10 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
-
 from datetime import datetime, timezone
+from json import loads
+
+import pytest
 
 from cloudevents.core.formats.json import JSONFormat
 from cloudevents.core.v1.event import CloudEvent
@@ -323,3 +325,24 @@ def test_read_cloud_event_from_string_input() -> None:
 
     assert result.get_id() == "123"
     assert result.get_source() == "source"
+
+
+@pytest.mark.parametrize(
+    "content_type", [None, "application/json", "application/octet-stream"]
+)
+def test_write_data_dict(content_type: str) -> None:
+    formatter = JSONFormat()
+    data = {"key": "value", "nested": {"a": 1}}
+    result = formatter.write_data(data, datacontenttype=content_type)
+
+    assert isinstance(result, bytes)
+    assert loads(result) == data
+
+
+@pytest.mark.parametrize("content_type", [None, "application/json"])
+def test_read_data_json_body(content_type: str) -> None:
+    formatter = JSONFormat()
+    body = b'{"key": "value"}'
+    result = formatter.read_data(body, content_type)
+
+    assert result == {"key": "value"}