Signals & Resize

# Send SIGINT to interrupt/cancel the currently running command
await stream.send_signal("SIGINT")
 
# Alternative: send the raw Ctrl+C byte directly to the PTY
await stream.send_input(b"\x03")

# Send SIGTERM for graceful termination (process can clean up before exiting)
await stream.send_signal("SIGTERM")

# Send SIGKILL to force-kill immediately (cannot be caught or ignored)
await stream.send_signal("SIGKILL")

# Detect the current local terminal dimensions
import shutil
cols, rows = shutil.get_terminal_size()
 
# Send the new dimensions to the remote PTY so it reflows output correctly
await stream.resize(rows=rows, cols=cols)

import signal
import shutil
 
def handle_resize(signum, frame):
    # Called automatically when the local terminal window is resized (SIGWINCH)
    cols, rows = shutil.get_terminal_size()
    # Schedule the async resize call from within the sync signal handler
    asyncio.create_task(stream.resize(rows, cols))
 
# Register the SIGWINCH handler to detect terminal resize events
signal.signal(signal.SIGWINCH, handle_resize)

# Start a long-running process in the foreground
await stream.send_input(b"./long_task.sh\n")
 
# Suspend the process with Ctrl+Z (sends SIGTSTP)
await stream.send_input(b"\x1a")
 
# Move the suspended process to the background so it continues running
await stream.send_input(b"bg\n")
 
# The shell prompt returns; you can run other commands now
await stream.send_input(b"ls -la\n")

# Bring the most recently backgrounded process back to the foreground
await stream.send_input(b"fg\n")

# Show all background and suspended jobs in the current shell session
await stream.send_input(b"jobs\n")

import asyncio
 
async def run_with_timeout(command: str, timeout: int):
    # Send the command to the remote terminal
    await stream.send_input(f"{command}\n".encode())
 
    try:
        # Wait for the shell prompt to reappear (indicates command finished)
        await asyncio.wait_for(
            wait_for_prompt(),
            timeout=timeout
        )
    except asyncio.TimeoutError:
        # Force-kill the process if it exceeds the timeout duration
        await stream.send_signal("SIGKILL")
        raise TimeoutError(f"Command timed out after {timeout}s")

# Launch vim to edit a file on the remote machine
await stream.send_input(b"vim file.txt\n")
 
# Enter insert mode to start typing
await stream.send_input(b"i")
 
# Type content into the file
await stream.send_input(b"Hello, world!")
 
# Press ESC to exit insert mode and return to normal mode
await stream.send_input(b"\x1b")
 
# Save the file and quit vim
await stream.send_input(b":wq\n")

# Open a large file in the less pager
await stream.send_input(b"less large_file.txt\n")
 
# Navigate through the file
await stream.send_input(b" ")  # Space: scroll down one page
await stream.send_input(b"b")  # b: scroll up one page
await stream.send_input(b"G")  # G: jump to end of file
await stream.send_input(b"g")  # g: jump to beginning of file
 
# Exit the pager
await stream.send_input(b"q")

# Launch htop (interactive process viewer)
await stream.send_input(b"htop\n")
 
# Navigate the process list with arrow keys
await stream.send_input(b"\x1b[A")  # Up arrow: select previous process
await stream.send_input(b"\x1b[B")  # Down arrow: select next process
 
# Send F9 to open the kill menu for the selected process
await stream.send_input(b"\x1b[20~")
 
# Send F10 to quit htop
await stream.send_input(b"\x1b[21~")

import asyncio
import signal
import shutil
import sys
import tty
import termios
 
async def interactive_session():
    async with AsyncCMDOPClient.remote(api_key="cmd_xxx") as client:
        session = await client.terminal.get_active_session("my-server")
        stream = client.terminal.stream()
 
        # Write remote output directly to local stdout (raw bytes for accuracy)
        stream.on_output(lambda d: sys.stdout.buffer.write(d) or sys.stdout.flush())
 
        # Auto-resize: sync remote PTY dimensions when local terminal changes
        def on_resize(signum, frame):
            cols, rows = shutil.get_terminal_size()
            asyncio.create_task(stream.resize(rows, cols))
 
        # Listen for SIGWINCH (window change) events from the local terminal
        signal.signal(signal.SIGWINCH, on_resize)
 
        await stream.attach(session.session_id)
 
        # Set the remote PTY to match the current local terminal size
        cols, rows = shutil.get_terminal_size()
        await stream.resize(rows, cols)
 
        # Save terminal settings before switching to raw mode
        old_settings = termios.tcgetattr(sys.stdin)
 
        try:
            # Raw mode: forward every keypress immediately to the remote PTY
            tty.setraw(sys.stdin.fileno())
 
            while True:
                char = sys.stdin.read(1)
                if char == '\x11':  # Ctrl+Q to exit the local client
                    break
                await stream.send_input(char.encode())
 
        finally:
            # Restore original terminal settings and close the stream
            termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_settings)
            await stream.close()
 
asyncio.run(interactive_session())