diff --git a/ipykernel/comm/comm.py b/ipykernel/comm/comm.py
index b8f4586f..862e3bcf 100644
--- a/ipykernel/comm/comm.py
+++ b/ipykernel/comm/comm.py
@@ -33,6 +33,7 @@ def publish_msg(self, msg_type, data=None, metadata=None, buffers=None, **keys):
         if self.kernel is None:
             self.kernel = Kernel.instance()
 
+        assert self.kernel.session is not None
         self.kernel.session.send(
             self.kernel.iopub_socket,
             msg_type,
diff --git a/ipykernel/datapub.py b/ipykernel/datapub.py
index 4b1c7a50..5805c9af 100644
--- a/ipykernel/datapub.py
+++ b/ipykernel/datapub.py
@@ -48,6 +48,7 @@ def publish_data(self, data):
             The data to be published. Think of it as a namespace.
         """
         session = self.session
+        assert session is not None
         buffers = serialize_object(
             data,
             buffer_threshold=session.buffer_threshold,
diff --git a/ipykernel/displayhook.py b/ipykernel/displayhook.py
index 0727997c..4dd0bb5e 100644
--- a/ipykernel/displayhook.py
+++ b/ipykernel/displayhook.py
@@ -2,9 +2,11 @@
 
 # Copyright (c) IPython Development Team.
 # Distributed under the terms of the Modified BSD License.
+from __future__ import annotations
 
 import builtins
 import sys
+import typing as t
 
 from IPython.core.displayhook import DisplayHook
 from jupyter_client.session import Session, extract_header
@@ -61,6 +63,7 @@ class ZMQShellDisplayHook(DisplayHook):
     session = Instance(Session, allow_none=True)
     pub_socket = Any(allow_none=True)
     parent_header = Dict({})
+    msg: dict[str, t.Any] | None
 
     def set_parent(self, parent):
         """Set the parent for outbound messages."""
@@ -68,28 +71,31 @@ def set_parent(self, parent):
 
     def start_displayhook(self):
         """Start the display hook."""
-        self.msg = self.session.msg(
-            "execute_result",
-            {
-                "data": {},
-                "metadata": {},
-            },
-            parent=self.parent_header,
-        )
+        if self.session:
+            self.msg = self.session.msg(
+                "execute_result",
+                {
+                    "data": {},
+                    "metadata": {},
+                },
+                parent=self.parent_header,
+            )
 
     def write_output_prompt(self):
         """Write the output prompt."""
-        self.msg["content"]["execution_count"] = self.prompt_count
+        if self.msg:
+            self.msg["content"]["execution_count"] = self.prompt_count
 
     def write_format_data(self, format_dict, md_dict=None):
         """Write format data to the message."""
-        self.msg["content"]["data"] = json_clean(encode_images(format_dict))
-        self.msg["content"]["metadata"] = md_dict
+        if self.msg:
+            self.msg["content"]["data"] = json_clean(encode_images(format_dict))
+            self.msg["content"]["metadata"] = md_dict
 
     def finish_displayhook(self):
         """Finish up all displayhook activities."""
         sys.stdout.flush()
         sys.stderr.flush()
-        if self.msg["content"]["data"]:
+        if self.msg and self.msg["content"]["data"] and self.session:
             self.session.send(self.pub_socket, self.msg, ident=self.topic)
         self.msg = None
diff --git a/ipykernel/inprocess/client.py b/ipykernel/inprocess/client.py
index 1fa80bc2..0c4a1967 100644
--- a/ipykernel/inprocess/client.py
+++ b/ipykernel/inprocess/client.py
@@ -60,42 +60,43 @@ def _default_blocking_class(self):
     def get_connection_info(self):
         """Get the connection info for the client."""
         d = super().get_connection_info()
-        d["kernel"] = self.kernel
+        d["kernel"] = self.kernel  # type:ignore[assignment]
         return d
 
     def start_channels(self, *args, **kwargs):
         """Start the channels on the client."""
         super().start_channels()
-        self.kernel.frontends.append(self)
+        if self.kernel:
+            self.kernel.frontends.append(self)
 
     @property
     def shell_channel(self):
         if self._shell_channel is None:
-            self._shell_channel = self.shell_channel_class(self)
+            self._shell_channel = self.shell_channel_class(self)  # type:ignore[operator]
         return self._shell_channel
 
     @property
     def iopub_channel(self):
         if self._iopub_channel is None:
-            self._iopub_channel = self.iopub_channel_class(self)
+            self._iopub_channel = self.iopub_channel_class(self)  # type:ignore[operator]
         return self._iopub_channel
 
     @property
     def stdin_channel(self):
         if self._stdin_channel is None:
-            self._stdin_channel = self.stdin_channel_class(self)
+            self._stdin_channel = self.stdin_channel_class(self)  # type:ignore[operator]
         return self._stdin_channel
 
     @property
     def control_channel(self):
         if self._control_channel is None:
-            self._control_channel = self.control_channel_class(self)
+            self._control_channel = self.control_channel_class(self)  # type:ignore[operator]
         return self._control_channel
 
     @property
     def hb_channel(self):
         if self._hb_channel is None:
-            self._hb_channel = self.hb_channel_class(self)
+            self._hb_channel = self.hb_channel_class(self)  # type:ignore[operator]
         return self._hb_channel
 
     # Methods for sending specific messages
diff --git a/ipykernel/inprocess/ipkernel.py b/ipykernel/inprocess/ipkernel.py
index f3cac4a8..30daca88 100644
--- a/ipykernel/inprocess/ipkernel.py
+++ b/ipykernel/inprocess/ipkernel.py
@@ -51,7 +51,7 @@ class InProcessKernel(IPythonKernel):
     _underlying_iopub_socket = Instance(DummySocket, ())
     iopub_thread: IOPubThread = Instance(IOPubThread)  # type:ignore[assignment]
 
-    shell_stream = Instance(DummySocket, ())
+    shell_stream = Instance(DummySocket, ())  # type:ignore[arg-type]
 
     @default("iopub_thread")
     def _default_iopub_thread(self):
@@ -72,7 +72,8 @@ def __init__(self, **traits):
         super().__init__(**traits)
 
         self._underlying_iopub_socket.observe(self._io_dispatch, names=["message_sent"])
-        self.shell.kernel = self
+        if self.shell:
+            self.shell.kernel = self
 
     async def execute_request(self, stream, ident, parent):
         """Override for temporary IO redirection."""
@@ -81,7 +82,8 @@ async def execute_request(self, stream, ident, parent):
 
     def start(self):
         """Override registration of dispatchers for streams."""
-        self.shell.exit_now = False
+        if self.shell:
+            self.shell.exit_now = False
 
     def _abort_queues(self):
         """The in-process kernel doesn't abort requests."""
@@ -99,6 +101,7 @@ def _input_request(self, prompt, ident, parent, password=False):
 
         # Send the input request.
         content = json_clean(dict(prompt=prompt, password=password))
+        assert self.session is not None
         msg = self.session.msg("input_request", content, parent)
         for frontend in self.frontends:
             if frontend.session.session == parent["header"]["session"]:
@@ -132,6 +135,7 @@ def _redirected_io(self):
     def _io_dispatch(self, change):
         """Called when a message is sent to the IO socket."""
         assert self.iopub_socket.io_thread is not None
+        assert self.session is not None
         ident, msg = self.session.recv(self.iopub_socket.io_thread.socket, copy=False)
         for frontend in self.frontends:
             frontend.iopub_channel.call_handlers(msg)
diff --git a/ipykernel/inprocess/manager.py b/ipykernel/inprocess/manager.py
index 3388dbf6..3a3f92c3 100644
--- a/ipykernel/inprocess/manager.py
+++ b/ipykernel/inprocess/manager.py
@@ -49,8 +49,9 @@ def start_kernel(self, **kwds):
 
     def shutdown_kernel(self):
         """Shutdown the kernel."""
-        self.kernel.iopub_thread.stop()
-        self._kill_kernel()
+        if self.kernel:
+            self.kernel.iopub_thread.stop()
+            self._kill_kernel()
 
     def restart_kernel(self, now=False, **kwds):
         """Restart the kernel."""
diff --git a/ipykernel/ipkernel.py b/ipykernel/ipkernel.py
index f82831cc..990c5558 100644
--- a/ipykernel/ipkernel.py
+++ b/ipykernel/ipkernel.py
@@ -118,7 +118,7 @@ def __init__(self, **kwargs):
             )
 
         # Initialize the InteractiveShell subclass
