ops

Standard async operations on a running kernel, for ConKernelClient

conkernelclient.core makes concurrent execute(), reply(), and run() calls safe; this module adds the operations every client ends up needing on top of that: request-scoped output collection (exec_drain, over the shared run), 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. Collection is per-request throughout: each run files its entry before sending, so concurrent collections never touch each other’s traffic, and an attached JmsgQueues only ever holds what nothing matched.

Imports

import asyncio
from fastcore.test import test_eq, test_fail, ExceptionExpected
from fastcore.nbio import msg2out, msgs2outs
from jupywire.route import OUTPUT_MSGS, JmsgQueues
from jupywire.ops import output_types, parse_expr, EvalError
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()
qs = JmsgQueues(kc)
await kc.is_alive()

---

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.reply('1+1')
test_eq(parent_id(r), r['parent_header']['msg_id'])
assert parent_id({}) is None

Iopub collection

---

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

iopub_drain is the after-the-fact form, for a request that was sent without a run() (a bare execute, or one whose reply went to a reply() waiter): its iopub traffic reached the attached JmsgQueues, and the drain filters that queue by parent msg_id until the request’s idle. What it skips is discarded, so run one at a time; collectors that file before sending (exec_drain, run) are the concurrent-safe form.

---

source

ConKernelClient.iopub_drain

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

Collect unmatched iopub messages parented to msg_id from the attached JmsgQueues 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 pending iopub messages from the attached JmsgQueues, 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 qs.get('iopub', timeout=0.1)
    assert False, 'iopub should be empty after flush'
except Empty: pass

Collecting a request’s output means reading its messages until the kernel publishes the idle status for that request - the protocol’s “this request is done publishing” signal, far more reliable than sweeping whatever is pending after a fixed delay. exec_drain runs code through the shared run, so every message parented to the execute - the raw iopub messages, and the execute_reply itself - is collected by its own entry, and several drains can run concurrently without touching each other’s traffic (or anything on an attached JmsgQueues).

reply, msgs = await kc.exec_drain('print("hi"); 42')
test_eq(reply['content']['status'], 'ok')
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_ok

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

exec_drain, asserting the reply status is ok

exec_ok and exec_outs cover the other two common shapes: tests that just need success, and consumers that only want nbformat-style outputs (exec_outs comes from jupywire.route, over the shared run).

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

Streaming an execution: run

run comes from jupywire.route, shared with jupyasyncclient: it files its entry and sends at call time, and returns an async generator yielding every message parented to the execute as it arrives, ending once both the execute_reply and the idle status have been seen. Several runs in flight each collect only their own traffic:

types = [msg2out(m)['output_type'] async for m in kc.run("print('hi'); 6*7") if m['msg_type'] in OUTPUT_MSGS]
test_eq(types, ['stream', 'execute_result'])
test_eq((await kc.exec_outs('6*7'))[0]['data']['text/plain'], '42')

Concurrent submission means the kernel queues the cells, and with the wire default stop_on_error=True an error in one aborts the runs queued behind it. Independent runs pass stop_on_error=False, and each still collects only its own outputs (bounded, so a routing regression fails instead of hanging):

o1, o2, o3 = await asyncio.wait_for(asyncio.gather(*[kc.exec_outs(c, stop_on_error=False) for c in ('y = 6*7', '1/0', 'y')]), 30)
test_eq((o1, o2[0]['ename'], o3[0]['data']['text/plain']), ([], 'ZeroDivisionError', '42'))

Expression values

user_exprs and eval_expr live in jupywire.ops.EvalOps, shared with jupyasyncclient. 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 (parse_expr), and a kernel-side error raises EvalError. 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)
test_eq(await kc.eval_expr('sum(v[2].values())'), 2)

user_exprs returns the raw reply content, several expressions in one round trip; parse_expr reads each repr back:

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)

A repr that will not literal_eval comes back as the plain string, and a kernel-side error raises EvalError:

r = await kc.eval_expr('print')
assert isinstance(r, str) and 'print' in r, r
with ExceptionExpected(EvalError, 'NameError'): await kc.eval_expr('nope_undefined')

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. sidecar_=True tags a call for kernmini’s persistent named sidecar, which is created on first use. Service methods and variable operations (xpush, get_vars, eval_exprs, and retr) default to that serial lane; ordinary eval defaults to the main shell. Numeric priority execute metadata remains the separate queue-overtaking feature.

The notebook’s long-lived compatibility kernel is stock ipykernel, so sidecar-backed examples explicitly pass sidecar_=False. The real ipymini context near the end exercises the sidecar defaults.

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*", 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}]

The generated service wrapper makes the same call directly:

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

[{'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'], sidecar_=False), {'a': 1})
await kc.user_items(max_len=100, sidecar_=False)

({'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, sidecar_=False)
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'}]}

get_schemas returns docments-aware JSON schemas for named kernel functions:

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

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

xpush is the sync fire-and-forget sidecar setter for names; xenv does the same for environment variables on the main shell:

kc.xpush(asdf=4, sidecar_=False)

test_eq(await kc.retr('asdf', sidecar_=False), 4)

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

Generic requests

---

source

ConKernelClient.control_request

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

A named control request, content as keyword arguments; returns an awaitable of its reply

---

source

ConKernelClient.shell_request

def shell_request(
    msg_type, timeout:int=10, msg_id:NoneType=None, subshell_id:NoneType=None, metadata:NoneType=None,
    buffers:NoneType=None, **content
):

A named shell request, content as keyword arguments; returns an awaitable of its reply

Any non-execute shell request (kernel_info_request, complete_request, …) files with the same replies dict reply() uses - that is what shell_request does, with buffers for binary payloads and subshell_id/metadata for the header extras. control_request is the control-channel form; replies on both channels resolve by parent msg_id, so concurrent requests are fine on either. Comms are the fire-and-forget exception: comm_open and comm_msg (from jupywire.route) build and send without filing, returning the msg_id.

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

The typed verbs from jupywire.route (complete, inspect, check, history, comm_msg) ride the same request seam, with the same unwrapped return shapes as jupyasyncclient. They shadow jupyter_client’s senders of the same names, which returned a bare msg_id; the awaited answer is what callers actually want, and the raw form remains via request:

matches, start = await kc.complete('pri')
assert 'print' in matches
test_eq(await kc.check('for i in range(3):'), ('incomplete', '    '))
assert 'print' in await kc.inspect('print')

Read the input_request from the attached JmsgQueues, 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.reply("x = input('name? ')", timeout=10, allow_stdin=True)
msg = await qs.get('stdin', 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.comm_open('t', 'c1', 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.comm_open('tm', 'cm1', 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)
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.reply('import time; time.sleep(30)', 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, or display name when `argv` is given
    argv:NoneType=None, # Kernel argv; bypasses kernelspec discovery when provided
    manager_cls:MetaHasTraits=ConKernelManager, # Manager class, e.g. a subclass customizing launch
    **kwargs
):

Start a named or direct-argv kernel, yielding (km, kc), and always shut it down

A kernel normally reaches KernelManager through an installed kernelspec. Embedders and tests often already have the command they want to run, however, and should not need to install discovery metadata first. Pass argv to supply that command through an in-memory KernelSpec; KernelManager still owns connection-file creation, placeholder substitution, process launch, and shutdown. Here ipymini is launched directly, independently of the machine’s kernelspec registry:

async with run_kernel('ipymini', ['ipymini', '-f', '{connection_file}']) as (_, direct):
    test_eq((await direct.exec_outs('6*7'))[0]['data']['text/plain'], '42')
    await direct.exec_ok('import ipyfuncs')
    direct.xpush(sidecar_value=42)
    test_eq(await direct.get_vars(vs=['sidecar_value']), {'sidecar_value': 42})

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

---

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

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