fix(workflow): Allow workflow tasks access results from previous dependent tasks (#202)

Co-authored-by: Fox Qin <[email protected]>

Author: foxxqin

src/strands_tools/workflow.py

@@ -445,7 +445,11 @@ def execute_task(self, task: Dict, workflow: Dict) -> Dict:
 
             # Extract response content - handle both dict and custom object return types
             try:
-                content = result.get("content", []) if hasattr(result, "get") else getattr(result, "content", [])
+                content = (
+                    result.message.get("content", [])
+                    if hasattr(result.message, "get")
+                    else getattr(result.message, "content", [])
+                )
             except AttributeError:
                 content = [{"text": str(result)}]
 

tests/test_workflow.py

@@ -9,6 +9,7 @@
 
 import pytest
 from strands import Agent
+from strands.agent import AgentResult
 from strands_tools import workflow as workflow_module
 
 
@@ -426,14 +427,11 @@ def test_execute_task_success(self, mock_parent_agent, mock_agent_result):
         ):
             # Create a proper mock agent result that returns structured data
             mock_task_agent = MagicMock()
-            mock_result = MagicMock()
-            mock_result.__str__ = MagicMock(return_value="Task completed successfully")
-            mock_result.get = MagicMock(
-                side_effect=lambda k, default=None: {
-                    "content": [{"text": "Task completed successfully"}],
-                    "stop_reason": "completed",
-                    "metrics": None,
-                }.get(k, default)
+            mock_result = AgentResult(
+                message={"content": [{"text": "Task completed successfully"}]},
+                stop_reason="completed",
+                metrics=None,
+                state=MagicMock(),
             )
 
             mock_task_agent.return_value = mock_result