Index

The server management package provides classes and utilities for managing MCP server processes, including configuration, lifecycle operations, and persistent state tracking. It enables discovery, launching, monitoring, and safe shutdown of server instances, and supports robust state persistence across sessions.

Modules:

Name	Description
server_info	Data structures and enumerations for MCP server management.
server_manager	Server management utilities for MCP servers.
state_manager	Persistent state management for MCP server processes.

ServerInfo dataclass

Information about an MCP server.

Attributes:

Name	Type	Description
name	str	The name of the server.
package	str	The Python package that the server comes from.
module_path	str	The full Python module path for the server.
log_file	Optional[Path]	Path to the server's log file.
env_vars	Optional[Dict[str, Any]]	Environment variables for the server process.
status	ServerStatus	Current status of the server.
url	Optional[str]	Optional URL for accessing the server.
pid	Optional[int]	Process ID of the running server, if available.
host	str	Hostname or IP address where the server is running.
port	Optional[int]	Network port for the server, if applicable.
started_at	Optional[str]	ISO timestamp when the server was started.
last_checked	Optional[str]	ISO timestamp when the server status was last checked.

Methods:

Name	Description
to_dict	Serialize the ServerInfo instance to a dictionary suitable for JSON.
from_dict	Deserialize a dictionary into a ServerInfo instance.

Source code in mada_tools/server_management/server_info.py

@dataclass
class ServerInfo:
    """
    Information about an MCP server.

    Attributes:
        name (str): The name of the server.
        package (str): The Python package that the server comes from.
        module_path (str): The full Python module path for the server.
        log_file (Optional[Path]): Path to the server's log file.
        env_vars (Optional[Dict[str, Any]]): Environment variables for the server process.
        status (ServerStatus): Current status of the server.
        url (Optional[str]): Optional URL for accessing the server.
        pid (Optional[int]): Process ID of the running server, if available.
        host (str): Hostname or IP address where the server is running.
        port (Optional[int]): Network port for the server, if applicable.
        started_at (Optional[str]): ISO timestamp when the server was started.
        last_checked (Optional[str]): ISO timestamp when the server status was last checked.

    Methods:
        to_dict:
            Serialize the ServerInfo instance to a dictionary suitable for JSON.
        from_dict:
            Deserialize a dictionary into a ServerInfo instance.
    """

    name: str
    package: str
    module_path: str
    log_file: Optional[Path] = None
    env_vars: Optional[Dict[str, Any]] = None
    status: ServerStatus = ServerStatus.STOPPED
    url: Optional[str] = None
    pid: Optional[int] = None
    host: str = "localhost"
    port: Optional[int] = None
    started_at: Optional[str] = None
    last_checked: Optional[str] = None

    def to_dict(self) -> Dict[str, Any]:
        """
        Convert to dict for JSON serialization.

        Returns:

        """
        d = asdict(self)
        # Convert Path to string
        if self.log_file:
            d["log_file"] = str(self.log_file)
        # Convert Enum to value
        d["status"] = self.status.value
        return d

    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> "ServerInfo":
        """
        Create ServerInfo from dict.

        Returns:

        """
        # Convert status string back to Enum
        if "status" in data:
            data["status"] = ServerStatus(data["status"])
        # Convert log_file string back to Path
        if data.get("log_file"):
            data["log_file"] = Path(data["log_file"])
        return cls(**data)

from_dict(data) classmethod

Create ServerInfo from dict.

Returns:

Source code in mada_tools/server_management/server_info.py

@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "ServerInfo":
    """
    Create ServerInfo from dict.

    Returns:

    """
    # Convert status string back to Enum
    if "status" in data:
        data["status"] = ServerStatus(data["status"])
    # Convert log_file string back to Path
    if data.get("log_file"):
        data["log_file"] = Path(data["log_file"])
    return cls(**data)

to_dict()

Convert to dict for JSON serialization.

Returns:

Source code in mada_tools/server_management/server_info.py

def to_dict(self) -> Dict[str, Any]:
    """
    Convert to dict for JSON serialization.

    Returns:

    """
    d = asdict(self)
    # Convert Path to string
    if self.log_file:
        d["log_file"] = str(self.log_file)
    # Convert Enum to value
    d["status"] = self.status.value
    return d

ServerManager

Coordinate operations for MCP server processes.

The ServerManager provides operations to start, stop, restart, and query the status of MCP server processes. It uses ServerStateManager to track and validate running server instances across sessions.

Operations that start servers (start_servers, restart_servers) require a configuration file to be passed in. Operations that only need to interact with running servers (stop_servers, get_server_statuses) work purely from the state file.

Attributes:

Name	Type	Description
state_manager	ServerStateManager	Manager responsible for persisting and validating server runtime state.
_extension_registry	ExtensionRegistry	Registry used to resolve available MCP server registrations.

Methods:

Name	Description
start_servers	Start all or a subset of configured servers.
start_server	Start a single server process and register it with the state manager.
stop_servers	Stop all or a subset of currently running servers.
stop_server	Stop a single server, attempting graceful shutdown before forcing.
restart_servers	Restart all or a subset of configured servers.
restart_server	Restart a single server, stopping it if running then starting a fresh instance.
get_server_statuses	Retrieve ServerInfo objects describing current status of servers.
print_server_statuses	Print a simple table of server statuses to stdout.

Source code in mada_tools/server_management/server_manager.py

