grass-remote-cli

Remote command execution library for SSH-based cluster operations.

Features

• ClusterCmd: Execute commands on multiple servers concurrently
• SshCmd: Single-server SSH execution with retry logic
• StderrClassifier: Intelligent error classification (real errors vs innocent warnings)
• Notifications: Slack integration for alerts (optional)
• AnsibleInventory: Parse Ansible inventory files (optional)

Installation

# Core (SSH execution)
pip install grass-remote-cli

# With Ansible inventory support
pip install grass-remote-cli[inventory]

# With Slack notifications
pip install grass-remote-cli[notifications]

# Everything
pip install grass-remote-cli[all]

Quick Start

from grass_remote_cli import ClusterCmd

# Execute command on multiple servers
cluster = ClusterCmd(
    servers_ips=["<ip1>", "<ip2>", "<ip3>"],
    username="<ssh_user>",
    password="<ssh_password>",
    timeout_in_sec=300,
)
results, timed_out = cluster.execute("docker ps")

for ip, (_, stdout, stderr) in results:
    print(f"{ip}: {stdout}")

Single Server Execution

from grass_remote_cli import SshCmd

ssh = SshCmd()
ip, stdout, stderr = ssh.execute(
    host_ip="<server_ip>",
    username="<ssh_user>",
    password="<ssh_password>",
    commands="hostname && uptime"
)
print(stdout)

With Ansible Inventory

from grass_remote_cli import ClusterCmd
from grass_remote_cli.inventory import AnsibleInventory

# Load servers from Ansible inventory
inventory = AnsibleInventory("/path/to/ansible/inventories")
hosts = inventory.get_hosts_by_cluster("<cluster_name>")

# Deploy to cluster
ips = [host[1] for host in hosts]  # Extract IPs
cluster = ClusterCmd(ips, username="<ssh_user>", password="<ssh_password>")
results, _ = cluster.execute("docker pull <image>")

Error Classification

The library intelligently classifies stderr output:

from grass_remote_cli.stderr_classifier import StderrClassifier

classifier = StderrClassifier()
result = classifier.classify(stderr_output)

if result['is_error']:
    print(f"Real error: {result['summary']}")
else:
    print("Innocent warning, safe to ignore")

Slack Notifications

import os
from grass_remote_cli.notifications import post_to_slack

# Set webhook via environment variable
os.environ["SLACK_WEBHOOK_URL"] = "https://hooks.slack.com/services/..."

post_to_slack("Deployment complete!", channel="#deployments")

# Or pass webhook directly
post_to_slack(
    "Deployment complete!",
    channel="#deployments",
    webhook_url="https://hooks.slack.com/services/..."
)

Environment Variables

License

MIT