Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1"""Base class to manage the interaction with a running kernel"""
3# Copyright (c) Jupyter Development Team.
4# Distributed under the terms of the Modified BSD License.
5import asyncio
6import inspect
7import sys
8import time
9import typing as t
10from functools import partial
11from getpass import getpass
12from queue import Empty
14import zmq.asyncio
15from jupyter_core.utils import ensure_async
16from traitlets import Any, Bool, Instance, Type
18from .channels import major_protocol_version
19from .channelsabc import ChannelABC, HBChannelABC
20from .clientabc import KernelClientABC
21from .connect import ConnectionFileMixin
22from .session import Session
24# some utilities to validate message structure, these might get moved elsewhere
25# if they prove to have more generic utility
28def validate_string_dict(dct: t.Dict[str, str]) -> None:
29 """Validate that the input is a dict with string keys and values.
31 Raises ValueError if not."""
32 for k, v in dct.items():
33 if not isinstance(k, str):
34 raise ValueError("key %r in dict must be a string" % k)
35 if not isinstance(v, str):
36 raise ValueError("value %r in dict must be a string" % v)
39def reqrep(wrapped: t.Callable, meth: t.Callable, channel: str = "shell") -> t.Callable:
40 wrapped = wrapped(meth, channel)
41 if not meth.__doc__:
42 # python -OO removes docstrings,
43 # so don't bother building the wrapped docstring
44 return wrapped
46 basedoc, _ = meth.__doc__.split("Returns\n", 1)
47 parts = [basedoc.strip()]
48 if "Parameters" not in basedoc:
49 parts.append(
50 """
51 Parameters
52 ----------
53 """
54 )
55 parts.append(
56 """
57 reply: bool (default: False)
58 Whether to wait for and return reply
59 timeout: float or None (default: None)
60 Timeout to use when waiting for a reply
62 Returns
63 -------
64 msg_id: str
65 The msg_id of the request sent, if reply=False (default)
66 reply: dict
67 The reply message for this request, if reply=True
68 """
69 )
70 wrapped.__doc__ = "\n".join(parts)
71 return wrapped
74class KernelClient(ConnectionFileMixin):
75 """Communicates with a single kernel on any host via zmq channels.
77 There are five channels associated with each kernel:
79 * shell: for request/reply calls to the kernel.
80 * iopub: for the kernel to publish results to frontends.
81 * hb: for monitoring the kernel's heartbeat.
82 * stdin: for frontends to reply to raw_input calls in the kernel.
83 * control: for kernel management calls to the kernel.
85 The messages that can be sent on these channels are exposed as methods of the
86 client (KernelClient.execute, complete, history, etc.). These methods only
87 send the message, they don't wait for a reply. To get results, use e.g.
88 :meth:`get_shell_msg` to fetch messages from the shell channel.
89 """
91 # The PyZMQ Context to use for communication with the kernel.
92 context = Instance(zmq.Context)
94 _created_context = Bool(False)
96 def _context_default(self) -> zmq.Context:
97 self._created_context = True
98 return zmq.Context()
100 # The classes to use for the various channels
101 shell_channel_class = Type(ChannelABC)
102 iopub_channel_class = Type(ChannelABC)
103 stdin_channel_class = Type(ChannelABC)
104 hb_channel_class = Type(HBChannelABC)
105 control_channel_class = Type(ChannelABC)
107 # Protected traits
108 _shell_channel = Any()
109 _iopub_channel = Any()
110 _stdin_channel = Any()
111 _hb_channel = Any()
112 _control_channel = Any()
114 # flag for whether execute requests should be allowed to call raw_input:
115 allow_stdin: bool = True
117 def __del__(self) -> None:
118 """Handle garbage collection. Destroy context if applicable."""
119 if (
120 self._created_context
121 and self.context is not None # type:ignore[redundant-expr]
122 and not self.context.closed
123 ):
124 if self.channels_running:
125 if self.log:
126 self.log.warning("Could not destroy zmq context for %s", self)
127 else:
128 if self.log:
129 self.log.debug("Destroying zmq context for %s", self)
130 self.context.destroy(linger=100)
131 try:
132 super_del = super().__del__ # type:ignore[misc]
133 except AttributeError:
134 pass
135 else:
136 super_del()
138 # --------------------------------------------------------------------------
139 # Channel proxy methods
140 # --------------------------------------------------------------------------
142 async def _async_get_shell_msg(self, *args: t.Any, **kwargs: t.Any) -> t.Dict[str, t.Any]:
143 """Get a message from the shell channel"""
144 return await ensure_async(self.shell_channel.get_msg(*args, **kwargs))
146 async def _async_get_iopub_msg(self, *args: t.Any, **kwargs: t.Any) -> t.Dict[str, t.Any]:
147 """Get a message from the iopub channel"""
148 return await ensure_async(self.iopub_channel.get_msg(*args, **kwargs))
150 async def _async_get_stdin_msg(self, *args: t.Any, **kwargs: t.Any) -> t.Dict[str, t.Any]:
151 """Get a message from the stdin channel"""
152 return await ensure_async(self.stdin_channel.get_msg(*args, **kwargs))
154 async def _async_get_control_msg(self, *args: t.Any, **kwargs: t.Any) -> t.Dict[str, t.Any]:
155 """Get a message from the control channel"""
156 return await ensure_async(self.control_channel.get_msg(*args, **kwargs))
158 async def _async_wait_for_ready(self, timeout: float | None = None) -> None:
159 """Waits for a response when a client is blocked
161 - Sets future time for timeout
162 - Blocks on shell channel until a message is received
163 - Exit if the kernel has died
164 - If client times out before receiving a message from the kernel, send RuntimeError
165 - Flush the IOPub channel
166 """
167 if timeout is None:
168 timeout = float("inf")
169 abs_timeout = time.time() + timeout
171 from .manager import KernelManager
173 if not isinstance(self.parent, KernelManager):
174 # This Client was not created by a KernelManager,
175 # so wait for kernel to become responsive to heartbeats
176 # before checking for kernel_info reply
177 while not await self._async_is_alive():
178 if time.time() > abs_timeout:
179 raise RuntimeError(
180 "Kernel didn't respond to heartbeats in %d seconds and timed out" % timeout
181 )
182 await asyncio.sleep(0.2)
184 # Wait for kernel info reply on shell channel
185 while True:
186 self.kernel_info()
187 try:
188 msg = await ensure_async(self.shell_channel.get_msg(timeout=1))
189 except Empty:
190 pass
191 else:
192 if msg["msg_type"] == "kernel_info_reply":
193 # Checking that IOPub is connected. If it is not connected, start over.
194 try:
195 await ensure_async(self.iopub_channel.get_msg(timeout=0.2))
196 except Empty:
197 pass
198 else:
199 self._handle_kernel_info_reply(msg)
200 break
202 if not await self._async_is_alive():
203 msg = "Kernel died before replying to kernel_info"
204 raise RuntimeError(msg)
206 # Check if current time is ready check time plus timeout
207 if time.time() > abs_timeout:
208 raise RuntimeError("Kernel didn't respond in %d seconds" % timeout)
210 # Flush IOPub channel
211 while True:
212 try:
213 msg = await ensure_async(self.iopub_channel.get_msg(timeout=0.2))
214 except Empty:
215 break
217 async def _async_recv_reply(
218 self, msg_id: str, timeout: float | None = None, channel: str = "shell"
219 ) -> t.Dict[str, t.Any]:
220 """Receive and return the reply for a given request"""
221 if timeout is not None:
222 deadline = time.monotonic() + timeout
223 while True:
224 if timeout is not None:
225 timeout = max(0, deadline - time.monotonic())
226 try:
227 if channel == "control":
228 reply = await self._async_get_control_msg(timeout=timeout)
229 else:
230 reply = await self._async_get_shell_msg(timeout=timeout)
231 except Empty as e:
232 msg = "Timeout waiting for reply"
233 raise TimeoutError(msg) from e
234 if reply["parent_header"].get("msg_id") != msg_id:
235 # not my reply, someone may have forgotten to retrieve theirs
236 continue
237 return reply
239 async def _stdin_hook_default(self, msg: t.Dict[str, t.Any]) -> None:
240 """Handle an input request"""
241 content = msg["content"]
242 prompt = getpass if content.get("password", False) else input
244 try:
245 raw_data = prompt(content["prompt"])
246 except EOFError:
247 # turn EOFError into EOF character
248 raw_data = "\x04"
249 except KeyboardInterrupt:
250 sys.stdout.write("\n")
251 return
253 # only send stdin reply if there *was not* another request
254 # or execution finished while we were reading.
255 if not (await self.stdin_channel.msg_ready() or await self.shell_channel.msg_ready()):
256 self.input(raw_data)
258 def _output_hook_default(self, msg: t.Dict[str, t.Any]) -> None:
259 """Default hook for redisplaying plain-text output"""
260 msg_type = msg["header"]["msg_type"]
261 content = msg["content"]
262 if msg_type == "stream":
263 stream = getattr(sys, content["name"])
264 stream.write(content["text"])
265 elif msg_type in ("display_data", "execute_result"):
266 sys.stdout.write(content["data"].get("text/plain", ""))
267 elif msg_type == "error":
268 sys.stderr.write("\n".join(content["traceback"]))
270 def _output_hook_kernel(
271 self,
272 session: Session,
273 socket: zmq.sugar.socket.Socket,
274 parent_header: t.Any,
275 msg: t.Dict[str, t.Any],
276 ) -> None:
277 """Output hook when running inside an IPython kernel
279 adds rich output support.
280 """
281 msg_type = msg["header"]["msg_type"]
282 if msg_type in ("display_data", "execute_result", "error"):
283 session.send(socket, msg_type, msg["content"], parent=parent_header)
284 else:
285 self._output_hook_default(msg)
287 # --------------------------------------------------------------------------
288 # Channel management methods
289 # --------------------------------------------------------------------------
291 def start_channels(
292 self,
293 shell: bool = True,
294 iopub: bool = True,
295 stdin: bool = True,
296 hb: bool = True,
297 control: bool = True,
298 ) -> None:
299 """Starts the channels for this kernel.
301 This will create the channels if they do not exist and then start
302 them (their activity runs in a thread). If port numbers of 0 are
303 being used (random ports) then you must first call
304 :meth:`start_kernel`. If the channels have been stopped and you
305 call this, :class:`RuntimeError` will be raised.
306 """
307 if iopub:
308 self.iopub_channel.start()
309 if shell:
310 self.shell_channel.start()
311 if stdin:
312 self.stdin_channel.start()
313 self.allow_stdin = True
314 else:
315 self.allow_stdin = False
316 if hb:
317 self.hb_channel.start()
318 if control:
319 self.control_channel.start()
321 def stop_channels(self) -> None:
322 """Stops all the running channels for this kernel.
324 This stops their event loops and joins their threads.
325 """
326 if self.shell_channel.is_alive():
327 self.shell_channel.stop()
328 if self.iopub_channel.is_alive():
329 self.iopub_channel.stop()
330 if self.stdin_channel.is_alive():
331 self.stdin_channel.stop()
332 if self.hb_channel.is_alive():
333 self.hb_channel.stop()
334 if self.control_channel.is_alive():
335 self.control_channel.stop()
337 if self._created_context and not self.context.closed:
338 self.context.destroy(linger=100)
340 @property
341 def channels_running(self) -> bool:
342 """Are any of the channels created and running?"""
343 return (
344 (self._shell_channel and self.shell_channel.is_alive())
345 or (self._iopub_channel and self.iopub_channel.is_alive())
346 or (self._stdin_channel and self.stdin_channel.is_alive())
347 or (self._hb_channel and self.hb_channel.is_alive())
348 or (self._control_channel and self.control_channel.is_alive())
349 )
351 ioloop = None # Overridden in subclasses that use pyzmq event loop
353 @property
354 def shell_channel(self) -> t.Any:
355 """Get the shell channel object for this kernel."""
356 if self._shell_channel is None:
357 url = self._make_url("shell")
358 self.log.debug("connecting shell channel to %s", url)
359 socket = self.connect_shell(identity=self.session.bsession)
360 self._shell_channel = self.shell_channel_class( # type:ignore[call-arg,abstract]
361 socket, self.session, self.ioloop
362 )
363 return self._shell_channel
365 @property
366 def iopub_channel(self) -> t.Any:
367 """Get the iopub channel object for this kernel."""
368 if self._iopub_channel is None:
369 url = self._make_url("iopub")
370 self.log.debug("connecting iopub channel to %s", url)
371 socket = self.connect_iopub()
372 self._iopub_channel = self.iopub_channel_class( # type:ignore[call-arg,abstract]
373 socket, self.session, self.ioloop
374 )
375 return self._iopub_channel
377 @property
378 def stdin_channel(self) -> t.Any:
379 """Get the stdin channel object for this kernel."""
380 if self._stdin_channel is None:
381 url = self._make_url("stdin")
382 self.log.debug("connecting stdin channel to %s", url)
383 socket = self.connect_stdin(identity=self.session.bsession)
384 self._stdin_channel = self.stdin_channel_class( # type:ignore[call-arg,abstract]
385 socket, self.session, self.ioloop
386 )
387 return self._stdin_channel
389 @property
390 def hb_channel(self) -> t.Any:
391 """Get the hb channel object for this kernel."""
392 if self._hb_channel is None:
393 url = self._make_url("hb")
394 self.log.debug("connecting heartbeat channel to %s", url)
395 hb_kwargs = {}
396 if self.curve_publickey:
397 hb_kwargs["curve_serverkey"] = self.curve_publickey
398 try:
399 self._hb_channel = self.hb_channel_class( # type:ignore[call-arg,abstract]
400 self.context,
401 self.session,
402 url,
403 **hb_kwargs,
404 )
405 except TypeError as e:
406 if "curve_serverkey" in str(e):
407 msg = (
408 f"{self.hb_channel_class.__name__} does not support the "
409 "'curve_serverkey' parameter. Upgrade the heartbeat channel "
410 "class or disable CurveZMQ encryption."
411 )
412 raise RuntimeError(msg) from e
413 else:
414 raise
415 return self._hb_channel
417 @property
418 def control_channel(self) -> t.Any:
419 """Get the control channel object for this kernel."""
420 if self._control_channel is None:
421 url = self._make_url("control")
422 self.log.debug("connecting control channel to %s", url)
423 socket = self.connect_control(identity=self.session.bsession)
424 self._control_channel = self.control_channel_class( # type:ignore[call-arg,abstract]
425 socket, self.session, self.ioloop
426 )
427 return self._control_channel
429 async def _async_is_alive(self) -> bool:
430 """Is the kernel process still running?"""
431 from .manager import KernelManager
433 if isinstance(self.parent, KernelManager):
434 # This KernelClient was created by a KernelManager,
435 # we can ask the parent KernelManager:
436 return await self.parent._async_is_alive()
437 if self._hb_channel is not None:
438 # We don't have access to the KernelManager,
439 # so we use the heartbeat.
440 return self._hb_channel.is_beating()
441 # no heartbeat and not local, we can't tell if it's running,
442 # so naively return True
443 return True
445 async def _async_execute_interactive(
446 self,
447 code: str,
448 silent: bool = False,
449 store_history: bool = True,
450 user_expressions: t.Dict[str, t.Any] | None = None,
451 allow_stdin: bool | None = None,
452 stop_on_error: bool = True,
453 timeout: float | None = None,
454 output_hook: t.Callable | None = None,
455 stdin_hook: t.Callable | None = None,
456 ) -> t.Dict[str, t.Any]:
457 """Execute code in the kernel interactively
459 Output will be redisplayed, and stdin prompts will be relayed as well.
460 If an IPython kernel is detected, rich output will be displayed.
462 You can pass a custom output_hook callable that will be called
463 with every IOPub message that is produced instead of the default redisplay.
465 .. versionadded:: 5.0
467 Parameters
468 ----------
469 code : str
470 A string of code in the kernel's language.
472 silent : bool, optional (default False)
473 If set, the kernel will execute the code as quietly possible, and
474 will force store_history to be False.
476 store_history : bool, optional (default True)
477 If set, the kernel will store command history. This is forced
478 to be False if silent is True.
480 user_expressions : dict, optional
481 A dict mapping names to expressions to be evaluated in the user's
482 dict. The expression values are returned as strings formatted using
483 :func:`repr`.
485 allow_stdin : bool, optional (default self.allow_stdin)
486 Flag for whether the kernel can send stdin requests to frontends.
488 Some frontends (e.g. the Notebook) do not support stdin requests.
489 If raw_input is called from code executed from such a frontend, a
490 StdinNotImplementedError will be raised.
492 stop_on_error: bool, optional (default True)
493 Flag whether to abort the execution queue, if an exception is encountered.
495 timeout: float or None (default: None)
496 Timeout to use when waiting for a reply
498 output_hook: callable(msg)
499 Function to be called with output messages.
500 If not specified, output will be redisplayed.
502 stdin_hook: callable(msg)
503 Function or awaitable to be called with stdin_request messages.
504 If not specified, input/getpass will be called.
506 Returns
507 -------
508 reply: dict
509 The reply message for this request
510 """
511 if not self.iopub_channel.is_alive():
512 emsg = "IOPub channel must be running to receive output"
513 raise RuntimeError(emsg)
514 if allow_stdin is None:
515 allow_stdin = self.allow_stdin
516 if allow_stdin and not self.stdin_channel.is_alive():
517 emsg = "stdin channel must be running to allow input"
518 raise RuntimeError(emsg)
519 msg_id = await ensure_async(
520 self.execute(
521 code,
522 silent=silent,
523 store_history=store_history,
524 user_expressions=user_expressions,
525 allow_stdin=allow_stdin,
526 stop_on_error=stop_on_error,
527 )
528 )
529 if stdin_hook is None:
530 stdin_hook = self._stdin_hook_default
531 # detect IPython kernel
532 if output_hook is None and "IPython" in sys.modules:
533 from IPython import get_ipython
535 ip = get_ipython() # type:ignore[no-untyped-call]
536 in_kernel = getattr(ip, "kernel", False)
537 if in_kernel:
538 output_hook = partial(
539 self._output_hook_kernel,
540 ip.display_pub.session,
541 ip.display_pub.pub_socket,
542 ip.display_pub.parent_header,
543 )
544 if output_hook is None:
545 # default: redisplay plain-text outputs
546 output_hook = self._output_hook_default
548 # set deadline based on timeout
549 if timeout is not None:
550 deadline = time.monotonic() + timeout
551 else:
552 timeout_ms = None
554 poller = zmq.asyncio.Poller()
555 iopub_socket = self.iopub_channel.socket
556 poller.register(iopub_socket, zmq.POLLIN)
557 if allow_stdin:
558 stdin_socket = self.stdin_channel.socket
559 poller.register(stdin_socket, zmq.POLLIN)
560 else:
561 stdin_socket = None
563 # wait for output and redisplay it
564 while True:
565 if timeout is not None:
566 timeout = max(0, deadline - time.monotonic())
567 timeout_ms = int(1000 * timeout)
568 events = dict(await poller.poll(timeout_ms))
569 if not events:
570 emsg = "Timeout waiting for output"
571 raise TimeoutError(emsg)
572 if stdin_socket in events:
573 req = await ensure_async(self.stdin_channel.get_msg(timeout=0))
574 res = stdin_hook(req)
575 if inspect.isawaitable(res):
576 await res
577 continue
578 if iopub_socket not in events:
579 continue
581 msg = await ensure_async(self.iopub_channel.get_msg(timeout=0))
583 if msg["parent_header"].get("msg_id") != msg_id:
584 # not from my request
585 continue
586 output_hook(msg)
588 # stop on idle
589 if (
590 msg["header"]["msg_type"] == "status"
591 and msg["content"]["execution_state"] == "idle"
592 ):
593 break
595 # output is done, get the reply
596 if timeout is not None:
597 timeout = max(0, deadline - time.monotonic())
598 return await self._async_recv_reply(msg_id, timeout=timeout)
600 # Methods to send specific messages on channels
601 def execute(
602 self,
603 code: str,
604 silent: bool = False,
605 store_history: bool = True,
606 user_expressions: t.Dict[str, t.Any] | None = None,
607 allow_stdin: bool | None = None,
608 stop_on_error: bool = True,
609 ) -> str:
610 """Execute code in the kernel.
612 Parameters
613 ----------
614 code : str
615 A string of code in the kernel's language.
617 silent : bool, optional (default False)
618 If set, the kernel will execute the code as quietly possible, and
619 will force store_history to be False.
621 store_history : bool, optional (default True)
622 If set, the kernel will store command history. This is forced
623 to be False if silent is True.
625 user_expressions : dict, optional
626 A dict mapping names to expressions to be evaluated in the user's
627 dict. The expression values are returned as strings formatted using
628 :func:`repr`.
630 allow_stdin : bool, optional (default self.allow_stdin)
631 Flag for whether the kernel can send stdin requests to frontends.
633 Some frontends (e.g. the Notebook) do not support stdin requests.
634 If raw_input is called from code executed from such a frontend, a
635 StdinNotImplementedError will be raised.
637 stop_on_error: bool, optional (default True)
638 Flag whether to abort the execution queue, if an exception is encountered.
640 Returns
641 -------
642 The msg_id of the message sent.
643 """
644 if user_expressions is None:
645 user_expressions = {}
646 if allow_stdin is None:
647 allow_stdin = self.allow_stdin
649 # Don't waste network traffic if inputs are invalid
650 if not isinstance(code, str):
651 raise ValueError("code %r must be a string" % code)
652 validate_string_dict(user_expressions)
654 # Create class for content/msg creation. Related to, but possibly
655 # not in Session.
656 content = {
657 "code": code,
658 "silent": silent,
659 "store_history": store_history,
660 "user_expressions": user_expressions,
661 "allow_stdin": allow_stdin,
662 "stop_on_error": stop_on_error,
663 }
664 msg = self.session.msg("execute_request", content)
665 self.shell_channel.send(msg)
666 return msg["header"]["msg_id"]
668 def complete(self, code: str, cursor_pos: int | None = None) -> str:
669 """Tab complete text in the kernel's namespace.
671 Parameters
672 ----------
673 code : str
674 The context in which completion is requested.
675 Can be anything between a variable name and an entire cell.
676 cursor_pos : int, optional
677 The position of the cursor in the block of code where the completion was requested.
678 Default: ``len(code)``
680 Returns
681 -------
682 The msg_id of the message sent.
683 """
684 if cursor_pos is None:
685 cursor_pos = len(code)
686 content = {"code": code, "cursor_pos": cursor_pos}
687 msg = self.session.msg("complete_request", content)
688 self.shell_channel.send(msg)
689 return msg["header"]["msg_id"]
691 def inspect(self, code: str, cursor_pos: int | None = None, detail_level: int = 0) -> str:
692 """Get metadata information about an object in the kernel's namespace.
694 It is up to the kernel to determine the appropriate object to inspect.
696 Parameters
697 ----------
698 code : str
699 The context in which info is requested.
700 Can be anything between a variable name and an entire cell.
701 cursor_pos : int, optional
702 The position of the cursor in the block of code where the info was requested.
703 Default: ``len(code)``
704 detail_level : int, optional
705 The level of detail for the introspection (0-2)
707 Returns
708 -------
709 The msg_id of the message sent.
710 """
711 if cursor_pos is None:
712 cursor_pos = len(code)
713 content = {
714 "code": code,
715 "cursor_pos": cursor_pos,
716 "detail_level": detail_level,
717 }
718 msg = self.session.msg("inspect_request", content)
719 self.shell_channel.send(msg)
720 return msg["header"]["msg_id"]
722 def history(
723 self,
724 raw: bool = True,
725 output: bool = False,
726 hist_access_type: str = "range",
727 **kwargs: t.Any,
728 ) -> str:
729 """Get entries from the kernel's history list.
731 Parameters
732 ----------
733 raw : bool
734 If True, return the raw input.
735 output : bool
736 If True, then return the output as well.
737 hist_access_type : str
738 'range' (fill in session, start and stop params), 'tail' (fill in n)
739 or 'search' (fill in pattern param).
741 session : int
742 For a range request, the session from which to get lines. Session
743 numbers are positive integers; negative ones count back from the
744 current session.
745 start : int
746 The first line number of a history range.
747 stop : int
748 The final (excluded) line number of a history range.
750 n : int
751 The number of lines of history to get for a tail request.
753 pattern : str
754 The glob-syntax pattern for a search request.
756 Returns
757 -------
758 The ID of the message sent.
759 """
760 if hist_access_type == "range":
761 kwargs.setdefault("session", 0)
762 kwargs.setdefault("start", 0)
763 content = dict(raw=raw, output=output, hist_access_type=hist_access_type, **kwargs)
764 msg = self.session.msg("history_request", content)
765 self.shell_channel.send(msg)
766 return msg["header"]["msg_id"]
768 def kernel_info(self) -> str:
769 """Request kernel info
771 Returns
772 -------
773 The msg_id of the message sent
774 """
775 msg = self.session.msg("kernel_info_request")
776 self.shell_channel.send(msg)
777 return msg["header"]["msg_id"]
779 def comm_info(self, target_name: str | None = None) -> str:
780 """Request comm info
782 Returns
783 -------
784 The msg_id of the message sent
785 """
786 content = {} if target_name is None else {"target_name": target_name}
787 msg = self.session.msg("comm_info_request", content)
788 self.shell_channel.send(msg)
789 return msg["header"]["msg_id"]
791 def _handle_kernel_info_reply(self, msg: t.Dict[str, t.Any]) -> None:
792 """handle kernel info reply
794 sets protocol adaptation version. This might
795 be run from a separate thread.
796 """
797 adapt_version = int(msg["content"]["protocol_version"].split(".")[0])
798 if adapt_version != major_protocol_version:
799 self.session.adapt_version = adapt_version
801 def is_complete(self, code: str) -> str:
802 """Ask the kernel whether some code is complete and ready to execute.
804 Returns
805 -------
806 The ID of the message sent.
807 """
808 msg = self.session.msg("is_complete_request", {"code": code})
809 self.shell_channel.send(msg)
810 return msg["header"]["msg_id"]
812 def input(self, string: str) -> None:
813 """Send a string of raw input to the kernel.
815 This should only be called in response to the kernel sending an
816 ``input_request`` message on the stdin channel.
818 Returns
819 -------
820 The ID of the message sent.
821 """
822 content = {"value": string}
823 msg = self.session.msg("input_reply", content)
824 self.stdin_channel.send(msg)
826 def shutdown(self, restart: bool = False) -> str:
827 """Request an immediate kernel shutdown on the control channel.
829 Upon receipt of the (empty) reply, client code can safely assume that
830 the kernel has shut down and it's safe to forcefully terminate it if
831 it's still alive.
833 The kernel will send the reply via a function registered with Python's
834 atexit module, ensuring it's truly done as the kernel is done with all
835 normal operation.
837 Returns
838 -------
839 The msg_id of the message sent
840 """
841 # Send quit message to kernel. Once we implement kernel-side setattr,
842 # this should probably be done that way, but for now this will do.
843 msg = self.session.msg("shutdown_request", {"restart": restart})
844 self.control_channel.send(msg)
845 return msg["header"]["msg_id"]
848KernelClientABC.register(KernelClient)