diff --git a/backend/api_v2/deployment_helper.py b/backend/api_v2/deployment_helper.py
index 5fbec7999c..4d1094087a 100644
--- a/backend/api_v2/deployment_helper.py
+++ b/backend/api_v2/deployment_helper.py
@@ -191,7 +191,7 @@ def execute_workflow(
         include_metrics: bool = False,
         include_extracted_text: bool = False,
         use_file_history: bool = False,
-        tag_names: list[str] = [],
+        tag_names: list[str] | None = None,
         llm_profile_id: str | None = None,
         hitl_queue_name: str | None = None,
         hitl_packet_id: str | None = None,
@@ -222,6 +222,7 @@ def execute_workflow(
             Rate limiting is handled at the view layer. This method should be called
             after rate limit checks have passed, with a pre-acquired execution_id.
         """
+        tag_names = tag_names or []
         # Use provided execution_id or generate one (for backward compatibility)
         if execution_id is None:
             execution_id = str(uuid.uuid4())
diff --git a/backend/workflow_manager/endpoint_v2/queue_utils.py b/backend/workflow_manager/endpoint_v2/queue_utils.py
index 5e44998be5..fedc22f9b9 100644
--- a/backend/workflow_manager/endpoint_v2/queue_utils.py
+++ b/backend/workflow_manager/endpoint_v2/queue_utils.py
@@ -25,12 +25,13 @@ class QueueUtils:
     _hitl_connectors = {}
 
     @staticmethod
-    def get_queue_inst(connector_settings: dict[str, Any] = {}) -> UnstractQueue:
+    def get_queue_inst(connector_settings: dict[str, Any] | None = None) -> UnstractQueue:
         """Get queue connector instance based on configuration.
 
         For HITL operations, this can return PostgreSQL, Hybrid, or Redis connectors
         based on the HITL_QUEUE_BACKEND setting.
         """
+        connector_settings = connector_settings or {}
         # Check if caller explicitly wants the default (non-HITL) backend
         # This is used by HybridQueue to get the actual Redis connector without recursion
         force_default = connector_settings.get("force_default_backend", False)
@@ -73,7 +74,7 @@ def get_queue_inst(connector_settings: dict[str, Any] = {}) -> UnstractQueue:
 
     @staticmethod
     def get_hitl_queue_inst(
-        backend: str, connector_settings: dict[str, Any] = {}
+        backend: str, connector_settings: dict[str, Any] | None = None
     ) -> UnstractQueue:
         """Get HITL-specific queue connector instance with dynamic imports.
 
@@ -91,6 +92,7 @@ def get_hitl_queue_inst(
         Raises:
             UnstractQueueException: When HITL connectors are not available
         """
+        connector_settings = connector_settings or {}
         # For Redis backend, use default connector
         if backend == "redis":
             # Strip HITL flag to force default (non-HITL) connector path
diff --git a/backend/workflow_manager/endpoint_v2/source.py b/backend/workflow_manager/endpoint_v2/source.py
index 23b0d04f2c..a61cd3432e 100644
--- a/backend/workflow_manager/endpoint_v2/source.py
+++ b/backend/workflow_manager/endpoint_v2/source.py
@@ -912,7 +912,7 @@ def _create_file_hash(
         )
 
     def list_files_from_source(
-        self, file_hashes: dict[str, FileHash] = {}
+        self, file_hashes: dict[str, FileHash] | None = None
     ) -> tuple[dict[str, FileHash], int]:
         """List files from source connector.
 
@@ -922,6 +922,7 @@ def list_files_from_source(
             tuple[dict[str, FileHash], int]: A dictionary of FileHashes,
             along with the total count of matched files.
         """
+        file_hashes = file_hashes or {}
         connection_type = self.endpoint.connection_type
         if connection_type == WorkflowEndpoint.ConnectionType.FILESYSTEM:
             files = self.list_files_from_file_connector()
diff --git a/backend/workflow_manager/workflow_v2/dto.py b/backend/workflow_manager/workflow_v2/dto.py
index 97148f1e91..2c63400136 100644
--- a/backend/workflow_manager/workflow_v2/dto.py
+++ b/backend/workflow_manager/workflow_v2/dto.py
@@ -87,7 +87,9 @@ def _remove_item_top_metadata(item: dict, keys_to_remove: list[str]) -> None:
         else:
             item.pop("metadata", None)
 
-    def remove_result_metadata_keys(self, keys_to_remove: list[str] = []) -> None:
+    def remove_result_metadata_keys(
+        self, keys_to_remove: list[str] | None = None
+    ) -> None:
         """Removes specified keys from the 'metadata' dictionary within each
         'result' dictionary in the 'result' list attribute of the instance. If
         'keys_to_remove' is empty, the 'metadata' key itself is removed.
@@ -95,6 +97,7 @@ def remove_result_metadata_keys(self, keys_to_remove: list[str] = []) -> None:
         Args:
             keys_to_remove (List[str]): List of keys to be removed from 'metadata'.
         """
+        keys_to_remove = keys_to_remove or []
         if not isinstance(self.result, list):
             return
 
diff --git a/backend/workflow_manager/workflow_v2/views.py b/backend/workflow_manager/workflow_v2/views.py
index fefba8c21a..cb75a70772 100644
--- a/backend/workflow_manager/workflow_v2/views.py
+++ b/backend/workflow_manager/workflow_v2/views.py
@@ -302,9 +302,10 @@ def execute_workflow(
         execution_action: str | None = None,
         execution_id: str | None = None,
         pipeline_guid: str | None = None,
-        hash_values_of_files: dict[str, FileHash] = {},
+        hash_values_of_files: dict[str, FileHash] | None = None,
         use_file_history: bool = False,
     ) -> ExecutionResponse:
+        hash_values_of_files = hash_values_of_files or {}
         # Detect if this is an API execution by checking connector types
         is_api_execution = WorkflowEndpointUtils.is_api_workflow(workflow)
 
diff --git a/backend/workflow_manager/workflow_v2/workflow_helper.py b/backend/workflow_manager/workflow_v2/workflow_helper.py
index d22dcc85bb..7f7781d1b4 100644
--- a/backend/workflow_manager/workflow_v2/workflow_helper.py
+++ b/backend/workflow_manager/workflow_v2/workflow_helper.py
@@ -302,7 +302,7 @@ def validate_tool_instances_meta(
     def run_workflow(
         workflow: Workflow,
         workflow_execution: WorkflowExecution,
-        hash_values_of_files: dict[str, FileHash] = {},
+        hash_values_of_files: dict[str, FileHash] | None = None,
         organization_id: str | None = None,
         pipeline_id: str | None = None,
         scheduled: bool = False,
@@ -314,6 +314,7 @@ def run_workflow(
         packet_id: str | None = None,
         custom_data: dict[str, Any] | None = None,
     ) -> ExecutionResponse:
+        hash_values_of_files = hash_values_of_files or {}
         tool_instances: list[ToolInstance] = (
             ToolInstanceHelper.get_tool_instances_by_workflow(
                 workflow.id, ToolInstanceKey.STEP
@@ -880,11 +881,12 @@ def complete_execution(
         execution_id: str | None = None,
         pipeline_id: str | None = None,
         execution_mode: WorkflowExecution | None = WorkflowExecution.Mode.QUEUE,
-        hash_values_of_files: dict[str, FileHash] = {},
+        hash_values_of_files: dict[str, FileHash] | None = None,
         use_file_history: bool = False,
         timeout: int | None = None,
         is_api_execution: bool = False,
     ) -> ExecutionResponse:
+        hash_values_of_files = hash_values_of_files or {}
         if pipeline_id:
             logger.info(f"Executing pipeline: {pipeline_id}")
             # Create a new WorkflowExecution entity for each pipeline execution.
@@ -1024,8 +1026,9 @@ def step_execution(
         workflow: Workflow,
         execution_action: str,
         execution_id: str | None = None,
-        hash_values_of_files: dict[str, FileHash] = {},
+        hash_values_of_files: dict[str, FileHash] | None = None,
     ) -> ExecutionResponse:
+        hash_values_of_files = hash_values_of_files or {}
         if execution_action is Workflow.ExecutionAction.START.value:  # type: ignore
             if execution_id is None:
                 return WorkflowHelper.create_and_make_execution_response(