class ServerManager:
    """
    Coordinate operations for MCP server processes.

    The `ServerManager` provides operations to start, stop, restart, and query
    the status of MCP server processes. It uses `ServerStateManager` to track
    and validate running server instances across sessions.

    Operations that start servers (start_servers, restart_servers) require a
    configuration file to be passed in. Operations that only need to interact
    with running servers (stop_servers, get_server_statuses) work purely from
    the state file.

    Attributes:
        state_manager (ServerStateManager):
            Manager responsible for persisting and validating server runtime state.
        _extension_registry (ExtensionRegistry):
            Registry used to resolve available MCP server registrations.

    Methods:
        start_servers:
            Start all or a subset of configured servers.
        start_server:
            Start a single server process and register it with the state manager.
        stop_servers:
            Stop all or a subset of currently running servers.
        stop_server:
            Stop a single server, attempting graceful shutdown before forcing.
        restart_servers:
            Restart all or a subset of configured servers.
        restart_server:
            Restart a single server, stopping it if running then starting a fresh instance.
        get_server_statuses:
            Retrieve `ServerInfo` objects describing current status of servers.
        print_server_statuses:
            Print a simple table of server statuses to stdout.
    """

    def __init__(self, state_file: Optional[Path] = None):
        """
        Constructor for the ServerManager.

        Args:
            state_file (Optional[Path]): Optional path to state file for ServerStateManager.
        """
        self.state_manager = ServerStateManager(state_file=state_file)
        self._extension_registry = ExtensionRegistry()
        LOG.debug("ServerManager initialized")

    def _load_servers(self, config_file: Path) -> Dict[str, ServerInfo]:
        """
        Load server configurations from the config file and sync with current state.

        Args:
            config_file: Path to the JSON configuration file

        Returns:
            A dictionary mapping server names to `ServerInfo` instances.

        Raises:
            FileNotFoundError:
                Raised when the provided configuration path does not exist.
        """
        if not config_file.exists():
            raise FileNotFoundError(f"Config file not found: {config_file}")

        config = json.loads(config_file.read_text())

        # Discover all available server entry points
        available_servers = self._extension_registry.get_mcp_server_index()

        servers = {}
        server_configs = config.get("servers", {})

        # Get current running state from state manager
        running_servers = self.state_manager.get_running_servers(validate=True)

        for name, server_config in server_configs.items():
            # Verify the server is available
            if name not in available_servers:
                LOG.warning(f"Server '{name}' not found in discovered servers, skipping")
                continue
            registration = available_servers[name]

            # Get optional fields with defaults
            log_file = server_config.get("log_file")
            if log_file:
                log_file = Path(log_file).expanduser()
            else:
                log_file = Path.home() / ".mada" / "server_logs" / f"{name}.log"

            env_vars = server_config.get("env_vars", {})
            host = server_config.get("host", "localhost")
            port = server_config.get("port")

            # Check if server is currently running
            if name in running_servers:
                # Use the state manager's info which has current PID, status, etc.
                server_info = running_servers[name]
                # Update with any config changes
                server_info.package = registration.package
                server_info.module_path = registration.module_path
                server_info.log_file = log_file
                server_info.env_vars = env_vars
                server_info.host = host
                if port:
                    server_info.port = port
            else:
                # Create new ServerInfo instance
                server_info = ServerInfo(
                    name=name,
                    package=registration.package,
                    module_path=registration.module_path,
                    log_file=log_file,
                    env_vars=env_vars,
                    host=host,
                    port=port,
                    status=ServerStatus.STOPPED,
                )

            servers[name] = server_info

        return servers

    def start_servers(self, config_file: Path, server_names: Optional[List[str]] = None):
        """
        Start specified MCP servers or all servers if none specified.

        Args:
            config_file: Path to the JSON configuration file defining servers
            server_names: Optional list of server names to start. If None, starts all.

        Raises:
            ValueError:
                Raised when a requested server name is not present in the
                resolved configuration.
        """
        # Load servers from config
        servers = self._load_servers(config_file)

        if server_names:
            # Validate server names
            for name in server_names:
                if name not in servers:
                    raise ValueError(f"Unknown server: {name}")
            servers_to_start = {name: servers[name] for name in server_names}
        else:
            servers_to_start = servers

        for name, server_info in servers_to_start.items():
            try:
                self.start_server(config_file, name, server_info)
            except Exception as e:
                LOG.error(f"Failed to start server '{name}'.")
                raise e

    def start_server(self, config_file: Path, name: str, server_info: ServerInfo):
        """
        Start an individual MCP server.

        Args:
            config_file: Path to the JSON configuration file
            name: The server name
            server_info: A ServerInfo instance containing information about the server to start.
        """
        # Check if server is already running
        existing = self.state_manager.get_server(name)
        if existing and existing.pid:
            if self.state_manager._is_process_running(existing.pid):
                LOG.info(f"Server '{name}' is already running (PID: {existing.pid})")
                return

        LOG.info(f"Starting server '{name}'...")

        # Prepare log file
        server_info.log_file.parent.mkdir(parents=True, exist_ok=True)
        log_handle = open(server_info.log_file, "a")

        # Build command to run the server
        # Use the package's main entry point via python -m
        cmd = [
            sys.executable,
            "-m",
            server_info.module_path,
            "--config",
            str(config_file),
        ]

        # Prepare environment variables
        env = dict(os.environ)
        if server_info.env_vars:
            for key, value in server_info.env_vars.items():
                env[key] = str(value)

        # Verify that the port we're trying to use isn't already in use
        if server_info.port and self.state_manager._is_port_in_use(server_info.host, server_info.port) == 0:
            raise PortInUseError(
                f"Cannot start server '{name}', port {server_info.host}:{server_info.port} is already in use"
            )

        # Start the server process
        try:
            process = subprocess.Popen(
                cmd,
                stdout=log_handle,
                stderr=subprocess.STDOUT,
                env=env,
                start_new_session=True,  # Detach from parent process
            )

            # Update server info with process details
            server_info.pid = process.pid
            server_info.status = ServerStatus.STARTING
            server_info.started_at = datetime.now().isoformat()

            # Register with state manager
            config = json.loads(config_file.read_text())
            server_config = config.get("servers", {}).get(name, {})
            self.state_manager.register_server(server_info, server_config)

            # Give it a moment to start, then check health
            time.sleep(3)

            # Post-start check to see if the process is running
            if process.poll() is not None:
                self.state_manager.update_server_status(name, ServerStatus.FAILED)
                LOG.error(
                    f"Server '{name}' exited early with return code {process.returncode}. "
                    f"Check logs: {server_info.log_file}"
                )
                return

            LOG.info(f"Server '{name}' started with PID {process.pid}, logs: {server_info.log_file}")

            if server_info.port:
                return_code = self.state_manager._is_port_in_use(server_info.host, server_info.port)
                if return_code == 0:
                    self.state_manager.update_server_status(name, ServerStatus.RUNNING)
                    LOG.info(f"Server '{name}' is healthy and running")
                else:
                    self.state_manager.update_server_status(name, ServerStatus.UNHEALTHY)
                    LOG.warning(f"Server '{name}' started but health check failed with error code: {return_code}")

        except Exception as e:
            LOG.error(f"Failed to start server '{name}': {e}")
            server_info.status = ServerStatus.FAILED
            raise
        finally:
            if server_info.log_file and log_handle != subprocess.DEVNULL:
                log_handle.close()

    def stop_servers(
        self,
        server_names: Optional[List[str]] = None,
        config_file: Optional[Path] = None,
    ):
        """
        Stop specified running MCP servers or all servers if none specified.

        Args:
            server_names: Optional list of server names to stop. If None, stops all.
            config_file: Optional path to config file. If provided, shows all servers from config
                (even if stopped). If not provided, shows all servers from state.
        """
        # Get currently running servers from state
        running_servers = self.state_manager.get_running_servers(validate=True)

        # Determine allowed servers (filter by config if provided)
        if config_file:
            configured_servers = self._load_servers(config_file)
            allowed_servers = {
                name: running_servers[name] for name in configured_servers.keys() if name in running_servers
            }
        else:
            allowed_servers = running_servers

        # Further filter by requested server names
        if server_names:
            servers_to_stop = {name: allowed_servers[name] for name in server_names if name in allowed_servers}

            # Warn about servers that can't be stopped
            for name in server_names:
                if name not in allowed_servers:
                    if config_file:
                        if name not in configured_servers:
                            LOG.warning(f"Server '{name}' not found in config")
                        elif name not in running_servers:
                            LOG.warning(f"Server '{name}' is not running")
                    else:
                        LOG.warning(f"Server '{name}' is not running")
        else:
            servers_to_stop = allowed_servers

        # Stop each server
        for name, server_info in servers_to_stop.items():
            try:
                self.stop_server(name, server_info)
            except Exception as e:
                LOG.error(f"Failed to stop server '{name}': {e}")

    def stop_server(self, name: str, server_info: ServerInfo, timeout: int = 10) -> bool:
        """
        Stop an individual MCP server.

        Args:
            name: The server name to stop.
            server_info: ServerInfo with process details
            timeout: Seconds to wait for graceful shutdown before force kill

        Returns:
            bool: True if server was stopped, False if not running
        """
        LOG.info(f"Stopping server '{name}'...")

        if not server_info.pid:
            LOG.warning(f"Server '{name}' has no PID recorded")
            # Clean up state anyway
            self.state_manager.remove_server(name)
            return False

        pid = server_info.pid

        try:
            process = psutil.Process(pid)
        except psutil.NoSuchProcess:
            LOG.info(f"Server '{name}' (PID {pid}) is already stopped")
            self.state_manager.remove_server(name)
            return False
        except psutil.AccessDenied:
            LOG.error(f"Access denied when accessing server '{name}' (PID {pid})")
            return False

        # Check if this is actually our server process (safety check)
        try:
            # You could verify process name/cmdline matches expected server
            cmdline = process.cmdline()
            if not any(name in arg or server_info.package in arg for arg in cmdline):
                LOG.warning(f"PID {pid} doesn't appear to be server '{name}', skipping")
                self.state_manager.remove_server(name)
                return False
        except (psutil.AccessDenied, psutil.NoSuchProcess):
            pass  # Process might have died, continue with cleanup

        # Attempt graceful shutdown
        try:
            LOG.debug(f"Sending terminate signal to server '{name}' (PID {pid})")
            process.terminate()  # Cross-platform SIGTERM equivalent

            # Wait for process to exit gracefully
            try:
                process.wait(timeout=timeout)
                LOG.info(f"Server '{name}' stopped gracefully")
                self.state_manager.remove_server(name)
                return True
            except psutil.TimeoutExpired:
                # Process didn't stop, force kill
                LOG.warning(f"Server '{name}' did not stop gracefully after {timeout}s, forcing shutdown")

                # Kill all child processes too
                try:
                    parent = psutil.Process(pid)
                    children = parent.children(recursive=True)

                    # Kill children first
                    for child in children:
                        try:
                            LOG.debug(f"Killing child process {child.pid}")
                            child.kill()
                        except (psutil.NoSuchProcess, psutil.AccessDenied):
                            pass

                    # Then kill parent
                    parent.kill()

                    # Wait a moment for cleanup
                    parent.wait(timeout=2)

                    LOG.info(f"Server '{name}' force stopped")
                    self.state_manager.remove_server(name)
                    return True

                except psutil.NoSuchProcess:
                    # Process died during our kill attempt
                    LOG.info(f"Server '{name}' process terminated")
                    self.state_manager.remove_server(name)
                    return True

        except psutil.NoSuchProcess:
            # Process already gone
            LOG.info(f"Server '{name}' process already terminated")
            self.state_manager.remove_server(name)
            return True
        except psutil.AccessDenied:
            LOG.error(f"Permission denied when trying to stop server '{name}' (PID {pid})")
            return False
        except Exception as e:
            LOG.error(f"Error stopping server '{name}': {e}")
            return False

    def restart_servers(self, config_file: Path, server_names: Optional[List[str]] = None):
        """
        Restart specified MCP servers or all servers if none specified.

        Args:
            config_file: Path to the JSON configuration file defining servers
            server_names: Optional list of server names to restart. If None, restarts all configured servers.
        """
        # Load servers from config
        servers = self._load_servers(config_file)

        if server_names:
            # Validate server names exist in config
            for name in server_names:
                if name not in servers:
                    raise ValueError(f"Unknown server: {name}")
            servers_to_restart = server_names
        else:
            # Restart all configured servers
            servers_to_restart = list(servers.keys())

        for name in servers_to_restart:
            try:
                self.restart_server(config_file, name, servers[name])
            except Exception as e:
                LOG.error(f"Failed to restart server '{name}': {e}")

    def restart_server(self, config_file: Path, name: str, server_info: ServerInfo):
        """
        Restart a server (stop then start).

        Args:
            config_file: Path to the JSON configuration file
            name: Server name to restart
            server_info: ServerInfo instance from config
        """
        LOG.info(f"Restarting server '{name}'...")

        # Get current server info from state (if running)
        running_server = self.state_manager.get_server(name)

        if running_server and running_server.pid:
            # Stop if running
            stopped = self.stop_server(name, running_server)
            if stopped:
                LOG.info(f"Server '{name}' stopped, starting fresh...")
            else:
                LOG.warning(f"Failed to stop server '{name}', attempting to start anyway...")
        else:
            LOG.info(f"Server '{name}' is not running, starting fresh...")

        # Start fresh with config
        self.start_server(config_file, name, server_info)
        LOG.info(f"Server '{name}' restarted successfully")

    def get_server_statuses(
        self,
        server_names: Optional[List[str]] = None,
        config_file: Optional[Path] = None,
    ) -> Dict[str, ServerInfo]:
        """
        Get status information for specified servers or all servers.

        Args:
            server_names: Optional list of server names. If None, returns all running servers.
            config_file: Optional path to config file. If provided, shows all servers from config
                (even if stopped). If not provided, shows all servers from state.

        Returns:
            Dictionary mapping server names to their ServerInfo with current status.
        """
        servers: Dict[str, ServerInfo]
        if config_file is not None:
            # Get servers from config file
            servers = self._load_servers(config_file)
        else:
            # Get running servers from state manager
            servers = self.state_manager.get_servers(validate=True)

        # Determine which servers to check
        if server_names:
            for name in server_names[:]:
                if name not in servers:
                    LOG.warning(
                        f"Server '{name}' not found {'in config' if config_file is not None else 'by state manager'}."
                        "Removing this server from the filter."
                    )
                    server_names.remove(name)
            servers_to_check = server_names
        else:
            servers_to_check = list(servers.keys())

        statuses = {}
        for name in servers_to_check:
            if name in servers:
                statuses[name] = servers[name]
            else:
                LOG.warning(f"Server '{name}' not found in state.")

        return statuses

    def print_server_statuses(
        self,
        server_names: Optional[List[str]] = None,
        config_file: Optional[Path] = None,
    ):
        """
        Output the statuses of servers to the terminal.

        Args:
            server_names: Optional list of server names. If None, shows all running servers.
            config_file: Optional path to config file. If provided, shows all servers from config
                (even if stopped). If not provided, shows all servers from state.
        """
        statuses = self.get_server_statuses(server_names=server_names, config_file=config_file)
        console = Console()

        if not statuses:
            console.print("\nNo servers found.")
            return

        table = Table(title="MCP Server Status", show_header=True, header_style="bold magenta")

        # Add columns
        table.add_column("Server Name", style="cyan", no_wrap=True)
        table.add_column("Status", style="bold")
        table.add_column("PID", justify="right")
        table.add_column("Host:Port")

        # Add rows with status-based styling
        for name, server_info in statuses.items():
            pid_str = str(server_info.pid) if server_info.pid else "N/A"
            host_port = f"{server_info.host}:{server_info.port}" if server_info.port else "N/A"

            # Color-code status based on state
            status_display = server_info.status.value.upper()
            if server_info.status == ServerStatus.RUNNING:
                status_style = "[green]" + status_display + "[/green]"
            elif server_info.status == ServerStatus.STOPPED:
                status_style = "[dim]" + status_display + "[/dim]"
            elif server_info.status == ServerStatus.STARTING:
                status_style = "[yellow]" + status_display + "[/yellow]"
            elif server_info.status == ServerStatus.UNHEALTHY:
                status_style = "[red]" + status_display + "[/red]"
            elif server_info.status == ServerStatus.FAILED:
                status_style = "[bold red]" + status_display + "[/bold red]"
            else:
                status_style = status_display

            table.add_row(name, status_style, pid_str, host_port)

        console.print()
        console.print(table)
        console.print()

