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:VAR_POSITIONAL, kwargs:VAR_KEYWORD
):

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:VAR_POSITIONAL, kwargs:VAR_KEYWORD
):

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.

km = ConKernelManager(session=Session(key=b'x'))
await km.start_kernel()
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

'9f73c643-c0d062a945ec394c05c7c628_5169_1'

@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': '9f73c643-c0d062a945ec394c05c7c628_5169_1',
 'msg_type': 'execute_request',
 'username': 'jhoward',
 'session': '9f73c643-c0d062a945ec394c05c7c628',
 'date': datetime.datetime(2026, 6, 11, 0, 55, 34, 86393, 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': 'f322d2fc-83ead8c02338f8734d49896a_5171_21',
  'msg_type': 'execute_reply',
  'username': 'jhoward',
  'session': 'f322d2fc-83ead8c02338f8734d49896a',
  'date': datetime.datetime(2026, 6, 11, 0, 55, 35, 295246, tzinfo=tzutc()),
  'version': '5.4'},
 'msg_id': 'f322d2fc-83ead8c02338f8734d49896a_5171_21',
 'msg_type': 'execute_reply',
 'parent_header': {'msg_id': '9f73c643-c0d062a945ec394c05c7c628_5169_1',
  'msg_type': 'execute_request',
  'username': 'jhoward',
  'session': '9f73c643-c0d062a945ec394c05c7c628',
  'date': datetime.datetime(2026, 6, 11, 0, 55, 35, 289787, tzinfo=tzutc()),
  'version': '5.4'},
 'metadata': {'started': '2026-06-11T00:55:35.291931Z',
  'dependencies_met': True,
  'engine': '92e9022c-b4ef-4823-8ca8-5de2a9bb4e4b',
  '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']

'9f73c643-c0d062a945ec394c05c7c628_5169_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()