MCP Data Analytics Server

# Servidor MCP Optimizado from mcp.server.fastmcp import FastMCP from pathlib import Path from typing import Dict, List, Optional, Union, Any import httpx import pandas as pd import PyPDF2 import plotly.express as px import plotly.graph_objects as go from bs4 import BeautifulSoup import logging from datetime import datetime import json # Configuración de logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Instancia del servidor mcp = FastMCP("Demo") # ========== CLASES AUXILIARES ========== class FileHandler: """Manejador centralizado de archivos""" @staticmethod def validate_file_exists(path: Path) -> None: """Valida que el archivo exista""" if not path.exists(): raise FileNotFoundError(f"El archivo {path} no existe") @staticmethod def read_dataframe(path: Path) -> pd.DataFrame: """Lee un DataFrame desde CSV o Excel""" FileHandler.validate_file_exists(path) if path.suffix.lower() == ".csv": return pd.read_csv(path) elif path.suffix.lower() in [".xlsx", ".xls"]: return pd.read_excel(path) elif path.suffix.lower() == ".json": return pd.read_json(path) else: raise ValueError(f"Formato no soportado: {path.suffix}") @staticmethod def validate_columns(df: pd.DataFrame, columns: List[str]) -> None: """Valida que las columnas existan en el DataFrame""" missing_cols = [col for col in columns if col not in df.columns] if missing_cols: raise ValueError(f"Columnas no encontradas: {', '.join(missing_cols)}") class ResponseFormatter: """Formateador estándar de respuestas""" @staticmethod def success(data: Any, message: str = "") -> Dict: """Formato de respuesta exitosa""" return { "success": True, "data": data, "message": message, "timestamp": datetime.now().isoformat() } @staticmethod def error(error: str, details: str = "") -> Dict: """Formato de respuesta de error""" return { "success": False, "error": error, "details": details, "timestamp": datetime.now().isoformat() } # ========== HERRAMIENTAS BÁSICAS DE ARCHIVOS ========== @mcp.tool() def analizar_archivo(archivo: Path) -> Dict: """Analiza las propiedades completas de un archivo""" try: FileHandler.validate_file_exists(Path(archivo)) path = Path(archivo) stat = path.stat() data = { "nombre": path.name, "extension": path.suffix, "tamaño_bytes": stat.st_size, "tamaño_legible": _format_file_size(stat.st_size), "fecha_creacion": datetime.fromtimestamp(stat.st_ctime).isoformat(), "fecha_modificacion": datetime.fromtimestamp(stat.st_mtime).isoformat(), "es_archivo": path.is_file(), "es_directorio": path.is_dir(), } # Si es archivo de texto pequeño, incluir contenido if path.is_file() and path.suffix.lower() in ['.txt', '.md', '.py', '.json'] and stat.st_size < 10000: try: data["contenido_preview"] = path.read_text(encoding="utf-8")[:1000] except: data["contenido_preview"] = "No se pudo leer el contenido" return ResponseFormatter.success(data, f"Archivo {path.name} analizado exitosamente") except Exception as e: logger.error(f"Error analizando archivo: {e}") return ResponseFormatter.error(str(e)) @mcp.tool() def crear_archivo(ruta: Path, contenido: str, encoding: str = "utf-8") -> Dict: """Crea un archivo con el contenido especificado""" try: path = Path(ruta) # Crear directorio padre si no existe path.parent.mkdir(parents=True, exist_ok=True) path.write_text(contenido, encoding=encoding) return ResponseFormatter.success( {"ruta": str(path), "tamaño": len(contenido.encode(encoding))}, f"Archivo {path.name} creado exitosamente" ) except Exception as e: logger.error(f"Error creando archivo: {e}") return ResponseFormatter.error(str(e)) # ========== HERRAMIENTAS DE LECTURA DE DOCUMENTOS ========== @mcp.tool() def leer_documento(ruta: Path, limite_caracteres: int = 10000) -> Dict: """Lee documentos de texto, PDF con límite configurable""" try: path = Path(ruta) FileHandler.validate_file_exists(path) if path.suffix.lower() == ".pdf": contenido = _extract_pdf_text(path) elif path.suffix.lower() in [".txt", ".md", ".csv", ".json", ".py"]: contenido = path.read_text(encoding="utf-8") else: raise ValueError(f"Formato no soportado: {path.suffix}") # Aplicar límite de caracteres contenido_limitado = contenido[:limite_caracteres] fue_truncado = len(contenido) > limite_caracteres data = { "contenido": contenido_limitado, "longitud_total": len(contenido), "longitud_retornada": len(contenido_limitado), "fue_truncado": fue_truncado } return ResponseFormatter.success(data, f"Documento {path.name} leído exitosamente") except Exception as e: logger.error(f"Error leyendo documento: {e}") return ResponseFormatter.error(str(e)) # ========== HERRAMIENTAS DE ANÁLISIS DE DATOS ========== @mcp.tool() def analizar_datos(ruta_archivo: Path, incluir_muestra: bool = True) -> Dict: """Análisis completo de archivos de datos (CSV/Excel/JSON)""" try: df = FileHandler.read_dataframe(Path(ruta_archivo)) # Información básica info_basica = { "filas": len(df), "columnas": len(df.columns), "lista_columnas": list(df.columns), "tipos_datos": df.dtypes.to_dict(), } # Estadísticas descriptivas para columnas numéricas numericas = df.select_dtypes(include=['number']) estadisticas = {} if not numericas.empty: estadisticas = numericas.describe().to_dict() # Información de valores faltantes valores_faltantes = df.isnull().sum().to_dict() # Muestra de datos (opcional) muestra = {} if incluir_muestra: muestra = { "primeras_5_filas": df.head().to_dict('records'), "ultimas_5_filas": df.tail().to_dict('records') } data = { "info_basica": info_basica, "estadisticas_numericas": estadisticas, "valores_faltantes": valores_faltantes, "muestra_datos": muestra } return ResponseFormatter.success(data, f"Análisis de {Path(ruta_archivo).name} completado") except Exception as e: logger.error(f"Error analizando datos: {e}") return ResponseFormatter.error(str(e)) @mcp.tool() def tabla_dinamica_avanzada( ruta_archivo: Path, index_cols: Union[str, List[str]], columns_cols: Union[str, List[str]], values_col: str, aggfunc: Union[str, List[str]] = "mean", fill_value: Optional[float] = None ) -> Dict: """Crea tablas dinámicas avanzadas con múltiples funciones de agregación""" try: df = FileHandler.read_dataframe(Path(ruta_archivo)) # Normalizar inputs if isinstance(index_cols, str): index_cols = [index_cols] if isinstance(columns_cols, str): columns_cols = [columns_cols] if isinstance(aggfunc, str): aggfunc = [aggfunc] # Validar columnas todas_columnas = index_cols + columns_cols + [values_col] FileHandler.validate_columns(df, todas_columnas) # Validar funciones de agregación funciones_validas = ["mean", "sum", "min", "max", "count", "std", "var", "median"] funciones_invalidas = [f for f in aggfunc if f not in funciones_validas] if funciones_invalidas: raise ValueError(f"Funciones no válidas: {funciones_invalidas}. Use: {funciones_validas}") # Crear tabla dinámica pivot_table = df.pivot_table( index=index_cols, columns=columns_cols, values=values_col, aggfunc=aggfunc, fill_value=fill_value ) # Convertir a formato serializable if len(aggfunc) == 1: resultado = pivot_table.to_dict() else: resultado = {func: pivot_table[func].to_dict() for func in aggfunc} data = { "tabla_dinamica": resultado, "metadata": { "index_columnas": index_cols, "columnas_pivote": columns_cols, "columna_valores": values_col, "funciones_agregacion": aggfunc, "dimensiones": pivot_table.shape } } return ResponseFormatter.success(data, "Tabla dinámica creada exitosamente") except Exception as e: logger.error(f"Error creando tabla dinámica: {e}") return ResponseFormatter.error(str(e)) # ========== HERRAMIENTAS DE VISUALIZACIÓN ========== @mcp.tool() def crear_visualizacion( ruta_archivo: Path, tipo_grafico: str, configuracion: Dict[str, Any] ) -> Dict: """Sistema unificado de visualización con Plotly""" try: df = FileHandler.read_dataframe(Path(ruta_archivo)) # Validar configuración mínima if 'columnas' not in configuracion: raise ValueError("Debe especificar 'columnas' en la configuración") columnas = configuracion['columnas'] FileHandler.validate_columns(df, columnas) # Crear figura según tipo fig = _create_plotly_figure(df, tipo_grafico, configuracion) # Personalizar figura titulo = configuracion.get('titulo', f'Gráfico {tipo_grafico}') fig.update_layout(title=titulo, template=configuracion.get('template', 'plotly')) # Guardar ruta_salida = Path(ruta_archivo).with_suffix('.html') fig.write_html(ruta_salida) data = { "archivo_salida": str(ruta_salida), "tipo_grafico": tipo_grafico, "columnas_utilizadas": columnas } return ResponseFormatter.success(data, f"Visualización {tipo_grafico} creada exitosamente") except Exception as e: logger.error(f"Error creando visualización: {e}") return ResponseFormatter.error(str(e)) # ========== HERRAMIENTAS WEB ========== @mcp.tool() async def buscar_repositorios_github( query: str, sort: str = "stars", order: str = "desc", per_page: int = 10 ) -> Dict: """Búsqueda avanzada de repositorios en GitHub""" try: async with httpx.AsyncClient(timeout=30) as client: url = "https://api.github.com/search/repositories" params = { "q": query, "sort": sort, "order": order, "per_page": min(per_page, 100) # Límite de API } response = await client.get(url, params=params) response.raise_for_status() data_json = response.json() repositorios = [] for repo in data_json.get("items", []): repositorios.append({ "nombre_completo": repo["full_name"], "descripcion": repo.get("description", "Sin descripción"), "url": repo["html_url"], "estrellas": repo["stargazers_count"], "forks": repo["forks_count"], "lenguaje": repo.get("language", "No especificado"), "actualizado": repo["updated_at"], "temas": repo.get("topics", []) }) data = { "repositorios": repositorios, "total_encontrados": data_json["total_count"], "query_utilizada": query } return ResponseFormatter.success(data, f"Se encontraron {len(repositorios)} repositorios") except Exception as e: logger.error(f"Error buscando repositorios: {e}") return ResponseFormatter.error(str(e)) @mcp.tool() async def extraer_contenido_web(url: str, selector_css: Optional[str] = None) -> Dict: """Extrae contenido de páginas web con opciones avanzadas""" try: async with httpx.AsyncClient(timeout=30) as client: response = await client.get(url) response.raise_for_status() soup = BeautifulSoup(response.text, "html.parser") if selector_css: elementos = soup.select(selector_css) contenido = "\n".join([elem.get_text(strip=True) for elem in elementos]) else: contenido = soup.get_text(strip=True) data = { "url": url, "contenido": contenido[:10000], # Límite para evitar respuestas muy grandes "longitud_total": len(contenido), "titulo": soup.title.string if soup.title else "Sin título", "selector_usado": selector_css } return ResponseFormatter.success(data, "Contenido web extraído exitosamente") except Exception as e: logger.error(f"Error extrayendo contenido web: {e}") return ResponseFormatter.error(str(e)) @mcp.tool() async def descargar_archivo_web(url: str, ruta_destino: Path, sobrescribir: bool = False) -> Dict: """Descarga archivos desde URLs con validaciones mejoradas""" try: path = Path(ruta_destino) # Verificar si el archivo existe if path.exists() and not sobrescribir: raise FileExistsError(f"El archivo {path} ya existe. Use sobrescribir=True para reemplazarlo") async with httpx.AsyncClient(timeout=60) as client: response = await client.get(url) response.raise_for_status() # Crear directorio padre si no existe path.parent.mkdir(parents=True, exist_ok=True) # Guardar archivo path.write_bytes(response.content) data = { "url": url, "archivo_destino": str(path), "tamaño_bytes": len(response.content), "tamaño_legible": _format_file_size(len(response.content)), "tipo_contenido": response.headers.get("content-type", "desconocido") } return ResponseFormatter.success(data, "Archivo descargado exitosamente") except Exception as e: logger.error(f"Error descargando archivo: {e}") return ResponseFormatter.error(str(e)) # ========== HERRAMIENTAS DE CONVERSIÓN ========== @mcp.tool() def convertir_formato_datos( archivo_entrada: Path, formato_salida: str, opciones: Dict[str, Any] = None ) -> Dict: """Conversión mejorada entre formatos de datos""" try: if opciones is None: opciones = {} df = FileHandler.read_dataframe(Path(archivo_entrada)) # Generar nombre de archivo de salida archivo_salida = Path(archivo_entrada).with_suffix(f".{formato_salida}") # Opciones por defecto para cada formato opciones_defecto = { "json": {"orient": "records", "indent": 2}, "csv": {"index": False, "encoding": "utf-8"}, "xlsx": {"index": False}, "parquet": {"index": False} } # Combinar opciones opts_finales = {**opciones_defecto.get(formato_salida, {}), **opciones} # Realizar conversión if formato_salida == "json": df.to_json(archivo_salida, **opts_finales) elif formato_salida == "csv": df.to_csv(archivo_salida, **opts_finales) elif formato_salida == "xlsx": df.to_excel(archivo_salida, **opts_finales) elif formato_salida == "parquet": df.to_parquet(archivo_salida, **opts_finales) else: formatos_soportados = ["json", "csv", "xlsx", "parquet"] raise ValueError(f"Formato '{formato_salida}' no soportado. Use: {formatos_soportados}") data = { "archivo_entrada": str(archivo_entrada), "archivo_salida": str(archivo_salida), "formato_origen": Path(archivo_entrada).suffix, "formato_destino": f".{formato_salida}", "filas_procesadas": len(df) } return ResponseFormatter.success(data, f"Conversión a {formato_salida} completada") except Exception as e: logger.error(f"Error en conversión: {e}") return ResponseFormatter.error(str(e)) # ========== FUNCIONES AUXILIARES ========== def _format_file_size(size_bytes: int) -> str: """Convierte bytes a formato legible""" for unit in ['B', 'KB', 'MB', 'GB']: if size_bytes < 1024: return f"{size_bytes:.1f} {unit}" size_bytes /= 1024 return f"{size_bytes:.1f} TB" def _extract_pdf_text(path: Path) -> str: """Extrae texto de archivos PDF""" with open(path, "rb") as file: reader = PyPDF2.PdfReader(file) return " ".join([page.extract_text() for page in reader.pages]) def _create_plotly_figure(df: pd.DataFrame, tipo: str, config: Dict) -> go.Figure: """Crea figuras de Plotly según tipo y configuración""" columnas = config['columnas'] if tipo == "bar": return px.bar(df, x=columnas[0], y=columnas[1], **config.get('extra_params', {})) elif tipo == "line": return px.line(df, x=columnas[0], y=columnas[1], **config.get('extra_params', {})) elif tipo == "scatter": return px.scatter(df, x=columnas[0], y=columnas[1], **config.get('extra_params', {})) elif tipo == "pie": return px.pie(df, names=columnas[0], values=columnas[1], **config.get('extra_params', {})) elif tipo == "histogram": return px.histogram(df, x=columnas[0], **config.get('extra_params', {})) elif tipo == "box": return px.box(df, x=columnas[0], y=columnas[1] if len(columnas) > 1 else None, **config.get('extra_params', {})) else: tipos_soportados = ["bar", "line", "scatter", "pie", "histogram", "box"] raise ValueError(f"Tipo '{tipo}' no soportado. Use: {tipos_soportados}") # ========== EJECUCIÓN DEL SERVIDOR ========== if __name__ == "__main__": mcp.run(transport="stdio")