__init__(state_file=None)

Constructor for the ServerManager.

Parameters:

Name	Type	Description	Default
state_file	Optional[Path]	Optional path to state file for ServerStateManager.	None

Source code in mada_tools/server_management/server_manager.py

def __init__(self, state_file: Optional[Path] = None):
    """
    Constructor for the ServerManager.

    Args:
        state_file (Optional[Path]): Optional path to state file for ServerStateManager.
    """
    self.state_manager = ServerStateManager(state_file=state_file)
    self._extension_registry = ExtensionRegistry()
    LOG.debug("ServerManager initialized")

get_server_statuses(server_names=None, config_file=None)

Get status information for specified servers or all servers.

Parameters:

Name	Type	Description	Default
server_names	Optional[List[str]]	Optional list of server names. If None, returns all running servers.	None
config_file	Optional[Path]	Optional path to config file. If provided, shows all servers from config (even if stopped). If not provided, shows all servers from state.	None

Returns:

Type	Description
Dict[str, ServerInfo]	Dictionary mapping server names to their ServerInfo with current status.

Source code in mada_tools/server_management/server_manager.py

def get_server_statuses(
    self,
    server_names: Optional[List[str]] = None,
    config_file: Optional[Path] = None,
) -> Dict[str, ServerInfo]:
    """
    Get status information for specified servers or all servers.

    Args:
        server_names: Optional list of server names. If None, returns all running servers.
        config_file: Optional path to config file. If provided, shows all servers from config
            (even if stopped). If not provided, shows all servers from state.

    Returns:
        Dictionary mapping server names to their ServerInfo with current status.
    """
    servers: Dict[str, ServerInfo]
    if config_file is not None:
        # Get servers from config file
        servers = self._load_servers(config_file)
    else:
        # Get running servers from state manager
        servers = self.state_manager.get_servers(validate=True)

    # Determine which servers to check
    if server_names:
        for name in server_names[:]:
            if name not in servers:
                LOG.warning(
                    f"Server '{name}' not found {'in config' if config_file is not None else 'by state manager'}."
                    "Removing this server from the filter."
                )
                server_names.remove(name)
        servers_to_check = server_names
    else:
        servers_to_check = list(servers.keys())

    statuses = {}
    for name in servers_to_check:
        if name in servers:
            statuses[name] = servers[name]
        else:
            LOG.warning(f"Server '{name}' not found in state.")

    return statuses

