feat: implement standalone mode support with mock modules and path handling

py/config.py

@@ -3,6 +3,11 @@ import platform
 import folder_paths # type: ignore
 from typing import List
 import logging
+import sys
+import json
+
+# Check if running in standalone mode
+standalone_mode = 'nodes' not in sys.modules
 
 logger = logging.getLogger(__name__)
 
@@ -18,10 +23,47 @@ class Config:
         self._route_mappings = {}
         self.loras_roots = self._init_lora_paths()
         self.checkpoints_roots = self._init_checkpoint_paths()
-        self.temp_directory = folder_paths.get_temp_directory()
         # 在初始化时扫描符号链接
         self._scan_symbolic_links()
         
+        if not standalone_mode:
+            # Save the paths to settings.json when running in ComfyUI mode
+            self.save_folder_paths_to_settings()
+
+    def save_folder_paths_to_settings(self):
+        """Save folder paths to settings.json for standalone mode to use later"""
+        try:
+            # Check if we're running in ComfyUI mode (not standalone)
+            if hasattr(folder_paths, "get_folder_paths") and not isinstance(folder_paths, type):
+                # Get all relevant paths
+                lora_paths = folder_paths.get_folder_paths("loras")
+                checkpoint_paths = folder_paths.get_folder_paths("checkpoints")
+                diffuser_paths = folder_paths.get_folder_paths("diffusers")
+                unet_paths = folder_paths.get_folder_paths("unet")
+                
+                # Load existing settings
+                settings_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'settings.json')
+                settings = {}
+                if os.path.exists(settings_path):
+                    with open(settings_path, 'r', encoding='utf-8') as f:
+                        settings = json.load(f)
+                
+                # Update settings with paths
+                settings['folder_paths'] = {
+                    'loras': lora_paths,
+                    'checkpoints': checkpoint_paths,
+                    'diffusers': diffuser_paths,
+                    'unet': unet_paths
+                }
+                
+                # Save settings
+                with open(settings_path, 'w', encoding='utf-8') as f:
+                    json.dump(settings, f, indent=2)
+                
+                logger.info("Saved folder paths to settings.json")
+        except Exception as e:
+            logger.warning(f"Failed to save folder paths: {e}")
+
     def _is_link(self, path: str) -> bool:
         try:
             if os.path.islink(path):
@@ -103,59 +145,67 @@ class Config:
 
     def _init_lora_paths(self) -> List[str]:
         """Initialize and validate LoRA paths from ComfyUI settings"""
-        raw_paths = folder_paths.get_folder_paths("loras")
+        try:
+            raw_paths = folder_paths.get_folder_paths("loras")
             
-        # Normalize and resolve symlinks, store mapping from resolved -> original
+            # Normalize and resolve symlinks, store mapping from resolved -> original
-        path_map = {}
+            path_map = {}
-        for path in raw_paths:
+            for path in raw_paths:
-            if os.path.exists(path):
+                if os.path.exists(path):
-                real_path = os.path.normpath(os.path.realpath(path)).replace(os.sep, '/')
+                    real_path = os.path.normpath(os.path.realpath(path)).replace(os.sep, '/')
-                path_map[real_path] = path_map.get(real_path, path.replace(os.sep, "/"))  # preserve first seen
+                    path_map[real_path] = path_map.get(real_path, path.replace(os.sep, "/"))  # preserve first seen
             
-        # Now sort and use only the deduplicated real paths
+            # Now sort and use only the deduplicated real paths
-        unique_paths = sorted(path_map.values(), key=lambda p: p.lower())
+            unique_paths = sorted(path_map.values(), key=lambda p: p.lower())
-        print("Found LoRA roots:", "\n - " + "\n - ".join(unique_paths))
+            logger.info("Found LoRA roots:" + ("\n - " + "\n - ".join(unique_paths) if unique_paths else "[]"))
             
-        if not unique_paths:
+            if not unique_paths:
-            raise ValueError("No valid loras folders found in ComfyUI configuration")
+                logger.warning("No valid loras folders found in ComfyUI configuration")
+                return []
             
-        for original_path in unique_paths:
+            for original_path in unique_paths:
-            real_path = os.path.normpath(os.path.realpath(original_path)).replace(os.sep, '/')
+                real_path = os.path.normpath(os.path.realpath(original_path)).replace(os.sep, '/')
-            if real_path != original_path:
+                if real_path != original_path:
-                self.add_path_mapping(original_path, real_path)
+                    self.add_path_mapping(original_path, real_path)
-        
-        return unique_paths
             
