diff --git a/CondaPkg.toml b/CondaPkg.toml
new file mode 100644
index 00000000..9108042d
--- /dev/null
+++ b/CondaPkg.toml
@@ -0,0 +1,3 @@
+[deps]
+jupyter_client = ""
+jupyter_kernel_test = ""
diff --git a/docs/src/library/internals.md b/docs/src/library/internals.md
index 94908414..bae28639 100644
--- a/docs/src/library/internals.md
+++ b/docs/src/library/internals.md
@@ -59,8 +59,6 @@ IJulia.waitloop
 
 ```@docs
 IJulia.IJuliaStdio
-IJulia.capture_stdout
-IJulia.capture_stderr
 IJulia.watch_stream
 ```
 
diff --git a/src/IJulia.jl b/src/IJulia.jl
index c61b7497..971b063b 100644
--- a/src/IJulia.jl
+++ b/src/IJulia.jl
@@ -33,13 +33,16 @@ The `IJulia` module is used in three ways
 module IJulia
 export notebook, jupyterlab, installkernel
 
+import MbedTLS
 using ZMQ, JSON, SoftGlobalScope
-import Base.invokelatest
+import Base: invokelatest, RefValue
 import Dates
 using Dates: now, format, UTC, ISODateTimeFormat
 import Random
+import Random: seed!
 using Base64: Base64EncodePipe
 import REPL
+import Logging
 
 # InteractiveUtils is not used inside IJulia, but loaded in src/kernel.jl
 # and this import makes it possible to load InteractiveUtils from the IJulia namespace
@@ -49,11 +52,161 @@ const depfile = joinpath(dirname(@__FILE__), "..", "deps", "deps.jl")
 isfile(depfile) || error("IJulia not properly installed. Please run Pkg.build(\"IJulia\")")
 include(depfile) # generated by Pkg.build("IJulia")
 
-#######################################################################
-# Debugging IJulia
+# use our own random seed for msg_id so that we
+# don't alter the user-visible random state (issue #336)
+const IJulia_RNG = seed!(Random.MersenneTwister(0))
+import UUIDs
+uuid4() = string(UUIDs.uuid4(IJulia_RNG))
+
+"""
+IPython message struct.
+"""
+mutable struct Msg
+    idents::Vector{String}
+    header::Dict
+    content::Dict
+    parent_header::Dict
+    metadata::Dict
+    function Msg(idents, header::Dict, content::Dict,
+                 parent_header=Dict{String,Any}(), metadata=Dict{String,Any}())
+        new(idents,header,content,parent_header,metadata)
+    end
+end
+
+@kwdef mutable struct Kernel
+    verbose::Bool = IJULIA_DEBUG
+    inited::Bool = false
+    current_module::Module = Main
+
+    # These fields are special and are mirrored to their corresponding global
+    # variables.
+    In::Dict{Int, String} = Dict{Int, String}()
+    Out::Dict{Int, Any} = Dict{Int, Any}()
+    ans::Any = nothing
+    n::Int = 0
+    capture_stdout::Bool = true
+    capture_stderr::Bool = !IJULIA_DEBUG
+
+    postexecute_hooks::Vector{Function} = Function[]
+    preexecute_hooks::Vector{Function} = Function[]
+    posterror_hooks::Vector{Function} = Function[]
+    shutdown::Function = exit
+
+    # the following constants need to be initialized in init().
+    publish::RefValue{Socket} = Ref{Socket}()
+    raw_input::RefValue{Socket} = Ref{Socket}()
+    requests::RefValue{Socket} = Ref{Socket}()
+    control::RefValue{Socket} = Ref{Socket}()
+    heartbeat::RefValue{Socket} = Ref{Socket}()
+    heartbeat_context::RefValue{Context} = Ref{Context}()
+    profile::Dict{String, Any} = Dict{String, Any}()
+    connection_file::Union{String, Nothing} = nothing
+    read_stdout::RefValue{Base.PipeEndpoint} = Ref{Base.PipeEndpoint}()
+    read_stderr::RefValue{Base.PipeEndpoint} = Ref{Base.PipeEndpoint}()
+    socket_locks = Dict{Socket, ReentrantLock}()
+    hmacstate::RefValue{MbedTLS.MD{true}} = Ref{MbedTLS.MD{true}}()
+
+    stop_event::Base.Event = Base.Event()
+    waitloop_task::RefValue{Task} = Ref{Task}()
+
+    requests_task::RefValue{Task} = Ref{Task}()
+    watch_stdout_task::RefValue{Task} = Ref{Task}()
+    watch_stderr_task::RefValue{Task} = Ref{Task}()
+    watch_stdout_timer::RefValue{Timer} = Ref{Timer}()
+    watch_stderr_timer::RefValue{Timer} = Ref{Timer}()
+
+    # name=>iobuffer for each stream ("stdout","stderr") so they can be sent in flush
+    bufs::Dict{String, IOBuffer} = Dict{String, IOBuffer}()
+    # max output per code cell is 512 kb by default
+    max_output_per_request::RefValue{Int} = Ref(1 << 19)
+
+    # Variable so that display can be done in the correct Msg context
+    execute_msg::Msg = Msg(["julia"], Dict("username"=>"jlkernel", "session"=>uuid4()), Dict())
+    # Variable tracking the number of bytes written in the current execution request
+    stdio_bytes::RefValue{Int} = Ref(0)
+    # Use an array to accumulate "payloads" for the execute_reply message
+    execute_payloads::Vector{Dict} = Dict[]
+
+    heartbeat_threadid::Vector{Int} = zeros(Int, 128) # sizeof(uv_thread_t) <= 8 on Linux, OSX, Win
+
+    # queue of objects to display at end of cell execution
+    displayqueue::Vector{Any} = Any[]
+end
+
+function Base.setproperty!(kernel::Kernel, name::Symbol, x)
+    # These fields need to be assigned explicitly to their global counterparts
+    if name ∈ (:ans, :n, :In, :Out, :inited)
+        setproperty!(IJulia, name, x)
+
+        if name ∈ (:ans, :In, :Out)
+            if hasproperty(kernel.current_module, name)
+                setproperty!(kernel.current_module, name, x)
+            end
+        end
+    end
+
+    setfield!(kernel, name, x)
+end
+
+function Base.wait(kernel::Kernel)
+    if isassigned(kernel.waitloop_task)
+        wait(kernel.waitloop_task[])
+    end
+end
+
+function Base.close(kernel::Kernel)
+    # Reset the IO streams first so that any later errors get printed
+    if kernel.capture_stdout
+        redirect_stdout(orig_stdout[])
+        close(kernel.watch_stdout_timer[])
+        close(kernel.read_stdout[])
+        wait(kernel.watch_stdout_task[])
+    end
+    if kernel.capture_stderr
+        redirect_stderr(orig_stderr[])
+        close(kernel.watch_stderr_timer[])
+        close(kernel.read_stderr[])
+        wait(kernel.watch_stderr_task[])
+    end
+    redirect_stdin(orig_stdin[])
+
+    # Reset the logger so that @log statements work and pop the InlineDisplay
+    Logging.global_logger(orig_logger[])
+    popdisplay()
+
+    # Close all sockets
+    close(kernel.requests[])
+    close(kernel.control[])
+    close(kernel.publish[])
+    close(kernel.raw_input[])
+
+    # Wait for the heartbeat thread
+    stop_heartbeat(kernel)
+
+    # The waitloop should now be ready to exit
+    kernel.inited = false
+    notify(kernel.stop_event)
+    wait(kernel)
+
+    # Reset global variables
+    IJulia.ans = nothing
+    IJulia.n = 0
+    IJulia._default_kernel = nothing
+end
+
+function Kernel(f::Function, profile; kwargs...)
+    kernel = Kernel(; kwargs...)
+    init([], kernel, profile)
+
+    try
+        f(kernel)
+    finally
+        close(kernel)
+    end
+end
+
+_default_kernel::Union{Kernel, Nothing} = nothing
 
-# in the Jupyter front-end, enable verbose output via IJulia.set_verbose()
-verbose = IJULIA_DEBUG
 """
     set_verbose(v=true)
 
@@ -63,8 +216,8 @@ This consists of log messages printed to the terminal window where
 `jupyter` was launched, displaying information about every message sent
 or received by the kernel.   Used for debugging IJulia.
 """
-function set_verbose(v::Bool=true)
-    global verbose = v
+function set_verbose(v::Bool=true, kernel=_default_kernel)
+    kernel.verbose = v
 end
 
 """
@@ -73,25 +226,9 @@ kernel is running, i.e. in a running IJulia notebook.  To test
 whether you are in an IJulia notebook, therefore, you can check
 `isdefined(Main, :IJulia) && IJulia.inited`.
 """
-inited = false
-
-const _capture_docstring = """
-The IJulia kernel captures all [stdout and stderr](https://en.wikipedia.org/wiki/Standard_streams)
-output and redirects it to the notebook.   When debugging IJulia problems,
-however, it can be more convenient to *not* capture stdout and stderr output
-(since the notebook may not be functioning). This can be done by editing
-`IJulia.jl` to set `capture_stderr` and/or `capture_stdout` to `false`.
-"""
+inited::Bool = false
 
-@doc _capture_docstring
-const capture_stdout = true
-
-# set this to false for debugging, to disable stderr redirection
-@doc _capture_docstring
-const capture_stderr = !IJULIA_DEBUG
-
-set_current_module(m::Module) = current_module[] = m
-const current_module = Ref{Module}(Main)
+set_current_module(m::Module; kernel=_default_kernel) = kernel.current_module = m
 
 #######################################################################
 include("jupyter.jl")
@@ -105,8 +242,8 @@ somewhat analogous to the `%load` magics in IPython. If the optional argument
 `replace` is `true`, then `s` replaces the *current* cell rather than creating
 a new cell.
 """
-function load_string(s::AbstractString, replace::Bool=false)
-    push!(execute_payloads, Dict(
+function load_string(s::AbstractString, replace::Bool=false, kernel=_default_kernel)
+    push!(kernel.execute_payloads, Dict(
         "source"=>"set_next_input",
         "text"=>s,
         "replace"=>replace
@@ -122,8 +259,8 @@ IJulia notebook, analogous to the `%load` magics in IPython.
 If the optional argument `replace` is `true`, then the file contents
 replace the *current* cell rather than creating a new cell.
 """
-load(filename::AbstractString, replace::Bool=false) =
-    load_string(read(filename, String), replace)
+load(filename::AbstractString, replace::Bool=false, kernel=_default_kernel) =
+    load_string(read(filename, String), replace, kernel)
 
 #######################################################################
 # History: global In/Out and other exported history variables