print_server_statuses(server_names=None, config_file=None)

Output the statuses of servers to the terminal.

Parameters:

Name	Type	Description	Default
server_names	Optional[List[str]]	Optional list of server names. If None, shows all running servers.	None
config_file	Optional[Path]	Optional path to config file. If provided, shows all servers from config (even if stopped). If not provided, shows all servers from state.	None

Source code in mada_tools/server_management/server_manager.py

def print_server_statuses(
    self,
    server_names: Optional[List[str]] = None,
    config_file: Optional[Path] = None,
):
    """
    Output the statuses of servers to the terminal.

    Args:
        server_names: Optional list of server names. If None, shows all running servers.
        config_file: Optional path to config file. If provided, shows all servers from config
            (even if stopped). If not provided, shows all servers from state.
    """
    statuses = self.get_server_statuses(server_names=server_names, config_file=config_file)
    console = Console()

    if not statuses:
        console.print("\nNo servers found.")
        return

    table = Table(title="MCP Server Status", show_header=True, header_style="bold magenta")

    # Add columns
    table.add_column("Server Name", style="cyan", no_wrap=True)
    table.add_column("Status", style="bold")
    table.add_column("PID", justify="right")
    table.add_column("Host:Port")

    # Add rows with status-based styling
    for name, server_info in statuses.items():
        pid_str = str(server_info.pid) if server_info.pid else "N/A"
        host_port = f"{server_info.host}:{server_info.port}" if server_info.port else "N/A"

        # Color-code status based on state
        status_display = server_info.status.value.upper()
        if server_info.status == ServerStatus.RUNNING:
            status_style = "[green]" + status_display + "[/green]"
        elif server_info.status == ServerStatus.STOPPED:
            status_style = "[dim]" + status_display + "[/dim]"
        elif server_info.status == ServerStatus.STARTING:
            status_style = "[yellow]" + status_display + "[/yellow]"
        elif server_info.status == ServerStatus.UNHEALTHY:
            status_style = "[red]" + status_display + "[/red]"
        elif server_info.status == ServerStatus.FAILED:
            status_style = "[bold red]" + status_display + "[/bold red]"
        else:
            status_style = status_display

        table.add_row(name, status_style, pid_str, host_port)

    console.print()
    console.print(table)
    console.print()

restart_server(config_file, name, server_info)

Restart a server (stop then start).

Parameters:

Name	Type	Description	Default
config_file	Path	Path to the JSON configuration file	required
name	str	Server name to restart	required
server_info	ServerInfo	ServerInfo instance from config	required

Source code in mada_tools/server_management/server_manager.py

def restart_server(self, config_file: Path, name: str, server_info: ServerInfo):
    """
    Restart a server (stop then start).

    Args:
        config_file: Path to the JSON configuration file
        name: Server name to restart
        server_info: ServerInfo instance from config
    """
    LOG.info(f"Restarting server '{name}'...")

    # Get current server info from state (if running)
    running_server = self.state_manager.get_server(name)

    if running_server and running_server.pid:
        # Stop if running
        stopped = self.stop_server(name, running_server)
        if stopped:
            LOG.info(f"Server '{name}' stopped, starting fresh...")
        else:
            LOG.warning(f"Failed to stop server '{name}', attempting to start anyway...")
    else:
        LOG.info(f"Server '{name}' is not running, starting fresh...")

    # Start fresh with config
    self.start_server(config_file, name, server_info)
    LOG.info(f"Server '{name}' restarted successfully")

restart_servers(config_file, server_names=None)

Restart specified MCP servers or all servers if none specified.

Parameters:

Name	Type	Description	Default
config_file	Path	Path to the JSON configuration file defining servers	required
server_names	Optional[List[str]]	Optional list of server names to restart. If None, restarts all configured servers.	None

Source code in mada_tools/server_management/server_manager.py

