Fix type-checking issues in to_structured_dict method

Co-Authored-By: Joe Moura <joao@crewai.com>

src/crewai/crew.py

@@ -1236,6 +1236,103 @@ class Crew(BaseModel):
     def __repr__(self):
         return f"Crew(id={self.id}, process={self.process}, number_of_agents={len(self.agents)}, number_of_tasks={len(self.tasks)})"
 
+    def _agent_to_dict(self, agent: BaseAgent) -> Dict[str, Any]:
+        """Convert an Agent object to a dictionary representation.
+        
+        Args:
+            agent: The Agent object to convert
+            
+        Returns:
+            Dict[str, Any]: Dictionary representation of the agent
+        """
+        return {
+            "id": str(agent.id),
+            "role": agent.role,
+            "goal": agent.goal,
+            "backstory": agent.backstory,
+            "allow_delegation": agent.allow_delegation,
+            "verbose": agent.verbose
+        }
+        
+    def _task_to_dict(self, task: Task) -> Dict[str, Any]:
+        """Convert a Task object to a dictionary representation.
+        
+        Args:
+            task: The Task object to convert
+            
+        Returns:
+            Dict[str, Any]: Dictionary representation of the task
+        """
+        task_info: Dict[str, Any] = {
+            "id": str(task.id),
+            "description": task.description,
+            "expected_output": task.expected_output
+        }
+        
+        # Safely add agent information if available
+        agent = getattr(task, 'agent', None)
+        if agent is not None:
+            task_info["agent"] = agent.role
+            
+        # Safely add async_execution if available
+        if hasattr(task, 'async_execution'):
+            async_execution = getattr(task, 'async_execution', None)
+            if async_execution is not None:
+                task_info["async_execution"] = async_execution
+            
+        return task_info
+        
+
+    def to_structured_dict(self) -> Dict[str, Any]:
+        """Return a structured dictionary representation of the Crew object.
+        
+        This method provides a frontend-friendly representation of the Crew's
+        structure and relationships, suitable for visualization and rendering.
+        
+        Returns:
+            Dict[str, Any]: A structured dictionary containing the Crew's configuration,
+                  agents, tasks, and their relationships.
+        """
+        # Basic crew information
+        result: Dict[str, Any] = {
+            "id": str(self.id),
+            "name": self.name,
+            "process": str(self.process),
+            "verbose": self.verbose,
+            "memory": self.memory,
+            "agents": [],
+            "tasks": [],
+            "task_relationships": []
+        }
+        
+        # Add agent information
+        for agent in self.agents:
+            result["agents"].append(self._agent_to_dict(agent))
+        
+        # Add task information - safely handle potential errors
+        for task in self.tasks:
+            try:
+                task_info = self._task_to_dict(task)
+                result["tasks"].append(task_info)
+                
+                # Add task relationships if context exists
+                context = getattr(task, 'context', None)
+                if context is not None:
+                    relationship = {
+                        "task_id": str(task.id),
+                        "depends_on": [str(context_task.id) for context_task in context]
+                    }
+                    result["task_relationships"].append(relationship)
+            except Exception as e:
+                self._logger.log("warning", f"Error processing task {getattr(task, 'id', 'unknown')}: {str(e)}")
+        
+        # Add manager agent if exists
+        manager_agent = getattr(self, 'manager_agent', None)
+        if manager_agent is not None:
+            result["manager_agent"] = self._agent_to_dict(manager_agent)
+        
+        return result
+
     def reset_memories(self, command_type: str) -> None:
         """Reset specific or all memories for the crew.
 

tests/test_crew_structured_dict.py

@@ -0,0 +1,176 @@
+import json
+import unittest
+import uuid
+from unittest.mock import MagicMock, PropertyMock, patch
+
+from crewai.agent import Agent
+from crewai.crew import Crew
+from crewai.process import Process
+from crewai.task import Task
+
+
+class TestCrewStructuredDict(unittest.TestCase):
+    def setUp(self):
+        # Create test agents
+        self.researcher = Agent(
+            role="Researcher",
+            goal="Research and gather information",
+            backstory="You are an expert researcher with a keen eye for detail."
+        )
+        
+        self.writer = Agent(
+            role="Writer",
+            goal="Write compelling content",
+            backstory="You are a skilled writer who can create engaging content."
+        )
+        
+        # Create test tasks
+        self.research_task = Task(
+            description="Research the latest AI developments",
+            expected_output="A summary of the latest AI developments",
+            agent=self.researcher
+        )
+        
+        self.writing_task = Task(
+            description="Write an article about AI developments",
+            expected_output="A well-written article about AI developments",
+            agent=self.writer,
+            context=[self.research_task]
+        )
+        
+        # Create a crew with tasks
+        self.crew = Crew(
+            agents=[self.researcher, self.writer],
+            tasks=[self.research_task, self.writing_task],
+            process=Process.sequential
+        )
+        
+        # Create a crew with manager
+        self.manager = Agent(
+            role="Manager",
+            goal="Manage the team",
+            backstory="You are an experienced manager who oversees projects."
+        )
+        
+        self.hierarchical_crew = Crew(
+            agents=[self.researcher, self.writer],
+            tasks=[self.research_task],
+            process=Process.hierarchical,
+            manager_agent=self.manager
+        )
+    
+    def test_to_structured_dict_basic_properties(self):
+        """Test that to_structured_dict includes basic Crew properties."""
+        result = self.crew.to_structured_dict()
+        
+        # Check basic properties
+        self.assertEqual(str(self.crew.id), result["id"])
+        self.assertEqual(self.crew.name, result["name"])
+        self.assertEqual(str(self.crew.process), result["process"])
+        self.assertEqual(self.crew.verbose, result["verbose"])
+        self.assertEqual(self.crew.memory, result["memory"])
+    
+    def test_to_structured_dict_agents(self):
+        """Test that to_structured_dict includes agent information."""
+        result = self.crew.to_structured_dict()
+        
+        # Check agents
+        self.assertEqual(len(self.crew.agents), len(result["agents"]))
+        
+        # Check first agent properties
+        agent_result = result["agents"][0]
+        agent = self.crew.agents[0]
+        self.assertEqual(str(agent.id), agent_result["id"])
+        self.assertEqual(agent.role, agent_result["role"])
+        self.assertEqual(agent.goal, agent_result["goal"])
+        self.assertEqual(agent.backstory, agent_result["backstory"])
+    
+    def test_to_structured_dict_tasks(self):
+        """Test that to_structured_dict includes task information."""
+        result = self.crew.to_structured_dict()
+        
+        # Check tasks
+        self.assertEqual(len(self.crew.tasks), len(result["tasks"]))
+        
+        # Check first task properties
+        task_result = result["tasks"][0]
+        task = self.crew.tasks[0]
+        self.assertEqual(str(task.id), task_result["id"])
+        self.assertEqual(task.description, task_result["description"])
+        self.assertEqual(task.expected_output, task_result["expected_output"])
+        self.assertEqual(task.agent.role, task_result["agent"])
+    
+    def test_to_structured_dict_task_relationships(self):
+        """Test that to_structured_dict includes task relationships."""
+        result = self.crew.to_structured_dict()
+        
+        # Check task relationships
+        self.assertTrue("task_relationships" in result)
+        self.assertEqual(1, len(result["task_relationships"]))
+        
+        # Check relationship properties
+        relationship = result["task_relationships"][0]
+        self.assertEqual(str(self.writing_task.id), relationship["task_id"])
+        self.assertEqual(1, len(relationship["depends_on"]))
+        self.assertEqual(str(self.research_task.id), relationship["depends_on"][0])
+    
+    def test_to_structured_dict_manager_agent(self):
+        """Test that to_structured_dict includes manager agent information."""
+        result = self.hierarchical_crew.to_structured_dict()
+        
+        # Check manager agent
+        self.assertTrue("manager_agent" in result)
+        manager_result = result["manager_agent"]
+        self.assertEqual(str(self.manager.id), manager_result["id"])
+        self.assertEqual(self.manager.role, manager_result["role"])
+        self.assertEqual(self.manager.goal, manager_result["goal"])
+        self.assertEqual(self.manager.backstory, manager_result["backstory"])
+    
+    def test_to_structured_dict_empty_tasks(self):
+        """Test that to_structured_dict handles empty tasks list."""
+        crew = Crew(
+            agents=[self.researcher],
+            tasks=[],
+            process=Process.sequential
+        )
+        
+        result = crew.to_structured_dict()
+        
+        # Check empty tasks
+        self.assertEqual(0, len(result["tasks"]))
+        self.assertEqual(0, len(result["task_relationships"]))
+    
+    def test_to_structured_dict_error_handling(self):
+        """Test that to_structured_dict handles errors gracefully."""
+        # Create a task with a valid agent
+        task = Task(
+            description="Test description",
+            expected_output="Test output",
+            agent=self.researcher
+        )
+        
+        # Create a crew with the task
+        crew = Crew(
+            agents=[self.researcher],
+            tasks=[task],
+            process=Process.sequential
+        )
+        
+        # Patch the task's context property to raise an exception
+        # We'll do this by patching the specific method that accesses context
+        with patch('crewai.crew.Crew._get_context', side_effect=Exception("Test exception")):
+            # The method should not raise an exception
+            result = crew.to_structured_dict()
+            
+            # Basic verification that the result still contains the task
+            self.assertEqual(1, len(result["tasks"]))
+            task_result = result["tasks"][0]
+            self.assertEqual(str(task.id), task_result["id"])
+            self.assertEqual(task.description, task_result["description"])
+            self.assertEqual(task.expected_output, task_result["expected_output"])
+            
+            # Verify no task relationships were added due to the exception
+            self.assertEqual(0, len(result["task_relationships"]))
+
+if __name__ == "__main__":
+    unittest.main()