Fix issue #2545: Pass inputs from Flow to Tasks for template interpolation

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

src/crewai/agent.py

@@ -23,7 +23,6 @@ from crewai.utilities.constants import TRAINED_AGENTS_DATA_FILE, TRAINING_DATA_F
 from crewai.utilities.converter import generate_model_description
 from crewai.utilities.token_counter_callback import TokenCalcHandler
 from crewai.utilities.training_handler import CrewTrainingHandler
-from crewai.utilities.typing import AgentConfig
 
 agentops = None
 
@@ -89,7 +88,6 @@ class Agent(BaseAgent):
     function_calling_llm: Optional[Any] = Field(
         description="Language model that will run the agent.", default=None
     )
-    config: Optional[Union[Dict[str, Any], AgentConfig]] = Field(default=None)
     system_template: Optional[str] = Field(
         default=None, description="System format for the agent."
     )

src/crewai/flow/flow.py

@@ -219,7 +219,11 @@ class Flow(Generic[T], metaclass=FlowMeta):
         """Returns the list of all outputs from executed methods."""
         return self._method_outputs
 
-    def _initialize_state(self, inputs: Dict[str, Any]) -> None:
+    def _initialize_state(self, inputs: Optional[Dict[str, Any]] = None) -> None:
+        """Initialize the state of the flow."""
+        if inputs is None:
+            return
+            
         if isinstance(self._state, BaseModel):
             # Structured state
             try:
@@ -245,6 +249,8 @@ class Flow(Generic[T], metaclass=FlowMeta):
             self._state.update(inputs)
         else:
             raise TypeError("State must be a BaseModel instance or a dictionary.")
