Add Python serialization examples

docs/modules/serialization.mdx

@@ -15,11 +15,33 @@ BeeAI framework provides robust serialization capabilities through its built-in
 - 📦 Snapshots: Create point-in-time captures of component state
 - 🔧 Reconstruction: Rebuild objects from their serialized representation
 
+The Python example prints the serialized payload, the restored ISO timestamp, and a boolean confirming the round trip.
+
 <CodeGroup>
 
-{/* <!-- comingsoon python/examples/serialization/base.py --> */}
+{/* <!-- embedme python/examples/serialization/base.py --> */}
 ```py Python [expandable]
-Example coming soon
+# Copyright 2025 © BeeAI a Series of LF Projects, LLC
+# SPDX-License-Identifier: Apache-2.0
+
+from datetime import UTC, datetime
+
+from beeai_framework.serialization import Serializer
+
+
+def main() -> None:
+    original = datetime(2024, 1, 1, tzinfo=UTC)
+    serialized = Serializer.serialize(original)
+    restored = Serializer.deserialize(serialized, expected_type=datetime)
+
+    print(serialized)
+    print(restored.isoformat())
+    print(restored == original)
+
+
+if __name__ == "__main__":
+    main()
+
 ```
 
 {/* <!-- embedme typescript/examples/serialization/base.ts --> */}
@@ -69,11 +91,37 @@ The serialization process involves:
 
 Most BeeAI components can be serialized out of the box. Here's an example using memory:
 
+Running the Python snippet restores the memory, appends an assistant reply, and prints the message count plus the first question.
+
 <CodeGroup>
 
-{/* <!-- comingsoon python/examples/serialization/memory.py --> */}
+{/* <!-- embedme python/examples/serialization/memory.py --> */}
 ```py Python [expandable]
-Example coming soon
+# Copyright 2025 © BeeAI a Series of LF Projects, LLC
+# SPDX-License-Identifier: Apache-2.0
+
+import asyncio
+
+from beeai_framework.backend import AssistantMessage, UserMessage
+from beeai_framework.memory import UnconstrainedMemory
+
+
+async def main() -> None:
+    memory = UnconstrainedMemory()
+    await memory.add(UserMessage("What is your name?"))
+
+    serialized = memory.serialize()
+    restored = UnconstrainedMemory.from_serialized(serialized)
+
+    await restored.add(AssistantMessage("Bee"))
+
+    print(len(restored.messages))
+    print(restored.messages[0].text)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
+
 ```
 
 {/* <!-- embedme typescript/examples/serialization/memory.ts --> */}
@@ -107,11 +155,52 @@ If you want to serialize a class that the `Serializer` does not know, you may re
 
 You can register external classes with the serializer:
 
+This example registers a lightweight data class and prints the reconstructed token along with its expiry timestamp.
+
 <CodeGroup>
 
-{/* <!-- comingsoon python/examples/serialization/custom_external.py --> */}
+{/* <!-- embedme python/examples/serialization/custom_external.py --> */}
 ```py Python [expandable]
-Example coming soon
+# Copyright 2025 © BeeAI a Series of LF Projects, LLC
+# SPDX-License-Identifier: Apache-2.0
+
+from dataclasses import dataclass
+from datetime import UTC, datetime
+
+from beeai_framework.serialization import Serializer
+
+
+@dataclass
+class ApiToken:
+    value: str
+    expires_at: datetime
+
+
+Serializer.register(
+    ApiToken,
+    to_plain=lambda token: {
+        "value": token.value,
+        "expires_at": token.expires_at,
+    },
+    from_plain=lambda payload: ApiToken(
+        value=payload["value"],
+        expires_at=payload["expires_at"],
+    ),
+)
+
+
+def main() -> None:
+    token = ApiToken("example-token", datetime(2025, 1, 1, tzinfo=UTC))
+    serialized = Serializer.serialize(token)
+    restored = Serializer.deserialize(serialized, expected_type=ApiToken)
+
+    print(restored)
+    print(restored.expires_at.isoformat())
+
+
+if __name__ == "__main__":
+    main()
+
 ```
 
 {/* <!-- embedme typescript/examples/serialization/customExternal.ts --> */}
@@ -150,11 +239,46 @@ console.info(deserialized);
 
 For deeper integration, extend the Serializable class:
 
+The Python variant increments a counter, serializes it, and prints both the live and restored values.
+
 <CodeGroup>
 
