File System MCP Server

"""Main entry point for the File System MCP Server.""" import argparse import asyncio import sys from pathlib import Path from typing import Optional from .config import ConfigManager from .server import FileSystemMCPServer from .logging_config import setup_logging, structured_logger def parse_arguments() -> argparse.Namespace: """Parse command line arguments.""" parser = argparse.ArgumentParser( description="File System MCP Server - Safe file operations for AI assistants", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Examples: %(prog)s # Run with default config %(prog)s --config custom.json # Run with custom config %(prog)s --log-level DEBUG # Run with debug logging %(prog)s --log-file server.log # Log to file """ ) parser.add_argument( "--config", "-c", type=str, help="Path to configuration file (default: config.json)" ) parser.add_argument( "--log-level", choices=["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"], help="Set logging level (overrides config file)" ) parser.add_argument( "--log-file", type=str, help="Path to log file (optional)" ) parser.add_argument( "--version", action="version", version="File System MCP Server 0.1.0" ) parser.add_argument( "--validate-config", action="store_true", help="Validate configuration and exit" ) parser.add_argument( "--create-example-config", type=str, metavar="PATH", help="Create example configuration file and exit" ) return parser.parse_args() def create_example_config(path: str) -> None: """Create an example configuration file.""" from .config import Config try: config_path = Path(path) config = Config() ConfigManager.save_config(config, config_path) print(f"Example configuration created at: {config_path}") # Add comments to the config file with open(config_path, 'r') as f: content = f.read() commented_content = '''// File System MCP Server Configuration // This file configures the behavior of the MCP server { // Directory where backup files are stored "backup_directory": "./.mcp_backups", // Maximum file size for read operations (in bytes) "max_file_size": 10485760, // Maximum recursion depth for directory listing "max_recursion_depth": 10, // Allowed file extensions (null = all allowed) "allowed_extensions": null, // Protected paths that cannot be modified "protected_paths": [ "/etc", "/usr", "/bin", "/System" ], // Enable automatic backup creation "enable_backups": true, // Logging level "log_level": "INFO" }''' # Write the commented version with open(config_path, 'w') as f: f.write(commented_content) print("\nConfiguration options:") print("- backup_directory: Where backup files are stored") print("- max_file_size: Maximum file size for read operations") print("- max_recursion_depth: Limit for recursive directory listing") print("- allowed_extensions: Restrict file types (null = all allowed)") print("- protected_paths: Directories that cannot be modified") print("- enable_backups: Whether to create backups before modifications") print("- log_level: Logging verbosity (DEBUG, INFO, WARNING, ERROR, CRITICAL)") except Exception as e: print(f"Error creating example config: {e}", file=sys.stderr) sys.exit(1) def validate_config(config_path: Optional[str]) -> None: """Validate configuration and report issues.""" try: config = ConfigManager.load_config(config_path) issues = ConfigManager.validate_paths(config) print("Configuration validation results:") print(f"Config file: {config_path or 'config.json'}") print(f"Backup directory: {config.backup_directory}") print(f"Max file size: {config.max_file_size:,} bytes") print(f"Max recursion depth: {config.max_recursion_depth}") print(f"Allowed extensions: {config.allowed_extensions or 'All'}") print(f"Protected paths: {len(config.protected_paths)} paths") print(f"Backups enabled: {config.enable_backups}") print(f"Log level: {config.log_level}") if issues: print("\nIssues found:") for issue in issues: print(f" ⚠️ {issue}") sys.exit(1) else: print("\n✅ Configuration is valid") except Exception as e: print(f"❌ Configuration validation failed: {e}", file=sys.stderr) sys.exit(1) async def run_server(args: argparse.Namespace) -> None: """Run the MCP server.""" try: # Load configuration config = ConfigManager.load_config(args.config) # Override log level if specified if args.log_level: config.log_level = args.log_level # Set up logging setup_logging( log_level=config.log_level, log_file=args.log_file ) structured_logger.log_server_start("0.1.0") structured_logger.log_config_loaded( args.config or "config.json", ConfigManager.validate_paths(config) ) # Create and run server server = FileSystemMCPServer(config) await server.run() except KeyboardInterrupt: structured_logger.log_server_stop() print("\nServer stopped by user") except Exception as e: structured_logger.logger.error(f"Server error: {e}", exc_info=True) print(f"Server error: {e}", file=sys.stderr) sys.exit(1) def main() -> None: """Main entry point.""" args = parse_arguments() # Handle special commands if args.create_example_config: create_example_config(args.create_example_config) return if args.validate_config: validate_config(args.config) return # Run the server try: asyncio.run(run_server(args)) except KeyboardInterrupt: pass if __name__ == "__main__": main()