258 lines
8.5 KiB
Python
258 lines
8.5 KiB
Python
"""Process Cleaning MCP Server - Tools for monitoring and cleaning system processes."""
|
|
|
|
import os
|
|
import signal
|
|
import subprocess
|
|
from dataclasses import asdict
|
|
from datetime import datetime, timezone
|
|
|
|
import psutil
|
|
from mcp.server import FastMCP
|
|
|
|
mcp = FastMCP(
|
|
name="procleaning-mcp",
|
|
instructions="Process monitoring and cleaning utilities. List, kill, and monitor system processes.",
|
|
)
|
|
|
|
|
|
def _proc_to_info(proc: psutil.Process) -> dict | None:
|
|
"""Convert a psutil.Process to a dict, handling exceptions."""
|
|
try:
|
|
with proc.oneshot():
|
|
return {
|
|
"pid": proc.pid,
|
|
"name": proc.name(),
|
|
"username": proc.username(),
|
|
"cpu_percent": round(proc.cpu_percent(), 2),
|
|
"memory_percent": round(proc.memory_percent(), 2),
|
|
"memory_rss_mb": round(proc.memory_info().rss / 1024 / 1024, 2),
|
|
"status": proc.status(),
|
|
"create_time": datetime.fromtimestamp(
|
|
proc.create_time(), tz=timezone.utc
|
|
).isoformat(),
|
|
"cmd_line": " ".join(proc.cmdline()) or "(unknown)",
|
|
}
|
|
except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess):
|
|
return None
|
|
|
|
|
|
@mcp.tool()
|
|
def list_processes(
|
|
name_filter: str | None = None,
|
|
username: str | None = None,
|
|
min_cpu: float | None = None,
|
|
min_memory: float | None = None,
|
|
min_rss_mb: float | None = None,
|
|
max_processes: int = 50,
|
|
) -> str:
|
|
"""List system processes with optional filtering by name, user, and resource usage thresholds."""
|
|
results = []
|
|
for proc in psutil.process_iter():
|
|
info = _proc_to_info(proc)
|
|
if info is None:
|
|
continue
|
|
|
|
if name_filter and name_filter.lower() not in info["name"].lower():
|
|
continue
|
|
if username and info["username"] != username:
|
|
continue
|
|
if min_cpu is not None and info["cpu_percent"] < min_cpu:
|
|
continue
|
|
if min_memory is not None and info["memory_percent"] < min_memory:
|
|
continue
|
|
if min_rss_mb is not None and info["memory_rss_mb"] < min_rss_mb:
|
|
continue
|
|
|
|
results.append(info)
|
|
if len(results) >= max_processes:
|
|
break
|
|
|
|
if not results:
|
|
return "No processes found matching the given criteria."
|
|
|
|
lines = [f"Found {len(results)} process(es):\n"]
|
|
for p in results:
|
|
lines.append(
|
|
f" PID: {p['pid']:>6} | Name: {p['name']:<25} | User: {p['username']:<12} "
|
|
f"| CPU: {p['cpu_percent']:>5}% | MEM: {p['memory_percent']:>4}% | "
|
|
f"RSS: {p['memory_rss_mb']:>7}MB | Status: {p['status']}"
|
|
)
|
|
return "\n".join(lines)
|
|
|
|
|
|
@mcp.tool()
|
|
def kill_process(pid: int, signal: str = "SIGTERM", confirm: bool = True) -> str:
|
|
"""Terminate a running process by PID. Default signal is SIGTERM (graceful); use SIGKILL for forceful termination."""
|
|
if not confirm:
|
|
return "Kill cancelled. Set confirm=true to proceed."
|
|
|
|
sig_num = getattr(signal, signal, signal.SIGTERM)
|
|
try:
|
|
proc = psutil.Process(pid)
|
|
proc_name = proc.name()
|
|
proc.kill(sig_num)
|
|
try:
|
|
proc.wait(timeout=5)
|
|
except psutil.TimeoutExpired:
|
|
return (
|
|
f"Warning: PID {pid} ({proc_name}) did not terminate within 5s. "
|
|
f"Consider using SIGKILL."
|
|
)
|
|
return f"Successfully killed process PID {pid} ({proc_name}) with {signal}."
|
|
except psutil.NoSuchProcess:
|
|
return f"Error: PID {pid} does not exist."
|
|
except psutil.AccessDenied:
|
|
return f"Error: permission denied killing PID {pid}."
|
|
except Exception as e:
|
|
return f"Error killing PID {pid}: {e}"
|
|
|
|
|
|
@mcp.tool()
|
|
def kill_by_name(
|
|
name_pattern: str,
|
|
signal: str = "SIGTERM",
|
|
max_kill: int = 10,
|
|
confirm: bool = True,
|
|
) -> str:
|
|
"""Terminate all processes matching a given name pattern. Default signal is SIGTERM (graceful)."""
|
|
if not confirm:
|
|
return "Kill cancelled. Set confirm=true to proceed."
|
|
|
|
sig_num = getattr(signal, signal, signal.SIGTERM)
|
|
|
|
targets = []
|
|
for proc in psutil.process_iter():
|
|
info = _proc_to_info(proc)
|
|
if info and name_pattern.lower() in info["name"].lower():
|
|
targets.append(info)
|
|
|
|
if not targets:
|
|
return f"No processes found matching '{name_pattern}'."
|
|
|
|
targets = targets[:max_kill]
|
|
killed = 0
|
|
results = []
|
|
for t in targets:
|
|
try:
|
|
p = psutil.Process(t["pid"])
|
|
p.kill(sig_num)
|
|
killed += 1
|
|
results.append(f" Killed PID {t['pid']} ({t['name']})")
|
|
except Exception as e:
|
|
results.append(f" Failed to kill PID {t['pid']}: {e}")
|
|
|
|
return (
|
|
f"Killed {killed}/{len(targets)} process(es) matching '{name_pattern}'.\n\n"
|
|
+ "\n".join(results)
|
|
)
|
|
|
|
|
|
@mcp.tool()
|
|
def system_resource_usage(top_n: int = 10, sort_by: str = "cpu") -> str:
|
|
"""Get current system resource usage overview including CPU, memory, disk, and top resource consumers."""
|
|
cpu_percent = psutil.cpu_percent(interval=1)
|
|
mem = psutil.virtual_memory()
|
|
disk = psutil.disk_usage("/")
|
|
load_avg = os.getloadavg()
|
|
cpu_count = psutil.cpu_count(logical=False) or 1
|
|
threads = psutil.cpu_count(logical=True) or 1
|
|
|
|
key_map = {
|
|
"cpu": lambda p: p.cpu_percent(),
|
|
"memory": lambda p: p.memory_percent(),
|
|
"rss": lambda p: p.memory_info().rss,
|
|
}
|
|
sort_key = key_map.get(sort_by, key_map["cpu"])
|
|
|
|
procs = []
|
|
for proc in psutil.process_iter():
|
|
info = _proc_to_info(proc)
|
|
if info:
|
|
procs.append(info)
|
|
|
|
procs.sort(key=sort_key, reverse=True)
|
|
top_procs = procs[:top_n]
|
|
|
|
summary = (
|
|
f"System Resource Usage\n"
|
|
f"=====================\n\n"
|
|
f"CPU Usage: {cpu_percent}% ({cpu_count} cores, {threads} threads)\n"
|
|
f"Memory Usage: {mem.percent}% ({mem.used / 1024**3:.1f}GB / {mem.total / 1024**3:.1f}GB)\n"
|
|
f"Disk Usage (/): {disk.percent}% ({disk.used / 1024**3:.1f}GB / {disk.total / 1024**3:.1f}GB)\n"
|
|
f"Load Average: {load_avg[0]:.2f}, {load_avg[1]:.2f}, {load_avg[2]:.2f}\n\n"
|
|
f"Top {top_n} Processes (sorted by {sort_by}):"
|
|
)
|
|
|
|
for i, p in enumerate(top_procs, 1):
|
|
summary += (
|
|
f"\n {i:2d}. PID: {p['pid']:>6} | Name: {p['name']:<25} "
|
|
f"| CPU: {p['cpu_percent']:>5}% | MEM: {p['memory_percent']:>4}% "
|
|
f"| RSS: {p['memory_rss_mb']:>7}MB"
|
|
)
|
|
|
|
return summary
|
|
|
|
|
|
@mcp.tool()
|
|
def find_zombies(include_details: bool = True) -> str:
|
|
"""Find and optionally clean zombie processes. Zombies are processes in 'zombie' status."""
|
|
zombies = []
|
|
for proc in psutil.process_iter():
|
|
try:
|
|
info = _proc_to_info(proc)
|
|
if info and info["status"] == "zombie":
|
|
zombies.append(info)
|
|
except Exception:
|
|
continue
|
|
|
|
if not zombies:
|
|
return "No zombie processes found."
|
|
|
|
if include_details:
|
|
lines = [f"Found {len(zombies)} zombie process(es):"]
|
|
for z in zombies:
|
|
lines.append(
|
|
f" PID: {z['pid']}, Name: {z['name']}, "
|
|
f"CMD: {z['cmd_line'][:80]}"
|
|
)
|
|
return "\n".join(lines)
|
|
else:
|
|
return f"Found {len(zombies)} zombie process(es): {', '.join(str(z['pid']) for z in zombies)}"
|
|
|
|
|
|
@mcp.tool()
|
|
def restart_service(service_name: str, confirm: bool = True) -> str:
|
|
"""Restart a systemd service by name. Requires sudo privileges for systemctl commands."""
|
|
if not confirm:
|
|
return "Restart cancelled. Set confirm=true to proceed."
|
|
|
|
if not service_name:
|
|
return "Error: service_name is required."
|
|
|
|
try:
|
|
result = subprocess.run(
|
|
["sudo", "systemctl", "restart", service_name],
|
|
capture_output=True, text=True, timeout=30,
|
|
)
|
|
if result.returncode == 0:
|
|
return f"Successfully restarted service '{service_name}'."
|
|
else:
|
|
return f"Failed to restart '{service_name}':\n{result.stderr.strip()}"
|
|
except FileNotFoundError:
|
|
return "Error: systemctl not found. Is this a systemd-based system?"
|
|
except subprocess.TimeoutExpired:
|
|
return "Error: restart operation timed out."
|
|
except PermissionError:
|
|
return "Error: permission denied. Ensure sudo access is configured."
|
|
except Exception as e:
|
|
return f"Error restarting service: {e}"
|
|
|
|
|
|
def main():
|
|
"""Run the MCP server."""
|
|
mcp.run()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|