ConKernelClient source

Concurrent-safe Jupyter KernelClient

Imports

from queue import Empty
from jupyter_client import KernelClient
from fastcore.test import test_eq
from fastcore.utils import patch

---

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

A KernelClient with async APIs

get_[channel]_msg() methods wait for and return messages on channels, raising :exc:queue.Empty if no message arrives within timeout seconds.

---

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

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'}

kc.stop_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()