Auto inject crewai_trigger_payload (#3351)

* feat: add props to inject trigger payload

* feat: auto-inject trigger_input in the first crew task

src/crewai/crew.py

@@ -503,6 +503,7 @@ class Crew(FlowTrackable, BaseModel):
                         )
         return self
 
+
     @property
     def key(self) -> str:
         source: List[str] = [agent.key for agent in self.agents] + [
@@ -639,6 +640,7 @@ class Crew(FlowTrackable, BaseModel):
                 self._inputs = inputs
                 self._interpolate_inputs(inputs)
             self._set_tasks_callbacks()
+            self._set_inject_trigger_input_for_first_task()
 
             i18n = I18N(prompt_file=self.prompt_file)
 
@@ -1508,3 +1510,10 @@ class Crew(FlowTrackable, BaseModel):
         """Reset crew and agent knowledge storage."""
         for ks in knowledges:
             ks.reset()
+
+    def _set_inject_trigger_input_for_first_task(self):
+        crewai_trigger_payload = self._inputs and self._inputs.get("crewai_trigger_payload")
+        able_to_inject = self.tasks and self.tasks[0].inject_trigger_input is None
+
+        if self.process == Process.sequential and crewai_trigger_payload and able_to_inject:
+            self.tasks[0].inject_trigger_input = True

src/crewai/task.py

@@ -72,6 +72,10 @@ class Task(BaseModel):
         output_pydantic: Pydantic model for task output.
         security_config: Security configuration including fingerprinting.
         tools: List of tools/resources limited for task execution.
+        inject_trigger_input: Optional flag to control crewai_trigger_payload injection.
+                              None (default): Auto-inject for first task only.
+                              True: Always inject trigger payload for this task.
+                              False: Never inject trigger payload, even for first task.
     """
 
     __hash__ = object.__hash__  # type: ignore
@@ -163,6 +167,10 @@ class Task(BaseModel):
     end_time: Optional[datetime.datetime] = Field(
         default=None, description="End time of the task execution"
     )
+    inject_trigger_input: Optional[bool] = Field(
+        default=None,
+        description="Whether this task should append 'Trigger Payload: {crewai_trigger_payload}' to the task description when crewai_trigger_payload exists in crew inputs.",
+    )
     model_config = {"arbitrary_types_allowed": True}
 
     @field_validator("guardrail")
@@ -548,12 +556,23 @@ class Task(BaseModel):
             str: The formatted prompt string containing the task description,
                  expected output, and optional markdown formatting instructions.
         """
-        tasks_slices = [self.description]
+        description = self.description
+
+        should_inject = self.inject_trigger_input
+
+        if should_inject and self.agent:
+            crew = getattr(self.agent, 'crew', None)
+            if crew and hasattr(crew, '_inputs') and crew._inputs:
+                trigger_payload = crew._inputs.get("crewai_trigger_payload")
+                if trigger_payload is not None:
+                    description += f"\n\nTrigger Payload: {trigger_payload}"
+
+        tasks_slices = [description]
 
         output = self.i18n.slice("expected_output").format(
             expected_output=self.expected_output
         )
-        tasks_slices = [self.description, output]
+        tasks_slices = [description, output]
 
         if self.markdown:
             markdown_instruction = """Your final answer MUST be formatted in Markdown syntax.

tests/agent_test.py

@@ -21,7 +21,7 @@ from crewai.utilities import RPMController
 from crewai.utilities.errors import AgentRepositoryError
 from crewai.utilities.events import crewai_event_bus
 from crewai.utilities.events.tool_usage_events import ToolUsageFinishedEvent
-
+from crewai.process import Process
 
 def test_agent_llm_creation_with_env_vars():
     # Store original environment variables
@@ -1209,6 +1209,181 @@ Thought:<|eot_id|>
         assert mock_format_prompt.return_value == expected_prompt
 
 
+@pytest.mark.vcr(filter_headers=["authorization"])
+def test_task_inject_trigger_input():
+    from crewai import Crew
+
+    agent = Agent(
+        role="test role",
+        goal="test goal",
+        backstory="test backstory"
+    )
+
+    task = Task(
+        description="Analyze the data",
+        expected_output="Analysis report",
+        agent=agent,
+        inject_trigger_input=True
+    )
+    crew = Crew(agents=[agent], tasks=[task])
+    crew.kickoff({"crewai_trigger_payload": "Important context data"})
+
+    prompt = task.prompt()
+
+    assert "Analyze the data" in prompt
+    assert "Trigger Payload: Important context data" in prompt
+
+
+@pytest.mark.vcr(filter_headers=["authorization"])
+def test_task_without_inject_trigger_input():
+    from crewai import Crew
+
+    agent = Agent(
+        role="test role",
+        goal="test goal",
+        backstory="test backstory"
+    )
+
+    task = Task(
+        description="Analyze the data",
+        expected_output="Analysis report",
+        agent=agent,
+        inject_trigger_input=False
+    )
+
+    crew = Crew(agents=[agent], tasks=[task])
+    crew.kickoff({"crewai_trigger_payload": "Important context data"})
+
+    prompt = task.prompt()
+
+    assert "Analyze the data" in prompt
+    assert "Trigger Payload:" not in prompt
+    assert "Important context data" not in prompt
+
+
+@pytest.mark.vcr(filter_headers=["authorization"])
+def test_task_inject_trigger_input_no_payload():
+    from crewai import Crew
+
+    agent = Agent(
+        role="test role",
+        goal="test goal",
+        backstory="test backstory"
+    )
+
+    task = Task(
+        description="Analyze the data",
+        expected_output="Analysis report",
+        agent=agent,
+        inject_trigger_input=True
+    )
+
+    crew = Crew(agents=[agent], tasks=[task])
+    crew.kickoff({"other_input": "other data"})
+
+
+    prompt = task.prompt()
+
+    assert "Analyze the data" in prompt
+    assert "Trigger Payload:" not in prompt
+
+
+@pytest.mark.vcr(filter_headers=["authorization"])
+def test_do_not_inject_trigger_input_for_first_task_hierarchical():
+    from crewai import Crew
+
+    agent1 = Agent(role="First Agent", goal="First goal", backstory="First backstory")
+    agent2 = Agent(role="Second Agent", goal="Second goal", backstory="Second backstory")
+
+    first_task = Task(
+        description="Process initial data",
+        expected_output="Initial analysis",
+        agent=agent1,
+    )
+
+
+    crew = Crew(
+        agents=[agent1, agent2],
+        tasks=[first_task],
+        process=Process.hierarchical,
+        manager_llm="gpt-4o"
+    )
+
+    crew.kickoff({"crewai_trigger_payload": "Initial context data"})
+
+    first_prompt = first_task.prompt()
+    assert "Process initial data" in first_prompt
+    assert "Trigger Payload: Initial context data" not in first_prompt
+
+
+@pytest.mark.vcr(filter_headers=["authorization"])
+def test_first_task_auto_inject_trigger():
+    from crewai import Crew
+
+    agent1 = Agent(role="First Agent", goal="First goal", backstory="First backstory")
+    agent2 = Agent(role="Second Agent", goal="Second goal", backstory="Second backstory")
+
+    first_task = Task(
+        description="Process initial data",
+        expected_output="Initial analysis",
+        agent=agent1,
+    )
+
+    second_task = Task(
+        description="Process secondary data",
+        expected_output="Secondary analysis",
+        agent=agent2,
+    )
+
+    crew = Crew(
+        agents=[agent1, agent2],
+        tasks=[first_task, second_task]
+    )
+    crew.kickoff({"crewai_trigger_payload": "Initial context data"})
+
+    first_prompt = first_task.prompt()
+    assert "Process initial data" in first_prompt
+    assert "Trigger Payload: Initial context data" in first_prompt
+
+    second_prompt = second_task.prompt()
+    assert "Process secondary data" in second_prompt
+    assert "Trigger Payload:" not in second_prompt
+
+@pytest.mark.vcr(filter_headers=["authorization"])
+def test_ensure_first_task_inject_trigger_input_is_false_does_not_inject():
+    from crewai import Crew
+
+    agent1 = Agent(role="First Agent", goal="First goal", backstory="First backstory")
+    agent2 = Agent(role="Second Agent", goal="Second goal", backstory="Second backstory")
+
+    first_task = Task(
+        description="Process initial data",
+        expected_output="Initial analysis",
+        agent=agent1,
+        inject_trigger_input=False
+    )
+
+    second_task = Task(
+        description="Process secondary data",
+        expected_output="Secondary analysis",
+        agent=agent2,
+        inject_trigger_input=True
+    )
+
+    crew = Crew(
+        agents=[agent1, agent2],
+        tasks=[first_task, second_task]
+    )
+    crew.kickoff({"crewai_trigger_payload": "Context data"})
+
+    first_prompt = first_task.prompt()
+    assert "Trigger Payload: Context data" not in first_prompt
+
+    second_prompt = second_task.prompt()
+    assert "Trigger Payload: Context data" in second_prompt
+
+
+
 @patch("crewai.agent.CrewTrainingHandler")
 def test_agent_training_handler(crew_training_handler):
     task_prompt = "What is 1 + 1?"