"""Logging configuration for the File System MCP Server."""
import logging
import logging.handlers
import sys
from pathlib import Path
from typing import Optional
def setup_logging(
log_level: str = "INFO",
log_file: Optional[str] = None,
max_file_size: int = 10 * 1024 * 1024, # 10MB
backup_count: int = 5
) -> None:
"""Set up logging configuration."""
# Create formatter
formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
# Get root logger
root_logger = logging.getLogger()
root_logger.setLevel(getattr(logging, log_level.upper(), logging.INFO))
# Clear existing handlers
root_logger.handlers.clear()
# Console handler
console_handler = logging.StreamHandler(sys.stdout)
console_handler.setFormatter(formatter)
console_handler.setLevel(logging.INFO)
root_logger.addHandler(console_handler)
# File handler (if specified)
if log_file:
try:
log_path = Path(log_file)
log_path.parent.mkdir(parents=True, exist_ok=True)
file_handler = logging.handlers.RotatingFileHandler(
log_file,
maxBytes=max_file_size,
backupCount=backup_count,
encoding='utf-8'
)
file_handler.setFormatter(formatter)
file_handler.setLevel(getattr(logging, log_level.upper(), logging.INFO))
root_logger.addHandler(file_handler)
except Exception as e:
root_logger.warning(f"Failed to set up file logging: {e}")
# Set specific logger levels
logging.getLogger("mcp").setLevel(logging.WARNING)
logging.getLogger("urllib3").setLevel(logging.WARNING)
logging.getLogger("requests").setLevel(logging.WARNING)
class StructuredLogger:
"""Structured logger for consistent error reporting."""
def __init__(self, name: str):
self.logger = logging.getLogger(name)
def log_operation_start(self, operation: str, path: str, **kwargs) -> None:
"""Log the start of a file operation."""
extra_info = " ".join(f"{k}={v}" for k, v in kwargs.items())
self.logger.info(f"Starting {operation} on {path} {extra_info}".strip())
def log_operation_success(self, operation: str, path: str, **kwargs) -> None:
"""Log successful completion of a file operation."""
extra_info = " ".join(f"{k}={v}" for k, v in kwargs.items())
self.logger.info(f"Completed {operation} on {path} {extra_info}".strip())
def log_operation_error(self, operation: str, path: str, error: Exception, **kwargs) -> None:
"""Log error during file operation."""
extra_info = " ".join(f"{k}={v}" for k, v in kwargs.items())
self.logger.error(
f"Failed {operation} on {path}: {str(error)} {extra_info}".strip(),
exc_info=True
)
def log_security_violation(self, operation: str, path: str, reason: str) -> None:
"""Log security violations."""
self.logger.warning(f"Security violation: {operation} on {path} - {reason}")
def log_backup_created(self, original_path: str, backup_path: str) -> None:
"""Log backup creation."""
self.logger.info(f"Backup created: {original_path} -> {backup_path}")
def log_backup_restored(self, backup_path: str, original_path: str) -> None:
"""Log backup restoration."""
self.logger.info(f"Backup restored: {backup_path} -> {original_path}")
def log_config_loaded(self, config_path: str, issues: list = None) -> None:
"""Log configuration loading."""
self.logger.info(f"Configuration loaded from {config_path}")
if issues:
for issue in issues:
self.logger.warning(f"Config issue: {issue}")
def log_server_start(self, version: str) -> None:
"""Log server startup."""
self.logger.info(f"File System MCP Server v{version} starting")
def log_server_stop(self) -> None:
"""Log server shutdown."""
self.logger.info("File System MCP Server stopping")
def log_tool_call(self, tool_name: str, arguments: dict) -> None:
"""Log MCP tool calls."""
# Don't log file content for privacy
safe_args = {k: v for k, v in arguments.items() if k != "content"}
if "content" in arguments:
safe_args["content_length"] = len(arguments["content"])
self.logger.debug(f"Tool call: {tool_name} with args: {safe_args}")
# Global structured logger instance
structured_logger = StructuredLogger("file_system_mcp_server")