mcp.servers.mcp_server_manager¶

MCP Server Manager for Model Context Protocol servers.

This module provides a robust way to start and manage MCP servers, handling the complexities of stdio transport, process management, and server lifecycle.

Key features:

  • Manages multiple MCP servers concurrently

  • Handles stdio transport servers correctly (stderr is normal output)

  • Provides health monitoring and automatic restart capabilities

  • Supports environment variable configuration for API keys

  • Clean shutdown handling with signal management

Example

Basic usage to start filesystem and time servers:

from haive.mcp.servers import MCPServerManager

manager = MCPServerManager()
manager.run(servers_to_start=["filesystem", "time"])

Non-blocking mode for integration:

manager = MCPServerManager()
manager.run(servers_to_start=["filesystem"], blocking=False)
# Servers run in background
status = manager.get_status()
print(f"Running servers: {status}")

With environment variables for API servers:

manager = MCPServerManager()
env_overrides = {"GITHUB_TOKEN": "your-token"}
manager.start_server("github", env_overrides=env_overrides)

Note

MCP servers using stdio transport write their status messages to stderr, which is normal behavior and not an error condition.

Attributes¶

logger

Classes¶

MCPServerManager

Manages MCP servers with proper stdio handling.

Functions¶

main()

Main entry point for command-line usage.

Module Contents¶

class mcp.servers.mcp_server_manager.MCPServerManager¶

Manages MCP servers with proper stdio handling.

This class provides comprehensive management of Model Context Protocol (MCP) servers, including starting, stopping, monitoring, and health checking.

servers¶

Dict mapping server names to their process information

available_servers¶

Dict of pre-configured server definitions

shutdown_requested¶

Flag indicating graceful shutdown was requested

Example

>>> manager = MCPServerManager()
>>> manager.start_server("filesystem")
True
>>> status = manager.get_status()
>>> print(status["filesystem"]["running"])
True
check_server_startup(process: subprocess.Popen, name: str, timeout: float = 5.0) → Tuple[bool, str]¶

Check if a server started successfully.

For stdio transport servers, output to stderr is normal behavior and not indicative of an error. This method handles the distinction between different transport types.

Parameters:

  • process – The subprocess.Popen instance for the server

  • name – Name of the server being checked

  • timeout – Maximum time to wait for startup confirmation

Returns:

bool, message: str) indicating startup status

Return type:

Tuple of (success

Note

Stdio servers are considered successfully started if they remain running after the initial startup period.

get_status() → Dict[str, dict]¶

Get status of all servers.

Returns current status information for all managed servers, including PID, running state, transport type, and description.

Returns:

  • pid: Process ID of the server

  • running: Boolean indicating if server is currently running

  • transport: Transport type (e.g., “stdio”)

  • description: Human-readable server description

Return type:

Dict mapping server names to status information dicts with keys

Example

>>> status = manager.get_status()
>>> print(status["filesystem"]["running"])
True
run(servers_to_start: List[str] = None, blocking: bool = True)¶

Run the MCP server manager.

Main entry point for starting and managing MCP servers. Can run in blocking mode (stays running until interrupted) or non-blocking mode (returns after starting servers).

Parameters:

  • servers_to_start – List of server names to start. Defaults to [“filesystem”, “time”] if not specified.

  • blocking – If True, blocks and monitors servers until interrupted. If False, returns immediately after starting servers.

Returns:

True if servers started successfully, False if critical

error occurred (e.g., npx not available)

Return type:

bool

Example

Blocking mode (typical usage):

manager = MCPServerManager()
manager.run()  # Runs until Ctrl+C

Non-blocking mode (for integration):

manager = MCPServerManager()
success = manager.run(blocking=False)
# Do other work while servers run
manager.stop_all_servers()  # Clean shutdown

Note

In blocking mode, installs signal handlers for SIGINT and SIGTERM to ensure clean shutdown of all servers.

show_status()¶

Show server status.

Prints a formatted status report of all servers to the logger, including running servers and available servers that aren’t running.

Note

Output is sent to the configured logger at INFO level.

start_server(name: str, env_overrides: Dict[str, str] = None) → bool¶

Start an MCP server by name.

Starts a configured MCP server process with proper environment setup and monitoring. Handles both stdio and other transport types.

Parameters:

  • name – Name of the server to start (must be in available_servers)

  • env_overrides – Optional dict of environment variables to set/override

Returns:

True if server started successfully, False otherwise

Return type:

bool

Raises:

None – Errors are logged but not raised

Example

>>> manager.start_server("filesystem")
True
>>> manager.start_server("github", {"GITHUB_TOKEN": "token"})
True
stop_all_servers()¶

Stop all running servers.

Gracefully shuts down all managed servers. This method is called automatically during signal handling for clean shutdown.

Note

Sets shutdown_requested flag to signal monitoring threads.

stop_server(name: str) → bool¶

Stop a server by name.

Attempts graceful termination first, then force kills if necessary. Cleans up all associated resources.

Parameters:

name – Name of the server to stop

Returns:

True if server was stopped, False if server wasn’t running

or stop failed

Return type:

bool

Example

>>> manager.stop_server("filesystem")
True
available_servers¶
servers: Dict[str, dict]¶
shutdown_requested = False¶
mcp.servers.mcp_server_manager.main()¶

Main entry point for command-line usage.

Provides a CLI interface for the MCP Server Manager with options for listing available servers, selecting which to start, and enabling debug logging.

Command-line arguments:

–servers: Space-separated list of servers to start –list: List available servers and exit –debug: Enable debug-level logging

Example

List available servers:

$ python mcp_server_manager.py --list

Start specific servers:

$ python mcp_server_manager.py --servers filesystem memory

Enable debug logging:

$ python mcp_server_manager.py --debug
mcp.servers.mcp_server_manager.logger¶