diff --git a/broker/commands.py b/broker/commands.py
index 11a4338a..d4c73cfd 100644
--- a/broker/commands.py
+++ b/broker/commands.py
@@ -36,6 +36,14 @@ def wrapper(*args, **kwargs):
     return decorator
 
 
+def parse_labels(provider_labels):
+    """Parse the provided label string and returns labels in a dict."""
+    return {
+        label[0]: "=".join(label[1:])
+        for label in [kv_pair.split("=") for kv_pair in provider_labels.split(",")]
+    }
+
+
 class ExceptionHandler(click.Group):
     """Wraps click group to catch and handle raised exceptions."""
 
@@ -189,6 +197,14 @@ 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(
+    "-l",
+    "--provider-labels",
+    type=str,
+    help="A string representing the list"
+    " of k=v pairs (comma-separated) to be used as provider resource"
+    " labels (e.g. '-l k1=v1,k2=v2,k3=v3=z4').",
+)
 @click.option(
     "--args-file",
     type=click.Path(exists=True),
@@ -196,7 +212,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, provider_labels, **kwargs):
     """Checkout or "create" a Virtual Machine broker instance.
 
     COMMAND: broker checkout --workflow "workflow-name" --workflow_arg1 something
@@ -210,6 +226,9 @@ def checkout(ctx, background, nick, count, args_file, **kwargs):
         broker_args["_count"] = count
     if args_file:
         broker_args["args_file"] = args_file
+    if provider_labels:
+        broker_args["provider_labels"] = parse_labels(provider_labels)
+
     # if additional arguments were passed, include them in the broker args
     # strip leading -- characters
     broker_args.update(
@@ -291,6 +310,14 @@ def inventory(details, sync, filter):
 @click.option("--all", "all_", is_flag=True, help="Select all VMs")
 @click.option("--sequential", is_flag=True, help="Run extends sequentially")
 @click.option("--filter", type=str, help="Extend only what matches the specified filter")
+@click.option(
+    "-l",
+    "--provider-labels",
+    type=str,
+    help="A string representing the list"
+    " of k=v pairs (comma-separated) to be used as provider resource"
+    " labels (e.g. '-l k1=v1,k2=v2,k3=v3=z4').",
+)
 @provider_options
 def extend(vm, background, all_, sequential, filter, **kwargs):
     """Extend a host's lease time.
@@ -322,9 +349,17 @@ def extend(vm, background, all_, sequential, filter, **kwargs):
     type=click.Path(exists=True),
     help="A json or yaml file mapping arguments to values",
 )
+@click.option(
+    "-l",
+    "--provider-labels",
+    type=str,
+    help="A string representing the list"
+    " of k=v pairs (comma-separated) to be used as provider resource"
+    " labels (e.g. '-l k1=v1,k2=v2,k3=v3=z4').",
+)
 @provider_options
 @click.pass_context
-def execute(ctx, background, nick, output_format, artifacts, args_file, **kwargs):
+def execute(ctx, background, nick, output_format, artifacts, args_file, provider_labels, **kwargs):
     """Execute an arbitrary provider action.
 
     COMMAND: broker execute --workflow "workflow-name" --workflow_arg1 something
@@ -338,6 +373,8 @@ def execute(ctx, background, nick, output_format, artifacts, args_file, **kwargs
         broker_args["artifacts"] = artifacts
     if args_file:
         broker_args["args_file"] = args_file
+    if provider_labels:
+        broker_args["provider_labels"] = parse_labels(provider_labels)
     # 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 2f051b3a..65052579 100644
--- a/broker/providers/ansible_tower.py
+++ b/broker/providers/ansible_tower.py
@@ -436,6 +436,25 @@ def _pull_extra_vars(extra_vars):
                 compiled[key] = val
             return compiled
 
+    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:
+            label_expanded = f"{label}={labels[label]}" if labels[label] else label
+            if result := self.v2.labels.get(name=label_expanded).results:
+                label_ids.append(result[0].id)
+            else:
+                # label does not exist yet, creating
+                result = self.v2.labels.post(
+                    {"name": label_expanded, "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."""
@@ -536,7 +555,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
@@ -572,12 +591,24 @@ 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)}")
+
         elif self.inventory:
             payload["inventory"] = self.inventory
             logger.info(f"Using tower inventory: {self._translate_inventory(self.inventory)}")
         else:
             logger.info("No inventory specified, Ansible Tower will use a default.")
 
+        # provider labels handling
+
+        provider_labels = kwargs.get("provider_labels", {})
+        # include eventual common labels, specified at each level of configuration
+        # typically imported from dynaconf env vars
+        provider_labels.update(settings.get("provider_labels", {}))
+        provider_labels.update(settings.ANSIBLETOWER.get("provider_labels", {}))
+        if provider_labels:
+            payload["labels"] = self._resolve_labels(provider_labels, target)
+            kwargs["provider_labels"] = provider_labels
+
         # Save custom, non-workflow extra vars to a named variable.
         # The workflow can save these values to job artifacts / host facts.
         workflow_extra_vars = self._pull_extra_vars(target.extra_vars)
