mcp.servers.mcp_server_manager_v2

MCP Server Manager V2 - Pydantic-based refactor.

This module provides a refactored MCPServerManager that inherits from the base server management framework, adding MCP-specific functionality with full Pydantic validation.

Key improvements:

  • Inherits from BaseServerManager for type-safe management

  • Full Pydantic validation for all configurations

  • Backward compatibility with existing MCPServerManager API

  • Enhanced error handling and logging

  • Better separation of concerns

Example

Basic usage with new typed API:

from haive.mcp.servers import MCPServerManagerV2

manager = MCPServerManagerV2()
manager.add_config("filesystem", {
    "command": ["npx", "-y", "@modelcontextprotocol/server-filesystem"],
    "description": "File system operations",
    "transport": "stdio"
})
await manager.start_server("filesystem")

Legacy compatibility mode:

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

Attributes

logger

Classes

MCPServerManagerV2

Refactored MCP Server Manager with Pydantic validation.

Functions

main()

Main entry point for command-line usage.

Module Contents

class mcp.servers.mcp_server_manager_v2.MCPServerManagerV2(**data)

Bases: haive.dataflow.server_management.base.BaseServerManager[mcp.servers.models.MCPServerConfig, mcp.servers.models.MCPServerInfo]

Refactored MCP Server Manager with Pydantic validation.

Inherits from BaseServerManager to provide type-safe server management with MCP-specific extensions for stdio transport, environment validation, and protocol handling.

config_class

MCPServerConfig class for validation

info_class

MCPServerInfo class for runtime state

output_monitors

Thread references for stdio output monitoring

legacy_mode

Enable backward compatibility behaviors

Example

>>> manager = MCPServerManagerV2()
>>> config = MCPServerConfig(
...     name="github",
...     command=["npx", "-y", "@modelcontextprotocol/server-github"],
...     transport=MCPTransport.STDIO,
...     requires_env=["GITHUB_TOKEN"]
... )
>>> manager.add_config("github", config)
>>> await manager.start_server("github", {"GITHUB_TOKEN": "token"})
async cleanup() → None

Clean up all resources including output monitors.

get_status() → Dict[str, dict]

Get legacy-format status.

async health_check(name: str) → bool

MCP-specific health check implementation.

async restart_server(name: str) → mcp.servers.models.MCPServerInfo

Restart server preserving environment configuration.

run(servers_to_start: List[str] | None = None, blocking: bool = True) → bool

Legacy run method for backward compatibility.

show_status() → None

Show status in legacy format.

async start_server(name: str, env_overrides: Dict[str, str] | None = None) → mcp.servers.models.MCPServerInfo

Start an MCP server with proper validation.

Extends base start_server with MCP-specific features like environment validation and stdio output monitoring.

Parameters:

  • name – Server name

  • env_overrides – Optional environment variable overrides

Returns:

MCPServerInfo with running server details

Raises:
start_server_sync(name: str, env_overrides: Dict[str, str] | None = None) → bool

Synchronous wrapper for legacy compatibility.

stop_all_servers() → None

Stop all servers synchronously for legacy compatibility.

async stop_server(name: str, force: bool = False) → bool

Stop server with cleanup of MCP-specific resources.

stop_server_sync(name: str) → bool

Synchronous wrapper for legacy compatibility.

property available_servers: Dict[str, dict]

Legacy property for backward compatibility.

config_class: type[mcp.servers.models.MCPServerConfig] = None
info_class: type[mcp.servers.models.MCPServerInfo] = None
legacy_mode: bool = None
output_monitors: Dict[str, threading.Thread] = None
shutdown_requested: bool = None
mcp.servers.mcp_server_manager_v2.main()

Main entry point for command-line usage.

mcp.servers.mcp_server_manager_v2.logger