Using Gemini CLI to automatically inspect server clusters: Python + SSH + Telegram push hands-on guide

npm install -g @google/gemini-cli
gemini auth   # Sign in with your Google account

echo "hello" | gemini

ssh-keygen -t ed25519 -f ~/.ssh/audit_key

ssh-copy-id -i ~/.ssh/audit_key.pub user@server-ip

#!/usr/bin/env python3
"""
Daily server cluster audit script
- SSH to each server, collect metrics
- Gemini CLI analyzes the report
- Telegram sends the daily report
"""

import subprocess
import json
import requests
from datetime import datetime

# === Config ===
SERVERS = [
    {"name": "aws-hk",  "host": "your-hk-server-ip", "user": "ubuntu"},
    {"name": "aws-us",  "host": "your-us-server-ip", "user": "ubuntu"},
    {"name": "home-1",  "host": "192.168.1.10",   "user": "as"},
]

SSH_KEY = "/root/.ssh/audit_key"   # or your key path
TG_TOKEN = "your-telegram-bot-token"
TG_CHAT_ID = "your-chat-id"
# ==============


def ssh_run(host: str, user: str, cmd: str) -> str:
    """Execute command on remote host, return output"""
    result = subprocess.run(
        ["ssh", "-i", SSH_KEY, "-o", "StrictHostKeyChecking=no",
         "-o", "ConnectTimeout=10", f"{user}@{host}", cmd],
        capture_output=True, text=True, timeout=30
    )
    return result.stdout.strip() if result.returncode == 0 else f"[ERROR] {result.stderr.strip()}"


def collect_metrics(server: dict) -> dict:
    """Collect metrics from a single server"""
    name = server["name"]
    host = server["host"]
    user = server["user"]

    print(f"  Collecting from {name} ({host})...")

    metrics = {
        "name": name,
        "host": host,
        "timestamp": datetime.now().isoformat(),
    }

    # CPU load (1-minute average)
    metrics["cpu_load"] = ssh_run(host, user,
        "cat /proc/loadavg | awk '{print $1, $2, $3}'")

    # Memory usage
    metrics["memory"] = ssh_run(host, user,
        "free -h | grep Mem | awk '{print \"total:\", $2, \"used:\", $3, \"free:\", $4}'")

    # Disk usage (root partition)
    metrics["disk"] = ssh_run(host, user,
        "df -h / | tail -1 | awk '{print \"total:\", $2, \"used:\", $3, \"avail:\", $4, \"use%:\", $5}'")

    # System uptime
    metrics["uptime"] = ssh_run(host, user, "uptime -p")

    # Service status (customize per your services)
    services = ["nginx", "xray", "hysteria-server"]
    service_status = {}
    for svc in services:
        status = ssh_run(host, user,
            f"systemctl is-active {svc} 2>/dev/null || echo 'not-found'")
        service_status[svc] = status
    metrics["services"] = service_status

    # Recent system errors (past hour)
    metrics["recent_errors"] = ssh_run(host, user,
        "journalctl -p err --since '1 hour ago' --no-pager -q 2>/dev/null | tail -20")

    # Network connections
    metrics["connections"] = ssh_run(host, user,
        "ss -s | grep 'TCP:' | head -3")

    return metrics


def format_report(all_metrics: list) -> str:
    """Format collected data into a text report"""
    lines = []
    lines.append(f"=== Cluster Audit Report {datetime.now().strftime('%Y-%m-%d %H:%M')} ===\n")

    for m in all_metrics:
        lines.append(f"[{m['name']}] {m['host']}")
        lines.append(f"  Load: {m.get('cpu_load', 'N/A')}")
        lines.append(f"  Memory: {m.get('memory', 'N/A')}")
        lines.append(f"  Disk: {m.get('disk', 'N/A')}")
        lines.append(f"  Uptime: {m.get('uptime', 'N/A')}")

        services = m.get('services', {})
        for svc, status in services.items():
            icon = "✓" if status == "active" else "✗"
            lines.append(f"  {icon} {svc}: {status}")

        errors = m.get('recent_errors', '')
        if errors and '[ERROR]' not in errors and len(errors) > 5:
            lines.append(f"  Recent errors:")
            for line in errors.split('\n')[:5]:
                lines.append(f"    {line}")

        lines.append("")

    return '\n'.join(lines)


def gemini_analyze(report: str) -> str:
    """Call Gemini CLI to analyze the report"""
    prompt = f"""Here's a server cluster audit report. Please:
1. Identify any anomalies or concerns
2. Summarize service status briefly
3. Flag any disk usage > 80% or inactive services as warnings
4. Give short recommendations if issues exist
5. If everything's fine, say "Cluster status normal"

Keep it under 200 words.

{report}"""

    result = subprocess.run(
        ["gemini", "--model", "gemini-2.5-flash", "-p", prompt],
        capture_output=True, text=True, timeout=60
    )

    if result.returncode == 0:
        return result.stdout.strip()
    else:
        return f"[Gemini analysis failed] {result.stderr[:200]}"


def send_telegram(message: str):
    """Send Telegram message"""
    url = f"https://api.telegram.org/bot{TG_TOKEN}/sendMessage"

    # Telegram message limit 4096 chars
    if len(message) > 4000:
        message = message[:4000] + "\n...[truncated]"

    resp = requests.post(url, json={
        "chat_id": TG_CHAT_ID,
        "text": message,
        "parse_mode": "HTML"
    }, timeout=15)

    if resp.status_code == 200:
        print("  Telegram sent successfully")
    else:
        print(f"  Telegram failed: {resp.text}")


def main():
    print(f"Starting audit of {len(SERVERS)} servers...")

    # Collect metrics from all servers
    all_metrics = []
    for server in SERVERS:
        try:
            metrics = collect_metrics(server)
            all_metrics.append(metrics)
        except Exception as e:
            print(f"  Failed to collect from {server['name']}: {e}")
            all_metrics.append({
                "name": server["name"],
                "host": server["host"],
                "error": str(e)
            })

    # Generate text report
    report = format_report(all_metrics)
    print("\n--- Raw Report ---")
    print(report)

    # Gemini analysis
    print("\nCalling Gemini for analysis...")
    analysis = gemini_analyze(report)
    print(f"Analysis: {analysis}")

    # Compose Telegram message
    tg_message = f"<b>Daily Cluster Audit</b>\n\n{analysis}\n\n<pre>{report[:2000]}</pre>"
    send_telegram(tg_message)

    print("\nAudit complete")


if __name__ == "__main__":
    main()

crontab -e

# Check Docker container status
metrics["docker"] = ssh_run(host, user,
    "docker ps --format '{{.Names}}: {{.Status}}' 2>/dev/null | head -10")

# Check if specific ports are listening
metrics["ports"] = ssh_run(host, user,
    "ss -tlnp | grep -E ':(443|80|8080|9090)' | awk '{print $4, $6}'")

# Recent nginx errors
metrics["nginx_errors"] = ssh_run(host, user,
    "tail -20 /var/log/nginx/error.log 2>/dev/null | grep -v '^$'")

# Disk inode usage (often overlooked)
metrics["inode"] = ssh_run(host, user,
    "df -i / | tail -1 | awk '{print \"inode use%:\", $5}'")

prompt = f"""Today's report:\n{today_report}\n\nYesterday's report:\n{yesterday_report}\n\nCompare the changes..."""

prompt = f"""...Pay special attention to disk usage approaching capacity and whether xray/hysteria services are online..."""

prompt = f"""...Return as JSON with fields: status (ok/warning/critical), issues (list), suggestions (list)..."""