ConKernelClient source

Concurrent-safe Jupyter KernelClient

Imports

import os
from jupyter_client import KernelClient
from jupyter_client.kernelspec import KernelSpec
from fastcore.test import test_eq, ExceptionExpected
from jupywire.route import JmsgQueues

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
):

Message handling over a transport that calls route(msg): reply(), run(), named requests, stdin, and death.

One _pump task per zmq socket drains into route, awaiting async handlers. A socket error is transport loss, and _watch_hb covers the kernel process: three consecutive unanswered heartbeat echoes deliver the same synthesized dead status.

start_channels waits for readiness, then launches the pumps and the heartbeat watcher; stop_channels fails every waiter and cancels them.

---

source

ConKernelClient.stop_channels

def stop_channels():

Stop channels and cancel the router pumps

---

source

ConKernelClient.start_channels

async def start_channels(
    shell:bool=True, iopub:bool=True, stdin:bool=True, hb:bool=True, control:bool=True
):

Start channels, wait for ready, then launch the router pumps and the heartbeat watcher

send transmits one built message dict, with buffers as extra frames; execute sends fire-and-forget and returns its msg_id.

---

source

ConKernelClient.execute

def execute(
    code, silent:bool=False, store_history:bool=True, user_expressions:NoneType=None, allow_stdin:NoneType=None,
    stop_on_error:bool=True, msg_id:NoneType=None, metadata:NoneType=None, subshell_id:NoneType=None,
    buffers:NoneType=None
):

Send an execute_request, fire-and-forget; returns its msg_id

---

source

ConKernelClient.send

def send(
    msg, channel
):

Transmit a built message dict on channel (buffers ride as extra frames); a closed socket raises DeadKernelError.

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 pumps without awaiting them, 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 each pump, awaits its exit, then closes the channels.

---

source

ConKernelClient.astop_channels

async def astop_channels():

Stop channels and cancel the router pumps, awaiting their exit

The get_*_msg accessor surface moves to JmsgQueues: attach one, and each get_*_msg reads its channel’s queue of routed-but-unmatched traffic, raising queue.Empty on timeout. A fire-and-forget request’s reply is such traffic, and so is every broadcast no run() collects. With no JmsgQueues attached (and no on_jmsg set), unmatched traffic is dropped, so nothing accumulates unread.

---

source

ConKernelManager

def ConKernelManager(
    *args, kernel_spec:NoneType=None, **kwargs
):

An async kernel manager.

jupyter_client normally resolves its read-only kernel_spec property by name. ConKernelManager also accepts a KernelSpec directly, allowing embedders to provide an in-memory launch description without installing it in Jupyter’s registry.

It also 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()
qs = JmsgQueues(kc)
await kc.is_alive()

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

execute sends the request and returns its msg_id without filing anything, so the matching execute_reply and the request’s iopub output are unmatched traffic: routed to on_jmsg, and readable through an attached JmsgQueues (as get_pubs below does). reply files a future for one execute_reply and returns its awaitable. run collects everything one execute causes.

mid = kc.execute('1+1')
mid

async def get_pubs(qs, timeout=0.2):
    "Retrieve all outstanding iopub messages"
    res = []
    try:
        while msg := await qs.get('iopub', timeout=timeout): res.append(msg)
    except Empty: pass
    return res

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

Every broadcast names its request in parent_header:

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()
qs = JmsgQueues(kc)

reply on the fresh channels, awaited for the whole reply message:

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

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

a = kc.reply('x=2')
b = kc.reply('y=3')
r = await asyncio.wait_for(asyncio.gather(a,b), timeout=2)
test_eq(len(r), 2)
r[0]['parent_header']['msg_id'] != r[1]['parent_header']['msg_id']

An error only affects what the kernel decides: stop_on_error=True on the wire makes it abort the queued cell, and the real aborted reply comes back to its own waiter.

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

async def g():
    for i in range(10): await kc.reply(f'a{i}={i}; a{i}')
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. The notebook supplies ipymini’s command through an in-memory KernelSpec, so running the source does not depend on ipymini also being installed in Jupyter’s kernelspec registry.

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:

mini_manager builds an ipymini ConKernelManager from a KernelSpec, the fixture both readiness paths share:

def mini_manager(**kwargs):
    spec = KernelSpec(argv=['ipymini', '-f', '{connection_file}'], display_name='ipymini')
    return ConKernelManager(kernel_spec=spec, **kwargs)

km2 = mini_manager()
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 get_pubs(JmsgQueues(kc3)), [])
assert elapsed < 0.35
elapsed

for c in (kc2, kc3): c.stop_channels()
await km2.shutdown_kernel(now=True)

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 = mini_manager()
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.reply('1+1', timeout=10)
test_eq(reply['content']['status'], 'ok')
elapsed

for c in (kc2, kc3): c.stop_channels()
await km2.shutdown_kernel(now=True)

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 = mini_manager(session=Session(key=b'x'))
await km3.start_kernel()
kc3 = await km3.client().start_channels()
qs3 = JmsgQueues(kc3)
fut = asyncio.ensure_future(kc3.reply("name = input('name: ')", allow_stdin=True, timeout=10))
req = await qs3.get('stdin', timeout=10)
kc3.input('Jeremy')
r = await fut
test_eq(req['content']['prompt'], 'name: ')
test_eq(r['content']['status'], 'ok')

Kernel death

Over zmq the dead status never arrives on the wire, so the client produces it. A _pump that hits a socket error routes the synthesized status, covering transport loss. The heartbeat watcher covers the kernel process itself: the kernel echoes heartbeats from a dedicated thread, so three consecutive unanswered echoes mean the process is gone. Either way route fails every pending reply() and run() with DeadKernelError. A kernel killed mid-cell never replies:

w = kc3.reply('import os, signal; os.kill(os.getpid(), signal.SIGKILL)', timeout=30)
with ExceptionExpected(DeadKernelError): await w
await kc3.astop_channels()
await km3.shutdown_kernel(now=True)