diff --git a/lib/galaxy/jobs/__init__.py b/lib/galaxy/jobs/__init__.py
index 8cfafec9291..c826f2de2e4 100644
--- a/lib/galaxy/jobs/__init__.py
+++ b/lib/galaxy/jobs/__init__.py
@@ -921,6 +921,10 @@ class JobWrapper(HasResourceParameters):
     def requires_containerization(self):
         return util.asbool(self.get_destination_configuration("require_container", "False"))
 
+    @property
+    def use_metadata_binary(self):
+        return util.asbool(self.get_destination_configuration('use_metadata_binary', "False"))
+
     def can_split(self):
         # Should the job handler split this job up?
         return self.app.config.use_tasked_jobs and self.tool.parallelism
diff --git a/lib/galaxy/jobs/command_factory.py b/lib/galaxy/jobs/command_factory.py
index db5cff813ce..adccf42b427 100644
--- a/lib/galaxy/jobs/command_factory.py
+++ b/lib/galaxy/jobs/command_factory.py
@@ -204,6 +204,7 @@ def __handle_metadata(commands_builder, job_wrapper, runner, remote_command_para
         datatypes_config=datatypes_config,
         compute_tmp_dir=compute_tmp_dir,
         resolve_metadata_dependencies=resolve_metadata_dependencies,
+        use_bin=job_wrapper.use_metadata_binary,
         kwds={'overwrite': False}
     ) or ''
     metadata_command = metadata_command.strip()
diff --git a/lib/galaxy/jobs/runners/pulsar.py b/lib/galaxy/jobs/runners/pulsar.py
index eda05554b6f..16829353477 100644
--- a/lib/galaxy/jobs/runners/pulsar.py
+++ b/lib/galaxy/jobs/runners/pulsar.py
@@ -699,7 +699,7 @@ class PulsarJobRunner(AsynchronousJobRunner):
 
     def __build_metadata_configuration(self, client, job_wrapper, remote_metadata, remote_job_config):
         metadata_kwds = {}
-        if remote_metadata:
+        if remote_metadata and not job_wrapper.use_metadata_binary:
             remote_system_properties = remote_job_config.get("system_properties", {})
             remote_galaxy_home = remote_system_properties.get("galaxy_home", None)
             if not remote_galaxy_home:
diff --git a/lib/galaxy/metadata/__init__.py b/lib/galaxy/metadata/__init__.py
index bd16701e6b0..7b78d39072c 100644
--- a/lib/galaxy/metadata/__init__.py
+++ b/lib/galaxy/metadata/__init__.py
@@ -46,7 +46,7 @@ class MetadataCollectionStrategy(object):
     @abc.abstractmethod
     def setup_external_metadata(self, datasets_dict, sa_session, exec_dir=None,
                                 tmp_dir=None, dataset_files_path=None,
-                                output_fnames=None, config_root=None,
+                                output_fnames=None, config_root=None, use_bin=False,
                                 config_file=None, datatypes_config=None,
                                 job_metadata=None, compute_tmp_dir=None,
                                 include_command=True, max_metadata_value_size=0,
@@ -99,7 +99,7 @@ class PortableDirectoryMetadataGenerator(MetadataCollectionStrategy):
 
     def setup_external_metadata(self, datasets_dict, sa_session, exec_dir=None,
                                 tmp_dir=None, dataset_files_path=None,
-                                output_fnames=None, config_root=None,
+                                output_fnames=None, config_root=None, use_bin=False,
                                 config_file=None, datatypes_config=None,
                                 job_metadata=None, compute_tmp_dir=None,
                                 include_command=True, max_metadata_value_size=0,
@@ -147,9 +147,12 @@ class PortableDirectoryMetadataGenerator(MetadataCollectionStrategy):
         if include_command:
             # return command required to build
             script_path = os.path.join(metadata_dir, "set.py")
-            with open(script_path, "w") as f:
-                f.write(SET_METADATA_SCRIPT)
-            return 'python "metadata/set.py"'
+            if use_bin:
+                return "galaxy-set-metadata"
+            else:
+                with open(script_path, "w") as f:
+                    f.write(SET_METADATA_SCRIPT)
+                return 'python "metadata/set.py"'
         else:
             # return args to galaxy_ext.metadata.set_metadata required to build
             return ''
@@ -214,7 +217,7 @@ class JobExternalOutputMetadataWrapper(MetadataCollectionStrategy):
 
     def setup_external_metadata(self, datasets_dict, sa_session, exec_dir=None,
                                 tmp_dir=None, dataset_files_path=None,
-                                output_fnames=None, config_root=None,
+                                output_fnames=None, config_root=None, use_bin=False,
                                 config_file=None, datatypes_config=None,
                                 job_metadata=None, compute_tmp_dir=None,
                                 include_command=True, max_metadata_value_size=0,
@@ -296,6 +299,7 @@ class JobExternalOutputMetadataWrapper(MetadataCollectionStrategy):
                                     job_metadata,
                                     " ".join(map(__metadata_files_list_to_cmd_line, metadata_files_list)),
                                     max_metadata_value_size)
+        assert not use_bin
         if include_command:
             # return command required to build
             fd, fp = tempfile.mkstemp(suffix='.py', dir=tmp_dir, prefix="set_metadata_")
diff --git a/test/unit/jobs/test_command_factory.py b/test/unit/jobs/test_command_factory.py
index 28f16c58f7b..f997190badf 100644
--- a/test/unit/jobs/test_command_factory.py
+++ b/test/unit/jobs/test_command_factory.py
@@ -178,6 +178,7 @@ class MockJobWrapper(object):
             )
         )
         self.shell = "/bin/sh"
+        self.use_metadata_binary = False
 
     def get_command_line(self):
         return self.command_line
diff --git a/test/unit/jobs/test_runner_local.py b/test/unit/jobs/test_runner_local.py
index 610da86a66f..7675b96a577 100644
--- a/test/unit/jobs/test_runner_local.py
+++ b/test/unit/jobs/test_runner_local.py
@@ -144,6 +144,7 @@ class MockJobWrapper(object):
         self.shell = "/bin/bash"
         self.cleanup_job = "never"
         self.tmp_dir_creation_statement = ""
+        self.use_metadata_binary = False
 
         # Cruft for setting metadata externally, axe at some point.
         self.external_output_metadata = bunch.Bunch(