tessl/pypi-jupyter-client
Jupyter protocol implementation and client libraries for kernel communication and management
—
Pending
kernel-management.mddocs/
Kernel Management
Comprehensive kernel lifecycle management including starting, stopping, restarting, and monitoring individual kernels or multiple kernels simultaneously. Supports both synchronous and asynchronous operations for different integration patterns.
Capabilities
Single Kernel Management
The KernelManager class provides complete lifecycle management for individual kernel instances, handling process management, connection setup, and kernel monitoring.
class KernelManager:
"""Manages the lifecycle of a single kernel process."""
def start_kernel(self, **kwargs):
"""
Start a new kernel process.
Parameters:
- kernel_name (str): Name of kernel to start
- cwd (str): Working directory for kernel
- env (dict): Environment variables
- **kwargs: Additional kernel arguments
Returns:
None
"""
def shutdown_kernel(self, now=False, restart=False):
"""
Shutdown the kernel process.
Parameters:
- now (bool): Force immediate shutdown if True
- restart (bool): Whether this is part of a restart
Returns:
None
"""
def restart_kernel(self, now=False, newports=False, **kwargs):
"""
Restart the kernel process.
Parameters:
- now (bool): Force immediate restart if True
- newports (bool): Use new ports for connection
- **kwargs: Additional arguments for new kernel
Returns:
None
"""
def kill_kernel(self):
"""
Force kill the kernel process.
Returns:
None
"""
def interrupt_kernel(self):
"""
Send interrupt signal to kernel.
Returns:
None
"""
def signal_kernel(self, signum):
"""
Send specific signal to kernel process.
Parameters:
- signum (int): Signal number to send
Returns:
None
"""
def is_alive(self):
"""
Check if kernel process is running.
Returns:
bool: True if kernel is alive
"""
def client(self, **kwargs):
"""
Create a client for communicating with this kernel.
Parameters:
- **kwargs: Client configuration options
Returns:
BlockingKernelClient: Client instance
"""
@property
def ready(self):
"""Future indicating when kernel is ready for communication."""
@property
def has_kernel(self):
"""Boolean indicating if kernel process exists."""
@property
def kernel_spec(self):
"""KernelSpec object for this kernel."""Async Kernel Management
The AsyncKernelManager provides async versions of kernel management operations for integration with asyncio-based applications.
class AsyncKernelManager(KernelManager):
"""Async version of KernelManager."""
def client(self, **kwargs):
"""
Create an async client for this kernel.
Parameters:
- **kwargs: Client configuration options
Returns:
AsyncKernelClient: Async client instance
"""Multi-Kernel Management
The MultiKernelManager class manages multiple kernel instances simultaneously, providing APIs for starting, stopping, and tracking multiple kernels with unique identifiers.
class MultiKernelManager:
"""Manages multiple kernel instances."""
def start_kernel(self, kernel_name=None, **kwargs):
"""
Start a new kernel and return its ID.
Parameters:
- kernel_name (str): Name of kernel to start
- **kwargs: Kernel configuration options
Returns:
str: Unique kernel ID
"""
def shutdown_kernel(self, kernel_id, now=False, restart=False):
"""
Shutdown a specific kernel.
Parameters:
- kernel_id (str): ID of kernel to shutdown
- now (bool): Force immediate shutdown
- restart (bool): Whether this is part of restart
Returns:
None
"""
def restart_kernel(self, kernel_id, now=False, **kwargs):
"""
Restart a specific kernel.
Parameters:
- kernel_id (str): ID of kernel to restart
- now (bool): Force immediate restart
- **kwargs: Additional kernel arguments
Returns:
None
"""
def kill_kernel(self, kernel_id):
"""
Force kill a specific kernel.
Parameters:
- kernel_id (str): ID of kernel to kill
Returns:
None
"""
def interrupt_kernel(self, kernel_id):
"""
Interrupt a specific kernel.
Parameters:
- kernel_id (str): ID of kernel to interrupt
Returns:
None
"""
def get_kernel(self, kernel_id):
"""
Get the KernelManager for a specific kernel.
Parameters:
- kernel_id (str): ID of kernel
Returns:
KernelManager: Manager instance for the kernel
"""
def list_kernel_ids(self):
"""
List all active kernel IDs.
Returns:
list: List of kernel ID strings
"""
def remove_kernel(self, kernel_id):
"""
Remove a kernel from management (must be shutdown first).
Parameters:
- kernel_id (str): ID of kernel to remove
Returns:
KernelManager: The removed kernel manager
"""
@property
def default_kernel_name(self):
"""Default kernel name for new kernels."""
@property
def kernel_manager_class(self):
"""Class used for individual kernel managers."""Async Multi-Kernel Management
class AsyncMultiKernelManager(MultiKernelManager):
"""Async version of MultiKernelManager with same interface."""Kernel Utilities
def run_kernel():
"""
Run a kernel in the current process.
Returns:
None
"""Usage Examples
Basic Single Kernel Management
from jupyter_client import KernelManager
# Create and configure kernel manager
km = KernelManager(kernel_name='python3')
# Start the kernel
km.start_kernel()
# Check if kernel is running
if km.is_alive():
print("Kernel is running")
# Create a client to communicate
kc = km.client()
kc.start_channels()
# Use the kernel...
# Clean shutdown
kc.stop_channels()
km.shutdown_kernel()Multi-Kernel Management
from jupyter_client import MultiKernelManager
# Create multi-kernel manager
mkm = MultiKernelManager()
# Start multiple kernels
python_id = mkm.start_kernel(kernel_name='python3')
r_id = mkm.start_kernel(kernel_name='ir')
# List all kernels
kernel_ids = mkm.list_kernel_ids()
print(f"Running kernels: {kernel_ids}")
# Get individual managers
python_km = mkm.get_kernel(python_id)
r_km = mkm.get_kernel(r_id)
# Clean shutdown
mkm.shutdown_kernel(python_id)
mkm.shutdown_kernel(r_id)Async Kernel Management
import asyncio
from jupyter_client import AsyncKernelManager
async def manage_kernel():
# Create async kernel manager
km = AsyncKernelManager()
# Start kernel
await km.start_kernel()
# Create async client
kc = km.client()
await kc.start_channels()
# Use kernel...
# Clean shutdown
await kc.stop_channels()
await km.shutdown_kernel()
# Run async function
asyncio.run(manage_kernel())Install with Tessl CLI
npx tessl i tessl/pypi-jupyter-client