ops

Standard async operations on a running kernel, for ConKernelClient

conkernelclient.core makes concurrent execute() calls safe; this module adds the operations every client ends up needing on top of that: request-scoped iopub collection, execute-and-drain composites, nbformat-style output conversion, interrupt, generic shell/control requests, and a kernel lifecycle context manager. It distills helpers that grew up independently in ipymini’s test suite and solveit’s gateway, keeping the strongest implementation of each.

One caveat applies throughout: shell replies are demultiplexed per-request by ConKernelClient, but iopub is a single broadcast stream. iopub_drain filters it by parent msg_id and discards what it skips, so run one drain at a time. Concurrent executes are fine; concurrent drains are not.

Imports

from fastcore.test import test_eq, test_fail
from jupyter_client.session import Session

Message basics

km = ConKernelManager(session=Session(key=b'x'))
await km.start_kernel()
kc = await km.client().start_channels()
await kc.is_alive()

True

---

source

iter_timeout

def iter_timeout(
    timeout:NoneType=None, default:int=10
):

Yield remaining seconds until timeout expires, using monotonic time

Every reply carries its request’s msg_id in the parent header; all routing below keys on it. iter_timeout drives every wait loop in this module: it yields the time still available, so a loop body can pass a shrinking timeout to each blocking call.

r = await kc.execute('1+1', reply=True)
test_eq(parent_id(r), r['parent_header']['msg_id'])
assert parent_id({}) is None

Iopub collection

---

source

ConKernelClient.iopub_drain

async def iopub_drain(
    msg_id, timeout:int=10
):

Collect iopub messages parented to msg_id until its idle status arrives. Other requests’ messages are discarded: one drain at a time

---

source

ConKernelClient.iopub_flush

async def iopub_flush(
    timeout:float=0.1
):

Discard all pending iopub messages, e.g. leftovers from fire-and-forget executes

iopub_flush is the counterpart for when you don’t want collection: clear the backlog before starting a push-based consumer or a fresh interaction.

kc.execute('print("junk")')
await kc.iopub_flush()
try:
    await kc.get_iopub_msg(timeout=0.1)
    assert False, 'iopub should be empty after flush'
except Empty: pass

Collecting a request’s output means draining iopub until the kernel publishes the idle status for that request - that is the protocol’s “this request is done publishing” signal, far more reliable than sweeping whatever is pending after a fixed delay.

mid = kc.execute('print("hi"); 42')
msgs = await kc.iopub_drain(mid)
test_eq([m['msg_type'] for m in msgs if m['msg_type']!='status'], ['execute_input', 'stream', 'execute_result'])

Outputs

---

source

iopub_streams

def iopub_streams(
    msgs, name:NoneType=None
):

The stream messages in msgs, optionally only stream name (‘stdout’/‘stderr’)

test_eq([m['msg_type'] for m in iopub_msgs(msgs, 'execute_result')], ['execute_result'])
test_eq([m['msg_type'] for m in iopub_msgs(msgs, output_types)], ['stream', 'execute_result'])
test_eq(iopub_streams(msgs)[0]['content']['text'], 'hi\n')
test_eq(iopub_streams(msgs, 'stderr'), [])

Raw-message filters for when the protocol wrapping is wanted (parent ids, per-message metadata) - msgs2outs is the lossy convenience, these are the lossless ones.

Iopub messages carry protocol wrapping (headers, status chatter, execute_input echoes) that consumers rarely want. fastcore.nbio.msgs2outs reduces them to the same output dicts a notebook file stores (mirroring nbformat.v4.output_from_msg without the dependency, minus schema validation), and exec_outs below composes it with a drain. output_types names the message types that carry outputs, for callers filtering by hand.

outs = msgs2outs(msgs)
test_eq(outs[0], dict(output_type='stream', name='stdout', text='hi\n'))
test_eq(outs[1]['data']['text/plain'], '42')

Execute composites

---

source

ConKernelClient.exec_outs

async def exec_outs(
    code, timeout:int=10, **kw
):

Execute code and return just its nbformat-style outputs

---

source

ConKernelClient.exec_ok

async def exec_ok(
    code, timeout:int=10, **kw
):

