Python version of the server with socket CLI.

gistfile1.txt

import socket
import os
import sys
import json
import signal
import pathlib
import tempfile
import time

SOCKET_PATH = pathlib.Path(tempfile.gettempdir()) / "python-app-daemon.sock"
START_TIME = time.time()

def main():
    """
    Sample script that showcases a supposed server application
    with integrated CLI. This sample foregoes the worker process.
    It simply opens the socket and listens for commands.
    Use this if you can guarantee the server is not blocked.
    """
    command = sys.argv[1] if len(sys.argv) > 1 else None
    args = sys.argv[2:]

    try:
        probe_socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        probe_socket.connect(str(SOCKET_PATH))
        probe_socket.close()

        if command:
            run_command(command, args)
        else:
            show_help_and_exit(is_running=True)

    except (ConnectionRefusedError, FileNotFoundError):
        if command:
            print("❌ Error: Server is not running. Cannot execute command.", file=sys.stderr)
            sys.exit(1)
        else:
            print("ℹ️ No active server found. Starting a new one...")
            if SOCKET_PATH.exists():
                SOCKET_PATH.unlink()  # Clean up stale socket if it exists.
            start_server_process()
    except Exception as e:
        print(f"An unexpected error occurred while probing the socket: {e}", file=sys.stderr)
        sys.exit(1)

def start_server_process():
    """
    Binds to the Unix socket, sets up signal handlers for graceful shutdown,
    and enters an infinite loop to accept and handle client connections.
    """
    print(f"🚀 Server process with PID: {os.getpid()} is starting.")

    def shutdown_handler(signum, frame):
        print("\n[Server] Shutting down, removing socket...")
        if SOCKET_PATH.exists():
            SOCKET_PATH.unlink()
        sys.exit(0)

    signal.signal(signal.SIGINT, shutdown_handler)
    signal.signal(signal.SIGTERM, shutdown_handler)

    server_socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    server_socket.bind(str(SOCKET_PATH))
    server_socket.listen(1)

    print("   ✅ Application logic is running. Listening for commands...")

    while True:
        try:
            conn, _ = server_socket.accept()
            handle_client(conn)
        except Exception as e:
            print(f"[Server] Error accepting connection: {e}", file=sys.stderr)
            break

def handle_client(conn):
    """
    Reads a JSON payload from a client, processes it, and sends back a response.
    """
    try:
        with conn:
            raw_data = conn.recv(1024)
            if not raw_data:
                return

            payload = json.loads(raw_data.decode('utf-8'))
            command = payload.get("command")
            args = payload.get("args")

            print(f'[Server] Received command: "{command}" with args: {args}')

            if command == "stop":
                response = "✅ Server is shutting down..."
                conn.sendall(response.encode('utf-8'))
                os.kill(os.getpid(), signal.SIGTERM)
                return

            response = handle_command(command, args)
            conn.sendall(response.encode('utf-8'))
    except Exception as e:
        print(f"[Server] Error handling client: {e}", file=sys.stderr)

def handle_command(command, args):
    """Contains the core logic for each CLI command."""
    if command == "status":
        uptime = time.time() - START_TIME
        status = {
            "processPid": os.getpid(),
            "status": "running",
            "uptime": f"{uptime:.2f}s",
        }
        return json.dumps(status, indent=2)
    elif command == "create:org":
        if not args:
            return "Error: Organization name is required."
        return f'Organization "{args[0]}" created successfully.'
    else:
        return f'Error: Unknown command "{command}"'

def show_help_and_exit(is_running):
    """Prints a help message and exits the program."""
    if is_running:
        print("ℹ️ Server is already running.")
    print(f"\nUsage: python {sys.argv[0]} <command> [arguments]\n")
    print("Available commands:")
    print("  status              Check the status of the running server")
    print("  create:org <name>   Create a new organization")
    print("  stop                Stop the running server daemon")
    sys.exit(0)

def run_command(command, args):
    """Connects to the server's socket and sends a command."""
    try:
        client_socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        client_socket.connect(str(SOCKET_PATH))
        with client_socket:
            payload = json.dumps({"command": command, "args": args})
            client_socket.sendall(payload.encode('utf-8'))

            response = client_socket.recv(4096).decode('utf-8')
            print(response)
    except Exception as e:
        print(f"❌ Error connecting to server to run command: {e}", file=sys.stderr)
        sys.exit(1)

if __name__ == "__main__":
    main()

Please see my blog post about this on dev.to.