# To create the session for the first time
tmux new -s bots # To detach from the session (it keeps running in the background)
Ctrl+b, then d # To re-attach to the session later from anywhere
tmux attach -s bots
# To create the session for the first time
tmux new -s bots # To detach from the session (it keeps running in the background)
Ctrl+b, then d # To re-attach to the session later from anywhere
tmux attach -s bots
# To create the session for the first time
tmux new -s bots # To detach from the session (it keeps running in the background)
Ctrl+b, then d # To re-attach to the session later from anywhere
tmux attach -s bots
import subprocess
import time # Configuration of all bots to be managed
# The key is a friendly name, the value is the command to run
BOTS = { "content_publisher": ["python", "bots/content_publisher.py"], "social_media_manager": ["python", "bots/social_media_manager.py"], "data_scraper_A": ["python", "bots/data_scraper_a.py"], # ... add all 12+ bots here
} # A dictionary to hold the running subprocess objects
running_bots = {} def start_all_bots(): print("--- MASTERCLAW: Starting all nanobots ---") for bot_name, bot_command in BOTS.items(): print(f"Starting bot: {bot_name}...") # We redirect stdout and stderr to a log file for each bot log_file = open(f"logs/{bot_name}.log", "a") process = subprocess.Popen( bot_command, stdout=log_file, stderr=log_file ) running_bots[bot_name] = (process, bot_command, log_file) print("--- MASTERCLAW: All nanobots started ---") # --- Main execution ---
if __name__ == "__main__": start_all_bots() # In the next step, we'll add the watchdog loop here while True: time.sleep(60) # Keep the main script alive
import subprocess
import time # Configuration of all bots to be managed
# The key is a friendly name, the value is the command to run
BOTS = { "content_publisher": ["python", "bots/content_publisher.py"], "social_media_manager": ["python", "bots/social_media_manager.py"], "data_scraper_A": ["python", "bots/data_scraper_a.py"], # ... add all 12+ bots here
} # A dictionary to hold the running subprocess objects
running_bots = {} def start_all_bots(): print("--- MASTERCLAW: Starting all nanobots ---") for bot_name, bot_command in BOTS.items(): print(f"Starting bot: {bot_name}...") # We redirect stdout and stderr to a log file for each bot log_file = open(f"logs/{bot_name}.log", "a") process = subprocess.Popen( bot_command, stdout=log_file, stderr=log_file ) running_bots[bot_name] = (process, bot_command, log_file) print("--- MASTERCLAW: All nanobots started ---") # --- Main execution ---
if __name__ == "__main__": start_all_bots() # In the next step, we'll add the watchdog loop here while True: time.sleep(60) # Keep the main script alive
import subprocess
import time # Configuration of all bots to be managed
# The key is a friendly name, the value is the command to run
BOTS = { "content_publisher": ["python", "bots/content_publisher.py"], "social_media_manager": ["python", "bots/social_media_manager.py"], "data_scraper_A": ["python", "bots/data_scraper_a.py"], # ... add all 12+ bots here
} # A dictionary to hold the running subprocess objects
running_bots = {} def start_all_bots(): print("--- MASTERCLAW: Starting all nanobots ---") for bot_name, bot_command in BOTS.items(): print(f"Starting bot: {bot_name}...") # We redirect stdout and stderr to a log file for each bot log_file = open(f"logs/{bot_name}.log", "a") process = subprocess.Popen( bot_command, stdout=log_file, stderr=log_file ) running_bots[bot_name] = (process, bot_command, log_file) print("--- MASTERCLAW: All nanobots started ---") # --- Main execution ---
if __name__ == "__main__": start_all_bots() # In the next step, we'll add the watchdog loop here while True: time.sleep(60) # Keep the main script alive
[Unit]
Description=Masterclaw Bot Management Service
After=network.target [Service]
User=your_user # IMPORTANT: run as a non-root user
Group=your_group
WorkingDirectory=/home/your_user/masterclaw_project
ExecStart=/usr/bin/python3 /home/your_user/masterclaw_project/masterclaw.py
Restart=always
RestartSec=10 [Install]
WantedBy=multi-user.target
[Unit]
Description=Masterclaw Bot Management Service
After=network.target [Service]
User=your_user # IMPORTANT: run as a non-root user
Group=your_group
WorkingDirectory=/home/your_user/masterclaw_project
ExecStart=/usr/bin/python3 /home/your_user/masterclaw_project/masterclaw.py
Restart=always
RestartSec=10 [Install]
WantedBy=multi-user.target
[Unit]
Description=Masterclaw Bot Management Service
After=network.target [Service]
User=your_user # IMPORTANT: run as a non-root user
Group=your_group
WorkingDirectory=/home/your_user/masterclaw_project
ExecStart=/usr/bin/python3 /home/your_user/masterclaw_project/masterclaw.py
Restart=always
RestartSec=10 [Install]
WantedBy=multi-user.target
# Reload systemd to recognize the new -weight: 500;">service file
-weight: 600;">sudo -weight: 500;">systemctl daemon-reload # Enable the -weight: 500;">service to -weight: 500;">start on boot
-weight: 600;">sudo -weight: 500;">systemctl -weight: 500;">enable masterclaw.-weight: 500;">service # Start the -weight: 500;">service immediately
-weight: 600;">sudo -weight: 500;">systemctl -weight: 500;">start masterclaw.-weight: 500;">service # Check its -weight: 500;">status
-weight: 600;">sudo -weight: 500;">systemctl -weight: 500;">status masterclaw.-weight: 500;">service
# Reload systemd to recognize the new -weight: 500;">service file
-weight: 600;">sudo -weight: 500;">systemctl daemon-reload # Enable the -weight: 500;">service to -weight: 500;">start on boot
-weight: 600;">sudo -weight: 500;">systemctl -weight: 500;">enable masterclaw.-weight: 500;">service # Start the -weight: 500;">service immediately
-weight: 600;">sudo -weight: 500;">systemctl -weight: 500;">start masterclaw.-weight: 500;">service # Check its -weight: 500;">status
-weight: 600;">sudo -weight: 500;">systemctl -weight: 500;">status masterclaw.-weight: 500;">service
# Reload systemd to recognize the new -weight: 500;">service file
-weight: 600;">sudo -weight: 500;">systemctl daemon-reload # Enable the -weight: 500;">service to -weight: 500;">start on boot
-weight: 600;">sudo -weight: 500;">systemctl -weight: 500;">enable masterclaw.-weight: 500;">service # Start the -weight: 500;">service immediately
-weight: 600;">sudo -weight: 500;">systemctl -weight: 500;">start masterclaw.-weight: 500;">service # Check its -weight: 500;">status
-weight: 600;">sudo -weight: 500;">systemctl -weight: 500;">status masterclaw.-weight: 500;">service
import subprocess
import time
import logging
from datetime import datetime # --- Configuration ---
LOG_FILE = "logs/masterclaw.log"
BOT_LOG_DIR = "logs/bots" # Define all bots to be managed
BOTS = { "publisher": ["python", "bots/publisher.py"], "scraper_A": ["python", "bots/scraper_a.py"], "scraper_B": ["python", "bots/scraper_b.py"], "social_poster": ["python", "bots/social.py"], # ... add as many as you need
} # --- Logging Setup ---
logging.basicConfig( level=logging.INFO, format="%(asctime)s - [%(levelname)s] - %(message)s", handlers=[ logging.FileHandler(LOG_FILE), logging.StreamHandler() # Also print to console ]
) # --- Core Logic ---
class MasterClaw: def __init__(self): # { bot_name: (process, command, log_file_handle) } self.running_bots = {} def start_bot(self, name): """Starts a single, specified bot.""" if name in self.running_bots: logging.warning(f"Bot '{name}' is already running. Cannot -weight: 500;">start.") return if name not in BOTS: logging.error(f"Bot '{name}' not found in configuration.") return command = BOTS[name] try: log_path = f"{BOT_LOG_DIR}/{name}.log" log_file = open(log_path, "a") process = subprocess.Popen( command, stdout=log_file, stderr=subprocess.STDOUT, text=True ) self.running_bots[name] = (process, command, log_file) logging.info(f"Successfully started bot '{name}' with PID {process.pid}.") except Exception as e: logging.error(f"Failed to -weight: 500;">start bot '{name}': {e}") def start_all(self): """Initial -weight: 500;">start of all configured bots.""" logging.info("--- MASTERCLAW Initializing ---") for bot_name in BOTS: self.start_bot(bot_name) logging.info("--- All bots have been launched ---") def watchdog_loop(self): """The main self-healing loop.""" logging.info("--- Watchdog is now active ---") while True: time.sleep(15) # Check every 15 seconds for name in list(self.running_bots.keys()): process, command, log_file = self.running_bots[name] return_code = process.poll() if return_code is not None: # Process has terminated logging.warning(f"WATCHDOG: Bot '{name}' has terminated with code {return_code}.") # Clean up old resources log_file.close() del self.running_bots[name] # Restart the bot logging.info(f"WATCHDOG: Attempting to -weight: 500;">restart bot '{name}'...") self.start_bot(name) def shutdown(self): """Gracefully shut down all bot processes.""" logging.info("--- MASTERCLAW Shutting Down ---") for name, (process, _, log_file) in self.running_bots.items(): logging.info(f"Terminating bot '{name}' (PID: {process.pid})") process.terminate() # Send SIGTERM try: process.wait(timeout=10) # Wait up to 10 seconds except subprocess.TimeoutExpired: logging.warning(f"Bot '{name}' did not terminate gracefully. Sending SIGKILL.") process.kill() log_file.close() logging.info("--- All bots have been shut down ---") if __name__ == "__main__": claw = MasterClaw() try: claw.start_all() claw.watchdog_loop() except KeyboardInterrupt: print("\nKeyboard interrupt received.") finally: claw.shutdown()
import subprocess
import time
import logging
from datetime import datetime # --- Configuration ---
LOG_FILE = "logs/masterclaw.log"
BOT_LOG_DIR = "logs/bots" # Define all bots to be managed
BOTS = { "publisher": ["python", "bots/publisher.py"], "scraper_A": ["python", "bots/scraper_a.py"], "scraper_B": ["python", "bots/scraper_b.py"], "social_poster": ["python", "bots/social.py"], # ... add as many as you need
} # --- Logging Setup ---
logging.basicConfig( level=logging.INFO, format="%(asctime)s - [%(levelname)s] - %(message)s", handlers=[ logging.FileHandler(LOG_FILE), logging.StreamHandler() # Also print to console ]
) # --- Core Logic ---
class MasterClaw: def __init__(self): # { bot_name: (process, command, log_file_handle) } self.running_bots = {} def start_bot(self, name): """Starts a single, specified bot.""" if name in self.running_bots: logging.warning(f"Bot '{name}' is already running. Cannot -weight: 500;">start.") return if name not in BOTS: logging.error(f"Bot '{name}' not found in configuration.") return command = BOTS[name] try: log_path = f"{BOT_LOG_DIR}/{name}.log" log_file = open(log_path, "a") process = subprocess.Popen( command, stdout=log_file, stderr=subprocess.STDOUT, text=True ) self.running_bots[name] = (process, command, log_file) logging.info(f"Successfully started bot '{name}' with PID {process.pid}.") except Exception as e: logging.error(f"Failed to -weight: 500;">start bot '{name}': {e}") def start_all(self): """Initial -weight: 500;">start of all configured bots.""" logging.info("--- MASTERCLAW Initializing ---") for bot_name in BOTS: self.start_bot(bot_name) logging.info("--- All bots have been launched ---") def watchdog_loop(self): """The main self-healing loop.""" logging.info("--- Watchdog is now active ---") while True: time.sleep(15) # Check every 15 seconds for name in list(self.running_bots.keys()): process, command, log_file = self.running_bots[name] return_code = process.poll() if return_code is not None: # Process has terminated logging.warning(f"WATCHDOG: Bot '{name}' has terminated with code {return_code}.") # Clean up old resources log_file.close() del self.running_bots[name] # Restart the bot logging.info(f"WATCHDOG: Attempting to -weight: 500;">restart bot '{name}'...") self.start_bot(name) def shutdown(self): """Gracefully shut down all bot processes.""" logging.info("--- MASTERCLAW Shutting Down ---") for name, (process, _, log_file) in self.running_bots.items(): logging.info(f"Terminating bot '{name}' (PID: {process.pid})") process.terminate() # Send SIGTERM try: process.wait(timeout=10) # Wait up to 10 seconds except subprocess.TimeoutExpired: logging.warning(f"Bot '{name}' did not terminate gracefully. Sending SIGKILL.") process.kill() log_file.close() logging.info("--- All bots have been shut down ---") if __name__ == "__main__": claw = MasterClaw() try: claw.start_all() claw.watchdog_loop() except KeyboardInterrupt: print("\nKeyboard interrupt received.") finally: claw.shutdown()
import subprocess
import time
import logging
from datetime import datetime # --- Configuration ---
LOG_FILE = "logs/masterclaw.log"
BOT_LOG_DIR = "logs/bots" # Define all bots to be managed
BOTS = { "publisher": ["python", "bots/publisher.py"], "scraper_A": ["python", "bots/scraper_a.py"], "scraper_B": ["python", "bots/scraper_b.py"], "social_poster": ["python", "bots/social.py"], # ... add as many as you need
} # --- Logging Setup ---
logging.basicConfig( level=logging.INFO, format="%(asctime)s - [%(levelname)s] - %(message)s", handlers=[ logging.FileHandler(LOG_FILE), logging.StreamHandler() # Also print to console ]
) # --- Core Logic ---
class MasterClaw: def __init__(self): # { bot_name: (process, command, log_file_handle) } self.running_bots = {} def start_bot(self, name): """Starts a single, specified bot.""" if name in self.running_bots: logging.warning(f"Bot '{name}' is already running. Cannot -weight: 500;">start.") return if name not in BOTS: logging.error(f"Bot '{name}' not found in configuration.") return command = BOTS[name] try: log_path = f"{BOT_LOG_DIR}/{name}.log" log_file = open(log_path, "a") process = subprocess.Popen( command, stdout=log_file, stderr=subprocess.STDOUT, text=True ) self.running_bots[name] = (process, command, log_file) logging.info(f"Successfully started bot '{name}' with PID {process.pid}.") except Exception as e: logging.error(f"Failed to -weight: 500;">start bot '{name}': {e}") def start_all(self): """Initial -weight: 500;">start of all configured bots.""" logging.info("--- MASTERCLAW Initializing ---") for bot_name in BOTS: self.start_bot(bot_name) logging.info("--- All bots have been launched ---") def watchdog_loop(self): """The main self-healing loop.""" logging.info("--- Watchdog is now active ---") while True: time.sleep(15) # Check every 15 seconds for name in list(self.running_bots.keys()): process, command, log_file = self.running_bots[name] return_code = process.poll() if return_code is not None: # Process has terminated logging.warning(f"WATCHDOG: Bot '{name}' has terminated with code {return_code}.") # Clean up old resources log_file.close() del self.running_bots[name] # Restart the bot logging.info(f"WATCHDOG: Attempting to -weight: 500;">restart bot '{name}'...") self.start_bot(name) def shutdown(self): """Gracefully shut down all bot processes.""" logging.info("--- MASTERCLAW Shutting Down ---") for name, (process, _, log_file) in self.running_bots.items(): logging.info(f"Terminating bot '{name}' (PID: {process.pid})") process.terminate() # Send SIGTERM try: process.wait(timeout=10) # Wait up to 10 seconds except subprocess.TimeoutExpired: logging.warning(f"Bot '{name}' did not terminate gracefully. Sending SIGKILL.") process.kill() log_file.close() logging.info("--- All bots have been shut down ---") if __name__ == "__main__": claw = MasterClaw() try: claw.start_all() claw.watchdog_loop() except KeyboardInterrupt: print("\nKeyboard interrupt received.") finally: claw.shutdown()
PID USER PRI NI VIRT RES SHR S CPU% MEM% TIME+ COMMAND 1234 myuser 20 0 180.2M 85.1M 15.2M S 1.3 4.3 1h25:12 python3 masterclaw.py 5678 myuser 20 0 110.5M 60.3M 12.1M S 0.7 3.0 0:45.33 python3 bots/publisher.py 5680 myuser 20 0 95.7M 45.8M 10.9M S 0.0 2.3 0:22.11 python3 bots/scraper_a.py 5682 myuser 20 0 98.2M 42.1M 11.5M S 0.0 2.1 0:18.45 python3 bots/social.py 5684 myuser 20 0 89.9M 38.5M 9.8M S 0.0 1.9 0:15.78 python3 bots/scraper_b.py
... (8 more similar processes) ...
PID USER PRI NI VIRT RES SHR S CPU% MEM% TIME+ COMMAND 1234 myuser 20 0 180.2M 85.1M 15.2M S 1.3 4.3 1h25:12 python3 masterclaw.py 5678 myuser 20 0 110.5M 60.3M 12.1M S 0.7 3.0 0:45.33 python3 bots/publisher.py 5680 myuser 20 0 95.7M 45.8M 10.9M S 0.0 2.3 0:22.11 python3 bots/scraper_a.py 5682 myuser 20 0 98.2M 42.1M 11.5M S 0.0 2.1 0:18.45 python3 bots/social.py 5684 myuser 20 0 89.9M 38.5M 9.8M S 0.0 1.9 0:15.78 python3 bots/scraper_b.py
... (8 more similar processes) ...
PID USER PRI NI VIRT RES SHR S CPU% MEM% TIME+ COMMAND 1234 myuser 20 0 180.2M 85.1M 15.2M S 1.3 4.3 1h25:12 python3 masterclaw.py 5678 myuser 20 0 110.5M 60.3M 12.1M S 0.7 3.0 0:45.33 python3 bots/publisher.py 5680 myuser 20 0 95.7M 45.8M 10.9M S 0.0 2.3 0:22.11 python3 bots/scraper_a.py 5682 myuser 20 0 98.2M 42.1M 11.5M S 0.0 2.1 0:18.45 python3 bots/social.py 5684 myuser 20 0 89.9M 38.5M 9.8M S 0.0 1.9 0:15.78 python3 bots/scraper_b.py
... (8 more similar processes) ... - Process Supervision: Something needs to be watching the bots.
- Automatic Restarts: If a bot dies, it must be brought back to life immediately.
- Isolation: A crash in one bot should not affect any of the others.
- Manageability: I need a simple way to view logs, see what's running, and manually -weight: 500;">start/-weight: 500;">stop/-weight: 500;">restart individual bots without bringing down the whole system.
- Boot Persistence: The whole system must automatically -weight: 500;">start up if the server reboots. - If .poll() returns None, the process is still running happily.
- If .poll() returns an integer (the exit code), the process has terminated. - WorkingDirectory: Sets the CWD so all relative paths in your script (like logs/ or bots/) work correctly.
- ExecStart: The full path to the Python interpreter and your gateway script.
- Restart=always: The magic directive. If the masterclaw.py script itself ever crashes for any reason, systemd will automatically -weight: 500;">restart it.
- RestartSec=10: Wait 10 seconds before attempting a -weight: 500;">restart. - Hire me on Upwork — Python automation, API integrations, trading systems
- Check my Fiverr gigs — Bot development, web scraping, data pipelines