+            
+        self._interpolate_inputs_in_crew(inputs)
 
     def kickoff(self, inputs: Optional[Dict[str, Any]] = None) -> Any:
         self.event_emitter.send(
@@ -406,6 +412,11 @@ class Flow(Generic[T], metaclass=FlowMeta):
 
             traceback.print_exc()
 
+    def _interpolate_inputs_in_crew(self, inputs: Dict[str, Any]) -> None:
+        """Interpolate inputs in the crew's tasks and agents if a crew is present."""
+        if hasattr(self, 'crew') and self.crew:
+            self.crew._interpolate_inputs(inputs)
+            
     def plot(self, filename: str = "crewai_flow") -> None:
         self._telemetry.flow_plotting_span(
             self.__class__.__name__, list(self._methods.keys())

src/crewai/project/annotations.py

@@ -16,12 +16,6 @@ def after_kickoff(func):
 
 
 def task(func):
-    """Decorator to mark a method as a task creator.
-    
-    When applied to a method in a class decorated with @CrewBase,
-    this makes the method's return value accessible as an element
-    of the self.tasks list.
-    """
     func.is_task = True
 
     @wraps(func)
@@ -35,12 +29,6 @@ def task(func):
 
 
 def agent(func):
-    """Decorator to mark a method as an agent creator.
-    
-    When applied to a method in a class decorated with @CrewBase,
-    this makes the method's return value accessible as an element
-    of the self.agents list.
-    """
     func.is_agent = True
     func = memoize(func)
     return func

src/crewai/project/crew_base.py

@@ -1,6 +1,6 @@
 import inspect
 from pathlib import Path
-from typing import Any, Callable, Dict, List, TypeVar, cast
+from typing import Any, Callable, Dict, TypeVar, cast
 
 import yaml
 from dotenv import load_dotenv
@@ -66,9 +66,6 @@ def CrewBase(cls: T) -> T:
             self._kickoff = self._filter_functions(
                 self._original_functions, "is_kickoff"
             )
-            
-            self.agents = []  # type: List[Any]
-            self.tasks = []   # type: List[Any]
 
         @staticmethod
         def load_yaml(config_path: Path):

src/crewai/task.py

@@ -41,7 +41,6 @@ from crewai.tools.base_tool import BaseTool
 from crewai.utilities.config import process_config
 from crewai.utilities.converter import Converter, convert_to_model
 from crewai.utilities.i18n import I18N
-from crewai.utilities.typing import TaskConfig
 
 
 class Task(BaseModel):
@@ -75,7 +74,7 @@ class Task(BaseModel):
     expected_output: str = Field(
         description="Clear definition of expected output for the task."
     )
-    config: Optional[Union[Dict[str, Any], TaskConfig]] = Field(
+    config: Optional[Dict[str, Any]] = Field(
         description="Configuration for the agent",
         default=None,
     )

src/crewai/utilities/typing/__init__.py

@@ -1,14 +0,0 @@
-from typing import Dict, List, Optional, Any, TypedDict, Union
-
-class AgentConfig(TypedDict, total=False):
-    """TypedDict for agent configuration loaded from YAML."""
-    role: str
-    goal: str
-    backstory: str
-    verbose: bool
-
-class TaskConfig(TypedDict, total=False):
-    """TypedDict for task configuration loaded from YAML."""
-    description: str
-    expected_output: str
-    agent: str  # Role of the agent to execute this task

tests/flow_test.py

@@ -322,3 +322,43 @@ def test_router_with_multiple_conditions():
 
     # final_step should run after router_and
     assert execution_order.index("log_final_step") > execution_order.index("router_and")
+
+
+def test_flow_inputs_passed_to_tasks():
+    """Test that inputs passed to Flow's kickoff method are correctly interpolated in task descriptions."""
+    from crewai import Agent, Crew, Task
+    from crewai.llm import LLM
+
+    agent = Agent(
+        role="Test Agent",
+        goal="Test Goal",
+        backstory="Test Backstory",
+        llm=LLM(model="gpt-4o-mini")
+    )
+
+    task = Task(
+        description="Process data about {topic}",
+        expected_output="Information about {topic}",
+        agent=agent
+    )
+
+    crew = Crew(
+        agents=[agent],
+        tasks=[task]
+    )
+
+    class TestFlow(Flow):
+        def __init__(self):
+            super().__init__()
+            self.crew = crew
+
+        @start()
+        def start_process(self):
+            pass
+
+    flow = TestFlow()
+    inputs = {"topic": "artificial intelligence"}
+    flow.kickoff(inputs=inputs)
+
+    assert task.description == "Process data about artificial intelligence"
+    assert task.expected_output == "Information about artificial intelligence"

tests/typing_test.py

@@ -1,55 +0,0 @@
-from typing import Dict, Any
-
-import pytest
-
-from crewai.agent import Agent
-from crewai.task import Task
-from crewai.utilities.typing import AgentConfig, TaskConfig
-
-
-def test_agent_with_config_dict():
-    config: AgentConfig = {
-        "role": "Test Agent", 
-        "goal": "Test Goal", 
-        "backstory": "Test Backstory",
-        "verbose": True
-    }
-    
-    agent = Agent(config=config)
-    
-    assert agent.role == "Test Agent"
-    assert agent.goal == "Test Goal"
-    assert agent.backstory == "Test Backstory"
-    assert agent.verbose is True
-
-
-def test_agent_with_yaml_config():
-    config: Dict[str, Any] = {
-        "researcher": {
-            "role": "Researcher",
-            "goal": "Research Goal",
-            "backstory": "Researcher Backstory",
-            "verbose": True
-        }
-    }
-    
-    agent = Agent(config=config["researcher"])
-    
-    assert agent.role == "Researcher"
-    assert agent.goal == "Research Goal"
-    assert agent.backstory == "Researcher Backstory"
-
-
-def test_task_with_config_dict():
-    config: TaskConfig = {
-        "description": "Test Task",
-        "expected_output": "Test Output",
-        "agent": "researcher"
-    }
-    
-    agent = Agent(role="Researcher", goal="Goal", backstory="Backstory")
-    task = Task(config=config, agent=agent)
-    
-    assert task.description == "Test Task"
-    assert task.expected_output == "Test Output"
-    assert task.agent == agent