Move kickoff callbacks to crew's domain

src/crewai/crew.py

@@ -5,7 +5,7 @@ import uuid
 import warnings
 from concurrent.futures import Future
 from hashlib import md5
-from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple, Union
+from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, Tuple, Union
 
 from pydantic import (
     UUID4,
@@ -36,9 +36,7 @@ from crewai.telemetry import Telemetry
 from crewai.tools.agent_tools.agent_tools import AgentTools
 from crewai.types.usage_metrics import UsageMetrics
 from crewai.utilities import I18N, FileHandler, Logger, RPMController
-from crewai.utilities.constants import (
-    TRAINING_DATA_FILE,
-)
+from crewai.utilities.constants import TRAINING_DATA_FILE
 from crewai.utilities.evaluators.crew_evaluator_handler import CrewEvaluator
 from crewai.utilities.evaluators.task_evaluator import TaskEvaluator
 from crewai.utilities.formatter import (
@@ -165,6 +163,16 @@ class Crew(BaseModel):
         default=None,
         description="Callback to be executed after each task for all agents execution.",
     )
+    before_kickoff_callbacks: List[
+        Callable[[Optional[Dict[str, Any]]], Optional[Dict[str, Any]]]
+    ] = Field(
+        default_factory=list,
+        description="List of callbacks to be executed before crew kickoff. It may be used to adjust inputs before the crew is executed.",
+    )
+    after_kickoff_callbacks: List[Callable[[CrewOutput], CrewOutput]] = Field(
+        default_factory=list,
+        description="List of callbacks to be executed after crew kickoff. It may be used to adjust the output of the crew.",
+    )
     max_rpm: Optional[int] = Field(
         default=None,
         description="Maximum number of requests per minute for the crew execution to be respected.",
@@ -478,6 +486,9 @@ class Crew(BaseModel):
         self,
         inputs: Optional[Dict[str, Any]] = None,
     ) -> CrewOutput:
+        for callback in self.before_kickoff_callbacks:
+            inputs = callback(inputs)
+
         """Starts the crew to work on its assigned tasks."""
         self._execution_span = self._telemetry.crew_execution_span(self, inputs)
         self._task_output_handler.reset()
@@ -520,6 +531,9 @@ class Crew(BaseModel):
                 f"The process '{self.process}' is not implemented yet."
             )
 
+        for callback in self.after_kickoff_callbacks:
+            result = callback(result)
+
         metrics += [agent._token_process.get_summary() for agent in self.agents]
 
         self.usage_metrics = UsageMetrics()