+            return unique_paths
+        except Exception as e:
+            logger.warning(f"Error initializing LoRA paths: {e}")
+            return []
 
     def _init_checkpoint_paths(self) -> List[str]:
         """Initialize and validate checkpoint paths from ComfyUI settings"""
-        # Get checkpoint paths from folder_paths
+        try:
-        checkpoint_paths = folder_paths.get_folder_paths("checkpoints")
+            # Get checkpoint paths from folder_paths
-        diffusion_paths = folder_paths.get_folder_paths("diffusers")
+            checkpoint_paths = folder_paths.get_folder_paths("checkpoints")
-        unet_paths = folder_paths.get_folder_paths("unet")
+            diffusion_paths = folder_paths.get_folder_paths("diffusers")
+            unet_paths = folder_paths.get_folder_paths("unet")
             
-        # Combine all checkpoint-related paths
+            # Combine all checkpoint-related paths
-        all_paths = checkpoint_paths + diffusion_paths + unet_paths
+            all_paths = checkpoint_paths + diffusion_paths + unet_paths
             
-        # Filter and normalize paths
+            # Filter and normalize paths
-        paths = sorted(set(path.replace(os.sep, "/") 
+            paths = sorted(set(path.replace(os.sep, "/") 
-                for path in all_paths 
+                    for path in all_paths 
-                if os.path.exists(path)), key=lambda p: p.lower())
+                    if os.path.exists(path)), key=lambda p: p.lower())
             
-        print("Found checkpoint roots:", paths)
+            logger.info("Found checkpoint roots:" + ("\n - " + "\n - ".join(paths) if paths else "[]"))
             
-        if not paths:
+            if not paths:
-            logger.warning("No valid checkpoint folders found in ComfyUI configuration")
+                logger.warning("No valid checkpoint folders found in ComfyUI configuration")
+                return []
+            
+            # 初始化路径映射，与 LoRA 路径处理方式相同
+            for path in paths:
+                real_path = os.path.normpath(os.path.realpath(path)).replace(os.sep, '/')
+                if real_path != path:
+                    self.add_path_mapping(path, real_path)
+            
+            return paths
+        except Exception as e:
+            logger.warning(f"Error initializing checkpoint paths: {e}")
             return []
 
-        # 初始化路径映射，与 LoRA 路径处理方式相同
-        for path in paths:
-            real_path = os.path.normpath(os.path.realpath(path)).replace(os.sep, '/')
-            if real_path != path:
-                self.add_path_mapping(path, real_path)
-        
-        return paths
-
     def get_preview_static_url(self, preview_path: str) -> str:
         """Convert local preview path to static URL"""
         if not preview_path:

py/lora_manager.py

@@ -9,9 +9,13 @@ from .routes.update_routes import UpdateRoutes
 from .routes.usage_stats_routes import UsageStatsRoutes
 from .services.service_registry import ServiceRegistry
 import logging
+import sys
 
 logger = logging.getLogger(__name__)
 
+# Check if we're in standalone mode
+STANDALONE_MODE = 'nodes' not in sys.modules
+
 class LoraManager:
     """Main entry point for LoRA Manager plugin"""
     
@@ -20,6 +24,9 @@ class LoraManager:
         """Initialize and register all routes"""
         app = PromptServer.instance.app
 
+        # Configure aiohttp access logger to be less verbose
+        logging.getLogger('aiohttp.access').setLevel(logging.WARNING)
+
         added_targets = set()  # Track already added target paths
         
         # Add static routes for each lora root
@@ -108,6 +115,9 @@ class LoraManager:
     async def _initialize_services(cls):
         """Initialize all services using the ServiceRegistry"""
         try:
+            # Ensure aiohttp access logger is configured with reduced verbosity
+            logging.getLogger('aiohttp.access').setLevel(logging.WARNING)
+            
             # Initialize CivitaiClient first to ensure it's ready for other services
             civitai_client = await ServiceRegistry.get_civitai_client()
             
@@ -137,6 +147,12 @@ class LoraManager:
             # Initialize recipe scanner if needed
             recipe_scanner = await ServiceRegistry.get_recipe_scanner()
             