def restart_servers(self, config_file: Path, server_names: Optional[List[str]] = None):
    """
    Restart specified MCP servers or all servers if none specified.

    Args:
        config_file: Path to the JSON configuration file defining servers
        server_names: Optional list of server names to restart. If None, restarts all configured servers.
    """
    # Load servers from config
    servers = self._load_servers(config_file)

    if server_names:
        # Validate server names exist in config
        for name in server_names:
            if name not in servers:
                raise ValueError(f"Unknown server: {name}")
        servers_to_restart = server_names
    else:
        # Restart all configured servers
        servers_to_restart = list(servers.keys())

    for name in servers_to_restart:
        try:
            self.restart_server(config_file, name, servers[name])
        except Exception as e:
            LOG.error(f"Failed to restart server '{name}': {e}")

start_server(config_file, name, server_info)

Start an individual MCP server.

Parameters:

Name	Type	Description	Default
config_file	Path	Path to the JSON configuration file	required
name	str	The server name	required
server_info	ServerInfo	A ServerInfo instance containing information about the server to start.	required

Source code in mada_tools/server_management/server_manager.py

def start_server(self, config_file: Path, name: str, server_info: ServerInfo):
    """
    Start an individual MCP server.

    Args:
        config_file: Path to the JSON configuration file
        name: The server name
        server_info: A ServerInfo instance containing information about the server to start.
    """
    # Check if server is already running
    existing = self.state_manager.get_server(name)
    if existing and existing.pid:
        if self.state_manager._is_process_running(existing.pid):
            LOG.info(f"Server '{name}' is already running (PID: {existing.pid})")
            return

    LOG.info(f"Starting server '{name}'...")

    # Prepare log file
    server_info.log_file.parent.mkdir(parents=True, exist_ok=True)
    log_handle = open(server_info.log_file, "a")

    # Build command to run the server
    # Use the package's main entry point via python -m
    cmd = [
        sys.executable,
        "-m",
        server_info.module_path,
        "--config",
        str(config_file),
    ]

    # Prepare environment variables
    env = dict(os.environ)
    if server_info.env_vars:
        for key, value in server_info.env_vars.items():
            env[key] = str(value)

    # Verify that the port we're trying to use isn't already in use
    if server_info.port and self.state_manager._is_port_in_use(server_info.host, server_info.port) == 0:
        raise PortInUseError(
            f"Cannot start server '{name}', port {server_info.host}:{server_info.port} is already in use"
        )

    # Start the server process
    try:
        process = subprocess.Popen(
            cmd,
            stdout=log_handle,
            stderr=subprocess.STDOUT,
            env=env,
            start_new_session=True,  # Detach from parent process
        )

        # Update server info with process details
        server_info.pid = process.pid
        server_info.status = ServerStatus.STARTING
        server_info.started_at = datetime.now().isoformat()

        # Register with state manager
        config = json.loads(config_file.read_text())
        server_config = config.get("servers", {}).get(name, {})
        self.state_manager.register_server(server_info, server_config)

        # Give it a moment to start, then check health
        time.sleep(3)

        # Post-start check to see if the process is running
        if process.poll() is not None:
            self.state_manager.update_server_status(name, ServerStatus.FAILED)
            LOG.error(
                f"Server '{name}' exited early with return code {process.returncode}. "
                f"Check logs: {server_info.log_file}"
            )
            return

        LOG.info(f"Server '{name}' started with PID {process.pid}, logs: {server_info.log_file}")

        if server_info.port:
            return_code = self.state_manager._is_port_in_use(server_info.host, server_info.port)
            if return_code == 0:
                self.state_manager.update_server_status(name, ServerStatus.RUNNING)
                LOG.info(f"Server '{name}' is healthy and running")
            else:
                self.state_manager.update_server_status(name, ServerStatus.UNHEALTHY)
                LOG.warning(f"Server '{name}' started but health check failed with error code: {return_code}")

    except Exception as e:
        LOG.error(f"Failed to start server '{name}': {e}")
        server_info.status = ServerStatus.FAILED
        raise
    finally:
        if server_info.log_file and log_handle != subprocess.DEVNULL:
            log_handle.close()

start_servers(config_file, server_names=None)

Start specified MCP servers or all servers if none specified.

Parameters:

Name	Type	Description	Default
config_file	Path	Path to the JSON configuration file defining servers	required
server_names	Optional[List[str]]	Optional list of server names to start. If None, starts all.	None

Raises:

Type	Description
ValueError	Raised when a requested server name is not present in the resolved configuration.

Source code in mada_tools/server_management/server_manager.py

def start_servers(self, config_file: Path, server_names: Optional[List[str]] = None):
    """
    Start specified MCP servers or all servers if none specified.

    Args:
        config_file: Path to the JSON configuration file defining servers
        server_names: Optional list of server names to start. If None, starts all.

    Raises:
        ValueError:
            Raised when a requested server name is not present in the
            resolved configuration.
    """
    # Load servers from config
    servers = self._load_servers(config_file)

    if server_names:
        # Validate server names
        for name in server_names:
            if name not in servers:
                raise ValueError(f"Unknown server: {name}")
        servers_to_start = {name: servers[name] for name in server_names}
    else:
        servers_to_start = servers

    for name, server_info in servers_to_start.items():
        try:
            self.start_server(config_file, name, server_info)
        except Exception as e:
            LOG.error(f"Failed to start server '{name}'.")
            raise e

stop_server(name, server_info, timeout=10)

Stop an individual MCP server.

Parameters:

Name	Type	Description	Default
name	str	The server name to stop.	required
server_info	ServerInfo	ServerInfo with process details	required
timeout	int	Seconds to wait for graceful shutdown before force kill	10

Returns:

Name	Type	Description
bool	bool	True if server was stopped, False if not running

Source code in mada_tools/server_management/server_manager.py

def stop_server(self, name: str, server_info: ServerInfo, timeout: int = 10) -> bool:
    """
    Stop an individual MCP server.

    Args:
        name: The server name to stop.
        server_info: ServerInfo with process details
        timeout: Seconds to wait for graceful shutdown before force kill

    Returns:
        bool: True if server was stopped, False if not running
    """
    LOG.info(f"Stopping server '{name}'...")

    if not server_info.pid:
        LOG.warning(f"Server '{name}' has no PID recorded")
        # Clean up state anyway
        self.state_manager.remove_server(name)
        return False

    pid = server_info.pid

    try:
        process = psutil.Process(pid)
    except psutil.NoSuchProcess:
        LOG.info(f"Server '{name}' (PID {pid}) is already stopped")
        self.state_manager.remove_server(name)
        return False
    except psutil.AccessDenied:
        LOG.error(f"Access denied when accessing server '{name}' (PID {pid})")
        return False

    # Check if this is actually our server process (safety check)
    try:
        # You could verify process name/cmdline matches expected server
        cmdline = process.cmdline()
        if not any(name in arg or server_info.package in arg for arg in cmdline):
            LOG.warning(f"PID {pid} doesn't appear to be server '{name}', skipping")
            self.state_manager.remove_server(name)
            return False
    except (psutil.AccessDenied, psutil.NoSuchProcess):
        pass  # Process might have died, continue with cleanup

    # Attempt graceful shutdown
    try:
        LOG.debug(f"Sending terminate signal to server '{name}' (PID {pid})")
        process.terminate()  # Cross-platform SIGTERM equivalent

        # Wait for process to exit gracefully
        try:
            process.wait(timeout=timeout)
            LOG.info(f"Server '{name}' stopped gracefully")
            self.state_manager.remove_server(name)
            return True
        except psutil.TimeoutExpired:
            # Process didn't stop, force kill
            LOG.warning(f"Server '{name}' did not stop gracefully after {timeout}s, forcing shutdown")

            # Kill all child processes too
            try:
                parent = psutil.Process(pid)
                children = parent.children(recursive=True)

                # Kill children first
                for child in children:
                    try:
                        LOG.debug(f"Killing child process {child.pid}")
                        child.kill()
                    except (psutil.NoSuchProcess, psutil.AccessDenied):
                        pass

                # Then kill parent
                parent.kill()

                # Wait a moment for cleanup
                parent.wait(timeout=2)

                LOG.info(f"Server '{name}' force stopped")
                self.state_manager.remove_server(name)
                return True

            except psutil.NoSuchProcess:
                # Process died during our kill attempt
                LOG.info(f"Server '{name}' process terminated")
                self.state_manager.remove_server(name)
                return True

    except psutil.NoSuchProcess:
        # Process already gone
        LOG.info(f"Server '{name}' process already terminated")
        self.state_manager.remove_server(name)
        return True
    except psutil.AccessDenied:
        LOG.error(f"Permission denied when trying to stop server '{name}' (PID {pid})")
        return False
    except Exception as e:
        LOG.error(f"Error stopping server '{name}': {e}")
        return False

