Python: Optimize shared serialization paths

python/CODING_STANDARD.md

@@ -688,6 +688,20 @@ message_data = message.to_dict(exclude_none=True)  # and this does so again!
 logger.info(message_data, extra={...})
 ```
 
+When converting arbitrary values for telemetry, protocol, or event payloads, reuse the optimized framework
+converter instead of adding a package-local recursive serializer:
+
+```python
+from agent_framework._serialization import make_json_safe  # pyright: ignore[reportPrivateUsage]
+
+payload = make_json_safe(value)
+```
+
+Use a model's `to_dict()` directly when its type is known. Use `make_json_safe()` for heterogeneous values that may
+contain framework models, Pydantic models, dataclasses, containers, or primitives. Keep provider-specific conversion
+local when an API requires exact aliases, JSON modes, or opaque JSON strings, and avoid `json.dumps()` followed by
+`json.loads()` unless crossing such a required wire-format boundary.
+
 ## Test Organization
 
 ### Test Directory Structure

python/packages/ag-ui/agent_framework_ag_ui/_http_service.py

@@ -33,6 +33,9 @@ def _serialize_available_interrupts(available_interrupts: Sequence[Any] | None)
         return None
     serialized: list[dict[str, Any]] = []
     for interrupt in available_interrupts:
+        if isinstance(interrupt, Interrupt):
+            serialized.append(cast(dict[str, Any], interrupt.model_dump(by_alias=True, exclude_none=True)))
+            continue
         if isinstance(interrupt, Mapping) and "reason" not in interrupt:
             interrupt = dict(interrupt)
             interrupt_type = interrupt.pop("type", None)
@@ -48,6 +51,9 @@ def _serialize_available_interrupts(available_interrupts: Sequence[Any] | None)
 
 def _serialize_resume_entry(entry: Any) -> dict[str, Any]:
     """Serialize one typed or legacy resume entry to canonical AG-UI JSON."""
+    if isinstance(entry, ResumeEntry):
+        return cast(dict[str, Any], entry.model_dump(by_alias=True, exclude_none=True))
+
     model_dump = getattr(entry, "model_dump", None)
     if callable(model_dump):
         entry = model_dump(by_alias=True, exclude_none=True)

python/packages/ag-ui/agent_framework_ag_ui/_utils.py

@@ -8,11 +8,10 @@
 import json
 import uuid
 from collections.abc import Callable, MutableMapping, Sequence
-from dataclasses import asdict, is_dataclass
-from datetime import date, datetime
 from typing import Any
 
 from agent_framework import AgentResponseUpdate, ChatResponseUpdate, FunctionTool
+from agent_framework._serialization import make_json_safe  # pyright: ignore[reportPrivateUsage]
 
 # Role mapping constants
 AGUI_TO_FRAMEWORK_ROLE: dict[str, str] = {
@@ -145,37 +144,6 @@ def merge_state(current: dict[str, Any], update: dict[str, Any]) -> dict[str, An
     return result
 
 
-def make_json_safe(obj: Any) -> Any:  # noqa: ANN401
-    """Make an object JSON serializable.
-
-    Args:
-        obj: Object to make JSON safe
-
-    Returns:
-        JSON-serializable version of the object
-    """
-    if obj is None or isinstance(obj, (str, int, float, bool)):
-        return obj
-    if isinstance(obj, (datetime, date)):
-        return obj.isoformat()
-    if is_dataclass(obj):
-        # asdict may return nested non-dataclass objects, so recursively make them safe
-        return make_json_safe(asdict(obj))  # type: ignore[arg-type]
-    if hasattr(obj, "model_dump"):
-        return make_json_safe(obj.model_dump())
-    if hasattr(obj, "to_dict"):
-        return make_json_safe(obj.to_dict())
-    if hasattr(obj, "dict"):
-        return make_json_safe(obj.dict())
-    if hasattr(obj, "__dict__"):
-        return {key: make_json_safe(value) for key, value in vars(obj).items()}  # type: ignore[misc]
-    if isinstance(obj, (list, tuple)):
-        return [make_json_safe(item) for item in obj]  # type: ignore[misc]
-    if isinstance(obj, dict):
-        return {key: make_json_safe(value) for key, value in obj.items()}  # type: ignore[misc]
-    return str(obj)
-
-
 def convert_agui_tools_to_agent_framework(
     agui_tools: list[dict[str, Any]] | None,
 ) -> list[FunctionTool] | None:

python/packages/core/agent_framework/_serialization.py

@@ -9,12 +9,15 @@
 from collections.abc import Mapping, MutableMapping
 from dataclasses import asdict, is_dataclass
 from datetime import date, datetime
-from typing import Any, ClassVar, Protocol, TypeVar, runtime_checkable
+from functools import lru_cache
+from typing import Any, ClassVar, Protocol, TypeGuard, TypeVar, cast, runtime_checkable
 
 logger = logging.getLogger("agent_framework")
 
 ClassT = TypeVar("ClassT", bound="SerializationMixin")
 ProtocolT = TypeVar("ProtocolT", bound="SerializationProtocol")
+_JSON_SCALAR_TYPES = (str, int, float, bool, type(None))
+_DIRECT_JSON_TYPES = (*_JSON_SCALAR_TYPES, list, dict)
 
 # Regex pattern for converting CamelCase to snake_case
 _CAMEL_TO_SNAKE_PATTERN = re.compile(r"(?<!^)(?=[A-Z])")
@@ -131,7 +134,20 @@ def is_serializable(value: Any) -> bool:
         that implement ``SerializationProtocol`` require conversion via ``to_dict()``
         before JSON serialization.
     """
-    return isinstance(value, (str, int, float, bool, type(None), list, dict))
+    return isinstance(value, _DIRECT_JSON_TYPES)
+
+
+@lru_cache(maxsize=128)
+def _implements_serialization_protocol(value_type: type[Any]) -> bool:
+    """Check structural serialization support once per concrete type."""
+    return callable(getattr(value_type, "to_dict", None)) and callable(getattr(value_type, "from_dict", None))
+
+
+def _is_serialization_protocol(value: Any) -> TypeGuard[SerializationProtocol]:
+    """Check whether a value structurally supports framework serialization."""
+    if _implements_serialization_protocol(cast(type[Any], type(value))):
+        return True
+    return callable(getattr(value, "to_dict", None)) and callable(getattr(value, "from_dict", None))
 
 
 class SerializationMixin:
@@ -318,22 +334,28 @@ def to_dict(self, *, exclude: set[str] | None = None, exclude_none: bool = True)
             if key not in combined_exclude and not key.startswith("_"):
                 if exclude_none and value is None:
                     continue
+                if type(value) in _JSON_SCALAR_TYPES:
+                    result[key] = value
+                    continue
                 # Recursively serialize SerializationProtocol objects
-                if isinstance(value, SerializationProtocol):
+                if _is_serialization_protocol(value):
                     result[key] = value.to_dict(exclude=exclude, exclude_none=exclude_none)
                     continue
                 # Handle lists containing SerializationProtocol objects
                 if isinstance(value, list):
                     value_as_list: list[Any] = []
-                    for item in value:  # pyright: ignore[reportUnknownVariableType]
-                        if isinstance(item, SerializationProtocol):
+                    for item in cast(list[Any], value):
+                        if type(item) in _JSON_SCALAR_TYPES:
+                            value_as_list.append(item)
+                            continue
+                        if _is_serialization_protocol(item):
                             value_as_list.append(item.to_dict(exclude=exclude, exclude_none=exclude_none))
                             continue
                         if is_serializable(item):
                             value_as_list.append(item)
                             continue
                         logger.debug(
-                            f"Skipping non-serializable item in list attribute '{key}' of type {type(item).__name__}"  # pyright: ignore[reportUnknownArgumentType]
+                            f"Skipping non-serializable item in list attribute '{key}' of type {type(item).__name__}"
                         )
                     result[key] = value_as_list
                     continue
@@ -342,22 +364,25 @@ def to_dict(self, *, exclude: set[str] | None = None, exclude_none: bool = True)
                     from datetime import date, datetime, time
 
                     serialized_dict: dict[str, Any] = {}
-                    for raw_key, v in value.items():  # pyright: ignore[reportUnknownVariableType]
-                        dict_key = str(raw_key)  # pyright: ignore[reportUnknownArgumentType]
-                        if isinstance(v, SerializationProtocol):
-                            serialized_dict[dict_key] = v.to_dict(exclude=exclude, exclude_none=exclude_none)
-                            continue
+                    for raw_key, v in cast(dict[Any, Any], value).items():
+                        dict_key = str(raw_key)
                         # Convert datetime objects to strings
                         if isinstance(v, (datetime, date, time)):
                             serialized_dict[dict_key] = str(v)
                             continue
+                        if type(v) in _JSON_SCALAR_TYPES:
+                            serialized_dict[dict_key] = v
+                            continue
+                        if _is_serialization_protocol(v):
+                            serialized_dict[dict_key] = v.to_dict(exclude=exclude, exclude_none=exclude_none)
+                            continue
                         # Check if the value is JSON serializable
                         if is_serializable(v):
                             serialized_dict[dict_key] = v
                             continue
                         logger.debug(
                             f"Skipping non-serializable value for key '{dict_key}' in dict attribute '{key}' "
-                            f"of type {type(v).__name__}"  # pyright: ignore[reportUnknownArgumentType]
+                            f"of type {type(v).__name__}"
                         )
                     result[key] = serialized_dict
                     continue
@@ -631,12 +656,16 @@ def make_json_safe(obj: Any) -> Any:
     Returns:
         A JSON-serializable version of the object.
     """
-    if obj is None or isinstance(obj, (str, int, float, bool)):
+    if isinstance(obj, _JSON_SCALAR_TYPES):
         return obj
     if isinstance(obj, (datetime, date)):
         return obj.isoformat()
     if is_dataclass(obj) and not isinstance(obj, type):
         return make_json_safe(asdict(obj))
+    if type(obj) is dict:
+        return {str(key): make_json_safe(value) for key, value in obj.items()}  # type: ignore[misc]
+    if type(obj) in (list, tuple):
+        return [make_json_safe(item) for item in obj]  # type: ignore[misc]
     if callable(getattr(obj, "model_dump", None)):
         try:
             return make_json_safe(obj.model_dump())  # type: ignore[no-any-return]

python/packages/core/agent_framework/observability.py

@@ -42,7 +42,9 @@
 from typing_extensions import Sentinel
 
 from . import __version__ as version_info
-from ._serialization import SerializationProtocol
+from ._serialization import (
+    _is_serialization_protocol,  # pyright: ignore[reportPrivateUsage]
+)
 from ._settings import load_settings
 
 if sys.version_info >= (3, 13):
@@ -2423,7 +2425,6 @@ def _build_tool_otel_definition(tool_item: Any) -> dict[str, Any] | None:
     from pydantic import BaseModel
 
     from ._mcp import MCPTool
-    from ._serialization import SerializationMixin
     from ._tools import FunctionTool
 
     if isinstance(tool_item, FunctionTool):
@@ -2444,7 +2445,7 @@ def _build_tool_otel_definition(tool_item: Any) -> dict[str, Any] | None:
     raw: Mapping[str, Any] | None = None
     if isinstance(tool_item, BaseModel):
         raw = tool_item.model_dump(exclude_none=True)
-    elif isinstance(tool_item, (SerializationMixin, SerializationProtocol)):
+    elif _is_serialization_protocol(tool_item):
         raw = tool_item.to_dict()
     elif isinstance(tool_item, Mapping):
         mapping_item = cast("Mapping[str, Any]", tool_item)

python/packages/core/tests/core/test_serializable_mixin.py

@@ -5,6 +5,8 @@
 import logging
 from typing import Any
 
+from typing_extensions import Self
+
 from agent_framework._serialization import SerializationMixin
 
 
@@ -224,6 +226,26 @@ def __init__(self, outer_value: str, inner: Any = None):
         assert data["outer_value"] == "outer_test"
         assert data["inner"]["inner_value"] == "inner_test"
 
+    def test_to_dict_with_nested_structural_serialization_protocol(self):
+        """Test to_dict handles a structural protocol implementation without the mixin."""
+
+        class InnerClass:
+            def __init__(self, inner_value: str):
+                self.inner_value = inner_value
+
+            def to_dict(self, **kwargs: Any) -> dict[str, Any]:
+                return {"inner_value": self.inner_value}
+
+            @classmethod
+            def from_dict(cls, value: dict[str, Any], **kwargs: Any) -> Self:
+                return cls(value["inner_value"])
+
+        class OuterClass(SerializationMixin):
+            def __init__(self, inner: InnerClass):
+                self.inner = inner
+
+        assert OuterClass(InnerClass("inner_test")).to_dict()["inner"] == {"inner_value": "inner_test"}
+
     def test_to_dict_with_list_of_serialization_protocol(self):
         """Test to_dict handles lists containing SerializationProtocol objects."""