+            # Initialize metadata collector if not in standalone mode
+            if not STANDALONE_MODE:
+                from .metadata_collector import init as init_metadata
+                init_metadata()
+                logger.debug("Metadata collector initialized")
+            
             # Create low-priority initialization tasks
             asyncio.create_task(lora_scanner.initialize_in_background(), name='lora_cache_init')
             asyncio.create_task(checkpoint_scanner.initialize_in_background(), name='checkpoint_cache_init')

py/metadata_collector/__init__.py

@@ -1,18 +1,32 @@
 import os
 import importlib
-from .metadata_hook import MetadataHook
+import sys
-from .metadata_registry import MetadataRegistry
 
-def init():
+# Check if running in standalone mode
-    # Install hooks to collect metadata during execution
+standalone_mode = 'nodes' not in sys.modules
-    MetadataHook.install()
 
-    # Initialize registry
+if not standalone_mode:
-    registry = MetadataRegistry()
+    from .metadata_hook import MetadataHook
+    from .metadata_registry import MetadataRegistry
 
-    print("ComfyUI Metadata Collector initialized")
+    def init():
+        # Install hooks to collect metadata during execution
+        MetadataHook.install()
         
-def get_metadata(prompt_id=None):
+        # Initialize registry
-    """Helper function to get metadata from the registry"""
+        registry = MetadataRegistry()
-    registry = MetadataRegistry()
+        
-    return registry.get_metadata(prompt_id)
+        print("ComfyUI Metadata Collector initialized")
+        
+    def get_metadata(prompt_id=None):
+        """Helper function to get metadata from the registry"""
+        registry = MetadataRegistry()
+        return registry.get_metadata(prompt_id)
+else:
+    # Standalone mode - provide dummy implementations
+    def init():
+        print("ComfyUI Metadata Collector disabled in standalone mode")
+        
+    def get_metadata(prompt_id=None):
+        """Dummy implementation for standalone mode"""
+        return {}

py/metadata_collector/metadata_processor.py

@@ -1,4 +1,8 @@
 import json
+import sys
+
+# Check if running in standalone mode
+standalone_mode = 'nodes' not in sys.modules
 
 from .constants import MODELS, PROMPTS, SAMPLING, LORAS, SIZE
 
@@ -276,6 +280,10 @@ class MetadataProcessor:
     @staticmethod
     def to_dict(metadata):
         """Convert extracted metadata to the ComfyUI output.json format"""           
+        if standalone_mode:
+            # Return empty dictionary in standalone mode
+            return {}
+        
         params = MetadataProcessor.extract_generation_params(metadata)
         
         # Convert all values to strings to match output.json format

py/routes/recipe_routes.py

@@ -10,16 +10,25 @@ from typing import Dict
 import tempfile
 import json
 import asyncio
+import sys
 from ..utils.exif_utils import ExifUtils
 from ..utils.recipe_parsers import RecipeParserFactory
 from ..utils.constants import CARD_PREVIEW_WIDTH
 
 from ..config import config
-from ..metadata_collector import get_metadata  # Add MetadataCollector import
+
-from ..metadata_collector.metadata_processor import MetadataProcessor  # Add MetadataProcessor import
+# Check if running in standalone mode
+standalone_mode = 'nodes' not in sys.modules
+
 from ..utils.utils import download_civitai_image
 from ..services.service_registry import ServiceRegistry  # Add ServiceRegistry import
-from ..metadata_collector.metadata_registry import MetadataRegistry
+
+# Only import MetadataRegistry in non-standalone mode
+if not standalone_mode:
+    # Import metadata_collector functions and classes conditionally
+    from ..metadata_collector import get_metadata  # Add MetadataCollector import
+    from ..metadata_collector.metadata_processor import MetadataProcessor  # Add MetadataProcessor import
+    from ..metadata_collector.metadata_registry import MetadataRegistry
 
 logger = logging.getLogger(__name__)
 
@@ -804,8 +813,11 @@ class RecipeRoutes:
                 return web.json_response({"error": "No generation metadata found"}, status=400)
             
             # Get the most recent image from metadata registry instead of temp directory
