fix: typeError when serializing multimodal prompts with binary content in Graph/Swarm session persistence (#1870)

Author: JackYPCOnline

src/strands/multiagent/graph.py

@@ -51,6 +51,7 @@
 from ..types.content import ContentBlock, Messages
 from ..types.event_loop import Metrics, Usage
 from ..types.multiagent import MultiAgentInput
+from ..types.session import decode_bytes_values, encode_bytes_values
 from ..types.traces import AttributeValue
 from .base import MultiAgentBase, MultiAgentResult, NodeResult, Status
 
@@ -1158,7 +1159,7 @@ def serialize_state(self) -> dict[str, Any]:
             "interrupted_nodes": [n.node_id for n in self.state.interrupted_nodes],
             "node_results": {k: v.to_dict() for k, v in (self.state.results or {}).items()},
             "next_nodes_to_execute": next_nodes,
-            "current_task": self.state.task,
+            "current_task": encode_bytes_values(self.state.task),
             "execution_order": [n.node_id for n in self.state.execution_order],
             "_internal_state": {
                 "interrupt_state": self._interrupt_state.to_dict(),
@@ -1248,7 +1249,7 @@ def _from_dict(self, payload: dict[str, Any]) -> None:
         self.state.execution_order = [self.nodes[node_id] for node_id in order_node_ids if node_id in self.nodes]
 
         # Task
-        self.state.task = payload.get("current_task", self.state.task)
+        self.state.task = decode_bytes_values(payload.get("current_task", self.state.task))
 
         # next nodes to execute
         next_nodes = [self.nodes[nid] for nid in (payload.get("next_nodes_to_execute") or []) if nid in self.nodes]

src/strands/multiagent/swarm.py

@@ -51,6 +51,7 @@
 from ..types.content import ContentBlock, Messages
 from ..types.event_loop import Metrics, Usage
 from ..types.multiagent import MultiAgentInput
+from ..types.session import decode_bytes_values, encode_bytes_values
 from ..types.traces import AttributeValue
 from .base import MultiAgentBase, MultiAgentResult, NodeResult, Status
 
@@ -965,7 +966,7 @@ def serialize_state(self) -> dict[str, Any]:
             "node_history": [n.node_id for n in self.state.node_history],
             "node_results": {k: v.to_dict() for k, v in self.state.results.items()},
             "next_nodes_to_execute": next_nodes,
-            "current_task": self.state.task,
+            "current_task": encode_bytes_values(self.state.task),
             "context": {
                 "shared_context": getattr(self.state.shared_context, "context", {}) or {},
                 "handoff_node": self.state.handoff_node.node_id if self.state.handoff_node else None,
@@ -1028,7 +1029,7 @@ def _from_dict(self, payload: dict[str, Any]) -> None:
                 logger.exception("Failed to hydrate NodeResult for node_id=%s; skipping.", node_id)
                 raise
         self.state.results = results
-        self.state.task = payload.get("current_task", self.state.task)
+        self.state.task = decode_bytes_values(payload.get("current_task", self.state.task))
 
         next_node_ids = payload.get("next_nodes_to_execute") or []
         if next_node_ids:

tests/strands/multiagent/test_graph.py

@@ -1986,7 +1986,10 @@ async def stream_without_result(*args, **kwargs):
 
 @pytest.mark.asyncio
 async def test_graph_persisted(mock_strands_tracer, mock_use_span):
-    """Test graph persistence functionality."""
+    """Test graph persistence functionality with multimodal input containing binary bytes."""
+    import base64
+    import json
+
     # Create mock session manager
     session_manager = Mock(spec=FileSessionManager)
     session_manager.read_multi_agent().return_value = None
@@ -2011,7 +2014,40 @@ async def test_graph_persisted(mock_strands_tracer, mock_use_span):
     assert "completed_nodes" in state
     assert "node_results" in state
 
-    # Test apply_state_from_dict with persisted state
+    # Build a multimodal prompt with inline binary PDF bytes (the problematic case)
+    pdf_bytes = b"%PDF-1.4 binary content"
+    multimodal_task = [
+        {"text": "Analyze this PDF"},
+        {
+            "document": {
+                "format": "pdf",
+                "name": "document.pdf",
+                "source": {
+                    "bytes": pdf_bytes,
+                },
+            }
+        },
+    ]
+
+    # Simulate graph having executed with a multimodal task
+    graph.state.task = multimodal_task
+
+    # serialize_state must not raise TypeError for bytes
+    serialized = graph.serialize_state()
+    assert json.dumps(serialized)  # must be JSON-serializable
+
+    # The bytes should be encoded in the serialized form
+    encoded_bytes = serialized["current_task"][1]["document"]["source"]["bytes"]
+    assert encoded_bytes == {"__bytes_encoded__": True, "data": base64.b64encode(pdf_bytes).decode()}
+
+    # deserialize_state must restore bytes back to original
+    serialized["next_nodes_to_execute"] = ["test_node"]
+    serialized["status"] = "executing"
+    graph.deserialize_state(serialized)
+    restored_bytes = graph.state.task[1]["document"]["source"]["bytes"]
+    assert restored_bytes == pdf_bytes
+
+    # Test apply_state_from_dict with plain string persisted state (backward compat)
     persisted_state = {
         "status": "executing",
         "completed_nodes": [],

tests/strands/multiagent/test_swarm.py

@@ -1106,7 +1106,10 @@ async def failing_execute_swarm(*args, **kwargs):
 
 @pytest.mark.asyncio
 async def test_swarm_persistence(mock_strands_tracer, mock_use_span):
-    """Test swarm persistence functionality."""
+    """Test swarm persistence functionality with multimodal input containing binary bytes."""
+    import base64
+    import json
+
     # Create mock session manager
     session_manager = Mock(spec=FileSessionManager)
     session_manager.read_multi_agent.return_value = None
@@ -1127,7 +1130,40 @@ async def test_swarm_persistence(mock_strands_tracer, mock_use_span):
     assert "node_results" in state
     assert "context" in state
 
-    # Test apply_state_from_dict with persisted state
+    # Build a multimodal prompt with inline binary PDF bytes (the problematic case)
+    pdf_bytes = b"%PDF-1.4 binary content"
+    multimodal_task = [
+        {"text": "Analyze this PDF"},
+        {
+            "document": {
+                "format": "pdf",
+                "name": "document.pdf",
+                "source": {
+                    "bytes": pdf_bytes,
+                },
+            }
+        },
+    ]
+
+    # Simulate swarm having executed with a multimodal task
+    swarm.state.task = multimodal_task
+
+    # serialize_state must not raise TypeError for bytes
+    serialized = swarm.serialize_state()
+    assert json.dumps(serialized)  # must be JSON-serializable
+
+    # The bytes should be encoded in the serialized form
+    encoded_bytes = serialized["current_task"][1]["document"]["source"]["bytes"]
+    assert encoded_bytes == {"__bytes_encoded__": True, "data": base64.b64encode(pdf_bytes).decode()}
+
+    # deserialize_state must restore bytes back to original
+    serialized["next_nodes_to_execute"] = ["test_agent"]
+    serialized["status"] = "executing"
+    swarm.deserialize_state(serialized)
+    restored_bytes = swarm.state.task[1]["document"]["source"]["bytes"]
+    assert restored_bytes == pdf_bytes
+
+    # Test apply_state_from_dict with plain string persisted state (backward compat)
     persisted_state = {
         "status": "executing",
         "node_history": [],