diff --git a/jupyter_client/_version.py b/jupyter_client/_version.py
index 3805b913..3e0649c0 100644
--- a/jupyter_client/_version.py
+++ b/jupyter_client/_version.py
@@ -2,7 +2,7 @@
 import re
 from typing import List, Union
 
-__version__ = "8.4.0"
+__version__ = "8.4.2"
 
 # Build up version_info tuple for backwards compatibility
 pattern = r'(?P<major>\d+).(?P<minor>\d+).(?P<patch>\d+)(?P<rest>.*)'
diff --git a/jupyter_client/multikernelmanager.py b/jupyter_client/multikernelmanager.py
index 2ebd0e9d..1b5d2013 100644
--- a/jupyter_client/multikernelmanager.py
+++ b/jupyter_client/multikernelmanager.py
@@ -18,7 +18,7 @@
 from .connect import KernelConnectionInfo
 from .kernelspec import NATIVE_KERNEL_NAME, KernelSpecManager
 from .manager import KernelManager
-from .utils import ensure_async, run_sync, utcnow
+from .utils import PORTS_ENV_MAP, ensure_async, run_sync, utcnow
 
 
 class DuplicateKernelError(Exception):
@@ -201,6 +201,12 @@ def pre_start_kernel(
         # subclass we are using. It can be configured as any Configurable,
         # including things like its transport and ip.
         constructor_kwargs = {}
+        # ------ set ports from env begin
+        env = kwargs.get("env", {})
+        for arg_key, env_key in PORTS_ENV_MAP.items():
+            if env_key in env:
+                constructor_kwargs[arg_key] = int(env[env_key])
+        # ------ set ports from env end
         if self.kernel_spec_manager:
             constructor_kwargs["kernel_spec_manager"] = self.kernel_spec_manager
         km = self.kernel_manager_factory(
diff --git a/jupyter_client/provisioning/local_provisioner.py b/jupyter_client/provisioning/local_provisioner.py
index ffe9b1f5..c0eec5de 100644
--- a/jupyter_client/provisioning/local_provisioner.py
+++ b/jupyter_client/provisioning/local_provisioner.py
@@ -10,6 +10,7 @@
 from ..connect import KernelConnectionInfo, LocalPortCache
 from ..launcher import launch_kernel
 from ..localinterfaces import is_local_ip, local_ips
+from ..utils import PORTS_ENV_MAP
 from .provisioner_base import KernelProvisionerBase
 
 
@@ -189,6 +190,10 @@ async def pre_launch(self, **kwargs: Any) -> Dict[str, Any]:
                 km.control_port = lpc.find_available_port(km.ip)
                 self.ports_cached = True
             if 'env' in kwargs:
+                env = kwargs.get("env", {})
+                for arg_key, env_key in PORTS_ENV_MAP.items():
+                    if env_key in env:
+                        setattr(km, arg_key, int(env[env_key]))
                 jupyter_session = kwargs['env'].get("JPY_SESSION_NAME", "")
                 km.write_connection_file(jupyter_session=jupyter_session)
             else:
diff --git a/jupyter_client/utils.py b/jupyter_client/utils.py
index 37eb3dc1..aaa5a9cf 100644
--- a/jupyter_client/utils.py
+++ b/jupyter_client/utils.py
@@ -9,6 +9,14 @@
 
 from .session import utcnow  # noqa
 
+PORTS_ENV_MAP = {
+    'hb_port': 'JUPYTER_SERVER_HB_PORT',
+    'shell_port': 'JUPYTER_SERVER_SHELL_PORT',
+    'iopub_port': 'JUPYTER_SERVER_IOPUB_PORT',
+    'stdin_port': 'JUPYTER_SERVER_STDIN_PORT',
+    'control_port': 'JUPYTER_SERVER_CONTROL_PORT',
+}
+
 
 def _filefind(filename, path_dirs=None):
     """Find a file by looking through a sequence of paths.