Fix issue #2526: Add utility to resolve ModuleNotFoundError when running flows from custom scripts

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

docs/concepts/flows.mdx

@@ -760,6 +760,64 @@ uv run kickoff
 
 The flow will execute, and you should see the output in the console.
 
+## Running a Flow from a Custom Script
+
+When running a Flow from a custom script (not using the CLI commands), you might encounter import errors like `ModuleNotFoundError`. This happens because Python can't find the modules due to how the import system works.
+
+To fix this issue, CrewAI provides a utility function that adds your project directory to the Python path:
+
+```python
+from crewai.utilities.path_utils import add_project_to_path
+
+# Call this function before importing your flow modules
+add_project_to_path()
+
+# Now you can import your flow modules without import errors
+from your_project.main import YourFlow
+
+# Create and kickoff your flow
+flow = YourFlow()
+result = flow.kickoff()
+```
+
+This utility function should be called before importing your flow modules. It adds your project directory to the Python path, allowing Python to find and import your modules correctly.
+
+Here's a complete example:
+
+```python
+#!/usr/bin/env python
+import os
+
+# Add this import to fix module import errors
+from crewai.utilities.path_utils import add_project_to_path
+
+# Call this function before importing your flow modules
+add_project_to_path()
+
+# Now you can import your flow modules without ModuleNotFoundError
+from my_flow.main import MyFlow
+
+def main():
+    """Run the flow from a custom script."""
+    # Create your flow instance
+    flow = MyFlow()
+    
+    # Kickoff the flow
+    result = flow.kickoff()
+    
+    # Process the result
+    print(f"Flow completed with result: {result}")
+
+if __name__ == "__main__":
+    main()
+```
+
+If your project is in a different directory, you can specify it:
+
+```python
+add_project_to_path('/path/to/your/project')
+```
+
 ## Plot Flows
 
 Visualizing your AI workflows can provide valuable insights into the structure and execution paths of your flows. CrewAI offers a powerful visualization tool that allows you to generate interactive plots of your flows, making it easier to understand and optimize your AI workflows.

examples/custom_flow_script.py

@@ -0,0 +1,25 @@
+"""
+Example script showing how to run a CrewAI flow from a custom script.
+
+This example demonstrates how to avoid the ModuleNotFoundError when 
+starting flows from custom scripts outside of the CLI command context.
+"""
+import os
+
+from crewai.utilities.path_utils import add_project_to_path
+
+add_project_to_path()
+
+from my_flow.main import MyFlow
+
+def main():
+    """Run the flow from a custom script."""
+    flow = MyFlow()
+    
+    result = flow.kickoff()
+    
+    print(f"Flow completed with result: {result}")
+
+
+if __name__ == "__main__":
+    main()

src/crewai/cli/kickoff_flow.py

@@ -2,11 +2,15 @@ import subprocess
 
 import click
 
+from crewai.utilities.path_utils import add_project_to_path
+
 
 def kickoff_flow() -> None:
     """
     Kickoff the flow by running a command in the UV environment.
     """
+    add_project_to_path()
+    
     command = ["uv", "run", "kickoff"]
 
     try:

src/crewai/utilities/path_utils.py

@@ -0,0 +1,43 @@
+import os
+import sys
+from pathlib import Path
+from typing import Optional
+
+
+def add_project_to_path(project_dir: Optional[str] = None) -> None:
+    """
+    Add the project directory to the Python path to resolve module imports.
+    
+    This function is especially useful when starting flows from custom scripts
+    outside of the CLI command context, to avoid ModuleNotFoundError.
+    
+    Args:
+        project_dir: Optional path to the project directory. If not provided,
+                     the current working directory is used.
+    
+    Example:
+        ```python
+        from crewai.utilities.path_utils import add_project_to_path
+        
+        add_project_to_path()
+        
+        from your_project.main import YourFlow
+        
+        flow = YourFlow()
+        flow.kickoff()
+        ```
+    """
+    if project_dir is None:
+        project_dir = os.getcwd()
+    
+    project_path = Path(project_dir).resolve()
+    
+    if (project_path / "src").exists() and (project_path / "src").is_dir():
+        if str(project_path) not in sys.path:
+            sys.path.insert(0, str(project_path))
+        
+        if str(project_path / "src") not in sys.path:
+            sys.path.insert(0, str(project_path / "src"))
+    else:
+        if str(project_path) not in sys.path:
+            sys.path.insert(0, str(project_path))

tests/utilities/test_path_utils.py

@@ -0,0 +1,80 @@
+import os
+import sys
+from pathlib import Path
+import unittest
+from unittest.mock import patch, MagicMock
+
+from crewai.utilities.path_utils import add_project_to_path
+
+
+class TestPathUtils(unittest.TestCase):
+    
+    @patch('os.getcwd')
+    @patch('pathlib.Path.exists')
+    @patch('pathlib.Path.is_dir')
+    def test_add_project_to_path_with_src(self, mock_is_dir, mock_exists, mock_getcwd):
+        mock_getcwd.return_value = "/home/user/project"
+        mock_exists.return_value = True
+        mock_is_dir.return_value = True
+        
+        original_sys_path = sys.path.copy()
+        
+        try:
+            if "/home/user/project" in sys.path:
+                sys.path.remove("/home/user/project")
+            if "/home/user/project/src" in sys.path:
+                sys.path.remove("/home/user/project/src")
+            
+            add_project_to_path()
+            
+            self.assertIn("/home/user/project", sys.path)
+            self.assertIn("/home/user/project/src", sys.path)
+            
+            self.assertTrue(
+                sys.path.index("/home/user/project/src") <= 1 and 
+                sys.path.index("/home/user/project") <= 1
+            )
+        finally:
+            sys.path = original_sys_path
+    
+    @patch('os.getcwd')
+    @patch('pathlib.Path.exists')
+    def test_add_project_to_path_without_src(self, mock_exists, mock_getcwd):
+        mock_getcwd.return_value = "/home/user/project"
+        mock_exists.return_value = False
+        
+        original_sys_path = sys.path.copy()
+        
+        try:
+            if "/home/user/project" in sys.path:
+                sys.path.remove("/home/user/project")
+            
+            add_project_to_path()
+            
+            self.assertIn("/home/user/project", sys.path)
+            
+            self.assertEqual(sys.path.index("/home/user/project"), 0)
+        finally:
+            sys.path = original_sys_path
+    
+    @patch('pathlib.Path.exists')
+    @patch('pathlib.Path.is_dir')
+    def test_add_project_to_path_with_custom_dir(self, mock_is_dir, mock_exists):
+        mock_exists.return_value = True
+        mock_is_dir.return_value = True
+        
+        original_sys_path = sys.path.copy()
+        
+        try:
+            custom_dir = "/home/user/custom_project"
+            if custom_dir in sys.path:
+                sys.path.remove(custom_dir)
+            if os.path.join(custom_dir, "src") in sys.path:
+                sys.path.remove(os.path.join(custom_dir, "src"))
+            
+            add_project_to_path(custom_dir)
+            
+            self.assertIn(custom_dir, sys.path)
+            self.assertIn(os.path.join(custom_dir, "src"), sys.path)
+        finally:
+            sys.path = original_sys_path