diff --git a/renku/core/workflow/plan.py b/renku/core/workflow/plan.py
index 3b2e4f67b0..db246d2c8a 100644
--- a/renku/core/workflow/plan.py
+++ b/renku/core/workflow/plan.py
@@ -151,6 +151,19 @@ def remove_plan(name_or_id: str, force: bool, plan_gateway: IPlanGateway, when:
     if latest_version.deleted:
         raise errors.ParameterError(f"The specified workflow '{name_or_id}' is already deleted.")
 
+    composites_containing_child = get_composite_plans_by_child(plan)
+
+    if composites_containing_child:
+        composite_names = "\n\t".join([c.name for c in composites_containing_child])
+
+        if not force:
+            raise errors.ParameterError(
+                f"The specified workflow '{name_or_id}' is part of the following composite workflows and won't be "
+                f"removed (use '--force' to remove anyways):\n\t{composite_names}"
+            )
+        else:
+            communication.warn(f"Removing '{name_or_id}', which is still used in these workflows:\n\t{composite_names}")
+
     if not force:
         prompt_text = f"You are about to remove the following workflow '{name_or_id}'.\n\nDo you wish to continue?"
         communication.confirm(prompt_text, abort=True, warning=True)
@@ -632,3 +645,16 @@ def is_plan_removed(plan: AbstractPlan) -> bool:
             return True
 
     return False
+
+
+@inject.autoparams()
+def get_composite_plans_by_child(plan: AbstractPlan, plan_gateway: IPlanGateway) -> List[CompositePlan]:
+    """Return all composite plans that contain a child plan."""
+
+    derivatives = {p.id for p in get_derivative_chain(plan=plan)}
+
+    composites = (p for p in plan_gateway.get_newest_plans_by_names().values() if isinstance(p, CompositePlan))
+
+    composites_containing_child = [c for c in composites if {p.id for p in c.plans}.intersection(derivatives)]
+
+    return composites_containing_child
diff --git a/renku/ui/cli/workflow.py b/renku/ui/cli/workflow.py
index 65ee029210..c993865ee3 100644
--- a/renku/ui/cli/workflow.py
+++ b/renku/ui/cli/workflow.py
@@ -806,7 +806,9 @@ def remove(name, force):
     """Remove a workflow named <name>."""
     from renku.command.workflow import remove_plan_command
 
-    remove_plan_command().build().execute(name_or_id=name, force=force)
+    communicator = ClickCallback()
+
+    remove_plan_command().with_communicator(communicator).build().execute(name_or_id=name, force=force)
 
 
 @workflow.command()
diff --git a/tests/api/test_plan.py b/tests/api/test_plan.py
index f84b423f29..cad4c5a794 100644
--- a/tests/api/test_plan.py
+++ b/tests/api/test_plan.py
@@ -36,7 +36,7 @@ def test_list_plans(client_with_runs):
 
 def test_list_deleted_plans(client_with_runs, runner):
     """Test listing deleted plans."""
-    result = runner.invoke(cli, ["workflow", "remove", "plan-1"])
+    result = runner.invoke(cli, ["workflow", "remove", "--force", "plan-1"])
     assert 0 == result.exit_code, format_result_exception(result)
 
     plans = Plan.list()
diff --git a/tests/cli/test_workflow.py b/tests/cli/test_workflow.py
index a0d3161d04..df95aaacae 100644
--- a/tests/cli/test_workflow.py
+++ b/tests/cli/test_workflow.py
@@ -369,6 +369,31 @@ def test_workflow_remove_command(runner, project):
     assert 2 == result.exit_code, format_result_exception(result)
 
 
+def test_workflow_remove_with_composite_command(runner, project):
+    """Test workflow remove with builder."""
+    workflow_name = "test_workflow"
+
+    result = runner.invoke(cli, ["workflow", "remove", workflow_name])
+    assert 2 == result.exit_code
+
+    result = runner.invoke(cli, ["run", "--success-code", "0", "--no-output", "--name", workflow_name, "echo", "foo"])
+    assert 0 == result.exit_code, format_result_exception(result)
+
+    result = runner.invoke(cli, ["workflow", "compose", "composed-workflow", workflow_name])
+    assert 0 == result.exit_code, format_result_exception(result)
+
+    result = runner.invoke(cli, ["workflow", "remove", workflow_name])
+    assert 2 == result.exit_code, format_result_exception(result)
+    assert (
+        "The specified workflow 'test_workflow' is part of the following composite workflows and won't be removed"
+        in result.stderr
+    )
+
+    result = runner.invoke(cli, ["workflow", "remove", "--force", workflow_name])
+    assert 0 == result.exit_code, format_result_exception(result)
+    assert "Removing 'test_workflow', which is still used in these workflows" in result.output
+
+
 def test_workflow_export_command(runner, project):
     """Test workflow export with builder."""
     result = runner.invoke(cli, ["run", "--success-code", "0", "--no-output", "--name", "run1", "touch", "data.csv"])