Make ServerKernelManager client traits configurable

Add config=True to client_class and client_factory traits in ServerKernelManager
to enable configuration of kernel client classes through configuration files and
command line arguments.

This enables custom kernel clients to be used for intercepting and routing
messages, which is required for server-side document handling.

- Override client_class and client_factory traits with config=True
- Add comprehensive unit tests for configuration via command line and Config object
- Move all imports to top of test file for better organization

Author: Zsailer

jupyter_server/services/kernels/kernelmanager.py

@@ -37,11 +37,13 @@
     Any,
     Bool,
     Dict,
+    DottedObjectName,
     Float,
     Instance,
     Integer,
     List,
     TraitError,
+    Type,
     Unicode,
     default,
     validate,
@@ -845,6 +847,29 @@ async def wrapped_method(self, *args, **kwargs):
 class ServerKernelManager(AsyncIOLoopKernelManager):
     """A server-specific kernel manager."""
 
+    # Override parent traits to make them configurable
+    client_class = DottedObjectName(
+        "jupyter_client.asynchronous.AsyncKernelClient",
+        config=True,
+        help="""The kernel client class to use for communicating with kernels.
+
+        This should be a subclass of KernelClient, and it should accept the
+        following arguments:
+        - kernel_manager
+        - blocking
+        - loop
+        """
+    )
+
+    client_factory = Type(
+        klass="jupyter_client.KernelClient",
+        config=True,
+        help="""The kernel client factory class to use for creating client instances.
+
+        This should be a subclass of KernelClient.
+        """
+    )
+
     # Define activity-related attributes:
     execution_state = Unicode(
         None, allow_none=True, help="The current execution state of the kernel"

tests/services/kernels/test_config.py

@@ -1,7 +1,9 @@
 import pytest
 from traitlets.config import Config
 
-from jupyter_server.services.kernels.kernelmanager import AsyncMappingKernelManager
+from jupyter_client.blocking.client import BlockingKernelClient
+from jupyter_server.services.kernels.kernelmanager import AsyncMappingKernelManager, ServerKernelManager
+from jupyter_server.utils import import_item
 
 
 @pytest.fixture
@@ -29,3 +31,40 @@ def test_not_server_kernel_manager(jp_configurable_serverapp):
     ]
     with pytest.warns(FutureWarning, match="is not a subclass of 'ServerKernelManager'"):
         jp_configurable_serverapp(argv=argv)
+
+
+def test_server_kernel_manager_client_class_configurable(jp_configurable_serverapp):
+    """Test that ServerKernelManager client_class trait is configurable."""
+    argv = [
+        "--AsyncMappingKernelManager.client_class=jupyter_client.asynchronous.AsyncKernelClient"
+    ]
+    app = jp_configurable_serverapp(argv=argv)
+    # Get a kernel manager instance to check the configuration
+    kernel_manager_class = app.kernel_manager.kernel_manager_class
+    km_class = import_item(kernel_manager_class)
+    km_instance = km_class()
+    assert km_instance.client_class == "jupyter_client.asynchronous.AsyncKernelClient"
+
+
+def test_server_kernel_manager_client_factory_configurable(jp_configurable_serverapp):
+    """Test that ServerKernelManager client_factory trait is configurable."""
+    argv = [
+        "--AsyncMappingKernelManager.client_factory=jupyter_client.blocking.client.BlockingKernelClient"
+    ]
+    app = jp_configurable_serverapp(argv=argv)
+    # Get a kernel manager instance to check the configuration
+    kernel_manager_class = app.kernel_manager.kernel_manager_class
+    km_class = import_item(kernel_manager_class)
+    km_instance = km_class()
+    assert km_instance.client_factory == BlockingKernelClient
+
+
+def test_server_kernel_manager_client_traits_via_config():
+    """Test that ServerKernelManager client traits can be configured via Config object."""
+    config = Config()
+    config.ServerKernelManager.client_class = "jupyter_client.blocking.client.BlockingKernelClient"
+    config.ServerKernelManager.client_factory = BlockingKernelClient
+
+    km = ServerKernelManager(config=config)
+    assert km.client_class == "jupyter_client.blocking.client.BlockingKernelClient"
+    assert km.client_factory == BlockingKernelClient