Check plugin registration status

distributed/client.py

@@ -5456,6 +5456,71 @@ def unregister_worker_plugin(self, name, nanny=None):
         """
         return self.sync(self._unregister_worker_plugin, name=name, nanny=nanny)
 
+    def has_plugin(
+        self, plugin: str | WorkerPlugin | SchedulerPlugin | NannyPlugin | list
+    ) -> bool | dict[str, bool]:
+        """Check if plugin(s) are registered
+
+        Parameters
+        ----------
+        plugin : str | plugin object | list
+            Plugin to check. You can use the plugin object directly or the plugin name. For plugin objects, they must have a 'name' attribute. You can also pass a list of plugin objects or names.
+
+        Returns
+        -------
+        bool or dict[str, bool]
+            If name is str: True if plugin is registered, False otherwise
+            If name is list: dict mapping names to registration status
+
+        Examples
+        --------
+        >>> logging_plugin = LoggingConfigPlugin()  # Has name = "logging-config"
+        >>> client.register_plugin(logging_plugin)
+        >>> client.has_plugin(logging_plugin)
+        True
+
+        >>> client.has_plugin('logging-config')
+        True
+
+        >>> client.has_plugin([logging_plugin, 'other-plugin'])
+        {'logging-config': True, 'other-plugin': False}
+        """
+        if isinstance(plugin, str):
+            result = self.sync(self._get_plugin_registration_status, names=[plugin])
+            return result[plugin]
+
+        elif isinstance(plugin, (WorkerPlugin, SchedulerPlugin, NannyPlugin)):
+            plugin_name = getattr(plugin, "name", None)
+            if plugin_name is None:
+                raise ValueError(
+                    f"Plugin {funcname(type(plugin))} has no 'name' attribute. "
+                    "Please add a 'name' attribute to your plugin class."
+                )
+            result = self.sync(
+                self._get_plugin_registration_status, names=[plugin_name]
+            )
+            return result[plugin_name]
+
+        elif isinstance(plugin, list):
+            names_to_check = []
+            for p in plugin:
+                if isinstance(p, str):
+                    names_to_check.append(p)
+                else:
+                    plugin_name = getattr(p, "name", None)
+                    if plugin_name is None:
+                        raise ValueError(
+                            f"Plugin {funcname(type(p))} has no 'name' attribute"
+                        )
+                    names_to_check.append(plugin_name)
+            return self.sync(self._get_plugin_registration_status, names=names_to_check)
+
+    async def _get_plugin_registration_status(
+        self, names: list[str]
+    ) -> dict[str, bool]:
+        """Async implementation for checking plugin registration"""
+        return await self.scheduler.get_plugin_registration_status(names=names)
+
     @property
     def amm(self):
         """Convenience accessors for the :doc:`active_memory_manager`"""

distributed/scheduler.py

@@ -4039,6 +4039,7 @@ async def post(self) -> None:
             "unregister_worker_plugin": self.unregister_worker_plugin,
             "register_nanny_plugin": self.register_nanny_plugin,
             "unregister_nanny_plugin": self.unregister_nanny_plugin,
+            "get_plugin_registration_status": self.get_plugin_registration_status,
             "adaptive_target": self.adaptive_target,
             "workers_to_close": self.workers_to_close,
             "subscribe_worker_status": self.subscribe_worker_status,
@@ -8696,6 +8697,50 @@ async def get_worker_monitor_info(
         )
         return dict(zip(self.workers, results))
 
+    async def get_plugin_registration_status(self, names: list[str]) -> dict[str, bool]:
+        """Check if plugins are registered in any plugin registry
+
+        Checks all plugin registries (worker, scheduler, nanny) and returns True
+        if the plugin is found in any of them.
+
+        Parameters
+        ----------
+        names : list[str]
+            List of plugin names to check
+
+        Returns
+        -------
+        dict[str, bool]
+            Dict mapping plugin names to their registration status across all registries
+        """
+        result = {}
+        for name in names:
+            # Check if plugin exists in any registry
+            result[name] = (
+                name in self.worker_plugins
+                or name in self.plugins
+                or name in self.nanny_plugins
+            )
+        return result
+
+    async def get_worker_plugin_registration_status(
+        self, names: list[str]
+    ) -> dict[str, bool]:
+        """Check if worker plugins are registered"""
+        return {name: name in self.worker_plugins for name in names}
+
+    async def get_scheduler_plugin_registration_status(
+        self, names: list[str]
+    ) -> dict[str, bool]:
+        """Check if scheduler plugins are registered"""
+        return {name: name in self.plugins for name in names}
+
+    async def get_nanny_plugin_registration_status(
+        self, names: list[str]
+    ) -> dict[str, bool]:
+        """Check if nanny plugins are registered"""
+        return {name: name in self.nanny_plugins for name in names}
+
     ###########
     # Cleanup #
     ###########