Terminal Service

from cmdop import AsyncCMDOPClient
 
# Connect to remote agent using async context manager (auto-closes on exit)
async with AsyncCMDOPClient.remote(api_key="cmd_xxx") as client:
    # Select the target machine for terminal commands
    await client.terminal.set_machine("my-server")
 
    # Execute a one-shot command and capture output + exit code
    output, exit_code = await client.terminal.execute("ls -la")
 
    print(output)                       # Command stdout/stderr as string
    print(f"Exit code: {exit_code}")    # 0 = success, non-zero = error

# Execute with custom options for timeout, directory, env, and shell
output, code = await client.terminal.execute(
    command="./script.sh",
    timeout=300,              # Max execution time in seconds (default: 30)
    cwd="/app",               # Working directory for the command
    env={"KEY": "value"},     # Additional environment variables to set
    shell="/bin/bash",        # Shell to use (default: system shell)
)

# Retrieve the active terminal session for the target machine
session = await client.terminal.get_active_session("my-server")
 
# Create a bidirectional stream for interactive I/O
stream = client.terminal.stream()
 
# Register a callback that fires on every chunk of terminal output
stream.on_output(lambda data: print(data.decode(), end=""))
 
# Connect the stream to the session (begins receiving output)
await stream.attach(session.session_id)

# Send a command (include \n to press Enter)
await stream.send_input(b"ls -la\n")
 
# Send special keys as raw bytes
await stream.send_input(b"\x03")   # Ctrl+C — interrupt running process
await stream.send_input(b"\t")     # Tab — trigger autocomplete
await stream.send_input(b"\x1b[A") # Up arrow — recall previous command

# Update terminal dimensions (triggers SIGWINCH on the remote process)
await stream.resize(rows=24, cols=80)

await stream.send_signal("SIGINT")   # Interrupt (like Ctrl+C)
await stream.send_signal("SIGTERM")  # Graceful termination request
await stream.send_signal("SIGKILL")  # Force kill (cannot be caught)

await stream.close()  # Detach without terminating session

stream = client.terminal.stream()
 
# Fires when terminal output data is received (bytes)
stream.on_output(lambda data: print(data.decode(), end=""))
 
# Fires when the stream connection is lost
stream.on_disconnect(lambda: print("Disconnected"))
 
# Fires when the stream automatically reconnects
stream.on_reconnect(lambda: print("Reconnected"))
 
# Fires on stream errors (connection issues, protocol errors)
stream.on_error(lambda e: print(f"Error: {e}"))

# Create a read-only stream — can watch but not type
stream = client.terminal.stream(mode="observer")
 
# Observers can still receive all terminal output
stream.on_output(handle_output)
 
# Attach to an existing session in read-only mode
await stream.attach(session.session_id)
 
# Attempting to send input as an observer raises PermissionError
# await stream.send_input(b"command")  # PermissionError

# Get the current active session for a specific machine
session = await client.terminal.get_active_session("my-server")
 
print(f"Session ID: {session.session_id}")       # Unique session UUID
print(f"Status: {session.status}")               # Connection status
print(f"Machine: {session.machine_hostname}")     # Target machine name
print(f"Created: {session.created_at}")           # Session creation timestamp

# List all terminal sessions across all machines
response = await client.terminal.list_sessions()
 
for session in response.sessions:
    print(f"{session.machine_hostname}: {session.status}")  # Machine name and status

# Fetch a specific session by its unique ID
session = await client.terminal.get_session(session_id="sess_abc123")

session = await client.terminal.get_active_session("my-server")
 
# All available session properties
session.session_id          # Unique UUID for this session
session.status              # PENDING, CONNECTED, GRACE_PERIOD, or DISCONNECTED
session.machine_hostname    # Hostname of the connected machine
session.workspace_id        # Workspace this session belongs to
session.created_at          # When the session was created
session.connected_at        # When the session first connected
session.last_heartbeat_at   # Last keepalive timestamp
session.attached_clients    # List of currently attached client streams

# Client A creates a stream and attaches to the session
stream_a = client.terminal.stream()
await stream_a.attach(session.session_id)
 
# Client B attaches to the same session (collaborative mode)
stream_b = client.terminal.stream()
await stream_b.attach(session.session_id)
 
# Both clients receive identical terminal output in real time
# Both can send input if they have operator permissions

import asyncio
 
# Wait until a specific prompt string appears in terminal output
async def wait_for_prompt(stream, prompt="$"):
    found = asyncio.Event()       # Event to signal when prompt is detected
    buffer = []                   # Accumulate output chunks
 
    def check(data):
        buffer.append(data)
        # Check if the prompt pattern appears in accumulated output
        if prompt.encode() in b"".join(buffer):
            found.set()           # Signal that prompt was found
 
    stream.on_output(check)       # Register the check callback
 
    # Wait up to 30 seconds for the prompt to appear
    await asyncio.wait_for(found.wait(), timeout=30)
    return b"".join(buffer).decode()  # Return all output as a string

# In SSH session
ccat main.py           # Python with colors
ccat config.yaml       # YAML with colors
ccat -n main.go        # With line numbers
ccat -t dracula app.js # Different theme

ccat <file>              # Auto-detect language
ccat -n <file>           # Show line numbers
ccat -l python <file>    # Force language
ccat -t dracula <file>   # Use different theme (default: monokai)
ccat --list-languages    # Show supported languages
ccat --list-themes       # Show available themes

NO_COLOR=1 ccat file.py  # Plain output (for piping)

# Sequential — each command runs in its own shell invocation
output1, _ = await client.terminal.execute("cd /app")
output2, _ = await client.terminal.execute("npm install")
output3, _ = await client.terminal.execute("npm run build")
 
# Chained — all commands run in a single shell (shares working directory)
output, code = await client.terminal.execute(
    "cd /app && npm install && npm run build"  # Stops on first failure
)

# Execute and check the exit code to detect failures
output, code = await client.terminal.execute("./deploy.sh")
 
if code != 0:  # Non-zero exit code indicates an error
    raise Exception(f"Deploy failed: {output}")

# Start a long-running process in the background with &
await stream.send_input(b"./long_task.sh &\n")
 
# Retrieve the PID of the last backgrounded process
await stream.send_input(b"echo $!\n")

from cmdop.exceptions import (
    SessionNotFoundError,   # No active session found for the machine
    PermissionError,        # Insufficient permissions (e.g., observer mode)
    TimeoutError            # Command exceeded its timeout limit
)
 
try:
    output, code = await client.terminal.execute("command")
except SessionNotFoundError:
    print("No active session for this machine")  # Machine not connected
except PermissionError:
    print("Not authorized")                      # Missing required permissions
except TimeoutError:
    print("Command timed out")                   # Exceeded timeout_seconds