stop_servers(server_names=None, config_file=None)

Stop specified running MCP servers or all servers if none specified.

Parameters:

Name	Type	Description	Default
server_names	Optional[List[str]]	Optional list of server names to stop. If None, stops all.	None
config_file	Optional[Path]	Optional path to config file. If provided, shows all servers from config (even if stopped). If not provided, shows all servers from state.	None

Source code in mada_tools/server_management/server_manager.py

def stop_servers(
    self,
    server_names: Optional[List[str]] = None,
    config_file: Optional[Path] = None,
):
    """
    Stop specified running MCP servers or all servers if none specified.

    Args:
        server_names: Optional list of server names to stop. If None, stops all.
        config_file: Optional path to config file. If provided, shows all servers from config
            (even if stopped). If not provided, shows all servers from state.
    """
    # Get currently running servers from state
    running_servers = self.state_manager.get_running_servers(validate=True)

    # Determine allowed servers (filter by config if provided)
    if config_file:
        configured_servers = self._load_servers(config_file)
        allowed_servers = {
            name: running_servers[name] for name in configured_servers.keys() if name in running_servers
        }
    else:
        allowed_servers = running_servers

    # Further filter by requested server names
    if server_names:
        servers_to_stop = {name: allowed_servers[name] for name in server_names if name in allowed_servers}

        # Warn about servers that can't be stopped
        for name in server_names:
            if name not in allowed_servers:
                if config_file:
                    if name not in configured_servers:
                        LOG.warning(f"Server '{name}' not found in config")
                    elif name not in running_servers:
                        LOG.warning(f"Server '{name}' is not running")
                else:
                    LOG.warning(f"Server '{name}' is not running")
    else:
        servers_to_stop = allowed_servers

    # Stop each server
    for name, server_info in servers_to_stop.items():
        try:
            self.stop_server(name, server_info)
        except Exception as e:
            LOG.error(f"Failed to stop server '{name}': {e}")

ServerStateManager

Manage persistent state and health information for MCP servers.

This class maintains a JSON backed registry of server processes, including their metadata and current status. It provides methods to register new servers, update their status, query running servers, and transparently remove entries for processes that are no longer alive or reachable.

State is stored in a single JSON file on disk, and all read or write operations are protected by a simple file based lock to avoid concurrent modification issues when multiple processes interact with the registry.

Attributes:

Name	Type	Description
state_file	Path	Path to the JSON file where server state is stored. If not explicitly provided at construction time, this defaults to ~/.mada/server_statuses.json. The parent directory is created automatically if it does not already exist.

Methods:

Name	Description
register_server	Register a newly started server and persist its initial state.
update_server_status	Update the status and last checked timestamp for an existing server entry.
remove_server	Remove a server by name from the state file. Returns True if an entry was removed, or False if no matching server exists.
get_server	Retrieve a single ServerInfo by server name, or None if the server is not registered.
get_running_servers	Return a mapping of server names to ServerInfo objects. When validate is True, each entry is checked for process liveness (via PID) and, if a port is known, TCP health. Stale or dead entries are removed from the persisted state.
cleanup_stale	Convenience method to remove dead or unreachable servers from the registry by invoking get_running_servers(validate=True).

Source code in mada_tools/server_management/state_manager.py

