diff --git a/azure/functions/_abc.py b/azure/functions/_abc.py
index 2f5cb69..c87b1f9 100644
--- a/azure/functions/_abc.py
+++ b/azure/functions/_abc.py
@@ -4,6 +4,7 @@
 import abc
 import datetime
 import io
+from ._jsonutils import json
 import threading
 import typing
 
@@ -12,6 +13,45 @@
 T = typing.TypeVar('T')
 
 
+class Serializable(abc.ABC):
+    """Mixin that provides a uniform serialization contract for binding types.
+
+    Binding classes should override :meth:`to_dict` to return a JSON-safe
+    representation.  A default :meth:`to_json` implementation is provided
+    that serializes the result of :meth:`to_dict` to a JSON string.
+
+    :meth:`to_dict` is intentionally non-abstract so that downstream
+    subclasses of the binding ABCs that predate this contract are not broken
+    at instantiation time.  Calling ``to_dict()`` on a class that has not
+    implemented it raises :exc:`NotImplementedError` at call time.
+    """
+
+    def to_dict(self) -> typing.Any:
+        """Return a JSON-safe representation of this binding object.
+
+        Non-list binding types return a ``dict``; collection types such as
+        ``DocumentList`` or ``SqlRowList`` return a ``list`` of ``dict``.
+        All ``bytes`` values are decoded as UTF-8 when possible, otherwise
+        represented as ``{"__encoding": "base64", "value": "<base64>"}``.
+        All ``datetime`` values are represented as ISO-8601 strings.
+
+        Subclasses that do not override this method will raise
+        :exc:`NotImplementedError` at call time rather than at instantiation.
+        """
+        raise NotImplementedError(
+            f"{type(self).__name__} does not implement to_dict(). "
+            "Override this method to enable serialization."
+        )
+
+    def to_json(self) -> str:
+        """Return the JSON string representation of this binding object.
+
+        Delegates to :meth:`to_dict`; subclasses should override
+        :meth:`to_dict` rather than this method.
+        """
+        return json.dumps(self.to_dict())
+
+
 class Out(abc.ABC, typing.Generic[T]):
     """An interface to set function output parameters."""
 
@@ -215,7 +255,7 @@ def get_body(self) -> bytes:
         pass
 
 
-class TimerRequest(abc.ABC):
+class TimerRequest(Serializable):
     """Timer request object."""
 
     @property
@@ -225,7 +265,7 @@ def past_due(self) -> bool:
         pass
 
 
-class InputStream(io.BufferedIOBase, abc.ABC):
+class InputStream(io.BufferedIOBase, Serializable):
     """File-like object representing an input blob."""
 
     @abc.abstractmethod
@@ -261,7 +301,7 @@ def uri(self) -> typing.Optional[str]:
         pass
 
 
-class QueueMessage(abc.ABC):
+class QueueMessage(Serializable):
 
     @property
     @abc.abstractmethod
@@ -302,7 +342,7 @@ def pop_receipt(self) -> typing.Optional[str]:
         pass
 
 
-class EventGridEvent(abc.ABC):
+class EventGridEvent(Serializable):
     @property
     @abc.abstractmethod
     def id(self) -> str:
@@ -338,7 +378,7 @@ def data_version(self) -> str:
         pass
 
 
-class EventGridOutputEvent(abc.ABC):
+class EventGridOutputEvent(Serializable):
     @property
     @abc.abstractmethod
     def id(self) -> str:
@@ -369,7 +409,7 @@ def data_version(self) -> str:
         pass
 
 
-class CloudEvent(abc.ABC):
+class CloudEvent(Serializable):
     """A CloudEvents v1.0 event message."""
 
     @property
@@ -417,7 +457,7 @@ def get_json(self) -> typing.Any:
         pass
 
 
-class Document(abc.ABC):
+class Document(Serializable):
 
     @classmethod
     @abc.abstractmethod
@@ -437,20 +477,16 @@ def __getitem__(self, key):
     def __setitem__(self, key, value):
         pass
 
-    @abc.abstractmethod
-    def to_json(self) -> str:
-        pass
-
     @abc.abstractmethod
     def to_dict(self) -> dict:
         pass
 
 
-class DocumentList(abc.ABC):
+class DocumentList(Serializable):
     pass
 
 
-class EventHubEvent(abc.ABC):
+class EventHubEvent(Serializable):
 
     @abc.abstractmethod
     def get_body(self) -> bytes:
@@ -489,7 +525,7 @@ def body(self) -> str:
         pass
 
 
-class ServiceBusMessage(abc.ABC):
+class ServiceBusMessage(Serializable):
 
     @abc.abstractmethod
     def get_body(self) -> typing.Union[str, bytes]:
diff --git a/azure/functions/_blob.py b/azure/functions/_blob.py
index 2cdf6ac..4ebd1bb 100644
--- a/azure/functions/_blob.py
+++ b/azure/functions/_blob.py
@@ -1,6 +1,7 @@
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # Licensed under the MIT License.
 
+import typing
 from typing import Optional
 
 from azure.functions import _abc as azf_abc
@@ -53,3 +54,14 @@ def read(self, size=-1) -> bytes:
             Bytes read from the input stream.
         """
         return self._io.read(size)
+
+    def to_dict(self) -> typing.Dict[str, typing.Any]:
+        """Return a JSON-safe dictionary of blob metadata fields.
+
+        Blob content is intentionally excluded; use :meth:`read` to access it.
+        """
+        return {
+            'name': self._name,
+            'uri': self._uri,
+            'length': self._length,
+        }
diff --git a/azure/functions/_cosmosdb.py b/azure/functions/_cosmosdb.py
index dcdf01c..941313c 100644
--- a/azure/functions/_cosmosdb.py
+++ b/azure/functions/_cosmosdb.py
@@ -46,4 +46,7 @@ def __repr__(self) -> str:
 
 class DocumentList(_abc.DocumentList, collections.UserList):
     "A ``UserList`` subclass containing a list of :class:`~Document` objects"
-    pass
+
+    def to_dict(self):
+        """Return a list of JSON-safe dicts, one per :class:`~Document`."""
+        return [doc.to_dict() for doc in self]
diff --git a/azure/functions/_eventgrid.py b/azure/functions/_eventgrid.py
index 9eb4092..4fa1ada 100644
--- a/azure/functions/_eventgrid.py
+++ b/azure/functions/_eventgrid.py
@@ -5,6 +5,7 @@
 import typing
 
 from azure.functions import _abc as azf_abc
+from ._utils import _serialize_value
 
 
 class EventGridEvent(azf_abc.EventGridEvent):
@@ -61,6 +62,21 @@ def __repr__(self) -> str:
             f'at 0x{id(self):0x}>'
         )
 
+    def to_dict(self) -> typing.Dict[str, typing.Any]:
+        """Return a JSON-safe dictionary of all EventGrid event fields.
+
+        ``event_time`` is represented as an ISO-8601 string.
+        """
+        return {
+            'id': self.id,
+            'topic': self.topic,
+            'subject': self.subject,
+            'event_type': self.event_type,
+            'event_time': _serialize_value(self.event_time),
+            'data_version': self.data_version,
+            'data': _serialize_value(self.get_json()),
+        }
+
 
 class EventGridOutputEvent(azf_abc.EventGridOutputEvent):
     """An EventGrid event message."""
@@ -110,6 +126,20 @@ def __repr__(self) -> str:
             f'at 0x{id(self):0x}>'
         )
 
+    def to_dict(self) -> typing.Dict[str, typing.Any]:
+        """Return a JSON-safe dictionary of all EventGrid output event fields.
+
+        ``event_time`` is represented as an ISO-8601 string.
+        """
+        return {
+            'id': self.id,
+            'subject': self.subject,
+            'event_type': self.event_type,
+            'event_time': _serialize_value(self.event_time),
+            'data_version': self.data_version,
+            'data': _serialize_value(self.get_json()),
+        }
+
 
 class CloudEvent(azf_abc.CloudEvent):
     """A CloudEvents v1.0 event message."""
@@ -182,3 +212,21 @@ def __repr__(self) -> str:
             f'type={self.type} '
             f'at 0x{id(self):0x}>'
         )
+
+    def to_dict(self) -> typing.Dict[str, typing.Any]:
+        """Return a JSON-safe dictionary of all CloudEvent fields.
+
+        ``time`` is represented as an ISO-8601 string.
+        """
+        return {
+            'id': self.id,
+            'source': self.source,
+            'type': self.type,
+            'specversion': self.specversion,
+            'subject': self.subject,
+            'time': _serialize_value(self.time),
+            'datacontenttype': self.datacontenttype,
+            'dataschema': self.dataschema,
+            'data': _serialize_value(self.get_json()),
+            'extension_attrs': _serialize_value(self.extension_attrs),
+        }
diff --git a/azure/functions/_eventhub.py b/azure/functions/_eventhub.py
index 47b25e1..ef483ff 100644
--- a/azure/functions/_eventhub.py
+++ b/azure/functions/_eventhub.py
@@ -6,6 +6,7 @@
 
 from azure.functions import _abc as func_abc
 from azure.functions import meta
+from ._utils import _serialize_value
 
 
 class EventHubEvent(func_abc.EventHubEvent):
@@ -80,6 +81,23 @@ def metadata(self) -> typing.Optional[typing.Mapping[str, typing.Any]]:
             }
         return self._trigger_metadata_pyobj
 
+    def to_dict(self) -> typing.Dict[str, typing.Any]:
+        """Return a JSON-safe dictionary of all Event Hub event fields.
+
+        ``body`` bytes are decoded as UTF-8 when valid, otherwise represented
+        as ``{"__encoding": "base64", "value": "<base64>"}``.
+        ``enqueued_time`` is an ISO-8601 string.
+        """
+        return {
+            'body': _serialize_value(self.get_body()),
+            'partition_key': self.partition_key,
+            'sequence_number': self.sequence_number,
+            'offset': self.offset,
+            'enqueued_time': _serialize_value(self.enqueued_time),
+            'iothub_metadata': (dict(self.iothub_metadata)
+                                if self.iothub_metadata else None),
+        }
+
     def __repr__(self) -> str:
         return (
             f'<azure.EventHubEvent '
diff --git a/azure/functions/_kafka.py b/azure/functions/_kafka.py
index 8d3b3ec..54e782e 100644
--- a/azure/functions/_kafka.py
+++ b/azure/functions/_kafka.py
@@ -4,8 +4,10 @@
 import abc
 import typing
 
+from . import _abc
 
-class AbstractKafkaEvent(abc.ABC):
+
+class AbstractKafkaEvent(_abc.Serializable):
 
     @abc.abstractmethod
     def get_body(self) -> bytes:
diff --git a/azure/functions/_mysql.py b/azure/functions/_mysql.py
index 1f6b407..d2bac0a 100644
--- a/azure/functions/_mysql.py
+++ b/azure/functions/_mysql.py
@@ -4,6 +4,7 @@
 import collections
 
 from ._jsonutils import json
+from ._utils import _serialize_value
 
 
 class BaseMySqlRow(abc.ABC):
@@ -53,7 +54,11 @@ def from_dict(cls, dct: dict) -> 'BaseMySqlRow':
 
     def to_json(self) -> str:
         """Return the JSON representation of the MySqlRow"""
-        return json.dumps(dict(self))
+        return json.dumps(self.to_dict())
+
+    def to_dict(self) -> dict:
+        """Return the MySqlRow as a plain dict."""
+        return _serialize_value(dict(self))
 
     def __getitem__(self, key):
         return collections.UserDict.__getitem__(self, key)
@@ -69,4 +74,11 @@ def __repr__(self) -> str:
 
 class MySqlRowList(BaseMySqlRowList, collections.UserList):
     "A ''UserList'' subclass containing a list of :class:'~MySqlRow' objects"
-    pass
+
+    def to_dict(self):
+        """Return a list of plain dicts, one per :class:`~MySqlRow`."""
+        return [row.to_dict() for row in self]
+
+    def to_json(self) -> str:
+        """Return the JSON representation of the MySqlRowList."""
+        return json.dumps(self.to_dict())
diff --git a/azure/functions/_queue.py b/azure/functions/_queue.py
index 85d574c..4f21596 100644
--- a/azure/functions/_queue.py
+++ b/azure/functions/_queue.py
@@ -6,6 +6,7 @@
 
 from . import _abc
 from ._jsonutils import json
+from ._utils import _serialize_value
 
 
 class QueueMessage(_abc.QueueMessage):
@@ -88,6 +89,23 @@ def get_json(self) -> typing.Any:
         """
         return json.loads(self.__body)
 