-        self.shell = self.shell_class.instance(
+        self.shell = self.shell_class.instance(  # type:ignore[attr-defined]
             parent=self,
             profile_dir=self.profile_dir,
             user_module=self.user_module,
@@ -206,7 +206,8 @@ def dispatch_debugpy(self, msg):
 
     @property
     def banner(self):
-        return self.shell.banner
+        if self.shell:
+            return self.shell.banner
 
     async def poll_stopped_queue(self):
         """Poll the stopped queue."""
@@ -215,7 +216,8 @@ async def poll_stopped_queue(self):
 
     def start(self):
         """Start the kernel."""
-        self.shell.exit_now = False
+        if self.shell:
+            self.shell.exit_now = False
         if self.debugpy_stream is None:
             self.log.warning("debugpy_stream undefined, debugging will not be enabled")
         else:
@@ -231,7 +233,7 @@ def set_parent(self, ident, parent, channel="shell"):
         about the parent message.
         """
         super().set_parent(ident, parent, channel)
-        if channel == "shell":
+        if channel == "shell" and self.shell:
             self.shell.set_parent(parent)
 
     def init_metadata(self, parent):
@@ -284,7 +286,8 @@ def _restore_input(self):
 
     @property
     def execution_count(self):
-        return self.shell.execution_count
+        if self.shell:
+            return self.shell.execution_count
 
     @execution_count.setter
     def execution_count(self, value):
@@ -348,6 +351,7 @@ async def do_execute(
     ):
         """Handle code execution."""
         shell = self.shell  # we'll need this a lot here
+        assert shell is not None
 
         self._forward_input(allow_stdin)
 
@@ -371,7 +375,7 @@ async def run_cell(*args, **kwargs):
             # not just asyncio
             preprocessing_exc_tuple = None
             try:
-                transformed_cell = self.shell.transform_cell(code)
+                transformed_cell = shell.transform_cell(code)
             except Exception:
                 transformed_cell = code
                 preprocessing_exc_tuple = sys.exc_info()
@@ -488,6 +492,7 @@ def do_complete(self, code, cursor_pos):
             cursor_pos = len(code)
         line, offset = line_at_cursor(code, cursor_pos)
         line_cursor = cursor_pos - offset
+        assert self.shell is not None
         txt, matches = self.shell.complete("", line, line_cursor)
         return {
             "matches": matches,
@@ -509,6 +514,7 @@ def _experimental_do_complete(self, code, cursor_pos):
         if cursor_pos is None:
             cursor_pos = len(code)
         with _provisionalcompleter():
+            assert self.shell is not None
             raw_completions = self.shell.Completer.completions(code, cursor_pos)
             completions = list(_rectify_completions(code, raw_completions))
 
@@ -548,6 +554,7 @@ def do_inspect(self, code, cursor_pos, detail_level=0, omit_sections=()):
         reply_content: t.Dict[str, t.Any] = {"status": "ok"}
         reply_content["data"] = {}
         reply_content["metadata"] = {}
+        assert self.shell is not None
         try:
             if release.version_info >= (8,):
                 # `omit_sections` keyword will be available in IPython 8, see
@@ -581,6 +588,7 @@ def do_history(
         unique=False,
     ):
         """Handle code history."""
+        assert self.shell is not None
         if hist_access_type == "tail":
             hist = self.shell.history_manager.get_tail(
                 n, raw=raw, output=output, include_latest=True
@@ -605,7 +613,8 @@ def do_history(
 
     def do_shutdown(self, restart):
         """Handle kernel shutdown."""
-        self.shell.exit_now = True
+        if self.shell:
+            self.shell.exit_now = True
         return dict(status="ok", restart=restart)
 
     def do_is_complete(self, code):
@@ -613,6 +622,7 @@ def do_is_complete(self, code):
         transformer_manager = getattr(self.shell, "input_transformer_manager", None)
         if transformer_manager is None:
             # input_splitter attribute is deprecated
+            assert self.shell is not None
             transformer_manager = self.shell.input_splitter
         status, indent_spaces = transformer_manager.check_complete(code)
         r = {"status": status}
@@ -628,6 +638,7 @@ def do_apply(self, content, bufs, msg_id, reply_metadata):
             from .serialize import serialize_object, unpack_apply_message
 
         shell = self.shell
+        assert shell is not None
         try:
             working = shell.user_ns
 
@@ -652,6 +663,7 @@ def do_apply(self, content, bufs, msg_id, reply_metadata):
                 for key in ns:
                     working.pop(key)
 
+            assert self.session is not None
             result_buf = serialize_object(
                 result,
                 buffer_threshold=self.session.buffer_threshold,
@@ -686,7 +698,8 @@ def do_apply(self, content, bufs, msg_id, reply_metadata):
 
     def do_clear(self):
         """Clear the kernel."""
-        self.shell.reset(False)
+        if self.shell:
+            self.shell.reset(False)
         return dict(status="ok")
 
 
diff --git a/ipykernel/kernelapp.py b/ipykernel/kernelapp.py
index 990d6ba4..2f51271f 100644
--- a/ipykernel/kernelapp.py
+++ b/ipykernel/kernelapp.py
@@ -547,7 +547,7 @@ def init_kernel(self):
         control_stream = ZMQStream(self.control_socket, self.control_thread.io_loop)
         debugpy_stream = ZMQStream(self.debugpy_socket, self.control_thread.io_loop)
         self.control_thread.start()
-        kernel_factory = self.kernel_class.instance
+        kernel_factory = self.kernel_class.instance  # type:ignore[attr-defined]
 
         kernel = kernel_factory(
             parent=self,
diff --git a/ipykernel/kernelbase.py b/ipykernel/kernelbase.py
index 8b45156a..6d06d4ab 100644
--- a/ipykernel/kernelbase.py
+++ b/ipykernel/kernelbase.py
@@ -304,7 +304,8 @@ async def _flush_control_queue(self):
 
         def _flush():
             # control_stream.flush puts messages on the queue
-            self.control_stream.flush()
+            if self.control_stream:
+                self.control_stream.flush()
             # put Future on the queue after all of those,
             # so we can wait for all queued messages to be processed
             self.control_queue.put(tracer_future)
@@ -314,6 +315,8 @@ def _flush():
 
     async def process_control(self, msg):
         """dispatch control requests"""
+        if not self.session:
+            return
         idents, msg = self.session.feed_identities(msg, copy=False)
         try:
             msg = self.session.deserialize(msg, content=True, copy=False)
@@ -345,7 +348,8 @@ async def process_control(self, msg):
         sys.stderr.flush()
         self._publish_status("idle", "control")
         # flush to ensure reply is sent
-        self.control_stream.flush(zmq.POLLOUT)
+        if self.control_stream:
+            self.control_stream.flush(zmq.POLLOUT)
 
     def should_handle(self, stream, msg, idents):
         """Check whether a shell-channel message should be handled
@@ -362,7 +366,8 @@ def should_handle(self, stream, msg, idents):
 
     async def dispatch_shell(self, msg):
         """dispatch shell requests"""
-
+        if not self.session:
+            return
         # flush control queue before handling shell requests
         await self._flush_control_queue()
 
@@ -385,7 +390,8 @@ async def dispatch_shell(self, msg):
             self._publish_status("idle", "shell")
             # flush to ensure reply is sent before
             # handling the next request
-            self.shell_stream.flush(zmq.POLLOUT)
+            if self.shell_stream:
+                self.shell_stream.flush(zmq.POLLOUT)
             return
 
         # Print some info about this message and leave a '--->' marker, so it's
@@ -426,7 +432,8 @@ async def dispatch_shell(self, msg):
         self._publish_status("idle", "shell")
         # flush to ensure reply is sent before
         # handling the next request
-        self.shell_stream.flush(zmq.POLLOUT)
+        if self.shell_stream:
+            self.shell_stream.flush(zmq.POLLOUT)
 
     def pre_handler_hook(self):
         """Hook to execute before calling message handler"""
@@ -486,7 +493,8 @@ async def do_one_iteration(self):
             This is now a coroutine
         """
         # flush messages off of shell stream into the message queue
-        self.shell_stream.flush()
+        if self.shell_stream:
+            self.shell_stream.flush()
         # process at most one shell message per iteration
         await self.process_one(wait=False)
 
@@ -546,19 +554,20 @@ def start(self):
         self.msg_queue: Queue[t.Any] = Queue()
         self.io_loop.add_callback(self.dispatch_queue)
 
-        self.control_stream.on_recv(self.dispatch_control, copy=False)
+        if self.control_stream:
+            self.control_stream.on_recv(self.dispatch_control, copy=False)
 
         control_loop = self.control_thread.io_loop if self.control_thread else self.io_loop
 
         asyncio.run_coroutine_threadsafe(self.poll_control_queue(), control_loop.asyncio_loop)
-
-        self.shell_stream.on_recv(
-            partial(
-                self.schedule_dispatch,
-                self.dispatch_shell,
-            ),
-            copy=False,
-        )
+        if self.shell_stream:
+            self.shell_stream.on_recv(
+                partial(
+                    self.schedule_dispatch,
+                    self.dispatch_shell,
+                ),
+                copy=False,
+            )
 
         # publish idle status
         self._publish_status("starting", "shell")
@@ -577,7 +586,8 @@ def record_ports(self, ports):
 
     def _publish_execute_input(self, code, parent, execution_count):
         """Publish the code request on the iopub stream."""
-
+        if not self.session:
+            return
         self.session.send(
             self.iopub_socket,
             "execute_input",
@@ -588,6 +598,8 @@ def _publish_execute_input(self, code, parent, execution_count):
 
     def _publish_status(self, status, channel, parent=None):
         """send status (busy/idle) on IOPub"""
+        if not self.session:
+            return
         self.session.send(
             self.iopub_socket,
             "status",
@@ -597,6 +609,8 @@ def _publish_status(self, status, channel, parent=None):
         )
 
     def _publish_debug_event(self, event):
+        if not self.session:
+            return
         self.session.send(
             self.iopub_socket,
             "debug_event",
@@ -662,6 +676,8 @@ def send_response(
         This relies on :meth:`set_parent` having been called for the current
         message.
         """
+        if not self.session:
+            return
         return self.session.send(
             stream,
             msg_or_type,
@@ -694,6 +710,8 @@ def finish_metadata(self, parent, metadata, reply_content):
 
     async def execute_request(self, stream, ident, parent):
         """handle an execute_request"""
+        if not self.session:
+            return
         try:
             content = parent["content"]
             code = content["code"]
@@ -752,7 +770,7 @@ async def execute_request(self, stream, ident, parent):
         reply_content = json_clean(reply_content)
         metadata = self.finish_metadata(parent, metadata, reply_content)
 
-        reply_msg = self.session.send(
+        reply_msg: dict[str, t.Any] = self.session.send(  # type:ignore[assignment]
             stream,
             "execute_reply",
             reply_content,
@@ -781,6 +799,8 @@ def do_execute(
 
     async def complete_request(self, stream, ident, parent):
         """Handle a completion request."""
+        if not self.session:
+            return
         content = parent["content"]
         code = content["code"]
         cursor_pos = content["cursor_pos"]
@@ -804,6 +824,8 @@ def do_complete(self, code, cursor_pos):
 
     async def inspect_request(self, stream, ident, parent):
         """Handle an inspect request."""
+        if not self.session:
+            return
         content = parent["content"]
 
         reply_content = self.do_inspect(
@@ -826,6 +848,8 @@ def do_inspect(self, code, cursor_pos, detail_level=0, omit_sections=()):
 
     async def history_request(self, stream, ident, parent):
         """Handle a history request."""
+        if not self.session:
+            return
         content = parent["content"]
 
         reply_content = self.do_history(**content)
@@ -853,7 +877,9 @@ def do_history(
 
     async def connect_request(self, stream, ident, parent):
         """Handle a connect request."""
-        content = self._recorded_ports.copy() if self._recorded_ports is not None else {}
+        if not self.session:
+            return
+        content = self._recorded_ports.copy() if self._recorded_ports else {}
         content["status"] = "ok"
         msg = self.session.send(stream, "connect_reply", content, parent, ident)
         self.log.debug("%s", msg)
@@ -871,6 +897,8 @@ def kernel_info(self):
 
     async def kernel_info_request(self, stream, ident, parent):
         """Handle a kernel info request."""
+        if not self.session:
+            return
         content = {"status": "ok"}
         content.update(self.kernel_info)
         msg = self.session.send(stream, "kernel_info_reply", content, parent, ident)
@@ -878,6 +906,8 @@ async def kernel_info_request(self, stream, ident, parent):
 
     async def comm_info_request(self, stream, ident, parent):
         """Handle a comm info request."""
+        if not self.session:
+            return
         content = parent["content"]
         target_name = content.get("target_name", None)
 
@@ -913,6 +943,8 @@ def _send_interrupt_children(self):
 
     async def interrupt_request(self, stream, ident, parent):
         """Handle an interrupt request."""
+        if not self.session:
+            return
         content: t.Dict[str, t.Any] = {"status": "ok"}
         try:
             self._send_interrupt_children()
@@ -931,6 +963,8 @@ async def interrupt_request(self, stream, ident, parent):
 
     async def shutdown_request(self, stream, ident, parent):
         """Handle a shutdown request."""
+        if not self.session:
+            return
         content = self.do_shutdown(parent["content"]["restart"])
         if inspect.isawaitable(content):
             content = await content
@@ -941,12 +975,14 @@ async def shutdown_request(self, stream, ident, parent):
         await self._at_shutdown()
 
         self.log.debug("Stopping control ioloop")
-        control_io_loop = self.control_stream.io_loop
-        control_io_loop.add_callback(control_io_loop.stop)
+        if self.control_stream:
+            control_io_loop = self.control_stream.io_loop
+            control_io_loop.add_callback(control_io_loop.stop)
 
         self.log.debug("Stopping shell ioloop")
-        shell_io_loop = self.shell_stream.io_loop
-        shell_io_loop.add_callback(shell_io_loop.stop)
+        if self.shell_stream:
+            shell_io_loop = self.shell_stream.io_loop
+            shell_io_loop.add_callback(shell_io_loop.stop)
 
     def do_shutdown(self, restart):
         """Override in subclasses to do things when the frontend shuts down the
@@ -956,6 +992,8 @@ def do_shutdown(self, restart):
 
     async def is_complete_request(self, stream, ident, parent):
         """Handle an is_complete request."""
+        if not self.session:
+            return
         content = parent["content"]
         code = content["code"]
 
@@ -972,6 +1010,8 @@ def do_is_complete(self, code):
 
     async def debug_request(self, stream, ident, parent):
         """Handle a debug request."""
+        if not self.session:
+            return
         content = parent["content"]
         reply_content = self.do_debug_request(content)
         if inspect.isawaitable(reply_content):
@@ -995,6 +1035,8 @@ def get_process_metric_value(self, process, name, attribute=None):
 
     async def usage_request(self, stream, ident, parent):
         """Handle a usage request."""
+        if not self.session:
+            return
         reply_content = {"hostname": socket.gethostname(), "pid": os.getpid()}
         current_process = psutil.Process()
         all_processes = [current_process, *current_process.children(recursive=True)]
@@ -1053,7 +1095,8 @@ async def apply_request(self, stream, ident, parent):  # pragma: no cover
         sys.stderr.flush()
 
         md = self.finish_metadata(parent, md, reply_content)
-
+        if not self.session:
+            return
         self.session.send(
             stream,
             "apply_reply",
@@ -1086,6 +1129,8 @@ async def abort_request(self, stream, ident, parent):  # pragma: no cover
             self.aborted.add(str(mid))
 
         content = dict(status="ok")
+        if not self.session:
+            return
         reply_msg = self.session.send(
             stream, "abort_reply", content=content, parent=parent, ident=ident
         )
@@ -1097,7 +1142,8 @@ async def clear_request(self, stream, idents, parent):  # pragma: no cover
             "clear_request is deprecated in kernel_base. It is only part of IPython parallel"
         )
         content = self.do_clear()
-        self.session.send(stream, "clear_reply", ident=idents, parent=parent, content=content)
+        if self.session:
+            self.session.send(stream, "clear_reply", ident=idents, parent=parent, content=content)
 
     def do_clear(self):
         """DEPRECATED since 4.0.3"""
@@ -1123,7 +1169,8 @@ def _abort_queues(self):
 
         # flush streams, so all currently waiting messages
         # are added to the queue
-        self.shell_stream.flush()
+        if self.shell_stream:
+            self.shell_stream.flush()
 
         # Callback to signal that we are done aborting
         # dispatch functions _must_ be async
@@ -1142,6 +1189,8 @@ async def stop_aborting():
 
     def _send_abort_reply(self, stream, msg, idents):
         """Send a reply to an aborted request"""
+        if not self.session:
+            return
         self.log.info(f"Aborting {msg['header']['msg_id']}: {msg['header']['msg_type']}")
         reply_type = msg["header"]["msg_type"].rsplit("_", 1)[0] + "_reply"
         status = {"status": "aborted"}
@@ -1222,6 +1271,7 @@ def _input_request(self, prompt, ident, parent, password=False):
                     raise
 
         # Send the input request.
+        assert self.session is not None
         content = json_clean(dict(prompt=prompt, password=password))
         self.session.send(self.stdin_socket, "input_request", content, parent, ident=ident)
 
@@ -1247,7 +1297,7 @@ def _input_request(self, prompt, ident, parent, password=False):
                 self.log.warning("Invalid Message:", exc_info=True)
 
         try:
-            value = reply["content"]["value"]
+            value = reply["content"]["value"]  # type:ignore[index]
         except Exception:
             self.log.error("Bad input_reply: %s", parent)
             value = ""
@@ -1325,11 +1375,12 @@ async def _at_shutdown(self):
             self.log.exception("Exception during subprocesses termination %s", e)
 
         finally:
-            if self._shutdown_message is not None:
+            if self._shutdown_message is not None and self.session:
                 self.session.send(
                     self.iopub_socket,
                     self._shutdown_message,
                     ident=self._topic("shutdown"),
                 )
                 self.log.debug("%s", self._shutdown_message)
-            self.control_stream.flush(zmq.POLLOUT)
+            if self.control_stream:
+                self.control_stream.flush(zmq.POLLOUT)
diff --git a/ipykernel/zmqshell.py b/ipykernel/zmqshell.py
index 61c5ee69..487008f5 100644
--- a/ipykernel/zmqshell.py
+++ b/ipykernel/zmqshell.py
@@ -115,6 +115,7 @@ def publish(
         # Use 2-stage process to send a message,
         # in order to put it through the transform
         # hooks before potentially sending.
+        assert self.session is not None
         msg = self.session.msg(msg_type, json_clean(content), parent=self.parent_header)
 
         # Each transform either returns a new
@@ -123,7 +124,7 @@ def publish(
         for hook in self._hooks:
             msg = hook(msg)
             if msg is None:
-                return
+                return  # type:ignore[unreachable]
 
         self.session.send(
             self.pub_socket,
@@ -144,13 +145,14 @@ def clear_output(self, wait=False):
         """
         content = dict(wait=wait)
         self._flush_streams()
+        assert self.session is not None
         msg = self.session.msg("clear_output", json_clean(content), parent=self.parent_header)
 
         # see publish() for details on how this works
         for hook in self._hooks:
             msg = hook(msg)
             if msg is None:
-                return
+                return  # type:ignore[unreachable]
 
         self.session.send(
             self.pub_socket,
@@ -538,7 +540,7 @@ def ask_exit(self):
             source="ask_exit",
             keepkernel=self.keepkernel_on_exit,
         )
-        self.payload_manager.write_payload(payload)
+        self.payload_manager.write_payload(payload)  # type:ignore[union-attr]
 
     def run_cell(self, *args, **kwargs):
         """Run a cell."""
@@ -583,7 +585,7 @@ def set_next_input(self, text, replace=False):
             text=text,
             replace=replace,
         )
-        self.payload_manager.write_payload(payload)
+        self.payload_manager.write_payload(payload)  # type:ignore[union-attr]
 
     def set_parent(self, parent):
         """Set the parent header for associating output with its triggering input"""
@@ -609,7 +611,7 @@ def init_magics(self):
         """Initialize magics."""
         super().init_magics()
         self.register_magics(KernelMagics)
-        self.magics_manager.register_alias("ed", "edit")
+        self.magics_manager.register_alias("ed", "edit")  # type:ignore[union-attr]
 
     def init_virtualenv(self):
         """Initialize virtual environment."""
diff --git a/pyproject.toml b/pyproject.toml
index 21181273..ae96b102 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -112,9 +112,9 @@ matrix.qt.features = [
 
 [tool.hatch.envs.typing]
 features = ["test"]
-dependencies = ["mypy>=1.5.1"]
+dependencies = ["mypy>=1.5.1", "traitlets>=5.10.1"]
 [tool.hatch.envs.typing.scripts]
-test = "mypy --install-types --non-interactive {args:.}"
+test = "mypy --install-types --non-interactive {args}"
 
 [tool.hatch.envs.lint]
 dependencies = ["black==23.3.0", "mdformat>0.7", "ruff==0.0.287"]
@@ -132,6 +132,7 @@ fmt = [
 ]
 
 [tool.mypy]
+files = "ipykernel"
 check_untyped_defs = true
 disallow_incomplete_defs = true
 disallow_untyped_decorators = true