-{/* <!-- comingsoon python/examples/serialization/custom_internal.py --> */}
+{/* <!-- embedme python/examples/serialization/custom_internal.py --> */}
 ```py Python [expandable]
-Example coming soon
+# Copyright 2025 © BeeAI a Series of LF Projects, LLC
+# SPDX-License-Identifier: Apache-2.0
+
+from beeai_framework.serialization import Serializable
+
+
+class Counter(Serializable[dict[str, int]]):
+    def __init__(self, value: int = 0) -> None:
+        self.value = value
+
+    def increment(self) -> None:
+        self.value += 1
+
+    def create_snapshot(self) -> dict[str, int]:
+        return {"value": self.value}
+
+    def load_snapshot(self, snapshot: dict[str, int]) -> None:
+        self.value = snapshot["value"]
+
+
+def main() -> None:
+    counter = Counter(3)
+    counter.increment()
+
+    serialized = counter.serialize()
+    restored = Counter.from_serialized(serialized)
+
+    print(counter.value)
+    print(restored.value)
+
+
+if __name__ == "__main__":
+    main()
+
 ```
 
 {/* <!-- embedme typescript/examples/serialization/customInternal.ts --> */}
@@ -199,11 +323,29 @@ Failure to register a class that the `Serializer` does not know will result in t
 
 ## Context matters
 
+Deserialize with the `extraClasses` equivalent to make sure message factories are registered; the Python snippet imports both `UserMessage` and `AssistantMessage` so the serializer can hydrate their content and then prints every restored user utterance.
+
 <CodeGroup>
 
-{/* <!-- comingsoon python/examples/serialization/context.py --> */}
+{/* <!-- embedme python/examples/serialization/context.py --> */}
 ```py Python [expandable]
-Example coming soon
+# Copyright 2025 © BeeAI a Series of LF Projects, LLC
+# SPDX-License-Identifier: Apache-2.0
+
+from beeai_framework.backend import AssistantMessage, UserMessage
+from beeai_framework.memory import UnconstrainedMemory
+
+SERIALIZED_MEMORY = """{"__version":"0.0.0","__root":{"__serializer":true,"__class":"UnconstrainedMemory","__value":{"messages":[{"__serializer":true,"__class":"UserMessage","__value":{"id":null,"meta":{"createdAt":{"__serializer":true,"__class":"datetime","__value":"2025-10-18T19:38:37.859543+00:00"}},"role":"user","content":[{"type":"text","text":"Hello!"}]}},{"__serializer":true,"__class":"AssistantMessage","__value":{"id":null,"meta":{"createdAt":{"__serializer":true,"__class":"datetime","__value":"2025-10-18T19:38:37.859665+00:00"}},"role":"assistant","content":[{"type":"text","text":"Hello, how can I help you?"}]}}]}}}"""
+
+
+def main() -> None:
+    memory = UnconstrainedMemory.from_serialized(SERIALIZED_MEMORY, extra_classes=[UserMessage, AssistantMessage])
+    print([message.text for message in memory.messages])
+
+
+if __name__ == "__main__":
+    main()
+
 ```
 
 {/* <!-- embedme typescript/examples/serialization/context.ts --> */}

python/beeai_framework/backend/message.py

@@ -4,7 +4,7 @@
 import enum
 import json
 from abc import ABC
-from collections.abc import Sequence
+from collections.abc import Iterable, Mapping, Sequence
 from datetime import UTC, datetime
 from enum import Enum
 from typing import Any, Generic, Literal, Required, Self, TypeAlias, TypeVar, cast
@@ -374,3 +374,89 @@ def dedupe_tool_calls(msg: AssistantMessage) -> None:
 
     for idx in sorted(excluded_indexes, reverse=True):
         msg.content.pop(idx)