+    def to_dict(self) -> typing.Dict[str, typing.Any]:
+        """Return a JSON-safe dictionary of all message fields.
+
+        ``body`` bytes are decoded as UTF-8 when valid, otherwise represented
+        as ``{"__encoding": "base64", "value": "<base64>"}``.
+        Datetime fields are ISO-8601 strings.
+        """
+        return {
+            'id': self.id,
+            'body': _serialize_value(self.get_body()),
+            'pop_receipt': self.pop_receipt,
+            'dequeue_count': self.dequeue_count,
+            'expiration_time': _serialize_value(self.expiration_time),
+            'insertion_time': _serialize_value(self.insertion_time),
+            'time_next_visible': _serialize_value(self.time_next_visible),
+        }
+
     def __repr__(self) -> str:
         return (
             f'<azure.QueueMessage id={self.id} at 0x{id(self):0x}>'
diff --git a/azure/functions/_servicebus.py b/azure/functions/_servicebus.py
index fb45c24..e7f793f 100644
--- a/azure/functions/_servicebus.py
+++ b/azure/functions/_servicebus.py
@@ -5,6 +5,7 @@
 from typing import Optional, Dict, Any, Union
 
 from . import _abc
+from ._utils import _serialize_value
 
 
 class ServiceBusMessage(_abc.ServiceBusMessage):
@@ -464,3 +465,43 @@ def __set_body(self, body):
     def get_body(self) -> bytes:
         """Return message content as bytes."""
         return self.__body
+
+    def to_dict(self) -> Dict[str, Any]:
+        """Return a JSON-safe dictionary of all Service Bus message fields.
+
+        ``body`` bytes are decoded as UTF-8 when valid, otherwise represented
+        as ``{"__encoding": "base64", "value": "<base64>"}``.
+        Datetime fields are ISO-8601 strings; ``time_to_live`` is a string
+        representation of the ``timedelta``.
+        """
+        ttl = self.time_to_live
+        return {
+            'body': _serialize_value(self.get_body()),
+            'application_properties': _serialize_value(self.application_properties),
+            'content_type': self.content_type,
+            'correlation_id': self.correlation_id,
+            'dead_letter_error_description': self.dead_letter_error_description,
+            'dead_letter_reason': self.dead_letter_reason,
+            'dead_letter_source': self.dead_letter_source,
+            'delivery_count': self.delivery_count,
+            'enqueued_sequence_number': self.enqueued_sequence_number,
+            'enqueued_time_utc': _serialize_value(self.enqueued_time_utc),
+            'expires_at_utc': _serialize_value(self.expires_at_utc),
+            'label': self.label,
+            'locked_until': _serialize_value(self.locked_until),
+            'lock_token': self.lock_token,
+            'message_id': self.message_id,
+            'partition_key': self.partition_key,
+            'reply_to': self.reply_to,
+            'reply_to_session_id': self.reply_to_session_id,
+            'scheduled_enqueue_time_utc': _serialize_value(
+                self.scheduled_enqueue_time_utc),
+            'sequence_number': self.sequence_number,
+            'session_id': self.session_id,
+            'state': self.state,
+            'subject': self.subject,
+            'time_to_live': str(ttl) if ttl is not None else None,
+            'to': self.to,
+            'transaction_partition_key': self.transaction_partition_key,
+            'user_properties': _serialize_value(self.user_properties),
+        }
diff --git a/azure/functions/_sql.py b/azure/functions/_sql.py
index 166b3af..a9769b4 100644
--- a/azure/functions/_sql.py
+++ b/azure/functions/_sql.py
@@ -4,6 +4,7 @@
 import collections
 
 from ._jsonutils import json
+from ._utils import _serialize_value
 
 
 class BaseSqlRow(abc.ABC):
@@ -53,7 +54,11 @@ def from_dict(cls, dct: dict) -> 'BaseSqlRow':
 
     def to_json(self) -> str:
         """Return the JSON representation of the SqlRow"""
-        return json.dumps(dict(self))
+        return json.dumps(self.to_dict())
+
+    def to_dict(self) -> dict:
+        """Return the SqlRow as a plain dict."""
+        return _serialize_value(dict(self))
 
     def __getitem__(self, key):
         return collections.UserDict.__getitem__(self, key)
@@ -69,4 +74,11 @@ def __repr__(self) -> str:
 
 class SqlRowList(BaseSqlRowList, collections.UserList):
     "A ''UserList'' subclass containing a list of :class:'~SqlRow' objects"
-    pass
+
+    def to_dict(self):
+        """Return a list of plain dicts, one per :class:`~SqlRow`."""
+        return [row.to_dict() for row in self]
+
+    def to_json(self) -> str:
+        """Return the JSON representation of the SqlRowList."""
+        return json.dumps(self.to_dict())
diff --git a/azure/functions/_timer.py b/azure/functions/_timer.py
index 191b14c..98ccbcf 100644
--- a/azure/functions/_timer.py
+++ b/azure/functions/_timer.py
@@ -18,3 +18,9 @@ def __init__(self, *, past_due: bool = False) -> None:
     def past_due(self) -> bool:
         """Whether the timer is past due."""
         return self.__past_due
+
+    def to_dict(self):
+        """Return a JSON-safe dictionary of timer request fields."""
+        return {
+            'past_due': self.past_due,
+        }
diff --git a/azure/functions/_utils.py b/azure/functions/_utils.py
index ebfe82a..1d9d401 100644
--- a/azure/functions/_utils.py
+++ b/azure/functions/_utils.py
@@ -1,10 +1,43 @@
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # Licensed under the MIT License.
 
-from typing import List, Tuple, Optional
+import base64
+import datetime as _dt
+from typing import Any, List, Tuple, Optional
 from datetime import datetime, timedelta
 
 
+def _serialize_value(v: Any) -> Any:
+    """Normalize a single field value to a JSON-safe type.
+
+    - ``bytes``: decoded as UTF-8 when valid, otherwise base64-encoded as
+      ``{"__encoding": "base64", "value": "<base64>"}``.
+    - ``datetime.datetime`` / ``datetime.date``: ISO-8601 string via
+      ``.isoformat()``.
+    - ``datetime.timedelta``: ``str(v)`` (e.g. ``"0:05:00"``).
+    - ``dict`` / ``list`` / ``tuple``: normalized recursively by applying this
+      function to nested values.
+    - All other types: returned unchanged.
+    """
+    if isinstance(v, bytes):
+        try:
+            return v.decode('utf-8')
+        except UnicodeDecodeError:
+            return {
+                '__encoding': 'base64',
+                'value': base64.b64encode(v).decode('ascii'),
+            }
+    if isinstance(v, (_dt.datetime, _dt.date)):
+        return v.isoformat()
+    if isinstance(v, _dt.timedelta):
+        return str(v)
+    if isinstance(v, dict):
+        return {k: _serialize_value(val) for k, val in v.items()}
+    if isinstance(v, (list, tuple)):
+        return [_serialize_value(item) for item in v]
+    return v
+
+
 def try_parse_datetime_with_formats(
     datetime_str: str,
     datetime_formats: List[str]
diff --git a/azure/functions/blob.py b/azure/functions/blob.py
index 1893507..a404bc5 100644
--- a/azure/functions/blob.py
+++ b/azure/functions/blob.py
@@ -2,11 +2,13 @@
 # Licensed under the MIT License.
 
 import io
+import typing
 from typing import Any, Optional, Union
 
 from azure.functions import _abc as azf_abc
 from azure.functions import _blob as azf_blob
 from . import meta
+from ._utils import _serialize_value
 
 
 class InputStream(azf_blob.InputStream):
@@ -56,6 +58,19 @@ def seekable(self) -> bool:
     def writable(self) -> bool:
         return False
 
+    def to_dict(self) -> typing.Dict[str, typing.Any]:
+        """Return a JSON-safe dictionary of blob metadata fields.
+
+        Blob content is intentionally excluded; use :meth:`read` to access it.
+        """
+        return {
+            'name': self._name,
+            'uri': self._uri,
+            'length': self._length,
+            'blob_properties': _serialize_value(self._blob_properties),
+            'metadata': _serialize_value(self._metadata),
+        }
+
 
 class BlobConverter(meta.InConverter,
                     meta.OutConverter,
diff --git a/azure/functions/kafka.py b/azure/functions/kafka.py
index f71bb2c..68efa14 100644
--- a/azure/functions/kafka.py
+++ b/azure/functions/kafka.py
@@ -9,6 +9,7 @@
 from ._jsonutils import json
 
 from ._kafka import AbstractKafkaEvent
+from ._utils import _serialize_value
 
 
 class KafkaEvent(AbstractKafkaEvent):
@@ -77,6 +78,22 @@ def metadata(self) -> typing.Optional[typing.Mapping[str, typing.Any]]:
 
         return self._trigger_metadata_pyobj
 
+    def to_dict(self) -> typing.Dict[str, typing.Any]:
+        """Return a JSON-safe dictionary of all Kafka event fields.
+
+        ``body`` bytes are decoded as UTF-8 when valid, otherwise represented
+        as ``{"__encoding": "base64", "value": "<base64>"}``.
+        """
+        return {
+            'body': _serialize_value(self.get_body()),
+            'key': self.key,
+            'offset': self.offset,
+            'partition': self.partition,
+            'topic': self.topic,
+            'timestamp': self.timestamp,
+            'headers': _serialize_value(self.headers),
+        }
+
     def __repr__(self) -> str:
         return (
             f'<azure.KafkaEvent '
diff --git a/azure/functions/timer.py b/azure/functions/timer.py
index 0d4fc36..7f1cc09 100644
--- a/azure/functions/timer.py
+++ b/azure/functions/timer.py
@@ -7,6 +7,7 @@
 from azure.functions import _timer as azf_timer
 from . import meta
 from ._jsonutils import json
+from ._utils import _serialize_value
 
 
 class TimerRequest(azf_timer.TimerRequest):
@@ -27,6 +28,15 @@ def schedule_status(self) -> dict:
     def schedule(self) -> dict:
         return self.__schedule
 
+    def to_dict(self) -> typing.Dict[str, typing.Any]:
+        """Return a JSON-safe dictionary of all timer request fields."""
+
+        return {
+            'past_due': self.past_due,
+            'schedule_status': _serialize_value(self.schedule_status),
+            'schedule': _serialize_value(self.schedule),
+        }
+
 
 class TimerRequestConverter(meta.InConverter,
                             binding='timerTrigger', trigger=True):
diff --git a/tests/test_serialization.py b/tests/test_serialization.py
new file mode 100644
index 0000000..6361ebd
--- /dev/null
+++ b/tests/test_serialization.py
@@ -0,0 +1,334 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+"""Tests for the uniform to_dict() / to_json() serialization contract."""
+
+import base64
+import datetime
+import json
+import unittest
+
+import azure.functions as func
+from azure.functions._utils import _serialize_value
+
+
+class TestSerializeValue(unittest.TestCase):
+    def test_bytes_utf8(self):
+        assert _serialize_value(b"hello") == "hello"
+
+    def test_bytes_non_utf8(self):
+        raw = bytes([0xFF, 0xFE])
+        result = _serialize_value(raw)
+        assert result == {
+            "__encoding": "base64",
+            "value": base64.b64encode(raw).decode("ascii"),
+        }
+
+    def test_datetime(self):
+        dt = datetime.datetime(2024, 1, 15, 12, 30, 0)
+        assert _serialize_value(dt) == "2024-01-15T12:30:00"
+
+    def test_timedelta(self):
+        td = datetime.timedelta(hours=1, minutes=30)
+        result = _serialize_value(td)
+        assert isinstance(result, str)
+        assert "1:30:00" in result
+
+    def test_passthrough_str(self):
+        assert _serialize_value("hello") == "hello"
+
+    def test_passthrough_int(self):
+        assert _serialize_value(42) == 42
+
+    def test_passthrough_none(self):
+        assert _serialize_value(None) is None
+
+
+class TestQueueMessageToDict(unittest.TestCase):
+    def test_to_dict_defaults(self):
+        msg = func.QueueMessage()
+        d = msg.to_dict()
+        assert d["id"] is None
+        assert d["body"] == ""  # b"" decodes to ""
+        assert d["pop_receipt"] is None
+        assert d["dequeue_count"] is None
+
+    def test_to_dict_with_body(self):
+        msg = func.QueueMessage(id="abc", body=b"hello")
+        d = msg.to_dict()
+        assert d["id"] == "abc"
+        assert d["body"] == "hello"
+
+    def test_to_dict_bytes_body_non_utf8(self):
+        raw = bytes([0xFF, 0xFE])
+        msg = func.QueueMessage(body=raw)
+        d = msg.to_dict()
+        assert d["body"]["__encoding"] == "base64"
+
+    def test_to_json_returns_string(self):
+        msg = func.QueueMessage(id="x", body=b"data")
+        s = msg.to_json()
+        assert isinstance(s, str)
+        parsed = json.loads(s)
+        assert parsed["id"] == "x"
+        assert parsed["body"] == "data"
+
+
+class TestEventHubEventToDict(unittest.TestCase):
+    def test_to_dict_basic(self):
+        from azure.functions._eventhub import EventHubEvent
+        evt = EventHubEvent(
+            body=b"payload",
+            partition_key="pk1",
+            sequence_number=42,
+            offset="100",
+            enqueued_time=datetime.datetime(2024, 6, 1, 10, 0, 0),
+        )
+        d = evt.to_dict()
+        assert d["body"] == "payload"
+        assert d["partition_key"] == "pk1"
+        assert d["sequence_number"] == 42
+        assert d["offset"] == "100"
+        assert d["enqueued_time"] == "2024-06-01T10:00:00"
+        assert d["iothub_metadata"] is None
+
+    def test_to_json_roundtrip(self):
+        from azure.functions._eventhub import EventHubEvent
+        evt = EventHubEvent(body=b"test", partition_key="pk")
+        parsed = json.loads(evt.to_json())
+        assert parsed["body"] == "test"
+        assert parsed["partition_key"] == "pk"
+
+
+class TestEventGridEventToDict(unittest.TestCase):
+    def test_to_dict(self):
+        evt = func.EventGridEvent(
+            id="evt1",
+            data={"key": "val"},
+            topic="/subscriptions/sub/providers/Microsoft.Storage",
+            subject="/blobServices/default/containers/test",
+            event_type="Microsoft.Storage.BlobCreated",
+            event_time=datetime.datetime(2024, 1, 1, 0, 0, 0),
+            data_version="1.0",
+        )
+        d = evt.to_dict()
+        assert d["id"] == "evt1"
+        assert d["topic"].startswith("/subscriptions")
+        assert d["event_time"] == "2024-01-01T00:00:00"
+        assert d["data"] == {"key": "val"}
+
+    def test_to_json(self):
+        evt = func.EventGridEvent(
+            id="e2",
+            data={},
+            topic="t",
+            subject="s",
+            event_type="et",
+            event_time=None,
+            data_version="1",
+        )
+        parsed = json.loads(evt.to_json())
+        assert parsed["id"] == "e2"
+        assert parsed["event_time"] is None
+
+
+class TestServiceBusMessageToDict(unittest.TestCase):
+    def test_stub_to_dict(self):
+        # The stub base class (_servicebus.ServiceBusMessage) also has to_dict
+        msg = func.ServiceBusMessage(body=b"hello")
+        d = msg.to_dict()
+        assert d["body"] == "hello"
+        assert d["message_id"] == ""  # stub default
+
+    def test_to_json(self):
+        msg = func.ServiceBusMessage(body=b"msg")
+        parsed = json.loads(msg.to_json())
+        assert parsed["body"] == "msg"
+
+
+class TestTimerRequestToDict(unittest.TestCase):
+    def test_base_timer_to_dict(self):
+        # The base _timer.TimerRequest
+        timer = func.TimerRequest(past_due=True)
+        d = timer.to_dict()
+        assert d["past_due"] is True
+
+    def test_to_json(self):
+        timer = func.TimerRequest()
+        parsed = json.loads(timer.to_json())
+        assert parsed["past_due"] is False
+
+
+class TestInputStreamToDict(unittest.TestCase):
+    def test_base_blob_to_dict(self):
+        blob = func.InputStream(name="myblob", uri="https://x/blob", length=42)
+        d = blob.to_dict()
+        assert d["name"] == "myblob"
+        assert d["uri"] == "https://x/blob"
+        assert d["length"] == 42
+
+    def test_to_json(self):
+        blob = func.InputStream(name="b", uri="u", length=10)
+        parsed = json.loads(blob.to_json())
+        assert parsed["name"] == "b"
+
+
+class TestDocumentToDict(unittest.TestCase):
+    def test_document_to_dict(self):
+        doc = func.Document.from_dict({"id": "1", "value": 42})
+        d = doc.to_dict()
+        assert d == {"id": "1", "value": 42}
+
+    def test_document_list_to_dict(self):
+        dl = func.DocumentList()
+        dl.append(func.Document.from_dict({"a": 1}))
+        dl.append(func.Document.from_dict({"b": 2}))
+        result = dl.to_dict()
+        assert result == [{"a": 1}, {"b": 2}]
+
+    def test_document_list_to_json(self):
+        dl = func.DocumentList()
+        dl.append(func.Document.from_dict({"x": 10}))
+        parsed = json.loads(dl.to_json())
+        assert parsed == [{"x": 10}]
+
+
+class TestSqlRowToDict(unittest.TestCase):
+    def test_sql_row_to_dict(self):
+        row = func.SqlRow.from_dict({"col1": "a", "col2": 1})
+        assert row.to_dict() == {"col1": "a", "col2": 1}
+
+    def test_sql_row_list_to_dict(self):
+        rl = func.SqlRowList()
+        rl.append(func.SqlRow.from_dict({"k": "v"}))
+        assert rl.to_dict() == [{"k": "v"}]
+
+    def test_sql_row_list_to_json(self):
+        rl = func.SqlRowList()
+        rl.append(func.SqlRow.from_dict({"n": 7}))
+        parsed = json.loads(rl.to_json())
+        assert parsed == [{"n": 7}]
+
+
+class TestMySqlRowToDict(unittest.TestCase):
+    def test_mysql_row_to_dict(self):
+        row = func.MySqlRow.from_dict({"col": "val"})
+        assert row.to_dict() == {"col": "val"}
+
+    def test_mysql_row_list_to_dict(self):
+        rl = func.MySqlRowList()
+        rl.append(func.MySqlRow.from_dict({"x": 5}))
+        assert rl.to_dict() == [{"x": 5}]
+
+    def test_mysql_row_list_to_json(self):
+        rl = func.MySqlRowList()
+        rl.append(func.MySqlRow.from_dict({"y": 9}))
+        parsed = json.loads(rl.to_json())
+        assert parsed == [{"y": 9}]
+
+
+class TestKafkaEventToDict(unittest.TestCase):
+    def test_to_dict(self):
+        from azure.functions.kafka import KafkaEvent
+        evt = KafkaEvent(
+            body=b"message",
+            key="key1",
+            offset=5,
+            partition=0,
+            topic="my-topic",
+            timestamp="2024-01-01T00:00:00",
+        )
+        d = evt.to_dict()
+        assert d["body"] == "message"
+        assert d["key"] == "key1"
+        assert d["topic"] == "my-topic"
+
+    def test_to_json(self):
+        from azure.functions.kafka import KafkaEvent
+        evt = KafkaEvent(body=b"k", topic="t", key="k1", offset=0, partition=1,
+                         timestamp="ts")
+        parsed = json.loads(evt.to_json())
+        assert parsed["body"] == "k"
+
+
+class TestCloudEventToDict(unittest.TestCase):
+    def test_to_dict(self):
+        evt = func.CloudEvent(
+            id="ce1",
+            source="https://example.com/source",
+            type="com.example.someevent",
+            specversion="1.0",
+            data={"key": "value"},
+            time=datetime.datetime(2024, 3, 1, 12, 0, 0),
+        )
+        d = evt.to_dict()
+        assert d["id"] == "ce1"
+        assert d["time"] == "2024-03-01T12:00:00"
+        assert d["data"] == {"key": "value"}
+
+    def test_to_json(self):
+        evt = func.CloudEvent(
+            id="ce2", source="s", type="t", specversion="1.0", data=None
+        )
+        parsed = json.loads(evt.to_json())
+        assert parsed["id"] == "ce2"
+
+
+class TestSerializableNonBreaking(unittest.TestCase):
+    """Verify that subclassing a binding ABC without implementing to_dict()
+    does not break instantiation - the error is deferred to call time."""
+
+    def _make_custom_queue_message(self):
+        """Return a minimal concrete subclass of _abc.QueueMessage that
+        implements only the pre-existing abstract surface, not to_dict()."""
+        import azure.functions._abc as azf_abc
+
+        class CustomQueueMessage(azf_abc.QueueMessage):
+            @property
+            def id(self):
+                return "x"
+
+            def get_body(self):
+                return b"body"
+
+            def get_json(self):
+                return "body"
+
+            @property
+            def dequeue_count(self):
+                return None
+
+            @property
+            def expiration_time(self):
+                return None
+
+            @property
+            def insertion_time(self):
+                return None
+
+            @property
+            def time_next_visible(self):
+                return None
+
+            @property
+            def pop_receipt(self):
+                return None
+
+        return CustomQueueMessage
+
+    def test_subclass_without_to_dict_can_be_instantiated(self):
+        # Must not raise TypeError at instantiation time.
+        cls = self._make_custom_queue_message()
+        msg = cls()
+        assert msg.id == "x"
+
+    def test_subclass_without_to_dict_raises_at_call_time(self):
+        cls = self._make_custom_queue_message()
+        msg = cls()
+        with self.assertRaises(NotImplementedError):
+            msg.to_dict()
+
+    def test_subclass_without_to_dict_to_json_raises_at_call_time(self):
+        cls = self._make_custom_queue_message()
+        msg = cls()
+        with self.assertRaises(NotImplementedError):
+            msg.to_json()