Skip to main content
Glama

Adversary MCP Server

by brettbergin
logger.py8.18 kB
"""Centralized logging configuration for Adversary MCP server.""" import logging import logging.handlers import os from pathlib import Path from typing import Any class AdversaryLogger: """Centralized logger for Adversary MCP server with file rotation.""" _instance = None _initialized = False def __new__(cls, test_mode: bool = False) -> "AdversaryLogger": """Ensure singleton pattern.""" if cls._instance is None: cls._instance = super().__new__(cls) return cls._instance def __init__(self, test_mode: bool = False) -> None: """Initialize the logger (only once). Args: test_mode: If True, use test-specific log file """ if self._initialized: return self._initialized = True self.log_dir = ( Path.home() / ".local" / "share" / "adversary-mcp-server" / "logs" ) # Use different log file for tests if test_mode: self.log_file = self.log_dir / "test-adversary.log" else: self.log_file = self.log_dir / "adversary-mcp.log" # Create log directory self.log_dir.mkdir(parents=True, exist_ok=True) # Set up logging self._setup_logging() # Apply environment variable log level if set env_log_level = os.environ.get("ADVERSARY_LOG_LEVEL", "").upper() if env_log_level in ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]: self.set_log_level(env_log_level) def _setup_logging(self) -> None: """Set up the logging configuration with file rotation.""" # Create logger self.logger = logging.getLogger("adversary_mcp") # Clear existing handlers to avoid duplicates self.logger.handlers.clear() # Create file handler with rotation (5MB max, keep 5 backups) file_handler = logging.handlers.RotatingFileHandler( self.log_file, maxBytes=5 * 1024 * 1024, # 5MB backupCount=5, encoding="utf-8", ) file_handler.setLevel(logging.DEBUG) # Create formatter formatter = logging.Formatter( "%(asctime)s - %(name)s - %(levelname)s - %(filename)s::%(funcName)s::%(lineno)d - %(message)s", datefmt="%Y-%m-%d %H:%M:%S", ) # Set formatter for handlers file_handler.setFormatter(formatter) # Add handlers to logger self.logger.addHandler(file_handler) # Set default log level to INFO for better visibility (was WARNING) self.logger.setLevel(logging.INFO) # Prevent propagation to root logger self.logger.propagate = False # ALSO CAPTURE MCP FRAMEWORK LOGGING # Set up root logger to capture all MCP server framework logs root_logger = logging.getLogger() root_logger.setLevel(logging.DEBUG) # Add our file handler to root logger to capture MCP framework logs root_logger.addHandler(file_handler) # Capture specific MCP-related loggers mcp_loggers = ["mcp", "mcp.server", "mcp.types", "anyio", "starlette"] for logger_name in mcp_loggers: mcp_logger = logging.getLogger(logger_name) mcp_logger.setLevel(logging.DEBUG) mcp_logger.addHandler(file_handler) mcp_logger.propagate = False # Log initialization self.logger.info("Adversary MCP logging system initialized") self.logger.info(f"Log file: {self.log_file}") self.logger.info("MCP framework logging enabled") def get_logger(self, name: str | None = None) -> logging.Logger: """Get a logger instance. Args: name: Optional logger name (will be appended to 'adversary_mcp') Returns: Logger instance """ if name: logger_name = f"adversary_mcp.{name}" else: logger_name = "adversary_mcp" return logging.getLogger(logger_name) def info(self, message: str, **kwargs: Any) -> None: """Log an info message.""" self.logger.info(message, **kwargs) def warning(self, message: str, **kwargs: Any) -> None: """Log a warning message.""" self.logger.warning(message, **kwargs) def error(self, message: str, **kwargs: Any) -> None: """Log an error message.""" self.logger.error(message, **kwargs) def debug(self, message: str, **kwargs: Any) -> None: """Log a debug message.""" self.logger.debug(message, **kwargs) def exception(self, message: str, **kwargs: Any) -> None: """Log an exception with traceback.""" self.logger.exception(message, **kwargs) def set_log_level(self, level: str) -> None: """Set the logging level. Args: level: Logging level ('DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL') """ numeric_level = getattr(logging, level.upper(), logging.INFO) self.logger.setLevel(numeric_level) # Update file handler level for handler in self.logger.handlers: if isinstance(handler, logging.handlers.RotatingFileHandler): handler.setLevel(numeric_level) def get_log_file_path(self) -> Path: """Get the path to the current log file. Returns: Path to the log file """ return self.log_file def get_log_stats(self) -> dict[str, Any]: """Get statistics about the log files. Returns: Dictionary with log file statistics """ current_size = 0 total_size = 0 backup_files = [] try: if self.log_file.exists(): current_size = self.log_file.stat().st_size total_size = current_size # Find backup files for i in range(1, 6): # Check for .1 to .5 backup files backup_file = Path(f"{self.log_file}.{i}") if backup_file.exists(): backup_size = backup_file.stat().st_size backup_files.append({"file": str(backup_file), "size": backup_size}) total_size += backup_size except OSError: pass # Ignore file system errors stats = { "log_file": str(self.log_file), "log_dir": str(self.log_dir), "current_size": current_size, "backup_files": backup_files, "total_size": total_size, } return stats @classmethod def reset_for_tests(cls) -> None: """Reset the logger singleton for test isolation.""" cls._instance = None cls._initialized = False # Global logger instance _logger_instance = AdversaryLogger() def get_logger(name: str | None = None) -> logging.Logger: """Get a logger instance. Args: name: Optional logger name Returns: Logger instance """ return _logger_instance.get_logger(name) def log_info(message: str, **kwargs: Any) -> None: """Log an info message.""" _logger_instance.info(message, **kwargs) def log_warning(message: str, **kwargs: Any) -> None: """Log a warning message.""" _logger_instance.warning(message, **kwargs) def log_error(message: str, **kwargs: Any) -> None: """Log an error message.""" _logger_instance.error(message, **kwargs) def log_debug(message: str, **kwargs: Any) -> None: """Log a debug message.""" _logger_instance.debug(message, **kwargs) def log_exception(message: str, **kwargs: Any) -> None: """Log an exception with traceback.""" _logger_instance.exception(message, **kwargs) def set_log_level(level: str) -> None: """Set the logging level.""" _logger_instance.set_log_level(level) def get_log_stats() -> dict[str, Any]: """Get log statistics.""" return _logger_instance.get_log_stats() def setup_test_logging() -> None: """Set up logging for tests with a separate log file.""" global _logger_instance AdversaryLogger.reset_for_tests() _logger_instance = AdversaryLogger(test_mode=True)

MCP directory API

We provide all the information about MCP servers via our MCP API.

curl -X GET 'https://glama.ai/api/mcp/v1/servers/brettbergin/adversary-mcp-server'

If you have feedback or need assistance with the MCP directory API, please join our Discord server