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