fix: replace mutable default arguments with None (#3429)

src/crewai/agents/crew_agent_executor.py

@@ -54,11 +54,11 @@ class CrewAgentExecutor(CrewAgentExecutorMixin):
         tools_description: str,
         tools_handler: ToolsHandler,
         step_callback: Any = None,
-        original_tools: List[Any] = [],
+        original_tools: List[Any] | None = None,
         function_calling_llm: Any = None,
         respect_context_window: bool = False,
         request_within_rpm_limit: Optional[Callable[[], bool]] = None,
-        callbacks: List[Any] = [],
+        callbacks: List[Any] | None = None,
     ):
         self._i18n: I18N = I18N()
         self.llm: BaseLLM = llm
@@ -70,10 +70,10 @@ class CrewAgentExecutor(CrewAgentExecutorMixin):
         self.tools_names = tools_names
         self.stop = stop_words
         self.max_iter = max_iter
-        self.callbacks = callbacks
+        self.callbacks = callbacks or []
         self._printer: Printer = Printer()
         self.tools_handler = tools_handler
-        self.original_tools = original_tools
+        self.original_tools = original_tools or []
         self.step_callback = step_callback
         self.use_stop_words = self.llm.supports_stop_words()
         self.tools_description = tools_description
@@ -122,7 +122,6 @@ class CrewAgentExecutor(CrewAgentExecutorMixin):
             handle_unknown_error(self._printer, e)
             raise
 
-
         if self.ask_for_human_input:
             formatted_answer = self._handle_human_feedback(formatted_answer)
 
@@ -156,7 +155,7 @@ class CrewAgentExecutor(CrewAgentExecutorMixin):
                     messages=self.messages,
                     callbacks=self.callbacks,
                     printer=self._printer,
-                    from_task=self.task
+                    from_task=self.task,
                 )
                 formatted_answer = process_llm_response(answer, self.use_stop_words)
 

src/crewai/cli/settings/main.py

@@ -10,8 +10,9 @@ console = Console()
 class SettingsCommand(BaseCommand):
     """A class to handle CLI configuration commands."""
 
-    def __init__(self, settings_kwargs: dict[str, Any] = {}):
+    def __init__(self, settings_kwargs: dict[str, Any] | None = None):
         super().__init__()
+        settings_kwargs = settings_kwargs or {}
         self.settings = Settings(**settings_kwargs)
 
     def list(self) -> None:

src/crewai/crew.py

@@ -559,9 +559,10 @@ class Crew(FlowTrackable, BaseModel):
         CrewTrainingHandler(filename).initialize_file()
 
     def train(
-        self, n_iterations: int, filename: str, inputs: Optional[Dict[str, Any]] = {}
+        self, n_iterations: int, filename: str, inputs: Optional[Dict[str, Any]] = None
     ) -> None:
         """Trains the crew for a given number of iterations."""
+        inputs = inputs or {}
         try:
             crewai_event_bus.emit(
                 self,
@@ -702,8 +703,11 @@ class Crew(FlowTrackable, BaseModel):
         self._task_output_handler.reset()
         return results
 
-    async def kickoff_async(self, inputs: Optional[Dict[str, Any]] = {}) -> CrewOutput:
+    async def kickoff_async(
+        self, inputs: Optional[Dict[str, Any]] = None
+    ) -> CrewOutput:
         """Asynchronous kickoff method to start the crew execution."""
+        inputs = inputs or {}
         return await asyncio.to_thread(self.kickoff, inputs)
 
     async def kickoff_for_each_async(self, inputs: List[Dict]) -> List[CrewOutput]:

src/crewai/llm.py

@@ -311,7 +311,7 @@ class LLM(BaseLLM):
         api_base: Optional[str] = None,
         api_version: Optional[str] = None,
         api_key: Optional[str] = None,
-        callbacks: List[Any] = [],
+        callbacks: List[Any] | None = None,
         reasoning_effort: Optional[Literal["none", "low", "medium", "high"]] = None,
         stream: bool = False,
         **kwargs,
@@ -351,7 +351,7 @@ class LLM(BaseLLM):
         else:
             self.stop = stop
 
-        self.set_callbacks(callbacks)
+        self.set_callbacks(callbacks or [])
         self.set_env_callbacks()
 
     def _is_anthropic_model(self, model: str) -> bool:

src/crewai/memory/storage/kickoff_task_outputs_storage.py

@@ -18,9 +18,7 @@ class KickoffTaskOutputsSQLiteStorage:
     An updated SQLite storage class for kickoff task outputs storage.
     """
 
-    def __init__(
-        self, db_path: Optional[str] = None
-    ) -> None:
+    def __init__(self, db_path: Optional[str] = None) -> None:
         if db_path is None:
             # Get the parent directory of the default db path and create our db file there
             db_path = str(Path(db_storage_path()) / "latest_kickoff_task_outputs.db")
@@ -67,7 +65,7 @@ class KickoffTaskOutputsSQLiteStorage:
         output: Dict[str, Any],
         task_index: int,
         was_replayed: bool = False,
-        inputs: Dict[str, Any] = {},
+        inputs: Dict[str, Any] | None = None,
     ) -> None:
         """Add a new task output record to the database.
 
@@ -81,6 +79,7 @@ class KickoffTaskOutputsSQLiteStorage:
         Raises:
             DatabaseOperationError: If saving the task output fails due to SQLite errors.
         """
+        inputs = inputs or {}
         try:
             with sqlite3.connect(self.db_path) as conn:
                 conn.execute("BEGIN TRANSACTION")
@@ -146,7 +145,9 @@ class KickoffTaskOutputsSQLiteStorage:
                 conn.commit()
 
                 if cursor.rowcount == 0:
-                    logger.warning(f"No row found with task_index {task_index}. No update performed.")
+                    logger.warning(
+                        f"No row found with task_index {task_index}. No update performed."
+                    )
         except sqlite3.Error as e:
             error_msg = DatabaseError.format_error(DatabaseError.UPDATE_ERROR, e)
             logger.error(error_msg)

src/crewai/utilities/task_output_storage_handler.py

@@ -10,8 +10,10 @@ from crewai.task import Task
 
 """Handles storage and retrieval of task execution outputs."""
 
+
 class ExecutionLog(BaseModel):
     """Represents a log entry for task execution."""
+
     task_id: str
     expected_output: Optional[str] = None
     output: Dict[str, Any]
@@ -26,6 +28,7 @@ class ExecutionLog(BaseModel):
 
 """Manages storage and retrieval of task outputs."""
 
+
 class TaskOutputStorageHandler:
     def __init__(self) -> None:
         self.storage = KickoffTaskOutputsSQLiteStorage()
@@ -55,9 +58,10 @@ class TaskOutputStorageHandler:
         task: Task,
         output: Dict[str, Any],
         task_index: int,
-        inputs: Dict[str, Any] = {},
+        inputs: Dict[str, Any] | None = None,
         was_replayed: bool = False,
     ):
+        inputs = inputs or {}
         self.storage.add(task, output, task_index, was_replayed, inputs)
 
     def reset(self):