ConKernelClient source

Concurrent-safe Jupyter KernelClient

Imports

import os
from jupyter_client import KernelClient
from fastcore.test import test_eq

---

source

DeadKernelError

def DeadKernelError(
    *args, **kwargs
):

Common base class for all non-exit exceptions.

Setup

---

source

apply_session_patch

def apply_session_patch():

Make Session.send compensate the ZMQ FD edge a sync send consumes (see _send comment). Idempotent; applied automatically when a ConKernelClient is constructed, and available for use with vanilla clients.

Importing this module must not change global state: the Session.send patch below is only applied when the first ConKernelClient is constructed (via apply_session_patch, which vanilla-client users such as tests/zmq_regression.py can also call directly). This keeps import conkernelclient side-effect free, so e.g. protocol test suites can exercise an unpatched client in the same process.

assert not hasattr(Session, '_orig_send')
assert Session.send is not _send

---

source

ConKernelClient

def ConKernelClient(
    *args, **kwargs
):

Kernel-client calling conventions over the inheritor’s reply and execute: see the module docstring.

Readiness

start_channels must not return until the kernel is genuinely ready: the shell socket answering requests, and our iopub subscription registered, since anything published before registration is silently dropped (zmq’s “slow joiner”). jupyter_client’s wait_for_ready proves the subscription indirectly: poke the kernel with kernel_info probes until some iopub traffic appears, then drain whatever the probes stirred up until 0.2s of silence. That works against any kernel, but costs settling waits on every connect, and “quiet for 0.2s” is a guess rather than a guarantee.

Kernels whose iopub socket is XPUB publish an iopub_welcome message the moment a subscription registers (JEP 65: ipykernel, ipymini). Nothing can be delivered to a subscriber before its subscription exists, so the welcome is guaranteed to be the first message a fresh client receives, and that makes readiness deterministic with no capability check needed. If the first iopub message is a welcome, nothing published afterwards can have been missed, so one final kernel_info acts as an end marker: once its reply arrives on shell every earlier probe reply has been consumed, and once its idle arrives on iopub every earlier probe’s status pulse has too, leaving both channels provably clean. If the first message is anything else, this kernel does not send welcomes, and that message is exactly the evidence jupyter_client’s loop waits for, so we continue with its classic semantics: take a kernel_info_reply, then drain iopub until silence.

---

source

ConKernelClient.wait_for_ready

async def wait_for_ready(
    timeout:NoneType=None
):

Wait for the kernel to be ready: deterministic via iopub_welcome (JEP 65) when the kernel sends one, else jupyter_client’s probe loop

stop_channels is synchronous: it closes the sockets and cancels the reader without awaiting it, which is fine when the caller shares the client’s event loop. Callers driving the client from another thread (such as sync code using fastcore’s run_sync) need teardown to run on the loop: astop_channels cancels the reader, awaits its exit, then closes the channels.

---

source

ConKernelClient.astop_channels

async def astop_channels():

Stop channels and cancel the background shell-reply reader, awaiting its exit

---

source

ConKernelManager

def ConKernelManager(
    *args:Any, **kwargs:Any
)->None:

An async kernel manager.

ConKernelManager defaults transport_encryption to 'auto': when the kernelspec declares supported_encryption: "curve" (as ipykernel≥7.3 does), the manager provisions CurveZMQ keys so kernel traffic is encrypted over TCP — which also silences ipykernel’s plain-TCP warning. For kernels without that metadata (e.g. ipymini) it’s a no-op.

km = ConKernelManager(session=Session(key=b'x'))
await km.start_kernel()
assert km.curve_secretkey is not None
await km.is_alive()

True

kc = await km.client().start_channels()
await kc.is_alive()

True

assert Session.send is _send
assert hasattr(Session, '_orig_send')

reply=False (the default) sends the request and returns its msg_id without registering it in _pending, so the matching execute_reply finds no entry in the reader’s routing table and is discarded. That is what makes fire-and-forget safe on the shell channel: the reader consumes every message either way, so nothing accumulates there. Iopub output from the request does queue up, which is what get_pubs below collects (and iopub_flush discards).

mid = kc.execute('1+1', reply=False)
mid

'1f83b809-b8efedc8ba1ecd01cfb4fb64_61485_2'

@patch
async def get_pubs(self:KernelClient, timeout=0.2):
    "Retrieve all outstanding iopub messages"
    res = []
    try:
        while msg := await self.get_iopub_msg(timeout=timeout): res.append(msg)
    except Empty: pass
    return res

pubs = await kc.get_pubs()
[(o['msg_type'],o['content']) for o in pubs]

[('status', {'execution_state': 'busy'}),
 ('execute_input', {'code': '1+1', 'execution_count': 1}),
 ('execute_result',
  {'data': {'text/plain': '2'},
   'metadata': {'__type': 'int'},
   'execution_count': 1}),
 ('status', {'execution_state': 'idle'})]

pubs[0]['parent_header']

{'msg_id': '1f83b809-b8efedc8ba1ecd01cfb4fb64_61485_2',
 'msg_type': 'execute_request',
 'username': 'jhoward',
 'session': '1f83b809-b8efedc8ba1ecd01cfb4fb64',
 'date': datetime.datetime(2026, 7, 7, 9, 24, 21, 820745, tzinfo=tzutc()),
 'version': '5.4'}

await kc.astop_channels()

kc = await km.client().start_channels()

r = await kc.execute('2+1', timeout=1, reply=True)
r