exec_drain, asserting the reply status is ok

---

source

ConKernelClient.exec_drain

async def exec_drain(
    code, timeout:int=10, **kw
):

Execute code; return (reply, outputs) where outputs are the request’s iopub messages

The composites cover the three common shapes: full protocol detail (exec_drain), tests that just need success (exec_ok), and consumers that only want outputs (exec_outs).

reply, outputs = await kc.exec_ok('x = 3; x')
test_eq(reply['content']['status'], 'ok')
test_eq((await kc.exec_outs('print(x*2)')), [dict(output_type='stream', name='stdout', text='6\n')])
reply, _ = await kc.exec_drain('1/0')
test_eq(reply['content']['ename'], 'ZeroDivisionError')

Expression values

---

source

ConKernelClient.eval_expr

async def eval_expr(
    expr:str, code:str='', timeout:int=10, **kw
):

Evaluate expr in the kernel (optionally after running code) and return its value: parsed via literal_eval when its repr allows, else the repr string. Raises EvalError if the kernel raises

---

source

ConKernelClient.user_exprs

async def user_exprs(
    exprs:dict, code:str='', timeout:int=10, **kw
):

Run code and evaluate each of the exprs expressions in the same round trip; returns the reply content (statuses and mimebundles unparsed)

---

source

parse_expr

def parse_expr(
    s
):

The literal_eval of s when its form allows, else s unchanged

---

source

EvalError

def EvalError(
    *args, **kwargs
):

An eval_expr expression raised in the kernel

The user_expressions round trip: one execute carries the expression, the kernel evaluates it after the (empty) cell and returns its repr inside the execute_reply - no iopub involved, so it works cleanly alongside streaming output. Values whose repr is not literal_eval-able come back as that repr string. This is the general core of solveit’s richer eval RPC.

await kc.exec_ok("v = [1, 'a', {'b': 2}]")
test_eq(await kc.eval_expr('v'), [1, 'a', {'b': 2}])
test_eq(await kc.eval_expr('w * 2', code='w = 21'), 42)
cts = await kc.user_exprs({'a': 'v[0]', 'b': 'len(v)'})
test_eq(cts['status'], 'ok')
test_eq(parse_expr(cts['user_expressions']['a']['data']['text/plain']), 1)
test_eq(parse_expr(cts['user_expressions']['b']['data']['text/plain']), 3)
test_eq(await kc.eval_expr('sum(v[2].values())'), 2)
r = await kc.eval_expr('print')
assert isinstance(r, str) and 'print' in r, r
try:
    await kc.eval_expr('nope_undefined')
    assert False, 'expected EvalError'
except EvalError as e: assert 'NameError' in str(e)

Calling kernel functions

eval turns the user_expressions round trip into a function call: run func(*args, **kw) kernel-side (awaiting coroutines), bring the result back by repr, and reconstruct it client-side — try_eval wraps primitive results in a dynamic class named after the kernel-side type, so Markdown reprs still compare equal while remembering what they were. _call=False skips the call and evaluates func as a bare expression. The whole family – eval, ipy (get_ipython() methods), the generated service methods mirroring what ipyfuncs patches onto the kernel’s shell (sig_help, get_schemas, ranked_complete, …), and xpush/retr/xenv – is inherited from jupywire’s EvalOps mixin over this module’s reply; _pre_ipy below adds the zmq liveness check. priority= sends kernmini’s priority: 1 execute metadata, so the call overtakes queued work on a kernel that honours it; ipykernel ignores the key. The service methods default it on.

---

source

ConKernelClient.reply

def reply(
    code, # A string of code in the kernel's language.
    user_expressions:NoneType=None, # A dict mapping names to expressions to be evaluated in the user's dict
    allow_stdin:NoneType=None, # Flag for whether the kernel can send stdin requests to frontends.
    cts_typ:str='code', timeout:int=None, msg_id:NoneType=None, **kw
):

The awaited transport seam: run code, returning an awaitable of the execute_reply message.

await kc.reply('def add(a, b): return a+b')
test_eq(await kc.eval('add', a=10, b=20), 30)

await kc.reply('async def add(a, b): return a+b')
test_eq(await kc.eval('add', a=10, b=20, _literal=False), '30')