@@ -132,33 +269,33 @@ load(filename::AbstractString, replace::Bool=false) =
 returns the string for input cell `n` of the notebook (as it was
 when it was *last evaluated*).
 """
-const In = Dict{Int,String}()
+In::Dict{String, Any} = Dict{String, Any}()
 """
 `Out` is a global dictionary of output values, where `Out[n]`
 returns the output from the last evaluation of cell `n` in the
 notebook.
 """
-const Out = Dict{Int,Any}()
+Out::Dict{String, Any} = Dict{String, Any}()
 """
 `ans` is a global variable giving the value returned by the last
 notebook cell evaluated.
 """
-ans = nothing
+ans::Any = nothing
 
 # execution counter
 """
 `IJulia.n` is the (integer) index of the last-evaluated notebook cell.
 """
-n = 0
+n::Int = 0
 
 #######################################################################
 # methods to clear history or any subset thereof
 
-function clear_history(indices)
+function clear_history(indices; kernel=_default_kernel)
     for n in indices
-        delete!(In, n)
-        if haskey(Out, n)
-            delete!(Out, n)
+        delete!(kernel.In, n)
+        if haskey(kernel.Out, n)
+            delete!(kernel.Out, n)
         end
     end
 end
@@ -167,10 +304,10 @@ end
 clear_history(r::AbstractRange{<:Integer}) =
     invoke(clear_history, Tuple{Any}, intersect(r, 1:n))
 
-function clear_history()
-    empty!(In)
-    empty!(Out)
-    global ans = nothing
+function clear_history(; kernel=_default_kernel)
+    empty!(kernel.In)
+    empty!(kernel.Out)
+    kernel.ans = nothing
 end
 
 """
@@ -189,18 +326,18 @@ clear_history
 
 #######################################################################
 # methods to print history or any subset thereof
-function history(io::IO, indices::AbstractVector{<:Integer})
-    for n in intersect(indices, 1:IJulia.n)
-      if haskey(In, n)
-        print(io, In[n])
+function history(io::IO, indices::AbstractVector{<:Integer}; kernel=_default_kernel)
+    for n in intersect(indices, 1:kernel.n)
+      if haskey(kernel.In, n)
+        println(io, kernel.In[n])
       end
     end
 end
 
-history(io::IO, x::Union{Integer,AbstractVector{<:Integer}}...) = history(io, vcat(x...))
-history(x...) = history(stdout, x...)
-history(io::IO, x...) = throw(MethodError(history, (io, x...,)))
-history() = history(1:n)
+history(io::IO, x::Union{Integer,AbstractVector{<:Integer}}...; kernel=_default_kernel) = history(io, vcat(x...); kernel)
+history(x...; kernel=_default_kernel) = history(stdout, x...; kernel)
+history(io::IO, x...; kernel=_default_kernel) = throw(MethodError(history, (io, x...,)))
+history(; kernel=_default_kernel) = history(1:kernel.n; kernel)
 """
     history([io], [indices...])
 
@@ -221,57 +358,54 @@ history
 # executing an input cell, e.g. to "close" the current plot in Pylab.
 # Modules should only use these if isdefined(Main, IJulia) is true.
 
-const postexecute_hooks = Function[]
 """
     push_postexecute_hook(f::Function)
 
 Push a function `f()` onto the end of a list of functions to
 execute after executing any notebook cell.
 """
-push_postexecute_hook(f::Function) = push!(postexecute_hooks, f)
+push_postexecute_hook(f::Function; kernel=_default_kernel) = push!(kernel.postexecute_hooks, f)
 """
     pop_postexecute_hook(f::Function)
 
 Remove a function `f()` from the list of functions to
 execute after executing any notebook cell.
 """
-pop_postexecute_hook(f::Function) =
-    splice!(postexecute_hooks, findlast(isequal(f), postexecute_hooks))
+pop_postexecute_hook(f::Function; kernel=_default_kernel) =
+    splice!(kernel.postexecute_hooks, findlast(isequal(f), kernel.postexecute_hooks))
 
-const preexecute_hooks = Function[]
 """
     push_preexecute_hook(f::Function)
 
 Push a function `f()` onto the end of a list of functions to
 execute before executing any notebook cell.
 """
-push_preexecute_hook(f::Function) = push!(preexecute_hooks, f)
+push_preexecute_hook(f::Function; kernel=_default_kernel) = push!(kernel.preexecute_hooks, f)
 """
     pop_preexecute_hook(f::Function)
 
 Remove a function `f()` from the list of functions to
 execute before executing any notebook cell.
 """
-pop_preexecute_hook(f::Function) =
-    splice!(preexecute_hooks, findlast(isequal(f), preexecute_hooks))
+pop_preexecute_hook(f::Function; kernel=_default_kernel) =
+    splice!(kernel.preexecute_hooks, findlast(isequal(f), kernel.preexecute_hooks))
 
 # similar, but called after an error (e.g. to reset plotting state)
-const posterror_hooks = Function[]
 """
     pop_posterror_hook(f::Function)
 
 Remove a function `f()` from the list of functions to
 execute after an error occurs when a notebook cell is evaluated.
 """
-push_posterror_hook(f::Function) = push!(posterror_hooks, f)
+push_posterror_hook(f::Function; kernel=_default_kernel) = push!(kernel.posterror_hooks, f)
 """
     pop_posterror_hook(f::Function)
 
 Remove a function `f()` from the list of functions to
 execute after an error occurs when a notebook cell is evaluated.
 """
-pop_posterror_hook(f::Function) =
-    splice!(posterror_hooks, findlast(isequal(f), posterror_hooks))
+pop_posterror_hook(f::Function; kernel=_default_kernel) =
+    splice!(kernel.posterror_hooks, findlast(isequal(f), kernel.posterror_hooks))
 
 #######################################################################
 
@@ -285,13 +419,14 @@ Call `clear_output()` to clear visible output from the current notebook
 cell.  Using `wait=true` clears the output only when new output is
 available, which reduces flickering and is useful for simple animations.
 """
-function clear_output(wait=false)
+function clear_output(wait=false, kernel=_default_kernel)
     # flush pending stdio
     flush_all()
-    empty!(displayqueue) # discard pending display requests
-    send_ipython(publish[], msg_pub(execute_msg::Msg, "clear_output",
-                                    Dict("wait" => wait)))
-    stdio_bytes[] = 0 # reset output throttling
+    empty!(kernel.displayqueue) # discard pending display requests
+    send_ipython(kernel.publish[], kernel, msg_pub(kernel.execute_msg::Msg, "clear_output",
+                                           Dict("wait" => wait)))
+    kernel.stdio_bytes[] = 0 # reset output throttling
+    return nothing
 end
 
 
@@ -302,8 +437,8 @@ Sets the maximum number of bytes, `max_output`, that can be written to stdout an
 stderr before getting truncated. A large value here allows a lot of output to be
 displayed in the notebook, potentially bogging down the browser.
 """
-function set_max_stdio(max_output::Integer)
-    max_output_per_request[] = max_output
+function set_max_stdio(max_output::Integer; kernel)
+    kernel.max_output_per_request[] = max_output
 end
 
 
diff --git a/src/comm_manager.jl b/src/comm_manager.jl
index 0d37f862..c84f67bb 100644
--- a/src/comm_manager.jl
+++ b/src/comm_manager.jl
@@ -44,7 +44,7 @@ end
 
 comm_target(comm :: Comm{target}) where {target} = target
 
-function comm_info_request(sock, msg)
+function comm_info_request(sock, kernel, msg)
     reply = if haskey(msg.content, "target_name")
         t = Symbol(msg.content["target_name"])
         filter(kv -> comm_target(kv.second) == t, comms)
@@ -59,7 +59,7 @@ function comm_info_request(sock, msg)
     end
     content = Dict(:comms => _comms)
 
-    send_ipython(sock,
+    send_ipython(sock, kernel,
                  msg_reply(msg, "comm_info_reply", content))
 end
 
@@ -75,18 +75,18 @@ function msg_comm(comm::Comm, m::IJulia.Msg, msg_type,
     return msg_pub(m, msg_type, content, metadata)
 end
 
-function send_comm(comm::Comm, data::Dict,
+function send_comm(comm::Comm, kernel, data::Dict,
                    metadata::Dict = Dict(); kwargs...)
     msg = msg_comm(comm, IJulia.execute_msg, "comm_msg", data,
                    metadata; kwargs...)
-    send_ipython(IJulia.publish[], msg)
+    send_ipython(kernel.publish[], kernel, msg)
 end
 
-function close_comm(comm::Comm, data::Dict = Dict(),
+function close_comm(comm::Comm, kernel, data::Dict = Dict(),
                     metadata::Dict = Dict(); kwargs...)
     msg = msg_comm(comm, IJulia.execute_msg, "comm_close", data,
                    metadata; kwargs...)
-    send_ipython(IJulia.publish[], msg)
+    send_ipython(kernel.publish[], kernel, msg)
 end
 
 function register_comm(comm::Comm, data)
@@ -97,7 +97,7 @@ end
 
 # handlers for incoming comm_* messages
 
-function comm_open(sock, msg)
+function comm_open(sock, msg, kernel)
     if haskey(msg.content, "comm_id")
         comm_id = msg.content["comm_id"]
         if haskey(msg.content, "target_name")
@@ -111,7 +111,7 @@ function comm_open(sock, msg)
         else
             # Tear down comm to maintain consistency
             # if a target_name is not present
-            send_ipython(IJulia.publish[],
+            send_ipython(kernel.publish[], kernel,
                          msg_comm(Comm(:notarget, comm_id),
                                   msg, "comm_close"))
         end
diff --git a/src/display.jl b/src/display.jl
index 3d8fb9e2..6c90c95a 100644
--- a/src/display.jl
+++ b/src/display.jl
@@ -137,13 +137,10 @@ function display_dict(x)
 
 end
 
-# queue of objects to display at end of cell execution
-const displayqueue = Any[]
-
 # remove x from the display queue
-function undisplay(x)
-    i = findfirst(isequal(x), displayqueue)
-    i !== nothing && splice!(displayqueue, i)
+function undisplay(x, kernel)
+    i = findfirst(isequal(x), kernel.displayqueue)
+    i !== nothing && splice!(kernel.displayqueue, i)
     return x
 end
 
diff --git a/src/eventloop.jl b/src/eventloop.jl
index 0fb8414a..a4ae6a47 100644
--- a/src/eventloop.jl
+++ b/src/eventloop.jl
@@ -1,17 +1,28 @@
 """