-            metadata_registry = MetadataRegistry()
+            if not standalone_mode:
-            latest_image = metadata_registry.get_first_decoded_image()
+                metadata_registry = MetadataRegistry()
+                latest_image = metadata_registry.get_first_decoded_image()
+            else:
+                latest_image = None
             
             if not latest_image:
                 return web.json_response({"error": "No recent images found to use for recipe. Try generating an image first."}, status=400)

py/utils/usage_stats.py

@@ -1,5 +1,6 @@
 import os
 import json
+import sys
 import time
 import asyncio
 import logging
@@ -7,8 +8,13 @@ from typing import Dict, Set
 
 from ..config import config
 from ..services.service_registry import ServiceRegistry
-from ..metadata_collector.metadata_registry import MetadataRegistry
+
-from ..metadata_collector.constants import MODELS, LORAS
+# Check if running in standalone mode
+standalone_mode = 'nodes' not in sys.modules
+
+if not standalone_mode:
+    from ..metadata_collector.metadata_registry import MetadataRegistry
+    from ..metadata_collector.constants import MODELS, LORAS
 
 logger = logging.getLogger(__name__)
 

standalone.py

@@ -0,0 +1,328 @@
+import os
+import sys
+import json
+
+# Create mock folder_paths module BEFORE any other imports
+class MockFolderPaths:
+    @staticmethod
+    def get_folder_paths(folder_name):
+        # Load paths from settings.json
+        settings_path = os.path.join(os.path.dirname(__file__), 'settings.json')
+        try:
+            if os.path.exists(settings_path):
+                with open(settings_path, 'r', encoding='utf-8') as f:
+                    settings = json.load(f)
+                
+                # Return paths if they exist in settings
+                if 'folder_paths' in settings and folder_name in settings['folder_paths']:
+                    paths = settings['folder_paths'][folder_name]
+                    # Filter out paths that don't exist
+                    valid_paths = [p for p in paths if os.path.exists(p)]
+                    if valid_paths:
+                        return valid_paths
+                    else:
+                        print(f"Warning: No valid paths found for {folder_name}")
+        except Exception as e:
+            print(f"Error loading folder paths from settings: {e}")
+        
+        # Fallback to empty list if no paths found
+        return []
+    
+    @staticmethod
+    def get_temp_directory():
+        return os.path.join(os.path.dirname(__file__), 'temp')
+    
+    @staticmethod
+    def set_temp_directory(path):
+        os.makedirs(path, exist_ok=True)
+        return path
+
+# Create mock server module with PromptServer
+class MockPromptServer:
+    def __init__(self):
+        self.app = None
+        
+    def send_sync(self, *args, **kwargs):
+        pass
+
+# Create mock metadata_collector module
+class MockMetadataCollector:
+    def init(self):
+        pass
+    
+    def get_metadata(self, prompt_id=None):
+        return {}
+
+# Initialize basic mocks before any imports
+sys.modules['folder_paths'] = MockFolderPaths()
+sys.modules['server'] = type('server', (), {'PromptServer': MockPromptServer()})
+sys.modules['py.metadata_collector'] = MockMetadataCollector()
+
+# Now we can safely import modules that depend on folder_paths and server
+import argparse
+import asyncio
+import logging
+from aiohttp import web
+
+# Setup logging
+logging.basicConfig(level=logging.INFO, 
+                    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+logger = logging.getLogger("lora-manager-standalone")
+
+# Configure aiohttp access logger to be less verbose
+logging.getLogger('aiohttp.access').setLevel(logging.WARNING)
+
+# Now we can import the global config from our local modules
+from py.config import config
+
+class StandaloneServer:
+    """Server implementation for standalone mode"""
+    
+    def __init__(self):
+        self.app = web.Application(logger=logger)
+        self.instance = self  # Make it compatible with PromptServer.instance pattern
+        
+        # Ensure the app's access logger is configured to reduce verbosity
+        self.app._subapps = []  # Ensure this exists to avoid AttributeError
+        
+        # Configure access logging for the app
+        self.app.on_startup.append(self._configure_access_logger)
+    
+    async def _configure_access_logger(self, app):
+        """Configure access logger to reduce verbosity"""
+        logging.getLogger('aiohttp.access').setLevel(logging.WARNING)
+        
+        # If using aiohttp>=3.8.0, configure access logger through app directly
+        if hasattr(app, 'access_logger'):
+            app.access_logger.setLevel(logging.WARNING)
+    
+    async def setup(self):
+        """Set up the standalone server"""
+        # Create placeholders for compatibility with ComfyUI's implementation
+        self.last_prompt_id = None
+        self.last_node_id = None
+        self.client_id = None
+        
+        # Set up routes
+        self.setup_routes()
+        
+        # Add startup and shutdown handlers
+        self.app.on_startup.append(self.on_startup)
+        self.app.on_shutdown.append(self.on_shutdown)
+    
+    def setup_routes(self):
+        """Set up basic routes"""
+        # Add a simple status endpoint
+        self.app.router.add_get('/', self.handle_status)
+    
+    async def handle_status(self, request):
+        """Handle status request"""
+        return web.json_response({
+            "status": "running",
+            "mode": "standalone",
+            "loras_roots": config.loras_roots,
+            "checkpoints_roots": config.checkpoints_roots
+        })
+    
+    async def on_startup(self, app):
+        """Startup handler"""
+        logger.info("LoRA Manager standalone server starting...")
+    
+    async def on_shutdown(self, app):
+        """Shutdown handler"""
+        logger.info("LoRA Manager standalone server shutting down...")
+    
+    def send_sync(self, event_type, data, sid=None):
+        """Stub for compatibility with PromptServer"""
+        # In standalone mode, we don't have the same websocket system
+        pass
+    
+    async def start(self, host='0.0.0.0', port=8188):
+        """Start the server"""
+        runner = web.AppRunner(self.app)
+        await runner.setup()
+        site = web.TCPSite(runner, host, port)
+        await site.start()
+        logger.info(f"Server started at http://{host}:{port}")
+        
+        # Keep the server running
+        while True:
+            await asyncio.sleep(3600)  # Sleep for a long time
+    
+    async def publish_loop(self):
+        """Stub for compatibility with PromptServer"""
+        # This method exists in ComfyUI's server but we don't need it
+        pass
+
+# After all mocks are in place, import LoraManager
+from py.lora_manager import LoraManager
+
+class StandaloneLoraManager(LoraManager):
+    """Extended LoraManager for standalone mode"""
+    
+    @classmethod
+    def add_routes(cls, server_instance):
+        """Initialize and register all routes for standalone mode"""
+        app = server_instance.app
+        
+        # Store app in a global-like location for compatibility
+        sys.modules['server'].PromptServer.instance = server_instance
+
+        # Configure aiohttp access logger to be less verbose
+        logging.getLogger('aiohttp.access').setLevel(logging.WARNING)
+        
+        added_targets = set()  # Track already added target paths
+        
+        # Add static routes for each lora root
+        for idx, root in enumerate(config.loras_roots, start=1):
+            if not os.path.exists(root):
+                logger.warning(f"Lora root path does not exist: {root}")
+                continue
+                
+            preview_path = f'/loras_static/root{idx}/preview'
+            
+            # Check if this root is a link path in the mappings
+            real_root = root
+            for target, link in config._path_mappings.items():
+                if os.path.normpath(link) == os.path.normpath(root):
+                    # If so, route should point to the target (real path)
+                    real_root = target
+                    break
+            
+            # Normalize and standardize path display for consistency
+            display_root = real_root.replace('\\', '/')
+            
+            # Add static route for original path - use the normalized path
+            app.router.add_static(preview_path, real_root)
+            logger.info(f"Added static route {preview_path} -> {display_root}")
+            
+            # Record route mapping with normalized path
+            config.add_route_mapping(real_root, preview_path)
+            added_targets.add(os.path.normpath(real_root))
+        
+        # Add static routes for each checkpoint root
+        for idx, root in enumerate(config.checkpoints_roots, start=1):
+            if not os.path.exists(root):
+                logger.warning(f"Checkpoint root path does not exist: {root}")
+                continue
+                
+            preview_path = f'/checkpoints_static/root{idx}/preview'
+            
+            # Check if this root is a link path in the mappings
+            real_root = root
+            for target, link in config._path_mappings.items():
+                if os.path.normpath(link) == os.path.normpath(root):
+                    # If so, route should point to the target (real path)
+                    real_root = target
+                    break
+            
+            # Normalize and standardize path display for consistency
+            display_root = real_root.replace('\\', '/')
+            
+            # Add static route for original path
+            app.router.add_static(preview_path, real_root)
+            logger.info(f"Added static route {preview_path} -> {display_root}")
+            
+            # Record route mapping
+            config.add_route_mapping(real_root, preview_path)
+            added_targets.add(os.path.normpath(real_root))
+        
+        # Add static routes for symlink target paths that aren't already covered
+        link_idx = {
+            'lora': 1,
+            'checkpoint': 1
+        }
+        
+        for target_path, link_path in config._path_mappings.items():
+            norm_target = os.path.normpath(target_path)
+            if norm_target not in added_targets:
+                # Determine if this is a checkpoint or lora link based on path
+                is_checkpoint = any(os.path.normpath(cp_root) in os.path.normpath(link_path) for cp_root in config.checkpoints_roots)
+                is_checkpoint = is_checkpoint or any(os.path.normpath(cp_root) in norm_target for cp_root in config.checkpoints_roots)
+                
+                if is_checkpoint:
+                    route_path = f'/checkpoints_static/link_{link_idx["checkpoint"]}/preview'
+                    link_idx["checkpoint"] += 1
+                else:
+                    route_path = f'/loras_static/link_{link_idx["lora"]}/preview'
+                    link_idx["lora"] += 1
+                
+                # Display path with forward slashes for consistency
+                display_target = target_path.replace('\\', '/')
+                
+                app.router.add_static(route_path, target_path)
+                logger.info(f"Added static route for link target {route_path} -> {display_target}")
+                config.add_route_mapping(target_path, route_path)
+                added_targets.add(norm_target)
+        
+        # Add static route for plugin assets
+        app.router.add_static('/loras_static', config.static_path)
+        
+        # Setup feature routes
+        from py.routes.lora_routes import LoraRoutes
+        from py.routes.api_routes import ApiRoutes
+        from py.routes.recipe_routes import RecipeRoutes
+        from py.routes.checkpoints_routes import CheckpointsRoutes
+        from py.routes.update_routes import UpdateRoutes
+        from py.routes.usage_stats_routes import UsageStatsRoutes
+        
+        lora_routes = LoraRoutes()
+        checkpoints_routes = CheckpointsRoutes()
+        
+        # Initialize routes
+        lora_routes.setup_routes(app)
+        checkpoints_routes.setup_routes(app)
+        ApiRoutes.setup_routes(app)
+        RecipeRoutes.setup_routes(app)
+        UpdateRoutes.setup_routes(app)
+        UsageStatsRoutes.setup_routes(app)
+        
+        # Schedule service initialization
+        app.on_startup.append(lambda app: cls._initialize_services())
+        
+        # Add cleanup
+        app.on_shutdown.append(cls._cleanup)
+        app.on_shutdown.append(ApiRoutes.cleanup)
+
+def parse_args():
+    """Parse command line arguments"""
+    parser = argparse.ArgumentParser(description="LoRA Manager Standalone Server")
+    parser.add_argument("--host", type=str, default="0.0.0.0", 
+                        help="Host to bind the server to")
+    parser.add_argument("--port", type=int, default=8188,
+                        help="Port to bind the server to")
+    parser.add_argument("--loras", type=str, nargs="+", 
+                        help="Additional paths to LoRA model directories (optional if settings.json has paths)")
+    parser.add_argument("--checkpoints", type=str, nargs="+",
+                        help="Additional paths to checkpoint model directories (optional if settings.json has paths)")
+    parser.add_argument("--log-level", type=str, default="INFO",
+                        choices=["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"],
+                        help="Logging level")
+    return parser.parse_args()
+
+async def main():
+    """Main entry point for standalone mode"""
+    args = parse_args()
+    
+    # Set log level
+    logging.getLogger().setLevel(getattr(logging, args.log_level))
+    
+    # Explicitly configure aiohttp access logger regardless of selected log level
+    logging.getLogger('aiohttp.access').setLevel(logging.WARNING)
+    
+    # Create the server instance
+    server = StandaloneServer()
+    
+    # Initialize routes via the standalone lora manager
+    StandaloneLoraManager.add_routes(server)
+    
+    # Set up and start the server
+    await server.setup()
+    await server.start(host=args.host, port=args.port)
+
+if __name__ == "__main__":
+    try:
+        # Run the main function
+        asyncio.run(main())
+    except KeyboardInterrupt:
+        logger.info("Server stopped by user")