class ServerStateManager:
    """
    Manage persistent state and health information for MCP servers.

    This class maintains a JSON backed registry of server processes,
    including their metadata and current status. It provides methods
    to register new servers, update their status, query running servers,
    and transparently remove entries for processes that are no longer
    alive or reachable.

    State is stored in a single JSON file on disk, and all read or write
    operations are protected by a simple file based lock to avoid
    concurrent modification issues when multiple processes interact
    with the registry.

    Attributes:
        state_file (Path):
            Path to the JSON file where server state is stored. If not
            explicitly provided at construction time, this defaults to
            `~/.mada/server_statuses.json`. The parent directory is
            created automatically if it does not already exist.

    Methods:
        register_server:
            Register a newly started server and persist its initial state.
        update_server_status:
            Update the status and last checked timestamp for an existing
            server entry.
        remove_server:
            Remove a server by name from the state file. Returns True if an
            entry was removed, or False if no matching server exists.
        get_server:
            Retrieve a single `ServerInfo` by server name, or `None` if the
            server is not registered.
        get_running_servers:
            Return a mapping of server names to `ServerInfo` objects. When
            `validate` is True, each entry is checked for process liveness
            (via PID) and, if a port is known, TCP health. Stale or dead
            entries are removed from the persisted state.
        cleanup_stale:
            Convenience method to remove dead or unreachable servers from
            the registry by invoking `get_running_servers(validate=True)`.
    """

    def __init__(self, state_file: Optional[Path] = None):
        """
        Constructor for `ServerStateManager`.

        Args:
            state_file (Optional[Path]): Optional state file. If not provided,
                "~/.mada/server_statuses.json" is used.
        """
        self.state_file = state_file or Path.home() / ".mada" / "server_statuses.json"
        self.state_file.parent.mkdir(parents=True, exist_ok=True)

    @contextmanager
    def _lock_state(self) -> Iterator[None]:
        """
        Acquire and release the state file lock around a critical section.

        Args:
            None.

        Returns:
            Iterator[None]: Context manager that yields while the lock is held.

        Raises:
            OSError: If the lock file cannot be opened.
            RuntimeError: If no supported file locking implementation is available.
        """
        lock_file = self.state_file.with_suffix(".lock")
        with open(lock_file, "a+b") as f:
            self._acquire_file_lock(f)
            try:
                yield
            finally:
                self._release_file_lock(f)

    @staticmethod
    def _acquire_file_lock(lock_handle: BinaryIO) -> None:
        """
        Acquire an exclusive lock using the platform's file locking API.

        Args:
            lock_handle (BinaryIO): Open lock file handle.

        Raises:
            OSError: If the underlying platform lock operation fails.
            RuntimeError: If no supported file locking implementation is available.
        """
        if fcntl is not None:
            fcntl.flock(lock_handle.fileno(), fcntl.LOCK_EX)
            return

        if msvcrt is not None:
            # Windows locks a byte range, so ensure the file has at least one byte.
            lock_handle.seek(0, 2)
            if lock_handle.tell() == 0:
                lock_handle.write(b"\0")
                lock_handle.flush()

            lock_handle.seek(0)
            msvcrt.locking(lock_handle.fileno(), msvcrt.LK_LOCK, 1)
            return

        raise RuntimeError("No supported file locking implementation is available")

    @staticmethod
    def _release_file_lock(lock_handle: BinaryIO) -> None:
        """
        Release a previously acquired file lock.

        Args:
            lock_handle (BinaryIO): Open lock file handle.

        Raises:
            OSError: If the underlying platform unlock operation fails.
            RuntimeError: If no supported file locking implementation is available.
        """
        if fcntl is not None:
            fcntl.flock(lock_handle.fileno(), fcntl.LOCK_UN)
            return

        if msvcrt is not None:
            lock_handle.seek(0)
            msvcrt.locking(lock_handle.fileno(), msvcrt.LK_UNLCK, 1)
            return

        raise RuntimeError("No supported file locking implementation is available")

    def _load_state(self) -> Dict[str, ServerInfo]:
        """
        Load in the state of the servers from the state file.

        Returns:
            Dict[str, ServerInfo]: Dictionary mapping server names to ServerInfo objects
        """
        if not self.state_file.exists():
            return {}

        data = json.loads(self.state_file.read_text(encoding="utf-8"))
        servers = {}
        for name, server_dict in data.get("servers", {}).items():
            try:
                servers[name] = ServerInfo.from_dict(server_dict)
            except Exception as e:
                # Log warning but continue - don't let one bad entry break everything
                LOG.warning(f"Failed to load server '{name}': {e}")

        return servers

    def _save_state(self, servers: Dict[str, ServerInfo]) -> None:
        """
        Save server state atomically.

        Args:
            servers (Dict[str, ServerInfo]): Dictionary mapping server names to
                ServerInfo objects.

        Returns:
            None.

        Raises:
            OSError: If writing the temporary state file or replacing the final
                state file fails.
            TypeError: If a server entry cannot be serialized to JSON.
        """
        state = {"servers": {name: server.to_dict() for name, server in servers.items()}}

        tmp = self.state_file.with_suffix(".tmp")
        tmp.write_text(json.dumps(state, indent=2), encoding="utf-8")
        tmp.replace(self.state_file)

    def _is_process_running(self, pid: int) -> bool:
        """
        Check if PID is still running.

        Args:
            pid: Process ID to check

        Returns:
            bool: True if process is running
        """
        try:
            return psutil.Process(pid).is_running()
        except (psutil.NoSuchProcess, psutil.AccessDenied, ProcessLookupError):
            return False

    def _is_port_in_use(self, host: str, port: int, timeout: int = 1, retries: int = 10, delay: int = 1) -> int | bool:
        """
        Check whether a TCP service is already listening on the given host and port.

        Args:
            host (str): Hostname or IP address.
            port (int): Port number.
            timeout (int): Connection timeout in seconds.
            retries (int): Number of retries to connect.
            delay (int): Time between retries in seconds.

        Returns:
            int | bool: Socket error code from connect_ex(), or False if an OSError occurs.
        """
        try:
            last_code = None
            for _ in range(retries):
                with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
                    sock.settimeout(timeout)
                    last_code = sock.connect_ex((host, port))
                    if last_code == 0:
                        break
                time.sleep(delay)
            return last_code
        except OSError as exc:
            # Covers typical network and socket errors
            LOG.error(f"Port check failed for {host}:{port}, error: {exc}")
            return False

    def register_server(self, server_info: ServerInfo, config: dict):
        """
        Register a newly started server.

        Args:
            server_info: ServerInfo object with server details
            config: Configuration dict used to start the server
        """
        with self._lock_state():
            servers = self._load_state()

            server_info.started_at = datetime.now().isoformat()
            server_info.status = ServerStatus.STARTING

            servers[server_info.name] = server_info
            self._save_state(servers)

    def update_server_status(self, name: str, status: ServerStatus):
        """
        Update server status after health check.

        Args:
            name: Server name
            status: New ServerStatus
        """
        with self._lock_state():
            servers = self._load_state()
            if name in servers:
                servers[name].status = status
                servers[name].last_checked = datetime.now().isoformat()
                self._save_state(servers)

    def remove_server(self, name: str) -> bool:
        """
        Remove a server from state.

        Args:
            name: Server name

        Returns:
            bool: True if server was removed, False if not found
        """
        with self._lock_state():
            servers = self._load_state()
            if name in servers:
                del servers[name]
                self._save_state(servers)
                return True
            return False

    def get_server(self, name: str) -> Optional[ServerInfo]:
        """
        Get a specific server by name.

        Args:
            name: Server name

        Returns:
            ServerInfo if found, None otherwise
        """
        servers = self._load_state()
        return servers.get(name)

    def get_servers(self, validate: bool = True) -> Dict[str, ServerInfo]:
        """
        Get all servers, optionally validating they're still running.

        Args:
            validate: If True, check process liveness and update statuses accordingly

        Returns:
            Dict[str, ServerInfo]: Dictionary of server names to ServerInfo objects
        """
        with self._lock_state():
            servers = self._load_state()

            if not validate:
                return servers

            # Validate and update statuses
            changed = False

            for _, server_info in servers.items():
                if server_info.pid and self._is_process_running(server_info.pid):
                    # Process is running, check health if we have port info
                    if server_info.port:
                        if self._is_port_in_use(server_info.host, server_info.port) == 0:
                            if server_info.status != ServerStatus.RUNNING:
                                server_info.status = ServerStatus.RUNNING
                                changed = True
                        else:
                            if server_info.status != ServerStatus.UNHEALTHY:
                                server_info.status = ServerStatus.UNHEALTHY
                                changed = True
                    else:
                        # No port to check, just trust PID
                        if server_info.status != ServerStatus.RUNNING:
                            server_info.status = ServerStatus.RUNNING
                            changed = True
                else:
                    # Process not running
                    if server_info.status != ServerStatus.STOPPED:
                        server_info.status = ServerStatus.STOPPED
                        server_info.pid = None
                        changed = True

            if changed:
                self._save_state(servers)

            return servers

    def get_running_servers(self, validate: bool = True) -> Dict[str, ServerInfo]:
        """
        Get all running servers, optionally validating they're still running.

        Args:
            validate: If True, check that processes are actually running and clean up stale entries

        Returns:
            Dict[str, ServerInfo]: Dictionary of server names to ServerInfo objects (only running servers)
        """
        all_servers = self.get_servers(validate=validate)

        # Filter to only running/unhealthy servers (servers with active PIDs)
        running = {
            name: server_info
            for name, server_info in all_servers.items()
            if server_info.pid
            and server_info.status in [ServerStatus.RUNNING, ServerStatus.UNHEALTHY, ServerStatus.STARTING]
        }

        return running

    def cleanup_stale(self):
        """Remove dead servers from state"""
        self.get_running_servers(validate=True)