-    eventloop(socket)
+    eventloop(socket, kernel)
 
 Generic event loop for one of the [kernel
 sockets](https://jupyter-client.readthedocs.io/en/latest/messaging.html#introduction).
 """
-function eventloop(socket)
+function eventloop(socket, kernel)
     task_local_storage(:IJulia_task, "write task")
     try
         while true
-            msg = recv_ipython(socket)
+            local msg
             try
-                send_status("busy", msg)
-                invokelatest(get(handlers, msg.header["msg_type"], unknown_request), socket, msg)
+                msg = recv_ipython(socket, kernel)
+            catch e
+                if isa(e, EOFError)
+                    # The socket was closed
+                    return
+                else
+                    rethrow()
+                end
+            end
+
+            try
+                send_status("busy", kernel, msg)
+                invokelatest(get(handlers, msg.header["msg_type"], unknown_request), socket, kernel, msg)
             catch e
                 # Try to keep going if we get an exception, but
                 # send the exception traceback to the front-ends.
@@ -20,45 +31,50 @@ function eventloop(socket)
                 if !isa(e, InterruptException)
                     content = error_content(e, msg="KERNEL EXCEPTION")
                     map(s -> println(orig_stderr[], s), content["traceback"])
-                    send_ipython(publish[], msg_pub(execute_msg, "error", content))
+                    send_ipython(kernel.publish[], kernel, msg_pub(kernel.execute_msg, "error", content))
                 end
             finally
                 flush_all()
-                send_status("idle", msg)
+                send_status("idle", kernel, msg)
             end
         end
     catch e
         # the Jupyter manager may send us a SIGINT if the user
         # chooses to interrupt the kernel; don't crash on this
         if isa(e, InterruptException)
-            eventloop(socket)
+            eventloop(socket, kernel)
+        elseif isa(e, ZMQ.StateError)
+            # This is almost certainly because of a closed socket
+            return
         else
             rethrow()
         end
     end
 end
 
-const requests_task = Ref{Task}()
-
 """
-    waitloop()
+    waitloop(kernel)
 
 Main loop of a kernel. Runs the event loops for the control and shell sockets
 (note: in IJulia the shell socket is called `requests`).
 """
-function waitloop()
-    @async eventloop(control[])
-    requests_task[] = @async eventloop(requests[])
-    while true
+function waitloop(kernel)
+    control_task = @async eventloop(kernel.control[], kernel)
+    kernel.requests_task[] = @async eventloop(kernel.requests[], kernel)
+
+    while kernel.inited
         try
-            wait()
+            wait(kernel.stop_event)
         catch e
             # send interrupts (user SIGINT) to the code-execution task
             if isa(e, InterruptException)
-                @async Base.throwto(requests_task[], e)
+                @async Base.throwto(kernel.requests_task[], e)
             else
                 rethrow()
             end
+        finally
+            wait(control_task)
+            wait(kernel.requests_task[])
         end
     end
 end
diff --git a/src/execute_request.jl b/src/execute_request.jl
index 3e49683b..9b8cd5dd 100644
--- a/src/execute_request.jl
+++ b/src/execute_request.jl
@@ -13,45 +13,36 @@ else # Pkg.jl#3777
         Pkg.REPLMode.do_cmds(cmd, stdout)
 end
 
-# global variable so that display can be done in the correct Msg context
-execute_msg = Msg(["julia"], Dict("username"=>"jlkernel", "session"=>uuid4()), Dict())
-# global variable tracking the number of bytes written in the current execution
-# request
-const stdio_bytes = Ref(0)
-
 import REPL: helpmode
 
-# use a global array to accumulate "payloads" for the execute_reply message
-const execute_payloads = Dict[]
 
 """
-    execute_request(socket, msg)
+    execute_request(socket, kernel, msg)
 
 Handle a [execute
 request](https://jupyter-client.readthedocs.io/en/latest/messaging.html#execute).
 This will execute Julia code, along with Pkg and shell commands.
 """
-function execute_request(socket, msg)
+function execute_request(socket, kernel, msg)
     code = msg.content["code"]
     @vprintln("EXECUTING ", code)
-    global execute_msg = msg
-    global n, In, Out, ans
-    stdio_bytes[] = 0
+    kernel.execute_msg = msg
+    kernel.stdio_bytes[] = 0
     silent = msg.content["silent"]
     store_history = get(msg.content, "store_history", !silent)
-    empty!(execute_payloads)
+    empty!(kernel.execute_payloads)
 
     if !silent
-        n += 1
-        send_ipython(publish[],
+        kernel.n += 1
+        send_ipython(kernel.publish[], kernel,
                      msg_pub(msg, "execute_input",
-                             Dict("execution_count" => n,
+                             Dict("execution_count" => kernel.n,
                                           "code" => code)))
     end
 
     silent = silent || REPL.ends_with_semicolon(code)
     if store_history
-        In[n] = code
+        kernel.In[kernel.n] = code
     end
 
     # "; ..." cells are interpreted as shell commands for run
@@ -69,32 +60,32 @@ function execute_request(socket, msg)
     hcode = replace(code, r"^\s*\?" => "")
 
     try
-        for hook in preexecute_hooks
+        for hook in kernel.preexecute_hooks
             invokelatest(hook)
         end
 
 
-        ans = result = if hcode != code # help request
+        kernel.ans = result = if hcode != code # help request
             Core.eval(Main, helpmode(hcode))
         else
             #run the code!
             occursin(magics_regex, code) && match(magics_regex, code).offset == 1 ? magics_help(code) :
-                SOFTSCOPE[] ? softscope_include_string(current_module[], code, "In[$n]") :
-                include_string(current_module[], code, "In[$n]")
+                SOFTSCOPE[] ? softscope_include_string(kernel.current_module, code, "In[$(kernel.n)]") :
+                include_string(kernel.current_module, code, "In[$(kernel.n)]")
         end
 
         if silent
             result = nothing
-        elseif (result !== nothing) && (result !== Out)
+        elseif (result !== nothing) && (result !== kernel.Out)
             if store_history
-                Out[n] = result
+                kernel.Out[kernel.n] = result
             end
         end
 
         user_expressions = Dict()
         for (v,ex) in msg.content["user_expressions"]
             try
-                value = include_string(current_module[], ex)
+                value = include_string(kernel.current_module, ex)
                 # Like the IPython reference implementation, we return
                 # something that looks like a `display_data` but also has a
                 # `status` field:
@@ -111,48 +102,55 @@ function execute_request(socket, msg)
             end
         end
 
-        for hook in postexecute_hooks
+        for hook in kernel.postexecute_hooks
             invokelatest(hook)
         end
 
         # flush pending stdio
         flush_all()
+        yield()
+        if haskey(kernel.bufs, "stdout")
+            send_stdout(kernel)
+        end
+        if haskey(kernel.bufs, "stderr")
+            send_stderr(kernel)
+        end
 
-        undisplay(result) # dequeue if needed, since we display result in pyout
-        invokelatest(display) # flush pending display requests
+        undisplay(result, kernel) # dequeue if needed, since we display result in pyout
+        @invokelatest display(kernel) # flush pending display requests
 
         if result !== nothing
             result_metadata = invokelatest(metadata, result)
             result_data = invokelatest(display_dict, result)
-            send_ipython(publish[],
+            send_ipython(kernel.publish[], kernel,
                          msg_pub(msg, "execute_result",
-                                 Dict("execution_count" => n,
-                                              "metadata" => result_metadata,
-                                              "data" => result_data)))
+                                 Dict("execution_count" => kernel.n,
+                                      "metadata" => result_metadata,
+                                      "data" => result_data)))
 
         end
-        send_ipython(requests[],
+        send_ipython(kernel.requests[], kernel,
                      msg_reply(msg, "execute_reply",
                                Dict("status" => "ok",
-                                            "payload" => execute_payloads,
-                                            "execution_count" => n,
-                                            "user_expressions" => user_expressions)))
-        empty!(execute_payloads)
+                                    "payload" => kernel.execute_payloads,
+                                    "execution_count" => kernel.n,
+                                    "user_expressions" => user_expressions)))
+        empty!(kernel.execute_payloads)
     catch e
         bt = catch_backtrace()
         try
             # flush pending stdio
             flush_all()
-            for hook in posterror_hooks
+            for hook in kernel.posterror_hooks
                 invokelatest(hook)
             end
         catch
         end
-        empty!(displayqueue) # discard pending display requests on an error
+        empty!(kernel.displayqueue) # discard pending display requests on an error
         content = error_content(e,bt)
-        send_ipython(publish[], msg_pub(msg, "error", content))
+        send_ipython(kernel.publish[], kernel, msg_pub(msg, "error", content))
         content["status"] = "error"
         content["execution_count"] = n
-        send_ipython(requests[], msg_reply(msg, "execute_reply", content))
+        send_ipython(kernel.requests[], kernel, msg_reply(msg, "execute_reply", content))
     end
 end
diff --git a/src/handlers.jl b/src/handlers.jl
index ab00fd3f..444b5c08 100644
--- a/src/handlers.jl
+++ b/src/handlers.jl
@@ -109,17 +109,17 @@ function complete_types(comps)
 end
 
 """
-    complete_request(socket, msg)
+    complete_request(socket, kernel, msg)
 
 Handle a [completion
 request](https://jupyter-client.readthedocs.io/en/latest/messaging.html#completion).
 """
-function complete_request(socket, msg)
+function complete_request(socket, kernel, msg)
     code = msg.content["code"]
     cursor_chr = msg.content["cursor_pos"]
     cursorpos = chr2ind(msg, code, cursor_chr)
     if all(isspace, code[1:cursorpos])
-        send_ipython(requests[], msg_reply(msg, "complete_reply",
+        send_ipython(kernel.requests[], kernel, msg_reply(msg, "complete_reply",
                                  Dict("status" => "ok",
                                               "metadata" => Dict(),
                                               "matches" => String[],
@@ -129,7 +129,7 @@ function complete_request(socket, msg)
     end
 
     codestart = find_parsestart(code, cursorpos)
-    comps_, positions, should_complete = REPLCompletions.completions(code[codestart:end], cursorpos-codestart+1, current_module[])
+    comps_, positions, should_complete = REPLCompletions.completions(code[codestart:end], cursorpos-codestart+1, kernel.current_module)
     comps = unique!(REPLCompletions.completion_text.(comps_)) # julia#26930
     # positions = positions .+ (codestart - 1) on Julia 0.7
     positions = (first(positions) + codestart - 1):(last(positions) + codestart - 1)
@@ -151,7 +151,7 @@ function complete_request(socket, msg)
             pushfirst!(comps, code[positions])
         end
     end
-    send_ipython(requests[], msg_reply(msg, "complete_reply",
+    send_ipython(kernel.requests[], kernel, msg_reply(msg, "complete_reply",
                                      Dict("status" => "ok",
                                                   "matches" => comps,
                                                   "metadata" => metadata,
@@ -160,17 +160,17 @@ function complete_request(socket, msg)
 end
 
 """
-    kernel_info_request(socket, msg)
+    kernel_info_request(socket, kernel, msg)
 
 Handle a [kernel info
 request](https://jupyter-client.readthedocs.io/en/latest/messaging.html#kernel-info).
 """
-function kernel_info_request(socket, msg)
-    send_ipython(socket,
+function kernel_info_request(socket, kernel, msg)
+    send_ipython(socket, kernel,
                  msg_reply(msg, "kernel_info_reply",
                            Dict("protocol_version" => "5.4",
                                         "implementation" => "ijulia",
-                                        "implementation_version" => pkgversion(@__MODULE__),
+                                        "implementation_version" => string(pkgversion(@__MODULE__)),
                                         "language_info" =>
                                         Dict("name" => "julia",
                                              "version" =>
@@ -192,13 +192,13 @@ function kernel_info_request(socket, msg)
 end
 
 """
-    connect_request(socket, msg)
+    connect_request(socket, kernel, msg)
 
 Handle a [connect
 request](https://jupyter-client.readthedocs.io/en/latest/messaging.html#connect).
 """
-function connect_request(socket, msg)
-    send_ipython(requests[],
+function connect_request(socket, kernel, msg)
+    send_ipython(kernel.requests[], kernel,
                  msg_reply(msg, "connect_reply",
                            Dict("shell_port" => profile["shell_port"],
                                 "iopub_port" => profile["iopub_port"],
@@ -207,20 +207,22 @@ function connect_request(socket, msg)
 end
 
 """
-    shutdown_request(socket, msg)
+    shutdown_request(socket, kernel, msg)
 
 Handle a [shutdown
 request](https://jupyter-client.readthedocs.io/en/latest/messaging.html#kernel-shutdown). After
 sending the reply this will exit the process.
 """
-function shutdown_request(socket, msg)
-    # stop heartbeat thread by closing the context
-    close(heartbeat_context[])
+function shutdown_request(socket, kernel, msg)
+    # stop heartbeat thread
+    stop_heartbeat(kernel)
 
-    send_ipython(requests[], msg_reply(msg, "shutdown_reply",
-                                       msg.content))
+    # In protocol 5.4 the shutdown reply moved to the control socket
+    shutdown_socket = VersionNumber(msg) >= v"5.4" ? kernel.control[] : kernel.requests[]
+    send_ipython(shutdown_socket, kernel,
+                 msg_reply(msg, "shutdown_reply", msg.content))
     sleep(0.1) # short delay (like in ipykernel), to hopefully ensure shutdown_reply is sent
-    exit()
+    kernel.shutdown()
 end
 
 docdict(s::AbstractString) = display_dict(Core.eval(Main, helpmode(devnull, s)))
@@ -261,12 +263,12 @@ function get_token(code, pos)
 end
 
 """
-    inspect_request(socket, msg)
+    inspect_request(socket, kernel, msg)
 
 Handle a [introspection
 request](https://jupyter-client.readthedocs.io/en/latest/messaging.html#introspection).
 """
-function inspect_request(socket, msg)
+function inspect_request(socket, kernel, msg)
     try
         code = msg.content["code"]
         s = get_token(code, chr2ind(msg, code, msg.content["cursor_pos"]))
@@ -276,56 +278,57 @@ function inspect_request(socket, msg)
             d = docdict(s)
             content = Dict("status" => "ok",
                            "found" => !isempty(d),
-                           "data" => d)
+                           "data" => d,
+                           "metadata" => Dict())
         end
-        send_ipython(requests[], msg_reply(msg, "inspect_reply", content))
+        send_ipython(kernel.requests[], kernel, msg_reply(msg, "inspect_reply", content))
     catch e
         content = error_content(e, backtrace_top=:inspect_request);
         content["status"] = "error"
-        send_ipython(requests[],
+        send_ipython(kernel.requests[], kernel,
                      msg_reply(msg, "inspect_reply", content))
     end
 end
 
 """
-    history_request(socket, msg)
+    history_request(socket, kernel, msg)
 
 Handle a [history
 request](https://jupyter-client.readthedocs.io/en/latest/messaging.html#history). This
 is currently only a dummy implementation that doesn't actually do anything.
 """
-function history_request(socket, msg)
+function history_request(socket, kernel, msg)
     # we will just send back empty history for now, pending clarification
     # as requested in ipython/ipython#3806
-    send_ipython(requests[],
+    send_ipython(kernel.requests[], kernel,
                  msg_reply(msg, "history_reply",
                            Dict("history" => [])))
 end
 
 """
-    is_complete_request(socket, msg)
+    is_complete_request(socket, kernel, msg)
 
 Handle a [completeness
 request](https://jupyter-client.readthedocs.io/en/latest/messaging.html#code-completeness).
 """
-function is_complete_request(socket, msg)
+function is_complete_request(socket, kernel, msg)
     ex = Meta.parse(msg.content["code"], raise=false)
     status = Meta.isexpr(ex, :incomplete) ? "incomplete" : Meta.isexpr(ex, :error) ? "invalid" : "complete"
-    send_ipython(requests[],
+    send_ipython(kernel.requests[], kernel,
                  msg_reply(msg, "is_complete_reply",
                            Dict("status"=>status, "indent"=>"")))
 end
 
 """
-    interrupt_request(socket, msg)
+    interrupt_request(socket, kernel, msg)
 
 Handle a [interrupt
 request](https://jupyter-client.readthedocs.io/en/latest/messaging.html#kernel-interrupt). This
 will throw an `InterruptException` to the currently executing request handler.
 """
-function interrupt_request(socket, msg)
-    @async Base.throwto(requests_task[], InterruptException())
-    send_ipython(socket, msg_reply(msg, "interrupt_reply", Dict()))
+function interrupt_request(socket, kernel, msg)
+    @async Base.throwto(kernel.requests_task[], InterruptException())
+    send_ipython(socket, kernel, msg_reply(msg, "interrupt_reply", Dict()))
 end
 
 function unknown_request(socket, msg)
diff --git a/src/heartbeat.jl b/src/heartbeat.jl
index 28fcaa12..7ba12539 100644
--- a/src/heartbeat.jl
+++ b/src/heartbeat.jl
@@ -4,8 +4,6 @@
 # call in libzmq, which simply blocks forever, so the usual lack of
 # thread safety in Julia should not be an issue here.
 
-const threadid = zeros(Int, 128) # sizeof(uv_thread_t) <= 8 on Linux, OSX, Win
-
 # entry point for new thread
 function heartbeat_thread(heartbeat::Ptr{Cvoid})
     @static if VERSION ≥ v"1.9.0-DEV.1588" # julia#46609
@@ -24,8 +22,29 @@ function heartbeat_thread(heartbeat::Ptr{Cvoid})
     return ret
 end
 
-function start_heartbeat(heartbeat)
+function start_heartbeat(heartbeat, kernel)
+    heartbeat.linger = 0
     heartbeat_c = @cfunction(heartbeat_thread, Cint, (Ptr{Cvoid},))
     ccall(:uv_thread_create, Cint, (Ptr{Int}, Ptr{Cvoid}, Ptr{Cvoid}),
-          threadid, heartbeat_c, heartbeat)
+          kernel.heartbeat_threadid, heartbeat_c, heartbeat)
+end
+
+function stop_heartbeat(kernel)
+    if !isopen(kernel.heartbeat_context[])
+        # Do nothing if it has already been stopped (which can happen in the tests)
+        return
+    end
+
+    # First we call zmq_ctx_shutdown() to ensure that the zmq_proxy() call
+    # returns. We don't call ZMQ.close(::Context) directly because that
+    # currently isn't threadsafe:
+    # https://github.com/JuliaInterop/ZMQ.jl/issues/256
+    ZMQ.lib.zmq_ctx_shutdown(kernel.heartbeat_context[])
+    @ccall uv_thread_join(kernel.heartbeat_threadid::Ptr{Int})::Cint
+
+    # Now that the heartbeat thread has joined and its guaranteed to no longer
+    # be working on the heartbeat socket, we can safely close it and then the
+    # context.
+    close(kernel.heartbeat[])
+    close(kernel.heartbeat_context[])
 end
diff --git a/src/hmac.jl b/src/hmac.jl
index c6fe856c..bc7b09cf 100644
--- a/src/hmac.jl
+++ b/src/hmac.jl
@@ -1,16 +1,13 @@
-using MbedTLS
-const hmacstate = Ref{MbedTLS.MD{true}}()
-
-function hmac(s1,s2,s3,s4)
-    if !isdefined(hmacstate, :x)
+function hmac(s1,s2,s3,s4, kernel)
+    if !isassigned(kernel.hmacstate)
         return ""
     else
-        MbedTLS.reset!(hmacstate[])
+        MbedTLS.reset!(kernel.hmacstate[])
         for s in (s1, s2, s3, s4)
-            write(hmacstate[], s)
+            write(kernel.hmacstate[], s)
         end
         # Take the digest (returned as a byte array) and convert it to hex string representation
-        digest = MbedTLS.finish!(hmacstate[])
+        digest = MbedTLS.finish!(kernel.hmacstate[])
         hexdigest = Vector{UInt8}(undef, length(digest)*2)
         for i = 1:length(digest)
             b = digest[i]
diff --git a/src/init.jl b/src/init.jl
index 4600d671..c021edef 100644
--- a/src/init.jl
+++ b/src/init.jl
@@ -1,36 +1,21 @@
 import Random: seed!
-import Logging: ConsoleLogger
-
-# use our own random seed for msg_id so that we
-# don't alter the user-visible random state (issue #336)
-const IJulia_RNG = seed!(Random.MersenneTwister(0))
-import UUIDs
-uuid4() = string(UUIDs.uuid4(IJulia_RNG))
+import Logging
+import Logging: AbstractLogger, ConsoleLogger
 
 const orig_stdin  = Ref{IO}()
 const orig_stdout = Ref{IO}()
 const orig_stderr = Ref{IO}()
+const orig_logger = Ref{AbstractLogger}()
 const SOFTSCOPE = Ref{Bool}()
 function __init__()
     seed!(IJulia_RNG)
     orig_stdin[]  = stdin
     orig_stdout[] = stdout
     orig_stderr[] = stderr
+    orig_logger[] = Logging.global_logger()
     SOFTSCOPE[] = lowercase(get(ENV, "IJULIA_SOFTSCOPE", "yes")) in ("yes", "true")
 end
 
-# the following constants need to be initialized in init().
-const publish = Ref{Socket}()
-const raw_input = Ref{Socket}()
-const requests = Ref{Socket}()
-const control = Ref{Socket}()
-const heartbeat = Ref{Socket}()
-const heartbeat_context = Ref{Context}()
-const profile = Dict{String,Any}()
-const read_stdout = Ref{Base.PipeEndpoint}()
-const read_stderr = Ref{Base.PipeEndpoint}()
-const socket_locks = Dict{Socket,ReentrantLock}()
-
 # needed for executing pkg commands on earlier Julia versions
 @static if VERSION < v"1.11"
     # similar to Pkg.REPLMode.MiniREPL, a minimal REPL-like emulator
@@ -46,22 +31,24 @@ const socket_locks = Dict{Socket,ReentrantLock}()
 end
 
 """
-    init(args)
+    init(args, kernel)
 
 Initialize a kernel. `args` may either be empty or have one element containing
 the path to an existing connection file. If `args` is empty a connection file
 will be generated.
 """
-function init(args)
-    inited && error("IJulia is already running")
+function init(args, kernel, profile=nothing)
+    !isnothing(_default_kernel) && error("IJulia is already running")
     if length(args) > 0
-        merge!(profile, open(JSON.parse,args[1]))
-        verbose && println("PROFILE = $profile")
-        global connection_file = args[1]
+        merge!(kernel.profile, open(JSON.parse,args[1]))
+        kernel.verbose && println("PROFILE = $profile")
+        kernel.connection_file = args[1]
+    elseif !isnothing(profile)
+        merge!(kernel.profile, profile)
     else
         # generate profile and save
         let port0 = 5678
-            merge!(profile, Dict{String,Any}(
+            merge!(kernel.profile, Dict{String,Any}(
                 "ip" => "127.0.0.1",
                 "transport" => "tcp",
                 "stdin_port" => port0,
@@ -72,14 +59,16 @@ function init(args)
                 "key" => uuid4()
             ))
             fname = "profile-$(getpid()).json"
-            global connection_file = "$(pwd())/$fname"
-            println("connect ipython with --existing $connection_file")
+            kernel.connection_file = "$(pwd())/$fname"
+            println("connect ipython with --existing $(kernel.connection_file)")
             open(fname, "w") do f
-                JSON.print(f, profile)
+                JSON.print(f, kernel.profile)
             end
         end
     end
 
+    profile = kernel.profile
+
     if !isempty(profile["key"])
         signature_scheme = get(profile, "signature_scheme", "hmac-sha256")
         isempty(signature_scheme) && (signature_scheme = "hmac-sha256")
@@ -87,46 +76,52 @@ function init(args)
         if sigschm[1] != "hmac" || length(sigschm) != 2
             error("unrecognized signature_scheme $signature_scheme")
         end
-        hmacstate[] = MbedTLS.MD(getfield(MbedTLS, Symbol("MD_", uppercase(sigschm[2]))),
-                                 profile["key"])
+        kernel.hmacstate[] = MbedTLS.MD(getfield(MbedTLS, Symbol("MD_", uppercase(sigschm[2]))),
+                                        profile["key"])
     end
 
-    publish[] = Socket(PUB)
-    raw_input[] = Socket(ROUTER)
-    requests[] = Socket(ROUTER)
-    control[] = Socket(ROUTER)
-    heartbeat_context[] = Context()
-    heartbeat[] = Socket(heartbeat_context[], ROUTER)
+    kernel.publish[] = Socket(PUB)
+    kernel.raw_input[] = Socket(ROUTER)
+    kernel.requests[] = Socket(ROUTER)
+    kernel.control[] = Socket(ROUTER)
+    kernel.heartbeat_context[] = Context()
+    kernel.heartbeat[] = Socket(kernel.heartbeat_context[], ROUTER)
     sep = profile["transport"]=="ipc" ? "-" : ":"
-    bind(publish[], "$(profile["transport"])://$(profile["ip"])$(sep)$(profile["iopub_port"])")
-    bind(requests[], "$(profile["transport"])://$(profile["ip"])$(sep)$(profile["shell_port"])")
-    bind(control[], "$(profile["transport"])://$(profile["ip"])$(sep)$(profile["control_port"])")
-    bind(raw_input[], "$(profile["transport"])://$(profile["ip"])$(sep)$(profile["stdin_port"])")
-    bind(heartbeat[], "$(profile["transport"])://$(profile["ip"])$(sep)$(profile["hb_port"])")
+    bind(kernel.publish[], "$(profile["transport"])://$(profile["ip"])$(sep)$(profile["iopub_port"])")
+    bind(kernel.requests[], "$(profile["transport"])://$(profile["ip"])$(sep)$(profile["shell_port"])")
+    bind(kernel.control[], "$(profile["transport"])://$(profile["ip"])$(sep)$(profile["control_port"])")
+    bind(kernel.raw_input[], "$(profile["transport"])://$(profile["ip"])$(sep)$(profile["stdin_port"])")
+    bind(kernel.heartbeat[], "$(profile["transport"])://$(profile["ip"])$(sep)$(profile["hb_port"])")
 
     # associate a lock with each socket so that multi-part messages
     # on a given socket don't get inter-mingled between tasks.
-    for s in (publish[], raw_input[], requests[], control[])
-        socket_locks[s] = ReentrantLock()
+    for s in (kernel.publish[], kernel.raw_input[], kernel.requests[], kernel.control[])
+        kernel.socket_locks[s] = ReentrantLock()
     end
 
-    start_heartbeat(heartbeat[])
-    if capture_stdout
-        read_stdout[], = redirect_stdout()
-        redirect_stdout(IJuliaStdio(stdout,"stdout"))
+    start_heartbeat(kernel.heartbeat[], kernel)
+    if kernel.capture_stdout
+        kernel.read_stdout[], = redirect_stdout()
+        redirect_stdout(IJuliaStdio(stdout, kernel, "stdout"))
     end
-    if capture_stderr
-        read_stderr[], = redirect_stderr()
-        redirect_stderr(IJuliaStdio(stderr,"stderr"))
+    if kernel.capture_stderr
+        kernel.read_stderr[], = redirect_stderr()
+        redirect_stderr(IJuliaStdio(stderr, kernel, "stderr"))
     end
-    redirect_stdin(IJuliaStdio(stdin,"stdin"))
+    redirect_stdin(IJuliaStdio(stdin, kernel, "stdin"))
     @static if VERSION < v"1.11"
         minirepl[] = MiniREPL(TextDisplay(stdout))
     end
 
+    watch_stdio(kernel)
+    pushdisplay(IJulia.InlineDisplay())
+
     logger = ConsoleLogger(Base.stderr)
     Base.CoreLogging.global_logger(logger)
+    IJulia._default_kernel = kernel
+
+    send_status("starting", kernel)
+    kernel.inited = true
 
-    send_status("starting")
-    global inited = true
+    kernel.waitloop_task[] = @async waitloop(kernel)
 end
diff --git a/src/inline.jl b/src/inline.jl
index a784e573..674b6724 100644
--- a/src/inline.jl
+++ b/src/inline.jl
@@ -72,9 +72,10 @@ end
 for mime in ipy_mime
     @eval begin
         function display(d::InlineDisplay, ::MIME{Symbol($mime)}, x)
+            kernel = _default_kernel
             flush_all() # so that previous stream output appears in order
-            send_ipython(publish[],
-                         msg_pub(execute_msg, "display_data",
+            send_ipython(kernel.publish[], kernel,
+                         msg_pub(kernel.execute_msg, "display_data",
                                  Dict(
                                   "metadata" => metadata(x), # optional
                                   "transient" => transient(x), # optional
@@ -94,13 +95,14 @@ display(d::InlineDisplay, m::MIME"text/javascript", x) = display(d, MIME("applic
 # the display message, also sending text/plain for text data.
 displayable(d::InlineDisplay, M::MIME) = istextmime(M)
 function display(d::InlineDisplay, M::MIME, x)
+    kernel = _default_kernel
     sx = limitstringmime(M, x)
     d = Dict(string(M) => sx)
     if istextmime(M)
         d["text/plain"] = sx # directly show text data, e.g. text/csv
     end
     flush_all() # so that previous stream output appears in order
-    send_ipython(publish[],
+    send_ipython(kernel.publish[], kernel,
                  msg_pub(execute_msg, "display_data",
                          Dict("metadata" => metadata(x), # optional
                               "transient" => transient(x), # optional
@@ -110,10 +112,11 @@ end
 # override display to send IPython a dictionary of all supported
 # output types, so that IPython can choose what to display.
 function display(d::InlineDisplay, x)
-    undisplay(x) # dequeue previous redisplay(x)
+    kernel = _default_kernel
+    undisplay(x, kernel) # dequeue previous redisplay(x)
     flush_all() # so that previous stream output appears in order
-    send_ipython(publish[],
-                 msg_pub(execute_msg, "display_data",
+    send_ipython(kernel.publish[], kernel,
+                 msg_pub(kernel.execute_msg, "display_data",
                          Dict("metadata" => metadata(x), # optional
                               "transient" => transient(x), # optional
                               "data" => display_dict(x))))
@@ -124,15 +127,16 @@ end
 # an input cell has finished executing.
 
 function redisplay(d::InlineDisplay, x)
-    if !in(x,displayqueue)
-        push!(displayqueue, x)
+    kernel = _default_kernel
+    if !in(x, kernel.displayqueue)
+        push!(kernel.displayqueue, x)
     end
 end
 
-function display()
-    q = copy(displayqueue)
-    empty!(displayqueue) # so that undisplay in display(x) is no-op
+function display(kernel::Kernel)
+    q = copy(kernel.displayqueue)
+    empty!(kernel.displayqueue) # so that undisplay in display(x) is no-op
     for x in q
-        display(x)
+        display(x, kernel)
     end
 end
diff --git a/src/kernel.jl b/src/kernel.jl
index bb7ad13d..5fdb1359 100644
--- a/src/kernel.jl
+++ b/src/kernel.jl
@@ -21,7 +21,8 @@ end
 ENV["LINES"] = get(ENV, "LINES", 30)
 ENV["COLUMNS"] = get(ENV, "COLUMNS", 80)
 
-IJulia.init(ARGS)
+println("Starting kernel event loops.")
+IJulia.init(ARGS, IJulia.Kernel())
 
 let startupfile = !isempty(DEPOT_PATH) ? abspath(DEPOT_PATH[1], "config", "startup_ijulia.jl") : ""
     isfile(startupfile) && Base.JLOptions().startupfile != 2 && Base.include(Main, startupfile)
@@ -30,13 +31,8 @@ end
 # import things that we want visible in IJulia but not in REPL's using IJulia
 import IJulia: ans, In, Out, clear_history
 
-pushdisplay(IJulia.InlineDisplay())
-
 ccall(:jl_exit_on_sigint, Cvoid, (Cint,), 0)
 
-println(IJulia.orig_stdout[], "Starting kernel event loops.")
-IJulia.watch_stdio()
-
 # workaround JuliaLang/julia#4259
 delete!(task_local_storage(),:SOURCE_PATH)
 
@@ -50,4 +46,4 @@ if isdefined(Main, :Revise)
     end
 end
 
-IJulia.waitloop()
+wait(IJulia._default_kernel)
diff --git a/src/msg.jl b/src/msg.jl
index 910bcaab..a38bad9f 100644
--- a/src/msg.jl
+++ b/src/msg.jl
@@ -2,21 +2,6 @@ import Base.show
 
 export Msg, msg_pub, msg_reply, send_status, send_ipython
 
-"""
-IPython message struct.
-"""
-mutable struct Msg
-    idents::Vector{String}
-    header::Dict
-    content::Dict
-    parent_header::Dict
-    metadata::Dict
-    function Msg(idents, header::Dict, content::Dict,
-                 parent_header=Dict{String,Any}(), metadata=Dict{String,Any}())
-        new(idents,header,content,parent_header,metadata)
-    end
-end
-
 """
     msg_header(m::Msg, msg_type::String)
 
@@ -29,6 +14,8 @@ msg_header(m::Msg, msg_type::String) = Dict("msg_id" => uuid4(),
                                             "msg_type" => msg_type,
                                             "version" => "5.4")
 
+Base.VersionNumber(m::Msg) = VersionNumber(m.header["version"])
+
 # PUB/broadcast messages use the msg_type as the ident, except for
 # stream messages which use the stream name (e.g. "stdout").
 # [According to minrk, "this isn't well defined, or even really part
@@ -48,12 +35,12 @@ function show(io::IO, msg::Msg)
 end
 
 """
-    send_ipython(socket, m::Msg)
+    send_ipython(socket, kernel, m::Msg)
 
 Send a message `m`. This will lock `socket`.
 """
-function send_ipython(socket, m::Msg)
-    lock(socket_locks[socket])
+function send_ipython(socket, kernel, m::Msg)
+    lock(kernel.socket_locks[socket])
     try
         @vprintln("SENDING ", m)
         for i in m.idents
@@ -64,23 +51,23 @@ function send_ipython(socket, m::Msg)
         parent_header = json(m.parent_header)
         metadata = json(m.metadata)
         content = json(m.content)
-        send(socket, hmac(header, parent_header, metadata, content), more=true)
+        send(socket, hmac(header, parent_header, metadata, content, kernel), more=true)
         send(socket, header, more=true)
         send(socket, parent_header, more=true)
         send(socket, metadata, more=true)
         send(socket, content)
     finally
-        unlock(socket_locks[socket])
+        unlock(kernel.socket_locks[socket])
     end
 end
 
 """
-    recv_ipython(socket)
+    recv_ipython(socket, kernel)
 
 Wait for and get a message. This will lock `socket`.
 """
-function recv_ipython(socket)
-    lock(socket_locks[socket])
+function recv_ipython(socket, kernel)
+    lock(kernel.socket_locks[socket])
     try
         idents = String[]
         s = recv(socket, String)
@@ -96,23 +83,24 @@ function recv_ipython(socket)
         parent_header = recv(socket, String)
         metadata = recv(socket, String)
         content = recv(socket, String)
-        if signature != hmac(header, parent_header, metadata, content)
+        if signature != hmac(header, parent_header, metadata, content, kernel)
             error("Invalid HMAC signature") # What should we do here?
         end
         m = Msg(idents, JSON.parse(header), JSON.parse(content), JSON.parse(parent_header), JSON.parse(metadata))
         @vprintln("RECEIVED $m")
         return m
     finally
-        unlock(socket_locks[socket])
+        unlock(kernel.socket_locks[socket])
     end
 end
 
 """
-    send_status(state::AbstractString, parent_msg::Msg=execute_msg)
+    send_status(state::AbstractString, kernel, parent_msg::Msg=execute_msg)
 
 Publish a status message.
 """
-function send_status(state::AbstractString, parent_msg::Msg=execute_msg)
-    send_ipython(publish[], Msg([ "status" ], msg_header(parent_msg, "status"),
-                                Dict("execution_state" => state), parent_msg.header))
+function send_status(state::AbstractString, kernel, parent_msg::Msg=kernel.execute_msg)
+    send_ipython(kernel.publish[], kernel,
+                 Msg([ "status" ], msg_header(parent_msg, "status"),
+                     Dict("execution_state" => state), parent_msg.header))
 end
diff --git a/src/stdio.jl b/src/stdio.jl
index bb5b7046..2f6d8d60 100644
--- a/src/stdio.jl
+++ b/src/stdio.jl
@@ -7,11 +7,12 @@ Wrapper type around redirected stdio streams, both for overloading things like
 """
 struct IJuliaStdio{IO_t <: IO} <: Base.AbstractPipe
     io::IOContext{IO_t}
+    kernel::Kernel
 end
-IJuliaStdio(io::IO, stream::AbstractString="unknown") =
+IJuliaStdio(io::IO, kernel::Kernel, stream::AbstractString="unknown") =
     IJuliaStdio{typeof(io)}(IOContext(io, :color=>Base.have_color,
                             :jupyter_stream=>stream,
-                            :displaysize=>displaysize()))
+                            :displaysize=>displaysize()), kernel)
 Base.pipe_reader(io::IJuliaStdio) = io.io.io
 Base.pipe_writer(io::IJuliaStdio) = io.io.io
 Base.lock(io::IJuliaStdio) = lock(io.io.io)
@@ -24,18 +25,6 @@ Base.displaysize(io::IJuliaStdio) = displaysize(io.io)
 Base.unwrapcontext(io::IJuliaStdio) = Base.unwrapcontext(io.io)
 Base.setup_stdio(io::IJuliaStdio, readable::Bool) = Base.setup_stdio(io.io.io, readable)
 
-if VERSION < v"1.7.0-DEV.254"
-    for s in ("stdout", "stderr", "stdin")
-        f = Symbol("redirect_", s)
-        sq = QuoteNode(Symbol(s))
-        @eval function Base.$f(io::IJuliaStdio)
-            io[:jupyter_stream] != $s && throw(ArgumentError(string("expecting ", $s, " stream")))
-            Core.eval(Base, Expr(:(=), $sq, io))
-            return io
-        end
-    end
-end
-
 # logging in verbose mode goes to original stdio streams.  Use macros
 # so that we do not even evaluate the arguments in no-verbose modes
 
@@ -48,7 +37,7 @@ end
 
 macro vprintln(x...)
     quote
-        if verbose::Bool
+        if _default_kernel.verbose
             println(orig_stdout[], get_log_preface(), $(map(esc, x)...))
         end
     end
@@ -56,19 +45,15 @@ end
 
 macro verror_show(e, bt)
     quote
-        if verbose::Bool
+        if _default_kernel.verbose
             showerror(orig_stderr[], $(esc(e)), $(esc(bt)))
         end
     end
 end
 
-#name=>iobuffer for each stream ("stdout","stderr") so they can be sent in flush
-const bufs = Dict{String,IOBuffer}()
 const stream_interval = 0.1
 # maximum number of bytes in libuv/os buffer before emptying
 const max_bytes = 10*1024
-# max output per code cell is 512 kb by default
-const max_output_per_request = Ref(1 << 19)
 
 """
     watch_stream(rd::IO, name::AbstractString)
@@ -79,20 +64,20 @@ when buffer contains more than `max_bytes` bytes. Otherwise, if data is availabl
 `stream_interval` seconds (see the `Timer`'s set up in `watch_stdio`). Truncate the output to `max_output_per_request`
 bytes per execution request since excessive output can bring browsers to a grinding halt.
 """
-function watch_stream(rd::IO, name::AbstractString)
+function watch_stream(rd::IO, name::AbstractString, kernel)
     task_local_storage(:IJulia_task, "read $name task")
     try
         buf = IOBuffer()
-        bufs[name] = buf
+        kernel.bufs[name] = buf
         while !eof(rd) # blocks until something is available
             nb = bytesavailable(rd)
             if nb > 0
-                stdio_bytes[] += nb
+                kernel.stdio_bytes[] += nb
                 # if this stream has surpassed the maximum output limit then ignore future bytes
-                if stdio_bytes[] >= max_output_per_request[]
+                if kernel.stdio_bytes[] >= kernel.max_output_per_request[]
                     read(rd, nb) # read from libuv/os buffer and discard
-                    if stdio_bytes[] - nb < max_output_per_request[]
-                        send_ipython(publish[], msg_pub(execute_msg, "stream",
+                    if kernel.stdio_bytes[] - nb < kernel.max_output_per_request[]
+                        send_ipython(kernel.publish[], kernel, msg_pub(execute_msg, "stream",
                                      Dict("name" => "stderr", "text" => "Excessive output truncated after $(stdio_bytes[]) bytes.")))
                     end
                 else
@@ -102,7 +87,7 @@ function watch_stream(rd::IO, name::AbstractString)
             if buf.size > 0
                 if buf.size >= max_bytes
                     #send immediately
-                    send_stream(name)
+                    send_stream(name, kernel)
                 end
             end
         end
@@ -110,22 +95,22 @@ function watch_stream(rd::IO, name::AbstractString)
         # the IPython manager may send us a SIGINT if the user
         # chooses to interrupt the kernel; don't crash on this
         if isa(e, InterruptException)
-            watch_stream(rd, name)
+            watch_stream(rd, name, kernel)
         else
             rethrow()
         end
     end
 end
 
-function send_stdio(name)
-    if verbose::Bool && !haskey(task_local_storage(), :IJulia_task)
+function send_stdio(name, kernel)
+    if kernel.verbose && !haskey(task_local_storage(), :IJulia_task)
         task_local_storage(:IJulia_task, "send $name task")
     end
-    send_stream(name)
+    send_stream(name, kernel)
 end
 
-send_stdout(t::Timer) = send_stdio("stdout")
-send_stderr(t::Timer) = send_stdio("stderr")
+send_stdout(kernel) = send_stdio("stdout", kernel)
+send_stderr(kernel) = send_stdio("stderr", kernel)
 
 """
 Jupyter associates cells with message headers. Once a cell's execution state has
@@ -137,12 +122,12 @@ is updating Signal graph state, it's execution state is busy, meaning Jupyter
 will not drop stream messages if Interact can set the header message under which
 the stream messages will be sent. Hence the need for this function.
 """
-function set_cur_msg(msg)
-    global execute_msg = msg
+function set_cur_msg(msg, kernel)
+    kernel.execute_msg = msg
 end
 
-function send_stream(name::AbstractString)
-    buf = bufs[name]
+function send_stream(name::AbstractString, kernel)
+    buf = kernel.bufs[name]
     if buf.size > 0
         d = take!(buf)
         n = num_utf8_trailing(d)
@@ -163,8 +148,8 @@ function send_stream(name::AbstractString)
             print(sbuf, '\n')
             s = String(take!(sbuf))
         end
-        send_ipython(publish[],
-             msg_pub(execute_msg, "stream",
+        send_ipython(kernel.publish[], kernel,
+             msg_pub(kernel.execute_msg, "stream",
                      Dict("name" => name, "text" => s)))
     end
 end
@@ -194,15 +179,15 @@ Display the `prompt` string, request user input,
 and return the string entered by the user.  If `password`
 is `true`, the user's input is not displayed during typing.
 """
-function readprompt(prompt::AbstractString; password::Bool=false)
-    if !execute_msg.content["allow_stdin"]
+function readprompt(prompt::AbstractString; kernel=_default_kernel, password::Bool=false)
+    if !kernel.execute_msg.content["allow_stdin"]
         error("IJulia: this front-end does not implement stdin")
     end
-    send_ipython(raw_input[],
-                 msg_reply(execute_msg, "input_request",
+    send_ipython(kernel.raw_input[], kernel,
+                 msg_reply(kernel.execute_msg, "input_request",
                            Dict("prompt"=>prompt, "password"=>password)))
     while true
-        msg = recv_ipython(raw_input[])
+        msg = recv_ipython(kernel.raw_input[], kernel)
         if msg.header["msg_type"] == "input_reply"
             return msg.content["value"]
         else
@@ -243,17 +228,19 @@ function readline(io::IJuliaStdio)
     end
 end
 
-function watch_stdio()
+function watch_stdio(kernel)
     task_local_storage(:IJulia_task, "init task")
-    if capture_stdout
-        read_task = @async watch_stream(read_stdout[], "stdout")
+    if kernel.capture_stdout
+        kernel.watch_stdout_task[] = @async watch_stream(kernel.read_stdout[], "stdout", kernel)
+        errormonitor(kernel.watch_stdout_task[])
         #send stdout stream msgs every stream_interval secs (if there is output to send)
-        Timer(send_stdout, stream_interval, interval=stream_interval)
+        kernel.watch_stdout_timer[] = Timer(_ -> send_stdout(kernel), stream_interval, interval=stream_interval)
     end
-    if capture_stderr
-        readerr_task = @async watch_stream(read_stderr[], "stderr")
+    if kernel.capture_stderr
+        kernel.watch_stderr_task[] = @async watch_stream(kernel.read_stderr[], "stderr", kernel)
+        errormonitor(kernel.watch_stderr_task[])
         #send STDERR stream msgs every stream_interval secs (if there is output to send)
-        Timer(send_stderr, stream_interval, interval=stream_interval)
+        kernel.watch_stderr_timer[] = Timer(_ -> send_stderr(kernel), stream_interval, interval=stream_interval)
     end
 end
 
@@ -277,5 +264,5 @@ import Base.flush
 function flush(io::IJuliaStdio)
     flush(io.io)
     oslibuv_flush()
-    send_stream(get(io,:jupyter_stream,"unknown"))
+    send_stream(get(io,:jupyter_stream,"unknown"), io.kernel)
 end
diff --git a/test/Project.toml b/test/Project.toml
index 4354dbcf..48b6c5a8 100644
--- a/test/Project.toml
+++ b/test/Project.toml
@@ -1,6 +1,12 @@
 [deps]
 Aqua = "4c88cf16-eb10-579e-8560-4a9242c79595"
 Base64 = "2a0f44e3-6c83-55bd-87e4-b1978d98bd5f"
+CondaPkg = "992eb4ea-22a4-4c89-a5bb-47a3300528ab"
 Dates = "ade2ca70-3891-5945-98fb-dc099432e06a"
+FileIO = "5789e2e9-d7fb-5bc7-8068-2c6fae9b9549"
+ImageIO = "82e4d734-157c-48bb-816b-45c225c6df19"
+ImageShow = "4e3cecfd-b093-5904-9786-8bbb286a6a31"
 JSON = "682c06a0-de6a-54ab-a142-c8b1cf79cde6"
+PythonCall = "6099a3de-0909-46bc-b1f4-468b9a2dfc0d"
+Sockets = "6462fe0b-24de-5631-8697-dd941f90decc"
 Test = "8dfed614-e22c-5e08-85e1-65c5234f0b40"
diff --git a/test/kernel.jl b/test/kernel.jl
new file mode 100644
index 00000000..1a8c247f
--- /dev/null
+++ b/test/kernel.jl
@@ -0,0 +1,369 @@
+ENV["JULIA_CONDAPKG_ENV"] = "@ijulia-tests"
+
+# If you're running the tests locally you could comment out the
+# JULIA_CONDAPKG_ENV line above and uncomment the two environment variables
+# below. This will be a bit faster since it stops CondaPkg from re-resolving the
+# environment each time (but you do need to run it at least once locally to
+# initialize the `@ijulia-tests` environment).
+# ENV["JULIA_PYTHONCALL_EXE"] = joinpath(Base.DEPOT_PATH[1], "conda_environments", "ijulia-tests", "bin", "python")
+# ENV["JULIA_CONDAPKG_BACKEND"] = "Null"
+
+
+using Test
+import Sockets
+import Sockets: listenany
+
+import PythonCall
+import PythonCall: Py, pyimport, pyconvert, pytype, pystr
+
+# A little bit of hackery to fix the version number sent by the client. See:
+# https://github.com/jupyter/jupyter_client/pull/1054
+jupyter_client_lib = pyimport("jupyter_client")
+jupyter_client_lib.session.protocol_version = "5.4"
+
+const BlockingKernelClient = jupyter_client_lib.BlockingKernelClient
+
+import IJulia: Kernel
+# These symbols are imported so that we can test that setproperty!(::Kernel)
+# will propagate changes from the corresponding Kernel fields to the
+# module-global variables.
+import IJulia: ans, In, Out
+
+
+function getports(port_hint, n)
+    ports = Int[]
+
+    for i in 1:n
+        port, server = listenany(Sockets.localhost, port_hint)
+        close(server)
+        push!(ports, port)
+        port_hint = port + 1
+    end
+
+    return ports
+end
+
+function create_profile(port_hint=8080)
+    ports = getports(port_hint, 5)
+
+    Dict(
+        "transport" => "tcp",
+        "ip" => "127.0.0.1",
+        "control_port" => ports[1],
+        "shell_port" => ports[2],
+        "stdin_port" => ports[3],
+        "hb_port" => ports[4],
+        "iopub_port" => ports[5],
+        "signature_scheme" => "hmac-sha256",
+        "key" => "a0436f6c-1916-498b-8eb9-e81ab9368e84"
+    )
+end
+
+function test_py_get!(get_func, result)
+    try
+        result[] = get_func(timeout=0)
+        return true
+    catch ex
+        exception_type = pyconvert(String, ex.t.__name__)
+        if exception_type != "Empty"
+            rethrow()
+        end
+
+        return false
+    end
+end
+
+function recursive_pyconvert(x)
+    x_type = pyconvert(String, pytype(x).__name__)
+
+    if x_type == "dict"
+        x = pyconvert(Dict{String, Any}, x)
+        for key in copy(keys(x))
+            if x[key] isa Py
+                x[key] = recursive_pyconvert(x[key])
+            elseif x[key] isa PythonCall.PyDict
+                x[key] = recursive_pyconvert(x[key].py)
+            end
+        end
+    elseif x_type == "str"
+        x = pyconvert(String, x)
+    end
+
+    return x
+end
+
+# Calling methods directly with `reply=true` on the BlockingKernelClient will
+# cause a deadlock because the client will block the whole thread while polling
+# the socket, which means that the thread will never enter a GC safepoint so any
+# other code that happens to allocate will get blocked. Instead, we send
+# requests by themselves and then poll the appropriate socket with `timeout=0`
+# so that the Python code will never block and we never get into a deadlock.
+function make_request(request_func, get_func, args...; wait=true, kwargs...)
+    request_func(args...; kwargs..., reply=false)
+    if !wait
+        return nothing
+    end
+
+    result = Ref{Py}()
+    timeout = haskey(ENV, "CI") ? 120 : 10
+    if timedwait(() -> test_py_get!(get_func, result), timeout) == :timed_out
+        error("Jupyter channel get timed out")
+    end
+
+    return recursive_pyconvert(result[])
+end
+
+kernel_info(client) = make_request(client.kernel_info, client.get_shell_msg)
+comm_info(client) = make_request(client.comm_info, client.get_shell_msg)
+history(client) = make_request(client.history, client.get_shell_msg)
+shutdown(client; wait=true) = make_request(client.shutdown, client.get_control_msg; wait)
+execute(client, code) = make_request(client.execute, client.get_shell_msg; code)
+inspect(client, code) = make_request(client.inspect, client.get_shell_msg; code)
+get_stdin_msg(client) = make_request(Returns(nothing), client.get_stdin_msg)
+get_iopub_msg(client) = make_request(Returns(nothing), client.get_iopub_msg)
+
+function get_execute_result(client)
+    while true
+        msg = get_iopub_msg(client)
+        if msg["header"]["msg_type"] == "execute_result"
+            return msg
+        end
+    end
+end
+
+function msg_ok(msg)
+    ok = msg["content"]["status"] == "ok"
+    if !ok
+        @error "Kernel is not ok" msg["content"]
+    end
+
+    return ok
+end
+
+msg_error(msg) = msg["content"]["status"] == "error"
+
+function jupyter_client(f, profile)
+    client = BlockingKernelClient()
+    client.load_connection_info(profile)
+    client.start_channels()
+
+    try
+        f(client)
+    finally
+        client.stop_channels()
+    end
+end
+
+@testset "Kernel" begin
+    profile = create_profile()
+    profile_kwargs = Dict([Symbol(key) => value for (key, value) in profile])
+    profile_kwargs[:key] = pystr(profile_kwargs[:key]).encode()
+
+    @testset "getproperty()/setproperty!()" begin
+        kernel = Kernel()
+
+        # Test setting special fields that should be mirrored to global variables
+        for field in (:n, :ans, :inited)
+            # Save the old value so we can restore them afterwards
+            old_value = getproperty(kernel, field)
+
+            test_value = field === :inited ? true : 10
+            setproperty!(kernel, field, test_value)
+            @test getproperty(IJulia, field) == test_value
+            @test getproperty(kernel, field) == test_value
+
+            setproperty!(kernel, field, old_value)
+        end
+    end
+
+    @testset "Explicit tests with jupyter_client" begin
+        # Some of these tests have their own kernel instance to avoid
+        # interfering with the state of other tests.
+
+        # Test clear_history()
+        Kernel(profile; capture_stdout=false, capture_stderr=false) do kernel
+            jupyter_client(profile) do client
+                for i in 1:10
+                    @test msg_ok(execute(client, "$(i)"))
+                end
+                @test length(kernel.In) == 10
+                @test msg_ok(execute(client, "IJulia.clear_history(-1:5)"))
+                @test Set(keys(kernel.In)) == Set(6:11) # The 11th entry is the call to clear_history()
+                @test msg_ok(execute(client, "IJulia.clear_history()"))
+                @test isempty(kernel.In)
+                @test isempty(kernel.Out)
+            end
+        end
+
+        # Test input
+        Kernel(profile; capture_stdout=false, capture_stderr=false) do kernel
+            jupyter_client(profile) do client
+                # The input system in Jupyter is a bit convoluted. First we
+                # make a request to the kernel:
+                client.execute("readline()")
+                # Then wait for readline(::IJuliaStdio) to send its own
+                # `input_request` message on the stdin socket.
+                @test msg_ok(get_stdin_msg(client))
+                # Send an `input_reply` back
+                client.input("foo")
+
+                # Wait for the original `execute_request` to complete and
+                # send an `execute_result` message with the 'input'.
+                msg = get_execute_result(client)
+                @test msg["content"]["data"]["text/plain"] == "\"foo\""
+            end
+        end
+
+        shutdown_called = false
+        Kernel(profile; capture_stdout=false, capture_stderr=false, shutdown=() -> shutdown_called = true) do kernel
+            jupyter_client(profile) do client
+                # Test load()/load_string()
+                mktemp() do path, _
+                    write(path, "42")
+
+                    msg = execute(client, "IJulia.load($(repr(path)))")
+                    @test msg_ok(msg)
+                    @test length(msg["content"]["payload"]) == 1
+                end
+
+                # Test hooks
+                @testset "Hooks" begin
+                    preexecute = false
+                    postexecute = false
+                    posterror = false
+                    preexecute_hook = () -> preexecute = !preexecute
+                    postexecute_hook = () -> postexecute = !postexecute
+                    posterror_hook = () -> posterror = !posterror
+                    IJulia.push_preexecute_hook(preexecute_hook)
+                    IJulia.push_postexecute_hook(postexecute_hook)
+                    IJulia.push_posterror_hook(posterror_hook)
+                    @test msg_ok(execute(client, "42"))
+
+                    # The pre/post hooks should've been called but not the posterror hook
+                    @test preexecute
+                    @test postexecute
+                    @test !posterror
+
+                    # With a throwing cell the posterror hook should be called
+                    @test msg_error(execute(client, "error(42)"))
+                    @test posterror
+
+                    # After popping the hooks they should no longer be executed
+                    preexecute = false
+                    postexecute = false
+                    posterror = false
+                    IJulia.pop_preexecute_hook(preexecute_hook)
+                    IJulia.pop_postexecute_hook(postexecute_hook)
+                    IJulia.pop_posterror_hook(posterror_hook)
+                    @test msg_ok(execute(client, "42"))
+                    @test msg_error(execute(client, "error(42)"))
+                    @test !preexecute
+                    @test !postexecute
+                    @test !posterror
+                end
+
+                # Smoke tests
+                @test msg_ok(kernel_info(client))
+                @test msg_ok(comm_info(client))
+                @test msg_ok(history(client))
+                @test msg_ok(execute(client, "IJulia.set_verbose(false)"))
+                @test msg_ok(execute(client, "flush(stdout)"))
+
+                # Test history(). This test requires `capture_stdout=false`.
+                IJulia.clear_history()
+                @test msg_ok(execute(client, "1"))
+                @test msg_ok(execute(client, "42"))
+                stdout_pipe = Pipe()
+                redirect_stdout(stdout_pipe) do
+                    IJulia.history()
+                end
+                close(stdout_pipe.in)
+                @test collect(eachline(stdout_pipe)) == ["1", "42"]
+
+                # Test that certain global variables are updated in kernel.current_module
+                @test msg_ok(execute(client, "42"))
+                @test msg_ok(execute(client, "ans == 42"))
+                @test kernel.ans
+
+                # Test shutdown_request
+                @test msg_ok(shutdown(client))
+                @test timedwait(() -> shutdown_called, 10) == :ok
+            end
+        end
+    end
+
+    @testset "jupyter_kernel_test" begin
+        stdout_pipe = Pipe()
+        stderr_pipe = Pipe()
+        Base.link_pipe!(stdout_pipe)
+        Base.link_pipe!(stderr_pipe)
+        stdout_str = ""
+        stderr_str = ""
+        test_proc = nothing
+
+        Kernel(profile; shutdown=Returns(nothing)) do kernel
+            test_file = joinpath(@__DIR__, "kernel_test.py")
+
+            mktemp() do connection_file, io
+                # Write the connection file
+                jupyter_client_lib.connect.write_connection_file(; fname=connection_file, profile_kwargs...)
+
+                try
+                    # Run jupyter_kernel_test
+                    cmd = ignorestatus(`$(PythonCall.C.python_executable_path()) $(test_file)`)
+                    cmd = addenv(cmd, "IJULIA_TESTS_CONNECTION_FILE" => connection_file)
+                    cmd = pipeline(cmd; stdout=stdout_pipe, stderr=stderr_pipe)
+                    test_proc = run(cmd)
+                finally
+                    close(stdout_pipe.in)
+                    close(stderr_pipe.in)
+                    stdout_str = read(stdout_pipe, String)
+                    stderr_str = read(stderr_pipe, String)
+                    close(stdout_pipe)
+                    close(stderr_pipe)
+                end
+            end
+        end
+
+        if !isempty(stdout_str)
+            @info "jupyter_kernel_test stdout:"
+            println(stdout_str)
+        end
+        if !isempty(stderr_str)
+            @info "jupyter_kernel_test stderr:"
+            println(stderr_str)
+        end
+        if !success(test_proc)
+            error("jupyter_kernel_test failed")
+        end
+    end
+
+    # kernel.jl is the script that's actually run by Jupyter
+    @testset "kernel.jl" begin
+        kernel_jl = joinpath(@__DIR__, "..", "src", "kernel.jl")
+        julia = joinpath(Sys.BINDIR, "julia")
+
+        mktemp() do connection_file, io
+            # Write the connection file
+            jupyter_client_lib.connect.write_connection_file(; fname=connection_file, profile_kwargs...)
+
+            cmd = `$julia --startup-file=no --project=$(Base.active_project()) $kernel_jl $(connection_file)`
+            kernel_proc = run(pipeline(cmd; stdout, stderr); wait=false)
+            try
+                jupyter_client(profile) do client
+                    @test msg_ok(kernel_info(client))
+                    @test msg_ok(execute(client, "42"))
+
+                    # Note that we don't wait for a reply because the kernel
+                    # will shut down almost immediately and it's not guaranteed
+                    # we'll receive the reply.
+                    shutdown(client; wait=false)
+                end
+
+                @test timedwait(() -> process_exited(kernel_proc), 10) == :ok
+            finally
+                kill(kernel_proc)
+            end
+        end
+    end
+end
diff --git a/test/kernel_test.py b/test/kernel_test.py
new file mode 100644
index 00000000..ef06003b
--- /dev/null
+++ b/test/kernel_test.py
@@ -0,0 +1,136 @@
+import os
+import unittest
+import typing as t
+
+import jupyter_kernel_test
+import jupyter_client
+from jupyter_client import KernelManager, BlockingKernelClient
+
+# A little bit of hackery to fix the version number sent by the client. See:
+# https://github.com/jupyter/jupyter_client/pull/1054
+jupyter_client.session.protocol_version = "5.4"
+
+# This is a modified version of jupyter_client.start_new_kernel() that uses an
+# existing kernel from a connection file rather than trying to launch one.
+def start_new_kernel2(
+    startup_timeout: float = 1, kernel_name: str = "python", **kwargs: t.Any
+) -> t.Tuple[KernelManager, BlockingKernelClient]:
+    """Start a new kernel, and return its Manager and Client"""
+    connection_file = os.environ["IJULIA_TESTS_CONNECTION_FILE"]
+
+    km = KernelManager(owns_kernel=False)
+    km.load_connection_file(connection_file)
+    km._connect_control_socket()
+
+    kc = BlockingKernelClient()
+    kc.load_connection_file(connection_file)
+
+    kc.start_channels()
+    try:
+        kc.wait_for_ready(timeout=startup_timeout)
+    except RuntimeError:
+        kc.stop_channels()
+        km.shutdown_kernel()
+        raise
+
+    return km, kc
+
+# Modified version of:
+# https://github.com/jupyter/jupyter_kernel_test/blob/main/test_ipykernel.py
+#
+# We skip the pager and history tests because they aren't supporteed.
+class IJuliaTests(jupyter_kernel_test.KernelTests):
+    # Required --------------------------------------
+
+    # The name identifying an installed kernel to run the tests against
+    kernel_name = "IJuliaKernel"
+
+    # Optional --------------------------------------
+
+    # language_info.name in a kernel_info_reply should match this
+    language_name = "julia"
+
+    # the normal file extension (including the leading dot) for this language
+    # checked against language_info.file_extension in kernel_info_reply
+    file_extension = ".jl"
+
+    # Code in the kernel's language to write "hello, world" to stdout
+    code_hello_world = 'println("hello, world")'
+
+    # code which should cause (any) text to be written to STDERR
+    code_stderr = 'println(stderr, "foo")'
+
+    # samples for the autocompletion functionality
+    # for each dictionary, `text` is the input to try and complete, and
+    # `matches` the list of all complete matching strings which should be found
+    completion_samples = [
+        {
+            "text": "zi",
+            "matches": {"zip"},
+        },
+    ]
+
+    # samples for testing code-completeness (used by console only)
+    # these samples should respectively be unambiguously complete statements
+    # (which should be executed on <enter>), incomplete statements or code
+    # which should be identified as invalid
+    complete_code_samples = ["1", 'print("hello, world")', "f(x) = x*2"]
+    incomplete_code_samples = ['print("hello', "f(x) = x*"]
+    invalid_code_samples = ["import = 7q"]
+
+    # code which should generate a (user-level) error in the kernel, and send
+    # a traceback to the client
+    code_generate_error = "error(42)"
+
+    # Samples of code which generate a result value (ie, some text
+    # displayed as Out[n])
+    code_execute_result = [{"code": "6*7", "result": "42"}]
+
+    # Samples of code which should generate a rich display output, and
+    # the expected MIME type.
+    # Note that we slice down the image so it doesn't display such a massive
+    # amount of text when debugging.
+    code_display_data = [
+        {
+            "code": 'using FileIO, ImageShow; display(load("mandrill.png")[1:5, 1:5])',
+            "mime": "image/png"
+        },
+        {
+            "code": 'display(MIME("image/svg+xml"), read("plus.svg", String))',
+            "mime": "image/svg+xml"
+        },
+        {
+            "code": 'display(MIME("text/latex"), "\\frac{1}{2}")',
+            "mime": "text/latex"
+        },
+        {
+            "code": 'display(MIME("text/markdown"), "# header")',
+            "mime": "text/markdown"
+        },
+        {
+            "code": 'display(MIME("text/html"), "<b>foo</b>")',
+            "mime": "text/html"
+        },
+        {
+            "code": 'display("foo")',
+            "mime": "text/plain"
+        }
+    ]
+
+    # test the support for object inspection
+    # the sample should be a name about which the kernel can give some help
+    # information (a built-in function is probably a good choice)
+    # only the default inspection level (equivalent to ipython "obj?")
+    # is currently tested
+    code_inspect_sample = "zip"
+
+    # a code sample which should cause a `clear_output` message to be sent to
+    # the client
+    code_clear_output = "IJulia.clear_output()"
+
+    @classmethod
+    def setUpClass(cls) -> None:
+        cls.km, cls.kc = start_new_kernel2(kernel_name=cls.kernel_name)
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/test/mandrill.png b/test/mandrill.png
new file mode 100644
index 00000000..8164b6f6
Binary files /dev/null and b/test/mandrill.png differ
diff --git a/test/plus.svg b/test/plus.svg
new file mode 100644
index 00000000..b74f17a6
--- /dev/null
+++ b/test/plus.svg
@@ -0,0 +1,6 @@
+<?xml version="1.0" encoding="iso-8859-1"?>
+<!-- Uploaded to: SVG Repo, www.svgrepo.com, Generator: SVG Repo Mixer Tools -->
+<svg fill="#000000" height="800px" width="800px" version="1.1" id="Capa_1" xmlns="http://www.w3.org/2000/svg" xmlns:xlink="http://www.w3.org/1999/xlink" 
+	 viewBox="0 0 42 42" xml:space="preserve">
+<polygon points="42,20 22,20 22,0 20,0 20,20 0,20 0,22 20,22 20,42 22,42 22,22 42,22 "/>
+</svg>
\ No newline at end of file
diff --git a/test/runtests.jl b/test/runtests.jl
index 523d062d..9ec3adf9 100644
--- a/test/runtests.jl
+++ b/test/runtests.jl
@@ -11,4 +11,11 @@ for file in TEST_FILES
     include(file)
 end
 
-Aqua.test_all(IJulia; piracies=(; broken=true))
+# MicroMamba (and thus CondaPkg and PythonCall) are not supported on 32bit
+if Sys.WORD_SIZE != 32
+    include("kernel.jl")
+else
+    @warn "Skipping the Kernel tests on 32bit"
+end
+
+Aqua.test_all(IJulia)
diff --git a/test/stdio.jl b/test/stdio.jl
index 3c22b3af..8710452a 100644
--- a/test/stdio.jl
+++ b/test/stdio.jl
@@ -2,9 +2,10 @@ using Test
 using IJulia
 
 @testset "stdio" begin
+    kernel = IJulia.Kernel()
 
     mktemp() do path, io
-        redirect_stdout(IJulia.IJuliaStdio(io, "stdout")) do
+        redirect_stdout(IJulia.IJuliaStdio(io, kernel, "stdout")) do
             println(Base.stdout, "stdout")
             println("print")
         end
@@ -12,17 +13,17 @@ using IJulia
         seek(io, 0)
         @test read(io, String) == "stdout\nprint\n"
         if VERSION < v"1.7.0-DEV.254"
-            @test_throws ArgumentError redirect_stdout(IJulia.IJuliaStdio(io, "stderr"))
-            @test_throws ArgumentError redirect_stdout(IJulia.IJuliaStdio(io, "stdin"))
-            @test_throws ArgumentError redirect_stderr(IJulia.IJuliaStdio(io, "stdout"))
-            @test_throws ArgumentError redirect_stderr(IJulia.IJuliaStdio(io, "stdin"))
-            @test_throws ArgumentError redirect_stdin(IJulia.IJuliaStdio(io, "stdout"))
-            @test_throws ArgumentError redirect_stdin(IJulia.IJuliaStdio(io, "stderr"))
+            @test_throws ArgumentError redirect_stdout(IJulia.IJuliaStdio(io, kernel, "stderr"))
+            @test_throws ArgumentError redirect_stdout(IJulia.IJuliaStdio(io, kernel, "stdin"))
+            @test_throws ArgumentError redirect_stderr(IJulia.IJuliaStdio(io, kernel, "stdout"))
+            @test_throws ArgumentError redirect_stderr(IJulia.IJuliaStdio(io, kernel, "stdin"))
+            @test_throws ArgumentError redirect_stdin(IJulia.IJuliaStdio(io, kernel, "stdout"))
+            @test_throws ArgumentError redirect_stdin(IJulia.IJuliaStdio(io, kernel, "stderr"))
         end
     end
 
     mktemp() do path, io
-        redirect_stderr(IJulia.IJuliaStdio(io, "stderr")) do
+        redirect_stderr(IJulia.IJuliaStdio(io, kernel, "stderr")) do
             println(Base.stderr, "stderr")
         end
         flush(io)
@@ -31,11 +32,11 @@ using IJulia
     end
 
     mktemp() do path, io
-        redirect_stdin(IJulia.IJuliaStdio(io, "stdin")) do
+        redirect_stdin(IJulia.IJuliaStdio(io, kernel, "stdin")) do
             # We can't actually do anything here because `IJuliaexecute_msg` has not
             # yet been initialized, so we just make sure that redirect_stdin does
             # not error.
         end
     end
 
-end
\ No newline at end of file
+end