{'header': {'msg_id': '0cc85378-9d1d9ce3859d6c5b00f14953_61488_25',
  'msg_type': 'execute_reply',
  'username': 'jhoward',
  'session': '0cc85378-9d1d9ce3859d6c5b00f14953',
  'date': datetime.datetime(2026, 7, 7, 9, 24, 22, 654854, tzinfo=tzutc()),
  'version': '5.4'},
 'msg_id': '0cc85378-9d1d9ce3859d6c5b00f14953_61488_25',
 'msg_type': 'execute_reply',
 'parent_header': {'msg_id': '1f83b809-b8efedc8ba1ecd01cfb4fb64_61485_1',
  'msg_type': 'execute_request',
  'username': 'jhoward',
  'session': '1f83b809-b8efedc8ba1ecd01cfb4fb64',
  'date': datetime.datetime(2026, 7, 7, 9, 24, 22, 650658, tzinfo=tzutc()),
  'version': '5.4'},
 'metadata': {'started': '2026-07-07T09:24:22.652210Z',
  'dependencies_met': True,
  'engine': 'e4e54c96-a1fd-4457-ba83-60d06c626bda',
  'status': 'ok'},
 'content': {'status': 'ok',
  'execution_count': 2,
  'user_expressions': {},
  'payload': []},
 'buffers': []}

await kc.get_pubs()
kc.execute('print("orphan")')
await asyncio.sleep(0.3)
slow, fast = await asyncio.gather(kc.execute('import time; time.sleep(0.3)', timeout=5, reply=True),
    kc.execute('1+1', timeout=5, reply=True), return_exceptions=True)
test_eq(type(slow), dict)
test_eq(type(fast), dict)

a = kc.execute('x=2', reply=True)
b = kc.execute('y=3', reply=True)

r = await asyncio.wait_for(asyncio.gather(a,b), timeout=2)
test_eq(len(r), 2)
r[0]['parent_header']['msg_id']

'1f83b809-b8efedc8ba1ecd01cfb4fb64_61485_5'

With the default fail_pending=False, an error only affects what the kernel decides (here: stop_on_error=True on the wire makes it abort the queued cell, whose real aborted reply comes back). Pass fail_pending=None to follow stop_on_error (or True to force): when this request errors, every other pending reply fails fast with a RuntimeError naming the root cause - the pipelining behavior solveit’s gateway was built on.

a = kc.execute("import time; time.sleep(0.2); raise ValueError('boom')", reply=True)
b = kc.execute("print('queued')", reply=True)
ra, rb = await asyncio.gather(a, b)
test_eq(ra['content']['status'], 'error')
test_eq(rb['content']['status'], 'aborted')

a = kc.execute("import time; time.sleep(0.2); raise ValueError('boom2')", reply=True, fail_pending=None)
b = kc.execute("print('queued2')", reply=True)
try:
    await asyncio.gather(a, b)
    assert False, 'expected RuntimeError from fail_pending'
except RuntimeError as e: assert 'boom2' in str(e)

async def g():
    for i in range(10): await kc.execute(f'a{i}={i}; a{i}', reply=True)

r = await asyncio.wait_for(asyncio.gather(g(),g(),g(),g()), timeout=10)

if await km.is_alive():
    kc.stop_channels()
    await km.shutdown_kernel()

Kernels with and without welcomes

Both readiness paths run against a real kernel. ipymini’s iopub socket is XPUB with XPUB_VERBOSE, so every subscriber gets its welcome, and its IPYMINI_IOPUB_XPUB=0 env flag reverts the socket to plain PUB, emulating a pre-JEP-65 kernel.

Connecting to a running welcoming kernel (the first client below absorbs kernel boot; the timed one connects to a warm kernel) completes in a few round-trips, no probe retries or drain windows, and hands over a provably clean iopub stream:

km2 = ConKernelManager(kernel_name='ipymini')
await km2.start_kernel()
kc2 = await km2.client().start_channels()
t0 = time.monotonic()
kc3 = await km2.client().start_channels()
elapsed = time.monotonic() - t0
test_eq(await kc3.get_pubs(), [])
assert elapsed < 0.35
for c in (kc2, kc3): c.stop_channels()
await km2.shutdown_kernel(now=True)
elapsed

The same kernel started with welcomes disabled sends the client down the fallback probe loop: it still gets ready and executes fine, but pays the drain’s settling wait, visible in the elapsed time:

km2 = ConKernelManager(kernel_name='ipymini')
await km2.start_kernel(env=dict(os.environ, IPYMINI_IOPUB_XPUB='0'))
kc2 = await km2.client().start_channels()
t0 = time.monotonic()
kc3 = await km2.client().start_channels()
elapsed = time.monotonic() - t0
reply = await kc3.execute('1+1', reply=True, timeout=10)
test_eq(reply['content']['status'], 'ok')
for c in (kc2, kc3): c.stop_channels()
await km2.shutdown_kernel(now=True)
elapsed

Stdin

input_request is the protocol’s one unsolicited routed send: the kernel learns a client’s identity from the shell request, then sends on its stdin ROUTER socket - a socket the client may not have finished connecting yet. Historically that send was silently discarded, so input() in the first instant after connect could lose its prompt. ipymini closes this at the source: its stdin router detects the unroutable send and retries until the client’s pipe is up, so the prompt below arrives even though the execute is fired immediately after start_channels returns. (Against kernels without that redelivery, such as ipykernel, this demo is timing-dependent - which is also its behavior with plain jupyter_client.)

km3 = ConKernelManager(kernel_name='ipymini', session=Session(key=b'x'))
await km3.start_kernel()
kc3 = await km3.client().start_channels()
fut = asyncio.ensure_future(kc3.execute("name = input('name: ')", reply=True, allow_stdin=True, timeout=10))
req = await kc3.get_stdin_msg(timeout=10)
kc3.input('Jeremy')
r = await fut
await kc3.astop_channels()
await km3.shutdown_kernel(now=True)
test_eq(req['content']['prompt'], 'name: ')
test_eq(r['content']['status'], 'ok')