__init__(state_file=None)

Constructor for ServerStateManager.

Parameters:

Name	Type	Description	Default
state_file	Optional[Path]	Optional state file. If not provided, "~/.mada/server_statuses.json" is used.	None

Source code in mada_tools/server_management/state_manager.py

def __init__(self, state_file: Optional[Path] = None):
    """
    Constructor for `ServerStateManager`.

    Args:
        state_file (Optional[Path]): Optional state file. If not provided,
            "~/.mada/server_statuses.json" is used.
    """
    self.state_file = state_file or Path.home() / ".mada" / "server_statuses.json"
    self.state_file.parent.mkdir(parents=True, exist_ok=True)

cleanup_stale()

Remove dead servers from state

Source code in mada_tools/server_management/state_manager.py

def cleanup_stale(self):
    """Remove dead servers from state"""
    self.get_running_servers(validate=True)

get_running_servers(validate=True)

Get all running servers, optionally validating they're still running.

Parameters:

Name	Type	Description	Default
validate	bool	If True, check that processes are actually running and clean up stale entries	True

Returns:

Type	Description
Dict[str, ServerInfo]	Dict[str, ServerInfo]: Dictionary of server names to ServerInfo objects (only running servers)

Source code in mada_tools/server_management/state_manager.py

def get_running_servers(self, validate: bool = True) -> Dict[str, ServerInfo]:
    """
    Get all running servers, optionally validating they're still running.

    Args:
        validate: If True, check that processes are actually running and clean up stale entries

    Returns:
        Dict[str, ServerInfo]: Dictionary of server names to ServerInfo objects (only running servers)
    """
    all_servers = self.get_servers(validate=validate)

    # Filter to only running/unhealthy servers (servers with active PIDs)
    running = {
        name: server_info
        for name, server_info in all_servers.items()
        if server_info.pid
        and server_info.status in [ServerStatus.RUNNING, ServerStatus.UNHEALTHY, ServerStatus.STARTING]
    }

    return running

get_server(name)

Get a specific server by name.

Parameters:

Name	Type	Description	Default
name	str	Server name	required

Returns:

Type	Description
Optional[ServerInfo]	ServerInfo if found, None otherwise

Source code in mada_tools/server_management/state_manager.py

def get_server(self, name: str) -> Optional[ServerInfo]:
    """
    Get a specific server by name.

    Args:
        name: Server name

    Returns:
        ServerInfo if found, None otherwise
    """
    servers = self._load_state()
    return servers.get(name)

get_servers(validate=True)

Get all servers, optionally validating they're still running.

Parameters:

Name	Type	Description	Default
validate	bool	If True, check process liveness and update statuses accordingly	True

Returns:

Type	Description
Dict[str, ServerInfo]	Dict[str, ServerInfo]: Dictionary of server names to ServerInfo objects

Source code in mada_tools/server_management/state_manager.py

def get_servers(self, validate: bool = True) -> Dict[str, ServerInfo]:
    """
    Get all servers, optionally validating they're still running.

    Args:
        validate: If True, check process liveness and update statuses accordingly

    Returns:
        Dict[str, ServerInfo]: Dictionary of server names to ServerInfo objects
    """
    with self._lock_state():
        servers = self._load_state()

        if not validate:
            return servers

        # Validate and update statuses
        changed = False

        for _, server_info in servers.items():
            if server_info.pid and self._is_process_running(server_info.pid):
                # Process is running, check health if we have port info
                if server_info.port:
                    if self._is_port_in_use(server_info.host, server_info.port) == 0:
                        if server_info.status != ServerStatus.RUNNING:
                            server_info.status = ServerStatus.RUNNING
                            changed = True
                    else:
                        if server_info.status != ServerStatus.UNHEALTHY:
                            server_info.status = ServerStatus.UNHEALTHY
                            changed = True
                else:
                    # No port to check, just trust PID
                    if server_info.status != ServerStatus.RUNNING:
                        server_info.status = ServerStatus.RUNNING
                        changed = True
            else:
                # Process not running
                if server_info.status != ServerStatus.STOPPED:
                    server_info.status = ServerStatus.STOPPED
                    server_info.pid = None
                    changed = True

        if changed:
            self._save_state(servers)

        return servers

register_server(server_info, config)

Register a newly started server.

Parameters:

Name	Type	Description	Default
server_info	ServerInfo	ServerInfo object with server details	required
config	dict	Configuration dict used to start the server	required

Source code in mada_tools/server_management/state_manager.py

def register_server(self, server_info: ServerInfo, config: dict):
    """
    Register a newly started server.

    Args:
        server_info: ServerInfo object with server details
        config: Configuration dict used to start the server
    """
    with self._lock_state():
        servers = self._load_state()

        server_info.started_at = datetime.now().isoformat()
        server_info.status = ServerStatus.STARTING

        servers[server_info.name] = server_info
        self._save_state(servers)

remove_server(name)

Remove a server from state.

Parameters:

Name	Type	Description	Default
name	str	Server name	required

Returns:

Name	Type	Description
bool	bool	True if server was removed, False if not found

Source code in mada_tools/server_management/state_manager.py

def remove_server(self, name: str) -> bool:
    """
    Remove a server from state.

    Args:
        name: Server name

    Returns:
        bool: True if server was removed, False if not found
    """
    with self._lock_state():
        servers = self._load_state()
        if name in servers:
            del servers[name]
            self._save_state(servers)
            return True
        return False

update_server_status(name, status)

Update server status after health check.

Parameters:

Name	Type	Description	Default
name	str	Server name	required
status	ServerStatus	New ServerStatus	required

Source code in mada_tools/server_management/state_manager.py

def update_server_status(self, name: str, status: ServerStatus):
    """
    Update server status after health check.

    Args:
        name: Server name
        status: New ServerStatus
    """
    with self._lock_state():
        servers = self._load_state()
        if name in servers:
            servers[name].status = status
            servers[name].last_checked = datetime.now().isoformat()
            self._save_state(servers)

ServerStatus

Bases: Enum

Server status enumeration.

Attributes:

Name	Type	Description
STOPPED	ServerStatus	The server is not running.
STARTING	ServerStatus	The server is in the process of starting.
RUNNING	ServerStatus	The server is running and healthy.
UNHEALTHY	ServerStatus	The server is running but failed health checks.
FAILED	ServerStatus	The server failed to start or encountered a fatal error.

Source code in mada_tools/server_management/server_info.py

class ServerStatus(Enum):
    """
    Server status enumeration.

    Attributes:
        STOPPED (ServerStatus): The server is not running.
        STARTING (ServerStatus): The server is in the process of starting.
        RUNNING (ServerStatus): The server is running and healthy.
        UNHEALTHY (ServerStatus): The server is running but failed health checks.
        FAILED (ServerStatus): The server failed to start or encountered a fatal error.
    """

    STOPPED = "stopped"
    STARTING = "starting"
    RUNNING = "running"
    UNHEALTHY = "unhealthy"
    FAILED = "failed"