await kc.reply('a = [1,2,3]')
test_eq(await kc.eval('a', _call=False), [1,2,3])

test_eq((await kc.eval('add', a=30, b=40, _literal=False), await kc.eval('add', a=30, b=40)), ('70', 70))

await kc.reply('from fastcore.xml import Safe')
r = await kc.eval('Safe', 'hello')
isinstance_str(r, 'Safe')

True

await kc.reply('from IPython.display import Markdown')
await kc.eval('Markdown', data=r"*a\a*", _literal=False)

'*a\\a*'

With literal=False, eval() skips literal_eval on the result:

Not evaluating allows getting non-builtin types as results:

await kc.reply("""
class Foo:
  def __init__(self, val: int): self.val = val
  def __repr__(self): return f'Foo instance with a value of {self.val}'
  
def make_foo(): return Foo(123)
""")
result = await kc.eval("Foo", val=99, _literal=False)
assert 'Foo instance' in result
test_eq(type(result), str)
result, type(result)

('Foo instance with a value of 99', str)

_r = await kc.eval('f', _timeout=10)
assert 'NameError' in _r, f'Expected NameError, got: {_r}'
print(_r)

---------------------------------------------------------------------------
NameError                                 Traceback (most recent call last)
Cell In[9], line 2
      1 import asyncio
----> 2 __254d11bb = f(*(), **{})
      3 if asyncio.iscoroutine(__254d11bb): __254d11bb = await __254d11bb

NameError: name 'f' is not defined

The _ipy_funcs services live kernel-side in ipyfuncs; importing it is the whole setup, so this battery doesn’t depend on a venv-level IPython config:

await kc.exec_ok('import ipyfuncs')

await kc.reply('''def range_ex(
    a:str  # some param
):
    "some func docstring"
    ...''')

{'header': {'msg_id': 'f40f78a3-249cb73f14dbb4b0d1640fd1_89345_172',
  'msg_type': 'execute_reply',
  'username': 'jhoward',
  'session': 'f40f78a3-249cb73f14dbb4b0d1640fd1',
  'date': datetime.datetime(2026, 8, 3, 7, 18, 32, 384722, tzinfo=tzutc()),
  'version': '5.4'},
 'msg_id': 'f40f78a3-249cb73f14dbb4b0d1640fd1_89345_172',
 'msg_type': 'execute_reply',
 'parent_header': {'msg_id': 'fc18205f-f426f5401e4dca936a58e529_89336_38',
  'msg_type': 'execute_request',
  'username': 'jhoward',
  'session': 'fc18205f-f426f5401e4dca936a58e529',
  'date': datetime.datetime(2026, 8, 3, 7, 18, 32, 383610, tzinfo=tzutc()),
  'version': '5.4'},
 'metadata': {},
 'content': {'status': 'ok',
  'execution_count': 21,
  'user_expressions': {},
  'payload': []},
 'buffers': []}

await kc.eval('get_ipython().ranked_complete', code='rang', line_no=1, col_no=5, _timeout=0.2)

[{'text': 'range',
  'type': 'class',
  'signature': '',
  'start': 0,
  'end': 4,
  'mod': None,
  'rank': 5},
 {'text': 'range_ex',
  'type': 'function',
  'signature': '(a: str)',
  'start': 0,
  'end': 4,
  'mod': '__main__',
  'rank': 2}]

await kc.ranked_complete(code='rang', line_no=1, col_no=5)

[{'text': 'range',
  'type': 'class',
  'signature': '',
  'start': 0,
  'end': 4,
  'mod': None,
  'rank': 5},
 {'text': 'range_ex',
  'type': 'function',
  'signature': '(a: str)',
  'start': 0,
  'end': 4,
  'mod': '__main__',
  'rank': 2}]

await kc.reply('a=1')
test_eq(await kc.get_vars(vs=['a']), {'a': 1})
await kc.user_items(max_len=100)

({'a': '1',
  'payl': "{'source': 'testing', 'foo': 'bar'}",
  'pm': '<IPython.core.payload.PayloadManager object>',
  'user_input': 'bbb',
  'test_var': '42',
  'another_var': 'hello'},
 {'add': '(a, b)', 'make_foo': '()', 'range_ex': '(a: str)'})

