Support for listing kernels

jupyter_kernel_client/client.py

@@ -228,6 +228,10 @@ def last_activity(self) -> datetime.datetime | None:
             else None
         )
 
+    def list_kernels(self) -> list[dict[str, t.Any]]:
+        """List the running kernels."""
+        return self._manager.list_kernels()
+
     @property
     def username(self) -> str:
         """Client owner username."""

jupyter_kernel_client/manager.py

@@ -190,6 +190,26 @@ def refresh_model(self, timeout: float = REQUEST_TIMEOUT) -> dict[str, t.Any] |
             self.__client = None
         return model
 
+    def list_kernels(self, timeout: float = REQUEST_TIMEOUT) -> list[dict[str, t.Any]]:
+        """List the running kernels.
+
+        Returns
+        -------
+            The list of running kernels.
+        """
+        kernels_url = url_path_join(self.server_url, "api/kernels")
+        self.log.debug("Request kernels at: %s", kernels_url)
+        try:
+            response = fetch(kernels_url, token=self.token, method="GET", timeout=timeout, headers=self.__extra_headers)
+        except HTTPError as error:
+            self.log.error("Error fetching kernels: %s", error)
+            return []
+        else:
+            models = response.json()
+        self.log.debug("Kernels retrieved: %s", models)
+        return models
+
+
     # --------------------------------------------------------------------------
     # Kernel management
     # --------------------------------------------------------------------------

jupyter_kernel_client/tests/test_client.py

@@ -61,6 +61,31 @@ def test_execution_no_context_manager(jupyter_server):
     assert reply["status"] == "ok"
 
 
+def test_list_kernels_client(jupyter_server):
+    port, token = jupyter_server
+
+    # Start a kernel to ensure the list is not empty
+    with KernelClient(server_url=f"http://localhost:{port}", token=token) as kernel:
+        kernel_id = kernel.id
+
+        # Use a new client to list the kernels
+        listing_client = KernelClient(server_url=f"http://localhost:{port}", token=token)
+        kernels = listing_client.list_kernels()
+
+        assert isinstance(kernels, list)
+        assert len(kernels) > 0
+
+        # Check that the kernel we started is in the list
+        found = False
+        for k in kernels:
+            assert "id" in k
+            assert "name" in k
+            if k["id"] == kernel_id:
+                found = True
+
+        assert found, f"Kernel with id {kernel_id} not found in the list of running kernels."
+
+
 def test_list_variables(jupyter_server):
     port, token = jupyter_server
 
@@ -212,15 +237,18 @@ def test_set_variables(jupyter_server, variable, set_variable, expected):
 async def test_multi_execution_in_event_loop(jupyter_server):
     port, token = jupyter_server
 
+    current_user = os.environ.get('USER', 'John Smith')
+    current_node = node()
+
     with KernelClient(server_url=f"http://localhost:{port}", token=token) as kernel:
         all = await asyncio.gather(
             asyncio.to_thread(
                 kernel.execute,
-                """import os
+                f"""import os
 from platform import node
 import time
 time.sleep(5)
-print(f"Hey {os.environ.get('USER', 'John Smith')} from {node()}.")
+print(f"Hey {{os.environ.get('USER', 'John Smith')}} from {{node()}}.")
 """
             ),
             asyncio.to_thread(
@@ -236,7 +264,7 @@ async def test_multi_execution_in_event_loop(jupyter_server):
             {
                 "output_type": "stream",
                 "name": "stdout",
-                "text": f"Hey {os.environ.get('USER', 'John Smith')} from {node()}.\n",
+                "text": f"Hey {current_user} from {current_node}.\n",
             }
         ]
         assert all[0]["status"] == "ok"

jupyter_kernel_client/tests/test_manager.py

@@ -0,0 +1,33 @@
+
+# Copyright (c) 2023-2024 Datalayer, Inc.
+# Copyright (c) 2025      Google
+#
+# BSD 3-Clause License
+
+from jupyter_kernel_client.manager import KernelHttpManager
+from jupyter_kernel_client import KernelClient
+
+
+def test_list_kernels(jupyter_server):
+    port, token = jupyter_server
+
+    # Start a kernel to ensure the list is not empty
+    with KernelClient(server_url=f"http://localhost:{port}", token=token) as kernel:
+        kernel_id = kernel.id
+        # The manager is created after the kernel is started to ensure we can list it.
+        manager = KernelHttpManager(server_url=f"http://localhost:{port}", token=token)
+        kernels = manager.list_kernels()
+
+        assert isinstance(kernels, list)
+        assert len(kernels) > 0
+
+        # Check that the kernel we started is in the list
+        found = False
+        for k in kernels:
+            assert "id" in k
+            assert "name" in k
+            if k["id"] == kernel_id:
+                found = True
+
+        assert found, f"Kernel with id {kernel_id} not found in the list of running kernels."
+