Fix issue #2347: Process tools with result_as_answer=True in Flow mode

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

src/crewai/flow/flow.py

@@ -787,6 +787,30 @@ class Flow(Generic[T], metaclass=FlowMeta):
         await asyncio.gather(*tasks)
 
         final_output = self._method_outputs[-1] if self._method_outputs else None
+        
+        # Check if any agent in the flow has tools_results with result_as_answer=True
+        for method_name, method in self._methods.items():
+            if hasattr(method, "__agent"):
+                agent = getattr(method, "__agent")
+                if hasattr(agent, "tools_results") and agent.tools_results:
+                    for tool_result in agent.tools_results:
+                        if tool_result.get("result_as_answer", False):
+                            final_output = tool_result["result"]
+                            break
+        
+        # Also check for any agents that might be stored as instance variables
+        for attr_name in dir(self):
+            if attr_name.startswith('_'):
+                continue
+            try:
+                attr = getattr(self, attr_name)
+                if hasattr(attr, "tools_results") and attr.tools_results:
+                    for tool_result in attr.tools_results:
+                        if tool_result.get("result_as_answer", False):
+                            final_output = tool_result["result"]
+                            break
+            except (AttributeError, TypeError):
+                continue
 
         crewai_event_bus.emit(
             self,
@@ -1070,6 +1094,26 @@ class Flow(Generic[T], metaclass=FlowMeta):
         elif level == "warning":
             logger.warning(message)
 
+    @classmethod
+    def with_agent(cls, agent):
+        """
+        Decorator to associate an agent with a flow method.
+        This allows tracking which agents are used in the flow.
+        
+        Parameters
+        ----------
+        agent : Agent
+            The agent to associate with the method
+            
+        Returns
+        -------
+        Callable
+            A decorator function that associates the agent with the method
+        """
+        def decorator(func):
+            func.__agent = agent
+            return func
+        return decorator
     def plot(self, filename: str = "crewai_flow") -> None:
         crewai_event_bus.emit(
             self,

tests/flow/test_flow_tool_result.py

@@ -0,0 +1,80 @@
+import pytest
+from unittest.mock import patch, MagicMock
+from crewai import Agent, Task, Crew, Flow
+from crewai.flow import start, listen
+from crewai.tools import BaseTool
+from typing import Type
+from pydantic import BaseModel, Field
+
+class TestToolInput(BaseModel):
+    query: str = Field(..., description='Query to process')
+
+class TestTool(BaseTool):
+    name: str = 'Test Tool'
+    description: str = 'A test tool to demonstrate the issue'
+    args_schema: Type[BaseModel] = TestToolInput
+    result_as_answer: bool = True
+    
+    def _run(self, query: str) -> str:
+        return f'Result for query: {query}'
+
+def test_flow_tool_result_as_answer():
+    """Test that tools with result_as_answer=True are properly processed in Flow mode."""
+    # Create a test tool
+    test_tool = TestTool()
+    
+    # Create a test agent with the tool
+    agent = Agent(
+        role='Tester',
+        goal='Test tools',
+        backstory='Testing tools in Flow vs Crew',
+        tools=[test_tool]
+    )
+
+    # Create a task with the tool
+    task = Task(
+        description='Test task using the tool',
+        expected_output='Test output',
+        agent=agent,
+        tools=[test_tool]
+    )
+    
+    # Create a simple Flow with direct access to the agent
+    class SimpleFlow(Flow):
+        def __init__(self):
+            super().__init__()
+            self.test_agent = agent
+            
+        @start()
+        def start_task(self):
+            return 'Task started'
+            
+        @listen('start_task')
+        @Flow.with_agent(agent)  # Associate the agent with this method
+        def execute_task(self):
+            # Simulate tool execution and setting tools_results
+            self.test_agent.tools_results = [{
+                "name": "Test Tool",
+                "input": {"query": "test"},
+                "result": "Result for query: test",
+                "result_as_answer": True
+            }]
+            return "Agent task execution result"
+
+    # Create a mock for Crew to return the same result
+    with patch('crewai.crew.Crew.kickoff') as mock_crew_kickoff:
+        mock_crew_kickoff.return_value = "Result for query: test"
+        
+        # Test Flow
+        flow = SimpleFlow()
+        flow_result = flow.kickoff()
+        
+        # Verify that Flow returns the tool result with our fix
+        assert flow_result == "Result for query: test", "Flow should return tool result with result_as_answer=True"
+        
+        # Test Crew
+        crew = Crew(agents=[agent], tasks=[task])
+        crew_result = crew.kickoff()
+        
+        # Verify that Crew returns the tool result
+        assert crew_result == "Result for query: test", "Crew should return tool result with result_as_answer=True"