res = await kc.sig_help(code='range(', line_no=1, col_no=6)
res[0]

{'label': 'class range',
 'typ': 'class',
 'mod': 'builtins',
 'doc': 'range(stop: SupportsIndex, /)\nrange(start: SupportsIndex, stop: SupportsIndex, step: SupportsIndex=1, /)\n\nrange(stop) -> range object\nrange(start, stop[, step]) -> range object\n\nReturn an object that produces a sequence of integers from start (inclusive)\nto stop (exclusive) by step.  range(i, j) produces i, i+1, i+2, ..., j-1.\nstart defaults to 0, and stop is omitted!  range(4) produces 0, 1, 2, 3.\nThese are exactly the valid indices for a list of 4 elements.\nWhen step is given, it specifies the increment (or decrement).',
 'idx': 0,
 'params': [{'name': 'stop', 'desc': 'param stop: SupportsIndex'}]}

await kc.get_schemas(fs=['range_ex'])

{'range_ex': {'type': 'function',
  'function': {'name': 'range_ex',
   'description': 'some func docstring',
   'parameters': {'type': 'object',
    'properties': {'a': {'description': 'some param', 'type': 'string'}},
    'required': ['a']}}}}

kc.xpush(asdf=4)

test_eq(await kc.retr('asdf'), 4)

kc.xenv(hi='johno')
test_eq(await kc.eval('__os.environ["hi"]', _call=False), 'johno')

Generic requests

---

source

ConKernelClient.control_request

async def control_request(
    msg_type, timeout:int=10, **content
):

Send a request on the control channel and await its matching reply

---

source

ConKernelClient.shell_request

def shell_request(
    msg_type, timeout:int=10, reply:bool=True, buffers:NoneType=None,
    msg_id:NoneType=None, # Override the auto-generated message id
    subshell_id:NoneType=None, # Route via this subshell (header field)
    metadata:NoneType=None, # Message metadata (e.g. ipywidgets control-comm version check)
    **content
):

Send an arbitrary shell request, routed through the reply reader like execute. Returns a coroutine for the reply, or just the msg_id when reply=False (for fire-and-forget types like comm_open that never get replies)

ConKernelClient’s background reader consumes every shell message, so a bare get_shell_msg would race it and lose. Any non-execute shell request (kernel_info_request, complete_request, …) must therefore register with the same routing table execute uses - that is what shell_request does. Pass reply=False for fire-and-forget message types the kernel never answers (comm_open, comm_msg, comm_close), and buffers for binary payloads. Control-channel replies have no reader competing for them, so control_request matches by parent id directly. Note control requests should not be issued concurrently with each other.

r = await kc.shell_request('kernel_info_request')
test_eq(r['header']['msg_type'], 'kernel_info_reply')
r = await kc.control_request('kernel_info_request')
test_eq(r['header']['msg_type'], 'kernel_info_reply')

Read the input_request from the stdin channel with get_stdin_msg, then answer it; the blocked execute’s reply follows.

---

source

ConKernelClient.input_reply

async def input_reply(
    value:str
):

Answer the kernel’s pending input_request (from input() in an execute sent with allow_stdin=True)

c = kc.execute("x = input('name? ')", reply=True, timeout=10, allow_stdin=True)
msg = await kc.get_stdin_msg(timeout=10)
test_eq(msg['header']['msg_type'], 'input_request')
await kc.input_reply('Ada')
test_eq((await c)['content']['status'], 'ok')
test_eq((await kc.exec_outs('x'))[0]['data']['text/plain'], "'Ada'")

code = """def _t(comm, msg):
    global nbuf; nbuf = len(msg.get('buffers', []))
get_ipython().kernel.comm_manager.register_target('t', _t)
"""
await kc.exec_ok(code)
mid = kc.shell_request('comm_open', reply=False, comm_id='c1', target_name='t', data={}, buffers=[b'12345'])
assert isinstance(mid, str)
await kc.exec_ok('import time; time.sleep(0.2)')
test_eq((await kc.exec_outs('nbuf'))[0]['data']['text/plain'], '1')

