monperrus · February 9, 2026 13:05
diff --git a/tmux_jupyterlab.py b/tmux_jupyterlab.py
 #!/usr/bin/env python3
 """
 JupyterLab Terminal Tmux Manager.

 This script provides multiple ways to interact with JupyterLab terminals:
 1. Tmux Orchestrator: Launches a tmux session where each window is connected to 
   an active JupyterHub terminal. Inside tmux, 'c' creates a new remote terminal.
 2. One-shot Execution: Creates a temporary terminal to run a single command 
   (passed as an argument or piped) and prints the output.
 3. Terminal Listing: Lists all active terminals and their last activity.
 4. Direct Connection: Connects interactively to a specific terminal by ID.

 Authentication is handled via the JUPYTERHUB_API_TOKEN environment variable, 
 system keyring, or an interactive prompt.

 URL: https://gist.github.com/monperrus/6c2d28dfa8b8519ddf05538f73015b09
 """

 import requests
 import json
 import sys
 import getpass
 import asyncio
 import websockets
 import ssl
 import keyring
 import tty
 import termios
 import os
 import argparse
 import subprocess
 import shutil

 # Configuration
 BASE_URL = "https://gpu1.eecs.kth.se"
 USERNAME = "monp" 

 def get_api_token():
    """Retrieve token from Env, Keyring, or Prompt."""
    # 1. Environment Variable
    token = os.environ.get("JUPYTERHUB_API_TOKEN")
    if token:
        return token

    # 2. Keyring
    try:
        login_keyring = keyring.get_keyring()
        token = login_keyring.get_password('login2', 'JUPYTERHUB_API_TOKEN')
        if token:
            return token
    except Exception:
        pass

    # 3. Prompt
    sys.stderr.write("Token not found in Env or Keyring.\n")
    token = getpass.getpass("Enter your JupyterHub API Token: ").strip()
    return token

 async def receive_messages(websocket):
    """Continuously receive and print messages from the WebSocket."""
    try:
        while True:
            message = await websocket.recv()
            data = json.loads(message)
            
            # Protocol: ['stdout', 'output_string']
            if data[0] == 'stdout':
                sys.stdout.write(data[1])
                sys.stdout.flush()
    except websockets.exceptions.ConnectionClosed:
        pass
    except asyncio.CancelledError:
        pass
    except Exception as e:
        if not isinstance(e, asyncio.CancelledError):
            sys.stderr.write(f"\r\nError receiving data: {e}\r\n")

 async def send_input_from_queue(websocket, queue):
    """Read from queue and send to WebSocket."""
    while True:
        data = await queue.get()
        if data == '\x04': # Ctrl-D
            break
        msg = json.dumps(['stdin', data])
        await websocket.send(msg)

 def stdin_reader(queue):
    """Callback for asyncio reader to read raw input."""
    try:
        data = os.read(sys.stdin.fileno(), 1024)
        if data:
            text = data.decode('utf-8', errors='ignore')
            asyncio.create_task(queue.put(text))
    except Exception:
        pass

 async def interact_with_terminal(terminal_name, token, command=None, cookies=None):
    ws_base = BASE_URL.replace("https://", "wss://").replace("http://", "ws://")
    ws_url = f"{ws_base}/user/{USERNAME}/terminals/websocket/{terminal_name}"
    
    headers = {
        "Authorization": f"token {token}",
    }
    
    if cookies:
        cookie_parts = [f"{k}={v}" for k, v in cookies.items()]
        if cookie_parts:
            headers["Cookie"] = "; ".join(cookie_parts)

    ssl_context = ssl.create_default_context()
    
    try:
        async with websockets.connect(ws_url, additional_headers=headers, origin=BASE_URL, ssl=ssl_context) as websocket:
            
            if command:
                # One-shot mode
                receive_task = asyncio.create_task(receive_messages(websocket))
                msg = json.dumps(['stdin', command + '; exit\r'])
                await websocket.send(msg)
                try:
                    await receive_task
                except asyncio.CancelledError:
                    pass
            else:
                # Interactive Raw Mode
                fd = sys.stdin.fileno()
                old_settings = termios.tcgetattr(fd)
                input_queue = asyncio.Queue()
                loop = asyncio.get_running_loop()
                
                receive_task = asyncio.create_task(receive_messages(websocket))
                send_task = asyncio.create_task(send_input_from_queue(websocket, input_queue))
                
                try:
                    tty.setraw(fd)
                    loop.add_reader(sys.stdin, stdin_reader, input_queue)
                    
                    done, pending = await asyncio.wait(
                        [send_task, receive_task],
                        return_when=asyncio.FIRST_COMPLETED
                    )
                    
                    for task in pending:
                        task.cancel()
                        try:
                            await task
                        except asyncio.CancelledError:
                            pass
                            
                finally:
                    loop.remove_reader(sys.stdin)
                    termios.tcsetattr(fd, termios.TCSADRAIN, old_settings)
                    print(f"\r\nDisconnected from terminal {terminal_name}.")

    except Exception as e:
        print(f"Connection failed: {e}")

 def create_terminal(token):
    """Creates a new terminal and returns its name."""
    create_url = f"{BASE_URL}/user/{USERNAME}/api/terminals"
    headers = {"Authorization": f"token {token}"}
    try:
        response = requests.post(create_url, headers=headers)
        response.raise_for_status()
        return response.json()["name"]
    except Exception as e:
        print(f"Failed to create terminal: {e}")
        sys.exit(1)

 def list_terminals(token):
    """Lists active terminals."""
    url = f"{BASE_URL}/user/{USERNAME}/api/terminals"
    headers = {"Authorization": f"token {token}"}
    try:
        response = requests.get(url, headers=headers)
        response.raise_for_status()
        return response.json() # Returns list of dicts: [{'name': '1'}, ...]
    except Exception as e:
        print(f"Failed to list terminals: {e}")
        sys.exit(1)

 def run_tmux_orchestrator(token):
    """Orchestrates the tmux session."""
    if not shutil.which("tmux"):
        print("Error: 'tmux' is not installed.")
        sys.exit(1)

    # Use a dedicated socket to isolate configuration (key bindings)
    # and prevent interference with user's standard tmux server.
    TMUX_CMD = ["tmux", "-L", "jupyterhub_cli"]

    terminals = list_terminals(token)
    
    if not terminals:
        print("No active terminals found. Creating one...")
        create_terminal(token)
        terminals = list_terminals(token)

    session_name = "jupyterhub_terminals"
    script_path = os.path.abspath(__file__)
    python_exe = sys.executable
    
    # Command to run in windows. 
    env_cmd = f"env JUPYTERHUB_API_TOKEN='{token}'"
    
    # 1. Check if session exists
    check = subprocess.run(TMUX_CMD + ["has-session", "-t", session_name], capture_output=True)
    session_exists = (check.returncode == 0)

    if not session_exists:
        first_term = terminals[0]['name']
        cmd = f"{env_cmd} '{python_exe}' '{script_path}' --terminal-id {first_term}"
        
        print(f"Starting tmux session '{session_name}' with terminal {first_term}...")
        
        # Create new session
        subprocess.run(TMUX_CMD + [
            "new-session", "-d", "-s", session_name, "-n", f"term_{first_term}", cmd
        ], check=True)

        # Add windows for remaining existing terminals
        for term in terminals[1:]:
            t_name = term['name']
            cmd = f"{env_cmd} '{python_exe}' '{script_path}' --terminal-id {t_name}"
            subprocess.run(TMUX_CMD + [
                "new-window", "-t", session_name, "-n", f"term_{t_name}", cmd
            ], check=True)
            print(f"Added window for terminal {t_name}")

    else:
        print(f"Session '{session_name}' already exists.")

    # 2. Configure Session (Run always to update token/bindings)
    # Set the token in the session environment
    subprocess.run(TMUX_CMD + ["set-environment", "-t", session_name, "JUPYTERHUB_API_TOKEN", token], check=True)
    
    # Bind 'c' to create a new window running this script with --create-new
    # Note: bind-key is server-wide for this socket (-L jupyterhub_cli), so no -t needed.
    new_window_cmd = f"'{python_exe}' '{script_path}' --create-new"
    subprocess.run(TMUX_CMD + ["bind-key", "c", "new-window", new_window_cmd], check=True)

    # 3. Attach to session
    print("Attaching to tmux...")
    # execvp replaces the current process
    os.execvp("tmux", TMUX_CMD + ["attach", "-t", session_name])

 def main():
    parser = argparse.ArgumentParser(
        description="JupyterHub Terminal Client & Manager",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
 Operating Modes:
  1. Tmux Orchestrator (Default):
     If no arguments are provided and stdin is a TTY, the script fetches all 
     active terminals on the server and opens each in a separate window 
     within a new tmux session.
     
     **New Feature**: Inside the tmux session, press 'c' (or Ctrl-b c) to 
     create a NEW remote terminal and connect to it automatically.

  2. One-Shot Command:
     Pass a command as a positional argument (e.g., python3 terminal_jupyterhub.py "ls -la")
     or pipe a command into the script (e.g., echo "uptime" | python3 terminal_jupyterhub.py).
     This creates a temporary terminal, executes the command, and prints the output.

  3. Terminal Listing:
     Use --list to see all active terminal IDs and their last activity timestamps.

  4. Direct Connection:
     Use --terminal-id <id> to connect interactively to a specific existing terminal.

 Authentication:
  The script looks for a JupyterHub API Token in:
  1. Environment variable: JUPYTERHUB_API_TOKEN
  2. System Keyring: service 'login2', key 'JUPYTERHUB_API_TOKEN'
  3. Interactive prompt (fallback)
 """
    )
    parser.add_argument("--terminal-id", help="Connect to a specific existing terminal ID (interactive)")
    parser.add_argument("--create-new", action="store_true", help="Create a new terminal and connect to it (used by tmux)")
    parser.add_argument("--list", action="store_true", help="List all active terminals on the server")
    parser.add_argument("command", nargs="?", help="Command to execute in one-shot mode (can also be piped via stdin)")
    args = parser.parse_args()

    token = get_api_token()
    if not token:
        return

    command = args.command
    # If no command arg, check if we're receiving a command via pipe
    if not command and not sys.stdin.isatty():
        piped_data = sys.stdin.read().strip()
        if piped_data:
            command = piped_data

    if args.list:
        terminals = list_terminals(token)
        print(f"{'Terminal ID':<15} | {'Last Activity'}")
        print("-" * 45)
        if not terminals:
            print("No active terminals found.")
        for t in terminals:
            # Jupyter API typically returns 'name' and 'last_activity'
            last_act = t.get('last_activity', 'N/A')
            print(f"{t['name']:<15} | {last_act}")
        return

    if args.create_new:
        # CREATE NEW MODE (Tmux 'c' binding)
        try:
            t_name = create_terminal(token)
            print(f"Created new terminal: {t_name}")
            
            # If running inside tmux, rename the window
            if os.environ.get("TMUX"):
                subprocess.run(["tmux", "rename-window", f"term_{t_name}"])
                
            asyncio.run(interact_with_terminal(t_name, token))
        except KeyboardInterrupt:
            pass
        return

    if args.terminal_id:
        # WORKER MODE: Connect to existing
        try:
            asyncio.run(interact_with_terminal(args.terminal_id, token))
        except KeyboardInterrupt:
            pass
            
    elif command:
        # ONE-SHOT MODE: Create new, run, exit (triggered by arg or pipe)
        t_name = create_terminal(token)
        if not args.command: # If it was piped, we don't want to mess up stdout too much
             sys.stderr.write(f"Created temporary terminal {t_name} for piped command.\n")
        else:
             print(f"Created temporary terminal {t_name} for command.")
             
        try:
            asyncio.run(interact_with_terminal(t_name, token, command))
        except KeyboardInterrupt:
            pass
            
    else:
        # ORCHESTRATOR MODE: Tmux
        run_tmux_orchestrator(token)

 if __name__ == "__main__":
    main()
	#!/usr/bin/env python3
	"""
	JupyterLab Terminal Tmux Manager.

	This script provides multiple ways to interact with JupyterLab terminals:
	1. Tmux Orchestrator: Launches a tmux session where each window is connected to
	an active JupyterHub terminal. Inside tmux, 'c' creates a new remote terminal.
	2. One-shot Execution: Creates a temporary terminal to run a single command
	(passed as an argument or piped) and prints the output.
	3. Terminal Listing: Lists all active terminals and their last activity.
	4. Direct Connection: Connects interactively to a specific terminal by ID.

	Authentication is handled via the JUPYTERHUB_API_TOKEN environment variable,
	system keyring, or an interactive prompt.

	URL: https://gist.github.com/monperrus/6c2d28dfa8b8519ddf05538f73015b09
	"""

	import requests
	import json
	import sys
	import getpass
	import asyncio
	import websockets
	import ssl
	import keyring
	import tty
	import termios
	import os
	import argparse
	import subprocess
	import shutil

	# Configuration
	BASE_URL = "https://gpu1.eecs.kth.se"
	USERNAME = "monp"

	def get_api_token():
	"""Retrieve token from Env, Keyring, or Prompt."""
	# 1. Environment Variable
	token = os.environ.get("JUPYTERHUB_API_TOKEN")
	if token:
	return token

	# 2. Keyring
	try:
	login_keyring = keyring.get_keyring()
	token = login_keyring.get_password('login2', 'JUPYTERHUB_API_TOKEN')
	if token:
	return token
	except Exception:
	pass

	# 3. Prompt
	sys.stderr.write("Token not found in Env or Keyring.\n")
	token = getpass.getpass("Enter your JupyterHub API Token: ").strip()
	return token

	async def receive_messages(websocket):
	"""Continuously receive and print messages from the WebSocket."""
	try:
	while True:
	message = await websocket.recv()
	data = json.loads(message)

	# Protocol: ['stdout', 'output_string']
	if data[0] == 'stdout':
	sys.stdout.write(data[1])
	sys.stdout.flush()
	except websockets.exceptions.ConnectionClosed:
	pass
	except asyncio.CancelledError:
	pass
	except Exception as e:
	if not isinstance(e, asyncio.CancelledError):
	sys.stderr.write(f"\r\nError receiving data: {e}\r\n")

	async def send_input_from_queue(websocket, queue):
	"""Read from queue and send to WebSocket."""
	while True:
	data = await queue.get()
	if data == '\x04': # Ctrl-D
	break
	msg = json.dumps(['stdin', data])
	await websocket.send(msg)

	def stdin_reader(queue):
	"""Callback for asyncio reader to read raw input."""
	try:
	data = os.read(sys.stdin.fileno(), 1024)
	if data:
	text = data.decode('utf-8', errors='ignore')
	asyncio.create_task(queue.put(text))
	except Exception:
	pass

	async def interact_with_terminal(terminal_name, token, command=None, cookies=None):
	ws_base = BASE_URL.replace("https://", "wss://").replace("http://", "ws://")
	ws_url = f"{ws_base}/user/{USERNAME}/terminals/websocket/{terminal_name}"

	headers = {
	"Authorization": f"token {token}",
	}

	if cookies:
	cookie_parts = [f"{k}={v}" for k, v in cookies.items()]
	if cookie_parts:
	headers["Cookie"] = "; ".join(cookie_parts)

	ssl_context = ssl.create_default_context()

	try:
	async with websockets.connect(ws_url, additional_headers=headers, origin=BASE_URL, ssl=ssl_context) as websocket:

	if command:
	# One-shot mode
	receive_task = asyncio.create_task(receive_messages(websocket))
	msg = json.dumps(['stdin', command + '; exit\r'])
	await websocket.send(msg)
	try:
	await receive_task
	except asyncio.CancelledError:
	pass
	else:
	# Interactive Raw Mode
	fd = sys.stdin.fileno()
	old_settings = termios.tcgetattr(fd)
	input_queue = asyncio.Queue()
	loop = asyncio.get_running_loop()

	receive_task = asyncio.create_task(receive_messages(websocket))
	send_task = asyncio.create_task(send_input_from_queue(websocket, input_queue))

	try:
	tty.setraw(fd)
	loop.add_reader(sys.stdin, stdin_reader, input_queue)

	done, pending = await asyncio.wait(
	[send_task, receive_task],
	return_when=asyncio.FIRST_COMPLETED
	)

	for task in pending:
	task.cancel()
	try:
	await task
	except asyncio.CancelledError:
	pass

	finally:
	loop.remove_reader(sys.stdin)
	termios.tcsetattr(fd, termios.TCSADRAIN, old_settings)
	print(f"\r\nDisconnected from terminal {terminal_name}.")

	except Exception as e:
	print(f"Connection failed: {e}")

	def create_terminal(token):
	"""Creates a new terminal and returns its name."""
	create_url = f"{BASE_URL}/user/{USERNAME}/api/terminals"
	headers = {"Authorization": f"token {token}"}
	try:
	response = requests.post(create_url, headers=headers)
	response.raise_for_status()
	return response.json()["name"]
	except Exception as e:
	print(f"Failed to create terminal: {e}")
	sys.exit(1)

	def list_terminals(token):
	"""Lists active terminals."""
	url = f"{BASE_URL}/user/{USERNAME}/api/terminals"
	headers = {"Authorization": f"token {token}"}
	try:
	response = requests.get(url, headers=headers)
	response.raise_for_status()
	return response.json() # Returns list of dicts: [{'name': '1'}, ...]
	except Exception as e:
	print(f"Failed to list terminals: {e}")
	sys.exit(1)

	def run_tmux_orchestrator(token):
	"""Orchestrates the tmux session."""
	if not shutil.which("tmux"):
	print("Error: 'tmux' is not installed.")
	sys.exit(1)

	# Use a dedicated socket to isolate configuration (key bindings)
	# and prevent interference with user's standard tmux server.
	TMUX_CMD = ["tmux", "-L", "jupyterhub_cli"]

	terminals = list_terminals(token)

	if not terminals:
	print("No active terminals found. Creating one...")
	create_terminal(token)
	terminals = list_terminals(token)

	session_name = "jupyterhub_terminals"
	script_path = os.path.abspath(__file__)
	python_exe = sys.executable

	# Command to run in windows.
	env_cmd = f"env JUPYTERHUB_API_TOKEN='{token}'"

	# 1. Check if session exists
	check = subprocess.run(TMUX_CMD + ["has-session", "-t", session_name], capture_output=True)
	session_exists = (check.returncode == 0)

	if not session_exists:
	first_term = terminals[0]['name']
	cmd = f"{env_cmd} '{python_exe}' '{script_path}' --terminal-id {first_term}"

	print(f"Starting tmux session '{session_name}' with terminal {first_term}...")

	# Create new session
	subprocess.run(TMUX_CMD + [
	"new-session", "-d", "-s", session_name, "-n", f"term_{first_term}", cmd
	], check=True)

	# Add windows for remaining existing terminals
	for term in terminals[1:]:
	t_name = term['name']
	cmd = f"{env_cmd} '{python_exe}' '{script_path}' --terminal-id {t_name}"
	subprocess.run(TMUX_CMD + [
	"new-window", "-t", session_name, "-n", f"term_{t_name}", cmd
	], check=True)
	print(f"Added window for terminal {t_name}")

	else:
	print(f"Session '{session_name}' already exists.")

	# 2. Configure Session (Run always to update token/bindings)
	# Set the token in the session environment
	subprocess.run(TMUX_CMD + ["set-environment", "-t", session_name, "JUPYTERHUB_API_TOKEN", token], check=True)

	# Bind 'c' to create a new window running this script with --create-new
	# Note: bind-key is server-wide for this socket (-L jupyterhub_cli), so no -t needed.
	new_window_cmd = f"'{python_exe}' '{script_path}' --create-new"
	subprocess.run(TMUX_CMD + ["bind-key", "c", "new-window", new_window_cmd], check=True)

	# 3. Attach to session
	print("Attaching to tmux...")
	# execvp replaces the current process
	os.execvp("tmux", TMUX_CMD + ["attach", "-t", session_name])

	def main():
	parser = argparse.ArgumentParser(
	description="JupyterHub Terminal Client & Manager",
	formatter_class=argparse.RawDescriptionHelpFormatter,
	epilog="""
	Operating Modes:
	1. Tmux Orchestrator (Default):
	If no arguments are provided and stdin is a TTY, the script fetches all
	active terminals on the server and opens each in a separate window
	within a new tmux session.

	New Feature: Inside the tmux session, press 'c' (or Ctrl-b c) to
	create a NEW remote terminal and connect to it automatically.

	2. One-Shot Command:
	Pass a command as a positional argument (e.g., python3 terminal_jupyterhub.py "ls -la")
	or pipe a command into the script (e.g., echo "uptime" \| python3 terminal_jupyterhub.py).
	This creates a temporary terminal, executes the command, and prints the output.

	3. Terminal Listing:
	Use --list to see all active terminal IDs and their last activity timestamps.

	4. Direct Connection:
	Use --terminal-id <id> to connect interactively to a specific existing terminal.

	Authentication:
	The script looks for a JupyterHub API Token in:
	1. Environment variable: JUPYTERHUB_API_TOKEN
	2. System Keyring: service 'login2', key 'JUPYTERHUB_API_TOKEN'
	3. Interactive prompt (fallback)
	"""
	)
	parser.add_argument("--terminal-id", help="Connect to a specific existing terminal ID (interactive)")
	parser.add_argument("--create-new", action="store_true", help="Create a new terminal and connect to it (used by tmux)")
	parser.add_argument("--list", action="store_true", help="List all active terminals on the server")
	parser.add_argument("command", nargs="?", help="Command to execute in one-shot mode (can also be piped via stdin)")
	args = parser.parse_args()

	token = get_api_token()
	if not token:
	return

	command = args.command
	# If no command arg, check if we're receiving a command via pipe
	if not command and not sys.stdin.isatty():
	piped_data = sys.stdin.read().strip()
	if piped_data:
	command = piped_data

	if args.list:
	terminals = list_terminals(token)
	print(f"{'Terminal ID':<15} \| {'Last Activity'}")
	print("-" * 45)
	if not terminals:
	print("No active terminals found.")
	for t in terminals:
	# Jupyter API typically returns 'name' and 'last_activity'
	last_act = t.get('last_activity', 'N/A')
	print(f"{t['name']:<15} \| {last_act}")
	return

	if args.create_new:
	# CREATE NEW MODE (Tmux 'c' binding)
	try:
	t_name = create_terminal(token)
	print(f"Created new terminal: {t_name}")

	# If running inside tmux, rename the window
	if os.environ.get("TMUX"):
	subprocess.run(["tmux", "rename-window", f"term_{t_name}"])

	asyncio.run(interact_with_terminal(t_name, token))
	except KeyboardInterrupt:
	pass
	return

	if args.terminal_id:
	# WORKER MODE: Connect to existing
	try:
	asyncio.run(interact_with_terminal(args.terminal_id, token))
	except KeyboardInterrupt:
	pass

	elif command:
	# ONE-SHOT MODE: Create new, run, exit (triggered by arg or pipe)
	t_name = create_terminal(token)
	if not args.command: # If it was piped, we don't want to mess up stdout too much
	sys.stderr.write(f"Created temporary terminal {t_name} for piped command.\n")
	else:
	print(f"Created temporary terminal {t_name} for command.")

	try:
	asyncio.run(interact_with_terminal(t_name, token, command))
	except KeyboardInterrupt:
	pass

	else:
	# ORCHESTRATOR MODE: Tmux
	run_tmux_orchestrator(token)

	if __name__ == "__main__":
	main()
No results found