""" Middleware Factory Provides factory methods for creating appropriate middleware instances based on target CLI tools and operation context. """ import logging from typing import Any from uuid import uuid4 from fastmcp import Context from .esptool_middleware import ESPToolMiddleware from .logger_interceptor import LoggerInterceptor, ToolNotFoundError logger = logging.getLogger(__name__) class MiddlewareFactory: """Factory for creating CLI tool middleware instances""" # Registry of available middleware classes _middleware_registry: dict[str, type[LoggerInterceptor]] = { "esptool": ESPToolMiddleware, } @classmethod def create_middleware( cls, tool_name: str, context: Context, operation_id: str | None = None, **kwargs ) -> LoggerInterceptor: """ Create middleware instance for specified CLI tool Args: tool_name: Name of the CLI tool (e.g., 'esptool') context: FastMCP context for logging and user interaction operation_id: Unique identifier for this operation **kwargs: Additional parameters for middleware initialization Returns: Configured middleware instance Raises: ToolNotFoundError: If tool is not supported """ if tool_name not in cls._middleware_registry: available_tools = ", ".join(cls._middleware_registry.keys()) raise ToolNotFoundError( f"No middleware available for tool: {tool_name}. Available tools: {available_tools}" ) # Generate operation ID if not provided if operation_id is None: operation_id = f"{tool_name}_{uuid4().hex[:8]}" # Get middleware class and create instance middleware_class = cls._middleware_registry[tool_name] try: middleware = middleware_class(context, operation_id, **kwargs) logger.info(f"Created {tool_name} middleware with operation ID: {operation_id}") return middleware except Exception as e: logger.error(f"Failed to create {tool_name} middleware: {e}") raise ToolNotFoundError(f"Failed to initialize {tool_name} middleware: {e}") @classmethod def register_middleware(cls, tool_name: str, middleware_class: type[LoggerInterceptor]) -> None: """ Register a new middleware class for a CLI tool Args: tool_name: Name of the CLI tool middleware_class: Middleware class that extends LoggerInterceptor """ if not issubclass(middleware_class, LoggerInterceptor): raise ValueError(f"Middleware class must extend LoggerInterceptor: {middleware_class}") cls._middleware_registry[tool_name] = middleware_class logger.info(f"Registered middleware for tool: {tool_name}") @classmethod def get_supported_tools(cls) -> dict[str, str]: """ Get list of supported CLI tools and their descriptions Returns: Dictionary mapping tool names to descriptions """ tool_descriptions = { "esptool": "ESP32/ESP8266 programming and debugging tool", } return { tool: tool_descriptions.get(tool, "CLI tool integration") for tool in cls._middleware_registry.keys() } @classmethod def is_tool_supported(cls, tool_name: str) -> bool: """Check if a CLI tool is supported by middleware""" return tool_name in cls._middleware_registry @classmethod def create_esptool_middleware( cls, context: Context, operation_id: str | None = None, **kwargs ) -> ESPToolMiddleware: """ Convenience method to create ESPTool middleware with proper typing Args: context: FastMCP context operation_id: Optional operation identifier **kwargs: Additional ESPTool-specific parameters Returns: Configured ESPToolMiddleware instance """ middleware = cls.create_middleware("esptool", context, operation_id, **kwargs) return middleware # Type checker knows this is ESPToolMiddleware @classmethod def get_middleware_info(cls, tool_name: str) -> dict[str, Any]: """ Get information about a specific middleware Args: tool_name: Name of the CLI tool Returns: Dictionary with middleware information """ if not cls.is_tool_supported(tool_name): return {"error": f"Tool not supported: {tool_name}"} middleware_class = cls._middleware_registry[tool_name] # Create temporary instance to get interaction points # (without context, for info purposes only) try: # Use a dummy context for information gathering class DummyContext: pass temp_instance = middleware_class(DummyContext(), "info_query") interaction_points = temp_instance.get_interaction_points() except Exception: interaction_points = [] return { "tool_name": tool_name, "middleware_class": middleware_class.__name__, "description": cls.get_supported_tools()[tool_name], "interaction_points": interaction_points, "module": middleware_class.__module__, }