code = """def _tm(comm, msg):
    global nmd; nmd = msg.get('metadata')
get_ipython().kernel.comm_manager.register_target('tm', _tm)
"""
await kc.exec_ok(code)
kc.shell_request('comm_open', reply=False, comm_id='cm1', target_name='tm', data={}, metadata={'version':'2.1.0'})
await kc.exec_ok('import time; time.sleep(0.2)')
test_eq((await kc.exec_outs('nmd'))[0]['data']['text/plain'], "{'version': '2.1.0'}")

Proxies

---

source

ConKernelClient.ctl

def ctl():

Control requests as methods: await kc.ctl.shutdown(restart=True) sends shutdown_request

---

source

ConKernelClient.cmd

def cmd():

Shell requests as methods: await kc.cmd.history(...) sends history_request

---

source

ConKernelClient.dap

def dap():

DAP debug requests as methods. Stateful (seq counter), so cached per client, unlike cmd/ctl

dap speaks the Debug Adapter Protocol over debug_request control messages. A trailing underscore escapes Python keywords (kc.dap.continue_(...) sends the DAP continue command); full=True returns the whole reply message instead of just its content.

The proxies make one-off protocol requests read like methods without hand-assembling messages. They are stateless, so no caching is needed.

r = await kc.cmd.complete(code='pri', cursor_pos=3)
assert 'print' in r['content']['matches']
r = await kc.cmd.is_complete(code='def f():')
test_eq(r['content']['status'], 'incomplete')

r = await kc.shell_request('kernel_info_request', msg_id='fixed-id-1')
test_eq(parent_id(r), 'fixed-id-1')
r = await kc.dap.debugInfo()
test_eq(r['command'], 'debugInfo')
assert r['success']
r2 = await kc.dap.debugInfo(full=True)
test_eq(parent_id(r2), parent_id(r2))  # full message shape
assert r2['header']['msg_type'] == 'debug_reply'

Interrupt

---

source

ConKernelClient.interrupt

async def interrupt(
    timeout:int=5
):

Interrupt the running request via interrupt_request on control; returns the reply

The interrupted execute returns an error reply with ename KeyboardInterrupt, and the kernel stays usable. (Solveit’s gateway historically followed the interrupt with an empty execute to wake a parked recv; the Session.send patch in core addresses that wake-up at the transport level, so it is not repeated here.)

task = asyncio.ensure_future(kc.execute('import time; time.sleep(30)', reply=True, timeout=15))
await asyncio.sleep(0.5)
r = await kc.interrupt()
test_eq(r['header']['msg_type'], 'interrupt_reply')
reply = await task
test_eq(reply['content']['ename'], 'KeyboardInterrupt')
test_eq((await kc.exec_outs('40+2'))[0]['data']['text/plain'], '42')

Lifecycle

---

source

run_kernel

def run_kernel(
    kernel_name:str='python3', # Kernelspec name to launch
    manager_cls:MetaHasTraits=ConKernelManager, # Manager class, e.g. a subclass customizing launch
    **kwargs
):

Start a kernel, yield (km, kc) with channels running, and always shut both down

---

source

reconnect

async def reconnect(
    km, kc:NoneType=None
):

Restart km’s kernel (fresh process, state discarded) and return a newly connected client, stopping kc if given

async with run_kernel() as (km3, kc3):
    await kc3.exec_ok('x = 42')
    pid1 = km3.provisioner.pid
    kc3 = await reconnect(km3, kc3)
    assert km3.provisioner.pid != pid1
    reply, _ = await kc3.exec_drain('x')
    test_eq(reply['content']['ename'], 'NameError')
    test_eq((await kc3.exec_outs('1+1'))[0]['data']['text/plain'], '2')

After a restart the old client’s channels and routed-reply reader are stale, so reconnect hands back a fresh client bound to the restarted process. State is gone: redo any imports/setup.

The context manager guarantees teardown even when a test body raises, which is what keeps a big protocol test suite from leaking kernel processes.

async with run_kernel() as (km2, kc2):
    test_eq((await kc2.exec_outs('1+1'))[0]['data']['text/plain'], '2')
    pid = km2.provisioner.pid
assert not await km2.is_alive()

Cleanup

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