Building an Autonomous AI-Powered OpenClaw Agent for Server Monitoring, Incident Investigation, and Root Cause Analysis

groups:
- name: openclaw-alerts

  rules:

  - alert: HighCPUUsage
    expr: 100 - (avg by(instance)(rate(node_cpu_seconds_total{mode="idle"}[1m])) * 100) > 80
    for: 1m
    labels:
      severity: critical
    annotations:
      summary: "CPU usage above 80%"

  - alert: HighMemoryUsage
    expr: ((node_memory_MemTotal_bytes - node_memory_MemAvailable_bytes) / node_memory_MemTotal_bytes) * 100 > 80
    for: 1m
    labels:
      severity: warning
    annotations:
      summary: "Memory usage above 80%"

  - alert: HighDiskUsage
    expr: ((node_filesystem_size_bytes{mountpoint="/"} - node_filesystem_free_bytes{mountpoint="/"}) / node_filesystem_size_bytes{mountpoint="/"}) * 100 > 80
    for: 1m
    labels:
      severity: warning
    annotations:
      summary: "Disk usage above 80%"

root@openclaw:/opt/openclaw-ai# cat listener.py
from flask import Flask, request
from openai import OpenAI
import paramiko
import yagmail
import os


# ==========================================
# CONFIGURATION
# ==========================================


OPENAI_API_KEY = "sk-proj-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"


EMAIL_USER = "xxxxxxxxx@gmail.com"
EMAIL_APP_PASSWORD = "xxxx xxxx xxxx xxx"


SSH_USERNAME = "xxxxxxxxxxxxxx"







SSH_KEY="/root/.ssh/id_ed25519"




client = OpenAI(api_key=OPENAI_API_KEY)


app = Flask(__name__)




# ==========================================
# SSH COMMAND EXECUTION
# ==========================================


def run_command(server, command):


    ssh = paramiko.SSHClient()


    ssh.set_missing_host_key_policy(
        paramiko.AutoAddPolicy()
    )


    ssh.connect(
        hostname=server,
        username=SSH_USERNAME,
        key_filename=SSH_KEY
    )


    stdin, stdout, stderr = ssh.exec_command(command)


    output = stdout.read().decode()
    error = stderr.read().decode()


    ssh.close()


    if error:
        return f"ERROR:\n{error}"


    return output




# ==========================================
# AI DECIDES WHAT TO INVESTIGATE
# ==========================================


def get_investigation_commands(alertname):


    response = client.chat.completions.create(
        model="gpt-4.1",
        messages=[
            {
                "role": "system",
                "content": """


You are an expert Linux Site Reliability Engineer.


A monitoring alert has been triggered.


Your task is to autonomously investigate the incident.


Think exactly like an experienced Linux administrator.


Based on ONLY the alert name,
decide what Linux commands should be executed.


Requirements:


- Decide the investigation commands yourself.
- Do NOT ask for clarification.
- Do NOT explain anything.
- Return ONLY executable Linux commands.
- One command per line.
- Maximum 10 commands.
- Choose commands that help identify the root cause.
- Prefer standard Linux utilities that are commonly available.
- If an advanced utility (such as mpstat, iostat, sar, pidstat, ss, lsof, etc.) is required, use it only if it is likely to exist on the system; otherwise choose an appropriate alternative.
- Avoid destructive commands.
- Avoid commands requiring user interaction.
- Never modify the server.
- Do not install packages.
- Do not assume specific monitoring tools are installed.


Your goal is to collect enough evidence for a root cause analysis.


"""
            },
            {
                "role": "user",
                "content": alertname
            }
        ]
    )


    return response.choices[0].message.content




# ==========================================
# AI DRIVEN INVESTIGATION
# ==========================================


def investigate_server(server, alertname):


    commands_text = get_investigation_commands(
        alertname
    )


    print()
    print("=" * 50)
    print("AI SELECTED COMMANDS")
    print("=" * 50)
    print(commands_text)


    report = ""


    commands = commands_text.splitlines()


    for cmd in commands:


        cmd = cmd.strip()


        if not cmd:
            continue


        try:


            print(f"Running: {cmd}")


            output = run_command(
                server,
                cmd
            )


            report += f"\n\n===== {cmd} =====\n"
            report += output


        except Exception as e:


            report += f"\n\n===== {cmd} =====\n"
            report += f"ERROR: {str(e)}"


    return report




# ==========================================
# ROOT CAUSE ANALYSIS
# ==========================================


