diff --git a/broker/commands.py b/broker/commands.py
index b4f165a6..9c8f3202 100644
--- a/broker/commands.py
+++ b/broker/commands.py
@@ -187,6 +187,9 @@ def cli(version):
 @click.option("-b", "--background", is_flag=True, help="Run checkout in the background")
 @click.option("-n", "--nick", type=str, help="Use a nickname defined in your settings")
 @click.option("-c", "--count", type=int, help="Number of times broker repeats the checkout")
+@click.option(
+    "--labels", type=str, help="List of strings (comma-separated) to be used as AAP labels"
+)  # fixme - this should be provider-agnostic
 @click.option(
     "--args-file",
     type=click.Path(exists=True),
@@ -194,7 +197,7 @@ def cli(version):
 )
 @provider_options
 @click.pass_context
-def checkout(ctx, background, nick, count, args_file, **kwargs):
+def checkout(ctx, background, nick, count, args_file, labels, **kwargs):
     """Checkout or "create" a Virtual Machine broker instance.
 
     COMMAND: broker checkout --workflow "workflow-name" --workflow_arg1 something
@@ -208,6 +211,8 @@ def checkout(ctx, background, nick, count, args_file, **kwargs):
         broker_args["_count"] = count
     if args_file:
         broker_args["args_file"] = args_file
+    if labels:
+        broker_args["labels"] = labels.split(",")
     # if additional arguments were passed, include them in the broker args
     # strip leading -- characters
     broker_args.update(
diff --git a/broker/providers/ansible_tower.py b/broker/providers/ansible_tower.py
index 72202368..bf15551c 100644
--- a/broker/providers/ansible_tower.py
+++ b/broker/providers/ansible_tower.py
@@ -418,6 +418,24 @@ def _get_broker_args_from_job(self, host):
         )
         return _broker_args
 
+    def _resolve_labels(self, labels, target):
+        """Fetch and return ids of the given labels.
+
+        If label does not exist, create it under the same org as the target template.
+        """
+        label_ids = []
+        for label in labels:
+            if result := self.v2.labels.get(name=label).results:
+                label_ids.append(result[0].id)
+            else:
+                # label does not exist yet, creating
+                result = self.v2.labels.post(
+                    {"name": label, "organization": target.summary_fields.organization.id}
+                )
+                if result:
+                    label_ids.append(result.id)
+        return label_ids
+
     @cached_property
     def inventory(self):
         """Return the current tower inventory."""
@@ -518,7 +536,7 @@ def _get_fields_from_facts(facts):
         return host_inst
 
     @Provider.register_action("workflow", "job_template")
-    def execute(self, **kwargs):  # noqa: PLR0912 - Possible TODO refactor
+    def execute(self, **kwargs):  # noqa: PLR0912,PLR0915 - Possible TODO refactor
         """Execute workflow or job template in Ansible Tower.
 
         :param kwargs: workflow or job template name passed in a string
@@ -554,6 +572,9 @@ def execute(self, **kwargs):  # noqa: PLR0912 - Possible TODO refactor
         if inventory := kwargs.pop("inventory", None):
             payload["inventory"] = inventory
             logger.info(f"Using tower inventory: {self._translate_inventory(inventory)}")
+        if labels := kwargs.pop("labels", None):
+            payload["labels"] = self._resolve_labels(labels, target)
+            kwargs["_labels"] = ",".joint(labels)
         elif self.inventory:
             payload["inventory"] = self.inventory
             logger.info(f"Using tower inventory: {self._translate_inventory(self.inventory)}")