#!/usr/bin/env python3
"""
Sistema de logging centralizado para MCP
"""
import logging
import logging.handlers
import os
from pathlib import Path
from datetime import datetime
class MCPLogger:
"""Logger centralizado para MCP con rotación automática"""
def __init__(self, name: str = "mcp-server", log_dir: str = "logs"):
self.name = name
self.log_dir = Path(log_dir)
self.log_dir.mkdir(exist_ok=True)
# Configurar logger
self.logger = logging.getLogger(name)
self.logger.setLevel(logging.DEBUG)
# Evitar duplicación de handlers
if self.logger.handlers:
return
# Formato de logs
formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(funcName)s:%(lineno)d - %(message)s'
)
# Handler para consola
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)
console_handler.setFormatter(formatter)
self.logger.addHandler(console_handler)
# Handler para archivo con rotación
log_file = self.log_dir / f"{name}.log"
file_handler = logging.handlers.RotatingFileHandler(
log_file,
maxBytes=5 * 1024 * 1024, # 5MB por archivo
backupCount=3, # 3 archivos de respaldo
encoding='utf-8'
)
file_handler.setLevel(logging.DEBUG)
file_handler.setFormatter(formatter)
self.logger.addHandler(file_handler)
# Log inicial
self.logger.info(f"Logger inicializado: {name}")
self.logger.info(f"Directorio de logs: {self.log_dir.absolute()}")
self.logger.info(f"Archivo principal: {log_file.absolute()}")
def get_logger(self):
"""Retorna el logger configurado"""
return self.logger
def log_mcp_request(self, request_type: str, data: dict = None):
"""Log específico para peticiones MCP"""
self.logger.debug(f"MCP {request_type}: {data}")
def log_mcp_response(self, response_type: str, data: dict = None):
"""Log específico para respuestas MCP"""
self.logger.debug(f"MCP {response_type}: {data}")
def log_error(self, error: Exception, context: str = ""):
"""Log de errores con contexto"""
self.logger.error(f"Error en {context}: {error}", exc_info=True)
def log_tool_call(self, tool_name: str, arguments: dict = None):
"""Log específico para llamadas a herramientas"""
self.logger.info(f"Llamada a herramienta: {tool_name}")
if arguments:
self.logger.debug(f"Argumentos: {arguments}")
def log_tool_result(self, tool_name: str, result: str = None):
"""Log específico para resultados de herramientas"""
self.logger.info(f"Resultado de {tool_name}: {len(result) if result else 0} caracteres")
if result and len(result) < 200: # Solo log completo si es corto
self.logger.debug(f"Contenido: {result}")
# Instancia global del logger
mcp_logger = MCPLogger()
logger = mcp_logger.get_logger()
def get_logger(name: str = None):
"""Función helper para obtener logger"""
if name:
return logging.getLogger(name)
return logger