def analyze_report(report):


    response = client.chat.completions.create(
        model="gpt-4.1",
        messages=[
            {
                "role": "system",
                "content": """
You are a Senior Linux SRE.


Analyze the investigation report.


Provide:


1. Root Cause
2. Problematic Process
3. Severity
4. Suggested Commands
5. Recommended Action


Be detailed and practical.
"""
            },
            {
                "role": "user",
                "content": report
            }
        ]
    )


    return response.choices[0].message.content




# ==========================================
# EMAIL REPORT
# ==========================================


def send_email(server, alertname, report, analysis):


    print("========== ENTERED send_email() ==========")


    import smtplib
    from email.mime.multipart import MIMEMultipart
    from email.mime.text import MIMEText


    print("Creating MIME message...")


    msg = MIMEMultipart()


    msg["From"] = EMAIL_USER
    msg["To"] = EMAIL_USER
    msg["Subject"] = f"OpenClaw Alert - {alertname}"


    print("Building HTML...")


    html = f"""
<html>
<body>
<pre>
Server: {server}


Alert: {alertname}


{report}


{analysis}
</pre>
</body>
</html>
"""


    msg.attach(MIMEText(html, "html"))


    print("Connecting to Gmail SMTP...")


    smtp = smtplib.SMTP("smtp.gmail.com", 587)


    print("Starting TLS...")


    smtp.starttls()


    print("Logging in...")


    smtp.login(EMAIL_USER, EMAIL_APP_PASSWORD)


    print("Sending mail...")


    smtp.sendmail(
        EMAIL_USER,
        EMAIL_USER,
        msg.as_string()
    )


    print("Mail sent!")


    smtp.quit()








# ==========================================
# ALERT ENDPOINT
# ==========================================


@app.route("/analyze", methods=["POST"])
def analyze():


    data = request.json


    server = data.get("server", "").split(":")[0]


    print(f"SSH Server: {server}")


    alertname = data.get(
        "alertname",
        "UnknownAlert"
    )


    print()
    print("=" * 50)
    print("PROMETHEUS ALERT RECEIVED")
    print("=" * 50)


    print(f"SERVER : {server}")
    print(f"ALERT  : {alertname}")


    print()
    print("Starting autonomous investigation...")


    report = investigate_server(
        server,
        alertname
    )


    with open(
        "investigation.txt",
        "w",
        encoding="utf-8"
    ) as f:
        f.write(report)


    print("Investigation completed")


    print()
    print("Analyzing with AI...")


    analysis = analyze_report(report)


    with open("analysis.txt", "w", encoding="utf-8") as f:
       f.write(analysis)


    print()
    print("=" * 50)
    print("AI ANALYSIS")
    print("=" * 50)


    print(analysis)


    try:


        send_email(
            server,
            alertname,
            report,
            analysis
        )


        print()
        print("Email sent successfully")


    except Exception as e:


        print()
        print("Email failed")
        print(str(e))


    return {
        "status": "received"
    }




# ==========================================
# START FLASK
# ==========================================


if __name__ == "__main__":


    app.run(
        host="0.0.0.0",
        port=8000
    )

from flask import Flask, request
import requests
import json


app = Flask(__name__)


# Change this if your listener runs on another IP/port
LISTENER_URL = "http://127.0.0.1:8000/analyze"




@app.route("/alert", methods=["POST"])
def alert():


    data = request.json


    print("=" * 60)
    print("ALERT RECEIVED FROM ALERTMANAGER")
    print("=" * 60)


    print(json.dumps(data, indent=2))


    if not data.get("alerts"):
        return {"status": "No alerts received"}, 400


    alert = data["alerts"][0]


    payload = {
        "server": alert["labels"].get("instance"),
        "alertname": alert["labels"].get("alertname"),
        "severity": alert["labels"].get("severity", "warning")
    }


    print()
    print("=" * 60)
    print("FORWARDING ALERT TO OPENCLAW AI")
    print("=" * 60)
    print(json.dumps(payload, indent=2))


    try:


        response = requests.post(
            LISTENER_URL,
            json=payload,
            timeout=300
        )


        print()
        print("Listener Response :", response.status_code)
        print(response.text)


        return {
            "status": "forwarded",
            "listener_status": response.status_code
        }, 200


    except Exception as e:


        print()
        print("ERROR forwarding to OpenClaw")
        print(str(e))


        return {
            "status": "failed",
            "error": str(e)
        }, 500




if __name__ == "__main__":


    app.run(
        host="0.0.0.0",
        port=5000
    )