@@ -628,7 +659,7 @@ def get_inventory(self, user=None):
             compiled_host_info = [self._compile_host_info(host) for host in hosts_bar]
         return compiled_host_info
 
-    def extend(self, target_vm, new_expire_time=None):
+    def extend(self, target_vm, new_expire_time=None, provider_labels=None):
         """Run the extend workflow with defaults args.
 
         :param target_vm: This should be a host object
@@ -643,6 +674,7 @@ def extend(self, target_vm, new_expire_time=None):
             workflow=settings.ANSIBLETOWER.extend_workflow,
             target_vm=target_vm.name,
             new_expire_time=new_expire_time or settings.ANSIBLETOWER.get("new_expire_time"),
+            provider_labels=provider_labels,
         )
 
     def provider_help(
diff --git a/broker/providers/container.py b/broker/providers/container.py
index 218c4358..ac61c994 100644
--- a/broker/providers/container.py
+++ b/broker/providers/container.py
@@ -274,7 +274,22 @@ def run_container(self, container_host, **kwargs):
         if origin[1]:
             envars["JENKINS_URL"] = origin[1]
         kwargs["environment"] = envars
-        kwargs["labels"] = envars
+
+        # process eventual provider labels for each setting level
+        kwargs["provider_labels"] = kwargs.get("provider_labels", {})
+        kwargs["provider_labels"].update(settings.get("provider_labels", {}))
+        kwargs["provider_labels"].update(settings.CONTAINER.get("provider_labels", {}))
+        # prefix eventual label keys with 'broker.' to conform to the docker guidelines
+        # https://docs.docker.com/config/labels-custom-metadata/#key-format-recommendations
+        kwargs["provider_labels"] = {
+            f"broker.{label[0]}": label[1] for label in kwargs.get("provider_labels", {}).items()
+        }
+        # process eventual labels that were passed externally, split by "="
+        kwargs["provider_labels"].update(
+            {"broker.origin": origin[0], "broker.jenkins.url": origin[1]}
+        )
+        # rename the dict key to the name of the arg recognized by provider
+        kwargs["labels"] = kwargs.pop("provider_labels")
         container_inst = self.runtime.create_container(container_host, **kwargs)
         container_inst.start()
         return container_inst
diff --git a/tests/functional/test_containers.py b/tests/functional/test_containers.py
index 027d99e4..4373863d 100644
--- a/tests/functional/test_containers.py
+++ b/tests/functional/test_containers.py
@@ -67,7 +67,7 @@ def test_containerhost_query():
 
 
 def test_container_e2e():
-    with Broker(container_host="ubi8:latest") as c_host:
+    with Broker(container_host="ubi8:latest", provider_labels={"l1": "v1", "l2": None}) as c_host:
         assert c_host._cont_inst.top()["Processes"]
         res = c_host.execute("hostname")
         assert res.stdout.strip() == c_host.hostname
@@ -87,6 +87,9 @@ def test_container_e2e():
                 SETTINGS_PATH.read_bytes() == data
             ), "Local file is different from the received one (return_data=True)"
             assert data == Path(tmp.file.name).read_bytes(), "Received files do not match"
+        # assert labels
+        assert c_host._cont_inst.labels.get("broker.l1") == "v1"
+        assert c_host._cont_inst.labels.get("broker.l2") == ""
         # test the tail_file context manager
         tailed_file = f"{remote_dir}/tail_me.txt"
         c_host.execute(f"echo 'hello world' > {tailed_file}")
diff --git a/tests/functional/test_satlab.py b/tests/functional/test_satlab.py
index a2997915..e108a274 100644
--- a/tests/functional/test_satlab.py
+++ b/tests/functional/test_satlab.py
@@ -124,3 +124,17 @@ def test_tower_host_mp():
         )
         res = r_hosts[1].execute(f"ls /root")
         assert SETTINGS_PATH.name in res.stdout
+
+
+def test_tower_provider_labels():
+    """Assert labels being created on AAP and OSP metadata 
+    being attached accordingly
+    """
+    with Broker(workflow="deploy-rhel", provider_labels={"l1": "v1", "l2": ""}) as r_host:
+        # check provider labels in the resulting host object
+        assert r_host.provider_labels.get("l1") == "v1"
+        assert r_host.provider_labels.get("l2") == ""
+        # assert the AAP labels got created on the provider
+        aap_labels = [l.name for l in r_host._prov_inst.v2.labels.get().results]
+        assert "l1=v1" in aap_labels
+        assert "l2" in aap_labels