+
+
+def _register_message_class(
+    message_cls: type[Message[Any]],
+    allowed_content: Mapping[str, type[BaseModel]],
+) -> None:
+    def _registrar() -> None:
+        from beeai_framework.serialization import Serializer, SerializerError
+
+        def _to_plain(message: Message[Any]) -> dict[str, Any]:
+            return {
+                "id": message.id,
+                "meta": dict(message.meta),
+                "role": str(message.role),
+                "content": [fragment.model_dump() for fragment in message.content],
+            }
+
+        def _from_plain(payload: Mapping[str, Any]) -> Message[Any]:
+            content_payload = payload.get("content", []) or []
+            content_models: list[Any] = []
+
+            for item in content_payload:
+                if not isinstance(item, Mapping):
+                    raise SerializerError("Serialized message content must be a mapping.")
+
+                content_type = item.get("type")
+                if not isinstance(content_type, str):
+                    raise SerializerError(
+                        f"Serialized message content is missing or has an invalid 'type': {content_type!r}",
+                    )
+
+                model_cls = allowed_content.get(content_type)
+                if model_cls is None:
+                    raise SerializerError(
+                        f"Unsupported message content '{content_type}' for {message_cls.__name__}.",
+                    )
+                content_models.append(model_cls.model_validate(dict(item)))
+
+            meta = payload.get("meta") or {}
+            meta_copy = dict(meta)
+            return message_cls(content_models, meta=meta_copy, id=payload.get("id"))
+
+        Serializer.register(
+            message_cls,
+            to_plain=_to_plain,
+            from_plain=_from_plain,
+        )
+
+    _registrar()
+
+    def _register_method(cls: type[Message[Any]], *, aliases: Iterable[str] | None = None) -> None:
+        _registrar()
+
+    type.__setattr__(message_cls, "register", classmethod(_register_method))
+
+
+_register_message_class(
+    SystemMessage,
+    {
+        "text": MessageTextContent,
+    },
+)
+
+_register_message_class(
+    UserMessage,
+    {
+        "text": MessageTextContent,
+        "image_url": MessageImageContent,
+        "file": MessageFileContent,
+    },
+)
+
+_register_message_class(
+    AssistantMessage,
+    {
+        "text": MessageTextContent,
+        "tool-call": MessageToolCallContent,
+    },
+)
+
+_register_message_class(
+    ToolMessage,
+    {
+        "tool-result": MessageToolResultContent,
+    },
+)

python/beeai_framework/memory/token_memory.py

@@ -2,10 +2,11 @@
 # SPDX-License-Identifier: Apache-2.0
 
 from math import ceil
-from typing import Any
+from typing import Any, Self
 
 from beeai_framework.backend.message import AnyMessage
 from beeai_framework.memory.base_memory import BaseMemory
+from beeai_framework.serialization import Serializable
 
 
 def simple_estimate(msg: AnyMessage) -> int:
@@ -16,12 +17,12 @@ def simple_tokenize(msgs: list[AnyMessage]) -> int:
     return sum(map(simple_estimate, msgs))
 
 
-class TokenMemory(BaseMemory):
+class TokenMemory(Serializable[dict[str, Any]], BaseMemory):
     """Memory implementation that respects token limits."""
 
     def __init__(
         self,
-        llm: Any,
+        llm: Any | None = None,
         max_tokens: int | None = None,
         sync_threshold: float = 0.25,
         capacity_threshold: float = 0.75,
@@ -117,8 +118,10 @@ def reset(self) -> None:
         self._tokens_by_message.clear()
 
     async def clone(self) -> "TokenMemory":
+        llm = self.llm
+        llm_clone = llm.clone() if llm is not None and hasattr(llm, "clone") else llm
         cloned = TokenMemory(
-            self.llm.clone(),
+            llm_clone,
             self._max_tokens,
             self._sync_threshold,
             self._threshold,
@@ -127,3 +130,37 @@ async def clone(self) -> "TokenMemory":
         cloned._messages = self._messages.copy()
         cloned._tokens_by_message = self._tokens_by_message.copy()
         return cloned
+
+    def create_snapshot(self) -> dict[str, Any]:
+        return {
+            "messages": list(self._messages),
+            "max_tokens": self._max_tokens,
+            "capacity_threshold": self._threshold,
+            "sync_threshold": self._sync_threshold,
+        }
+
+    def load_snapshot(self, snapshot: dict[str, Any]) -> None:
+        self._messages = list(snapshot.get("messages", []))
+        self._max_tokens = snapshot.get("max_tokens")
+        self._threshold = snapshot.get("capacity_threshold", self._threshold)
+        self._sync_threshold = snapshot.get("sync_threshold", self._sync_threshold)
+        self._tokens_by_message.clear()
+
+        for message in self._messages:
+            key = self._get_message_key(message)
+            estimated_tokens = self.handlers["estimate"](message)
+            self._tokens_by_message[key] = {
+                "tokens_count": estimated_tokens,
+                "dirty": True,
+            }
+
+    @classmethod
+    def from_snapshot(cls, snapshot: dict[str, Any]) -> Self:
+        instance = cls(
+            llm=None,
+            max_tokens=snapshot.get("max_tokens"),
+            sync_threshold=snapshot.get("sync_threshold", 0.25),
+            capacity_threshold=snapshot.get("capacity_threshold", 0.75),
+        )
+        instance.load_snapshot(snapshot)
+        return instance