feat: Add pluggable state serializers with transient state support

src/strands/__init__.py

@@ -2,11 +2,17 @@
 
 from . import agent, models, telemetry, types
 from .agent.agent import Agent
+from .agent.serializers import JSONSerializer, PickleSerializer, StateSerializer
+from .agent.state import AgentState
 from .tools.decorator import tool
 from .types.tools import ToolContext
 
 __all__ = [
     "Agent",
+    "AgentState",
+    "JSONSerializer",
+    "PickleSerializer",
+    "StateSerializer",
     "agent",
     "models",
     "tool",

src/strands/agent/__init__.py

@@ -4,6 +4,7 @@
 
 - Agent: The main interface for interacting with AI models and tools
 - ConversationManager: Classes for managing conversation history and context windows
+- Serializers: Pluggable serialization strategies for agent state (JSONSerializer, PickleSerializer)
 """
 
 from .agent import Agent
@@ -14,12 +15,18 @@
     SlidingWindowConversationManager,
     SummarizingConversationManager,
 )
+from .serializers import JSONSerializer, PickleSerializer, StateSerializer
+from .state import AgentState
 
 __all__ = [
     "Agent",
     "AgentResult",
+    "AgentState",
     "ConversationManager",
+    "JSONSerializer",
     "NullConversationManager",
+    "PickleSerializer",
     "SlidingWindowConversationManager",
+    "StateSerializer",
     "SummarizingConversationManager",
 ]

src/strands/agent/agent.py

@@ -66,6 +66,7 @@
     ConversationManager,
     SlidingWindowConversationManager,
 )
+from .serializers import StateSerializer
 from .state import AgentState
 
 logger = logging.getLogger(__name__)
@@ -121,6 +122,7 @@ def __init__(
         name: Optional[str] = None,
         description: Optional[str] = None,
         state: Optional[Union[AgentState, dict]] = None,
+        state_serializer: Optional[StateSerializer] = None,
         hooks: Optional[list[HookProvider]] = None,
         session_manager: Optional[SessionManager] = None,
         tool_executor: Optional[ToolExecutor] = None,
@@ -168,14 +170,18 @@ def __init__(
                 Defaults to None.
             state: stateful information for the agent. Can be either an AgentState object, or a json serializable dict.
                 Defaults to an empty AgentState object.
+            state_serializer: Serializer for state persistence (e.g., JSONSerializer, PickleSerializer).
+                Cannot be provided together with an AgentState object in 'state' parameter.
+                Defaults to JSONSerializer for backward compatibility.
             hooks: hooks to be added to the agent hook registry
                 Defaults to None.
             session_manager: Manager for handling agent sessions including conversation history and state.
                 If provided, enables session-based persistence and state management.
             tool_executor: Definition of tool execution strategy (e.g., sequential, concurrent, etc.).
 
         Raises:
-            ValueError: If agent id contains path separators.
+            ValueError: If agent id contains path separators, or if both state (AgentState) and state_serializer
+                are provided.
         """
         self.model = BedrockModel() if not model else BedrockModel(model_id=model) if isinstance(model, str) else model
         self.messages = messages if messages is not None else []
@@ -231,13 +237,18 @@ def __init__(
         # Initialize agent state management
         if state is not None:
             if isinstance(state, dict):
-                self.state = AgentState(state)
+                self.state = AgentState(state, serializer=state_serializer)
             elif isinstance(state, AgentState):
+                if state_serializer is not None:
+                    raise ValueError(
+                        "Cannot provide both state (AgentState) and state_serializer. "
+                        "Configure serializer on the AgentState object instead."
+                    )
                 self.state = state
             else:
                 raise ValueError("state must be an AgentState object or a dict")
         else:
-            self.state = AgentState()
+            self.state = AgentState(serializer=state_serializer)
 
         self.tool_caller = _ToolCaller(self)
 

src/strands/agent/serializers.py

@@ -0,0 +1,129 @@
+"""State serializers for agent state management.
+
+This module provides pluggable serialization strategies for AgentState:
+- JSONSerializer: Default serializer, backward compatible, validates on set()
+- PickleSerializer: Supports any Python object, no validation on set()
+- StateSerializer: Protocol for custom serializers
+"""
+
+import copy
+import json
+import pickle
+from typing import Any, Protocol, runtime_checkable
+
+
+@runtime_checkable
+class StateSerializer(Protocol):
+    """Protocol for state serializers.
+
+    Custom serializers can implement this protocol to provide
+    alternative serialization strategies for agent state.
+    """
+
+    def serialize(self, data: dict[str, Any]) -> bytes:
+        """Serialize state dict to bytes.
+
+        Args:
+            data: Dictionary of state data to serialize
+
+        Returns:
+            Serialized state as bytes
+        """
+        ...
+
+    def deserialize(self, data: bytes) -> dict[str, Any]:
+        """Deserialize bytes back to state dict.
+
+        Args:
+            data: Serialized state bytes
+
+        Returns:
+            Deserialized state dictionary
+        """
+        ...
+
+
+class JSONSerializer:
+    """JSON-based state serializer.
+
+    Default serializer that provides:
+    - Human-readable serialization format
+    - Validation on set() to maintain current behavior
+    - Backward compatibility with existing code
+    """
+
+    def serialize(self, data: dict[str, Any]) -> bytes:
+        """Serialize state dict to JSON bytes.
+
+        Args:
+            data: Dictionary of state data to serialize
+
+        Returns:
+            JSON serialized state as bytes
+        """
+        return json.dumps(data).encode("utf-8")
+
+    def deserialize(self, data: bytes) -> dict[str, Any]:
+        """Deserialize JSON bytes back to state dict.
+
+        Args:
+            data: JSON serialized state bytes
+
+        Returns:
+            Deserialized state dictionary
+        """
+        result: dict[str, Any] = json.loads(data.decode("utf-8"))
+        return result
+
+    def validate(self, value: Any) -> None:
+        """Validate that a value is JSON serializable.
+
+        Args:
+            value: The value to validate
+
+        Raises:
+            ValueError: If value is not JSON serializable
+        """
+        try:
+            json.dumps(value)
+        except (TypeError, ValueError) as e:
+            raise ValueError(
+                f"Value is not JSON serializable: {type(value).__name__}. "
+                f"Only JSON-compatible types (str, int, float, bool, list, dict, None) are allowed."
+            ) from e
+
+
+class PickleSerializer:
+    """Pickle-based state serializer.
+
+    Provides:
+    - Support for any Python object (datetime, UUID, dataclass, Pydantic models, etc.)
+    - No validation on set() (accepts anything)
+
+    Security Warning:
+        Pickle can execute arbitrary code during deserialization.
+        Only unpickle data from trusted sources.
+    """
+
+    def serialize(self, data: dict[str, Any]) -> bytes:
+        """Serialize state dict using pickle.
+
+        Args:
+            data: Dictionary of state data to serialize
+
+        Returns:
+            Pickle serialized state as bytes
+        """
+        return pickle.dumps(copy.deepcopy(data))
+
+    def deserialize(self, data: bytes) -> dict[str, Any]:
+        """Deserialize pickle bytes back to state dict.
+
+        Args:
+            data: Pickle serialized state bytes
+
+        Returns:
+            Deserialized state dictionary
+        """
+        result: dict[str, Any] = pickle.loads(data)  # noqa: S301
+        return result