How can I do a basic "hello world" using the jupyter_client API?

Kernels
1

I have tried to get the jupyter_client API working, and I cannot figure it out at all.
For example. Here’s me trying to make a basic AsyncKernelManager work:

import asyncio
from jupyter_client.manager import AsyncKernelManager
from queue import Empty

km = AsyncKernelManager(kernel_name="python3")

async def go():
    await km.start_kernel()
    client = km.client()
    msg_id = client.execute("time.sleep(1)")
    print(msg_id)
    while True:
        try:
            reply = await client.get_shell_msg(timeout=1)
            if reply:
                print(reply)
        except Empty:
            print("shell empty")

        try:
            iopub_msg = await client.get_iopub_msg(timeout=1)
            print(iopub_msg)
        except Empty:
            print("iopub empty")
            pass

# run async go
import asyncio
loop = asyncio.get_event_loop()
loop.run_until_complete(go())

This just loops forever, with no messages on either the shell or iopub channel.
Additionally, the non-async version (using either the standard KernelManager or the BlockingKernelManager) also do not work.

What am I doing wrong?

2

Got it working w/ ChatGPT:

import time
from jupyter_client import KernelManager

def main():
    # Create a new kernel manager
    km = KernelManager(kernel_name='python3')
    km.start_kernel()

    # Create a client to interact with the kernel
    kc = km.client()
    kc.start_channels()

    # Ensure the client is connected before executing code
    kc.wait_for_ready()

    # Execute the code
    code = 'print("Hello, World!")'
    msg_id = kc.execute(code)

    # Wait for the result and display it
    while True:
        try:
            msg = kc.get_iopub_msg(timeout=1)
            content = msg["content"]

            # When a message with the text stream comes and it's the result of our execution
            if msg["msg_type"] == "stream" and content["name"] == "stdout":
                print(content["text"])
                break
        except KeyboardInterrupt:
            print("Interrupted by user.")
            break
        except:
            # If no messages are available, we'll end up here, but we can just continue and try again.
            pass

    # Cleanup
    kc.stop_channels()
    km.shutdown_kernel()

if __name__ == '__main__':
    main()