Interactive Sessions

from cmdop import AsyncCMDOPClient
import asyncio
 
async def main():
    async with AsyncCMDOPClient.remote(api_key="cmd_xxx") as client:
        # Retrieve the active terminal session for the specified machine
        session = await client.terminal.get_active_session("my-server")
 
        # Create a new stream object for bidirectional communication
        stream = client.terminal.stream()
 
        # Register a callback to print output as it arrives in real-time
        stream.on_output(lambda data: print(data.decode(), end="", flush=True))
 
        # Attach the stream to the remote session (starts receiving output)
        await stream.attach(session.session_id)
 
        # Send a command to the remote terminal as raw bytes
        await stream.send_input(b"ls -la\n")
 
        # Wait briefly for the output to arrive over the network
        await asyncio.sleep(2)
 
        # Detach from the session (the remote session continues running)
        await stream.close()
 
asyncio.run(main())

import sys
import tty
import termios
import asyncio
from cmdop import AsyncCMDOPClient
 
async def interactive_terminal():
    async with AsyncCMDOPClient.remote(api_key="cmd_xxx") as client:
        session = await client.terminal.get_active_session("my-server")
        stream = client.terminal.stream()
 
        # Write raw bytes directly to stdout for accurate terminal rendering
        stream.on_output(lambda data: sys.stdout.buffer.write(data) or sys.stdout.flush())
 
        await stream.attach(session.session_id)
 
        # Save current terminal settings so we can restore them later
        old_settings = termios.tcgetattr(sys.stdin)
 
        try:
            # Switch to raw mode: each keypress is sent immediately (no line buffering)
            tty.setraw(sys.stdin.fileno())
 
            # Read one character at a time and forward to the remote session
            while True:
                char = sys.stdin.read(1)
                if char == '\x03':  # Ctrl+C to exit the local client
                    break
                await stream.send_input(char.encode())
 
        finally:
            # Restore the original terminal settings on exit
            termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_settings)
            await stream.close()
 
asyncio.run(interactive_terminal())

# Launch vim to edit a file on the remote machine
await stream.send_input(b"vim file.txt\n")
 
# Send arrow key escape sequences for navigation
await stream.send_input(b"\x1b[A")  # Up arrow
await stream.send_input(b"\x1b[B")  # Down arrow
 
# Force-quit vim without saving
await stream.send_input(b":q!\n")

# Client A starts a long-running process and then disconnects
await stream.send_input(b"./long_running_task.sh\n")
await stream.close()  # Disconnect from the session (process keeps running)
 
# ... hours later ...
 
# Client B reconnects to the same session and sees ongoing output
session = await client.terminal.get_active_session("my-server")
stream = client.terminal.stream()
await stream.attach(session.session_id)
# Process still running! Output continues from where Client A left off.

# Client A attaches to the session as an operator (read + write)
stream_a = client.terminal.stream()
await stream_a.attach(session.session_id)
 
# Client B also attaches to the same session as an operator
stream_b = client.terminal.stream()
await stream_b.attach(session.session_id)
 
# Both see the same output in real-time
# Both can send input (shared terminal)

# Create a stream in observer mode (read-only, no input allowed)
stream = client.terminal.stream(mode="observer")
await stream.attach(session.session_id)
 
# Output is visible to the observer
stream.on_output(print_output)
 
# Attempting to send input in observer mode raises an error:
# await stream.send_input(b"command")  # PermissionError

stream = client.terminal.stream()
 
# Called whenever the remote session produces output
stream.on_output(lambda data: process_output(data))
 
# Called when the WebSocket connection to the server drops
stream.on_disconnect(lambda: print("Disconnected"))
 
# Called when the connection is automatically re-established
stream.on_reconnect(lambda: print("Reconnected"))
 
# Called when an error occurs during streaming
stream.on_error(lambda e: print(f"Error: {e}"))
 
await stream.attach(session.session_id)

async def monitor_logs():
    session = await client.terminal.get_active_session("prod-server")
    # Use observer mode since we only need to read log output
    stream = client.terminal.stream(mode="observer")
 
    async def on_output(data):
        text = data.decode()
        # Check each output chunk for error patterns and send alerts
        if "ERROR" in text:
            send_alert(text)
 
    stream.on_output(on_output)
    await stream.attach(session.session_id)
    # Start tailing the application log file continuously
    await stream.send_input(b"tail -f /var/log/app.log\n")
 
    # Block forever to keep receiving log output
    await asyncio.Event().wait()

# Launch the Python debugger on the remote machine
await stream.send_input(b"python -m pdb script.py\n")
 
# Send debugger commands interactively
await stream.send_input(b"n\n")  # next: execute the next line
await stream.send_input(b"s\n")  # step: step into a function call
await stream.send_input(b"p variable\n")  # print: inspect a variable's value
await stream.send_input(b"c\n")  # continue: resume execution until next breakpoint

# Developer A shares their session ID with Developer B
session = await client.terminal.get_active_session("dev-machine")
print(f"Share this: {session.session_id}")
 
# Developer B attaches to the shared session using the session ID
await other_stream.attach(shared_session_id)
# Both developers see the same terminal and can type commands!

try:
    await stream.attach(session.session_id)
    # ... work with the session ...
finally:
    # Always close the stream in a finally block to avoid resource leaks
    await stream.close()

# Register handlers to react to network interruptions automatically
stream.on_disconnect(handle_disconnect)
stream.on_reconnect(handle_reconnect)

# Use observer mode when you only need to watch (e.g., monitoring, logging)
stream = client.terminal.stream(mode="observer")