server.py•27.2 kB
"""
Servidor MCP para automatización de navegadores con Selenium WebDriver.
"""
import asyncio
import json
from typing import Dict, Any, Optional
from contextlib import asynccontextmanager
from collections.abc import AsyncIterator
from mcp.server.fastmcp import FastMCP, Context
from config import ServerConfig, DEFAULT_CONFIG, STEALTH_CONFIG, PERFORMANCE_CONFIG
from session_manager import SessionManager
from selenium_tools import SeleniumTools
from stealth_features import StealthFeatures
from browser_detection import BrowserDetection
class SeleniumMCPServer:
"""Contexto de la aplicación para el servidor MCP Selenium."""
def __init__(self, config: ServerConfig = None):
self.config = config or DEFAULT_CONFIG
self.session_manager = SessionManager(self.config)
self.selenium_tools = SeleniumTools(self.session_manager)
self.stealth_features = StealthFeatures(self.session_manager)
self.browser_detection = BrowserDetection()
@asynccontextmanager
async def app_lifespan(server: FastMCP) -> AsyncIterator[SeleniumMCPServer]:
"""Gestiona el ciclo de vida de la aplicación."""
# Inicialización
app_context = SeleniumMCPServer()
try:
yield app_context
finally:
# Limpieza al cerrar
app_context.session_manager.close_all_sessions()
# Crear el servidor MCP
mcp = FastMCP(
"Selenium WebDriver Server",
dependencies=[
"selenium>=4.15.0",
"webdriver-manager>=4.0.0",
"undetected-chromedriver>=3.5.0",
"pydantic>=2.0.0"
],
lifespan=app_lifespan
)
@mcp.resource("config://server")
def get_server_config() -> str:
"""Obtiene la configuración actual del servidor."""
ctx = mcp.get_context()
app_context: SeleniumMCPServer = ctx.request_context.lifespan_context
config_dict = app_context.config.model_dump()
return json.dumps(config_dict, indent=2)
@mcp.resource("sessions://active")
def get_active_sessions() -> str:
"""Lista todas las sesiones activas."""
ctx = mcp.get_context()
app_context: SeleniumMCPServer = ctx.request_context.lifespan_context
sessions = app_context.session_manager.list_sessions()
return json.dumps(sessions, indent=2)
@mcp.tool()
def get_server_status() -> Dict[str, Any]:
"""Obtiene el estado actual del servidor."""
ctx = mcp.get_context()
app_context: SeleniumMCPServer = ctx.request_context.lifespan_context
return {
"status": "running",
"active_sessions": app_context.session_manager.get_session_count(),
"max_sessions": app_context.config.max_sessions,
"default_browser": app_context.config.default_browser,
"session_timeout": app_context.config.session_timeout
}
@mcp.tool()
def cleanup_expired_sessions() -> Dict[str, Any]:
"""Limpia las sesiones expiradas manualmente."""
ctx = mcp.get_context()
app_context: SeleniumMCPServer = ctx.request_context.lifespan_context
sessions_before = app_context.session_manager.get_session_count()
app_context.session_manager.cleanup_expired_sessions()
sessions_after = app_context.session_manager.get_session_count()
return {
"sessions_before": sessions_before,
"sessions_after": sessions_after,
"sessions_cleaned": sessions_before - sessions_after
}
@mcp.tool()
def update_server_config(config_preset: str = "default") -> Dict[str, Any]:
"""Actualiza la configuración del servidor con un preset predefinido."""
ctx = mcp.get_context()
app_context: SeleniumMCPServer = ctx.request_context.lifespan_context
config_presets = {
"default": DEFAULT_CONFIG,
"stealth": STEALTH_CONFIG,
"performance": PERFORMANCE_CONFIG
}
if config_preset not in config_presets:
return {
"success": False,
"error": f"Preset '{config_preset}' no encontrado. Disponibles: {list(config_presets.keys())}"
}
# Cerrar todas las sesiones existentes antes de cambiar la configuración
app_context.session_manager.close_all_sessions()
# Actualizar configuración
app_context.config = config_presets[config_preset]
app_context.session_manager.config = app_context.config
return {
"success": True,
"new_config": config_preset,
"message": "Configuración actualizada. Todas las sesiones anteriores fueron cerradas."
}
if __name__ == "__main__":
mcp.run()
# ===== HERRAMIENTAS DE SELENIUM WEBDRIVER =====
@mcp.tool()
def start_browser(
browser_type: str = "chrome",
options: Optional[Dict[str, Any]] = None,
proxy: Optional[Dict[str, str]] = None,
detection_evasion: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
"""
Inicia una nueva sesión de navegador.
Args:
browser_type: Tipo de navegador ("chrome" o "firefox")
options: Opciones del navegador (headless, window_size, etc.)
proxy: Configuración de proxy (http, https, socks)
detection_evasion: Configuraciones para evasión de detección
"""
ctx = mcp.get_context()
app_context: SeleniumMCPServer = ctx.request_context.lifespan_context
return app_context.selenium_tools.start_browser(
browser_type=browser_type,
options=options,
proxy=proxy,
detection_evasion=detection_evasion
)
@mcp.tool()
def navigate_to_url(session_id: str, url: str) -> Dict[str, Any]:
"""
Navega a una URL específica.
Args:
session_id: ID de la sesión del navegador
url: URL a la que navegar
"""
ctx = mcp.get_context()
app_context: SeleniumMCPServer = ctx.request_context.lifespan_context
return app_context.selenium_tools.navigate_to_url(session_id, url)
@mcp.tool()
def find_element(
session_id: str,
strategy: str,
value: str,
timeout: int = 10,
multiple: bool = False
) -> Dict[str, Any]:
"""
Encuentra uno o múltiples elementos usando la estrategia especificada.
Args:
session_id: ID de la sesión del navegador
strategy: Estrategia de localización (id, name, class_name, tag_name, css_selector, xpath, link_text, partial_link_text)
value: Valor del localizador
timeout: Tiempo de espera en segundos
multiple: Si buscar múltiples elementos
"""
ctx = mcp.get_context()
app_context: SeleniumMCPServer = ctx.request_context.lifespan_context
return app_context.selenium_tools.find_element(
session_id=session_id,
strategy=strategy,
value=value,
timeout=timeout,
multiple=multiple
)
@mcp.tool()
def click_element(
session_id: str,
strategy: str,
value: str,
timeout: int = 10,
element_index: int = 0
) -> Dict[str, Any]:
"""
Hace clic en un elemento.
Args:
session_id: ID de la sesión del navegador
strategy: Estrategia de localización
value: Valor del localizador
timeout: Tiempo de espera en segundos
element_index: Índice del elemento si hay múltiples coincidencias
"""
ctx = mcp.get_context()
app_context: SeleniumMCPServer = ctx.request_context.lifespan_context
return app_context.selenium_tools.click_element(
session_id=session_id,
strategy=strategy,
value=value,
timeout=timeout,
element_index=element_index
)
@mcp.tool()
def type_text(
session_id: str,
strategy: str,
value: str,
text: str,
timeout: int = 10,
element_index: int = 0,
clear_first: bool = True
) -> Dict[str, Any]:
"""
Escribe texto en un elemento.
Args:
session_id: ID de la sesión del navegador
strategy: Estrategia de localización
value: Valor del localizador
text: Texto a escribir
timeout: Tiempo de espera en segundos
element_index: Índice del elemento si hay múltiples coincidencias
clear_first: Si limpiar el campo antes de escribir
"""
ctx = mcp.get_context()
app_context: SeleniumMCPServer = ctx.request_context.lifespan_context
return app_context.selenium_tools.type_text(
session_id=session_id,
strategy=strategy,
value=value,
text=text,
timeout=timeout,
element_index=element_index,
clear_first=clear_first
)
@mcp.tool()
def take_screenshot(
session_id: str,
file_path: Optional[str] = None,
element_strategy: Optional[str] = None,
element_value: Optional[str] = None,
element_index: int = 0
) -> Dict[str, Any]:
"""
Toma una captura de pantalla.
Args:
session_id: ID de la sesión del navegador
file_path: Ruta donde guardar la imagen (opcional)
element_strategy: Estrategia para capturar solo un elemento específico
element_value: Valor del localizador del elemento
element_index: Índice del elemento si hay múltiples coincidencias
"""
ctx = mcp.get_context()
app_context: SeleniumMCPServer = ctx.request_context.lifespan_context
return app_context.selenium_tools.take_screenshot(
session_id=session_id,
file_path=file_path,
element_strategy=element_strategy,
element_value=element_value,
element_index=element_index
)
@mcp.tool()
def upload_file(
session_id: str,
strategy: str,
value: str,
file_path: str,
timeout: int = 10,
element_index: int = 0
) -> Dict[str, Any]:
"""
Sube un archivo a un elemento input[type=file].
Args:
session_id: ID de la sesión del navegador
strategy: Estrategia de localización del input file
value: Valor del localizador
file_path: Ruta del archivo a subir
timeout: Tiempo de espera en segundos
element_index: Índice del elemento si hay múltiples coincidencias
"""
ctx = mcp.get_context()
app_context: SeleniumMCPServer = ctx.request_context.lifespan_context
return app_context.selenium_tools.upload_file(
session_id=session_id,
strategy=strategy,
value=value,
file_path=file_path,
timeout=timeout,
element_index=element_index
)
@mcp.tool()
def execute_script(session_id: str, script: str, *args) -> Dict[str, Any]:
"""
Ejecuta JavaScript en el navegador.
Args:
session_id: ID de la sesión del navegador
script: Código JavaScript a ejecutar
*args: Argumentos para el script
"""
ctx = mcp.get_context()
app_context: SeleniumMCPServer = ctx.request_context.lifespan_context
return app_context.selenium_tools.execute_script(session_id, script, *args)
@mcp.tool()
def perform_mouse_action(
session_id: str,
action_type: str,
strategy: Optional[str] = None,
value: Optional[str] = None,
element_index: int = 0,
x_offset: int = 0,
y_offset: int = 0,
target_strategy: Optional[str] = None,
target_value: Optional[str] = None,
target_index: int = 0
) -> Dict[str, Any]:
"""
Realiza acciones de ratón (hover, drag and drop, etc.).
Args:
session_id: ID de la sesión del navegador
action_type: Tipo de acción (hover, drag_and_drop, right_click, double_click)
strategy: Estrategia de localización del elemento origen
value: Valor del localizador del elemento origen
element_index: Índice del elemento origen
x_offset: Offset X para movimientos relativos
y_offset: Offset Y para movimientos relativos
target_strategy: Estrategia del elemento destino (para drag_and_drop)
target_value: Valor del localizador del elemento destino
target_index: Índice del elemento destino
"""
ctx = mcp.get_context()
app_context: SeleniumMCPServer = ctx.request_context.lifespan_context
return app_context.selenium_tools.perform_mouse_action(
session_id=session_id,
action_type=action_type,
strategy=strategy,
value=value,
element_index=element_index,
x_offset=x_offset,
y_offset=y_offset,
target_strategy=target_strategy,
target_value=target_value,
target_index=target_index
)
@mcp.tool()
def send_keys(
session_id: str,
keys: str,
strategy: Optional[str] = None,
value: Optional[str] = None,
element_index: int = 0
) -> Dict[str, Any]:
"""
Envía teclas especiales (Enter, Tab, etc.) al navegador o a un elemento específico.
Args:
session_id: ID de la sesión del navegador
keys: Teclas a enviar (ENTER, TAB, ESCAPE, etc.)
strategy: Estrategia de localización del elemento (opcional)
value: Valor del localizador del elemento (opcional)
element_index: Índice del elemento si hay múltiples coincidencias
"""
ctx = mcp.get_context()
app_context: SeleniumMCPServer = ctx.request_context.lifespan_context
return app_context.selenium_tools.send_keys(
session_id=session_id,
keys=keys,
strategy=strategy,
value=value,
element_index=element_index
)
@mcp.tool()
def close_browser(session_id: str) -> Dict[str, Any]:
"""
Cierra una sesión de navegador.
Args:
session_id: ID de la sesión del navegador a cerrar
"""
ctx = mcp.get_context()
app_context: SeleniumMCPServer = ctx.request_context.lifespan_context
return app_context.selenium_tools.close_browser(session_id)
@mcp.tool()
def get_page_info(session_id: str) -> Dict[str, Any]:
"""
Obtiene información de la página actual.
Args:
session_id: ID de la sesión del navegador
"""
ctx = mcp.get_context()
app_context: SeleniumMCPServer = ctx.request_context.lifespan_context
return app_context.selenium_tools.get_page_info(session_id)
# ===== HERRAMIENTAS DE EVASIÓN DE DETECCIÓN =====
@mcp.tool()
def apply_stealth_mode(session_id: str) -> Dict[str, Any]:
"""
Aplica todas las técnicas de evasión de detección a una sesión.
Args:
session_id: ID de la sesión del navegador
"""
ctx = mcp.get_context()
app_context: SeleniumMCPServer = ctx.request_context.lifespan_context
return app_context.stealth_features.apply_stealth_mode(session_id)
@mcp.tool()
def randomize_user_agent(session_id: str) -> Dict[str, Any]:
"""
Cambia el user agent a uno aleatorio.
Args:
session_id: ID de la sesión del navegador
"""
ctx = mcp.get_context()
app_context: SeleniumMCPServer = ctx.request_context.lifespan_context
return app_context.stealth_features.randomize_user_agent(session_id)
@mcp.tool()
def randomize_viewport(session_id: str) -> Dict[str, Any]:
"""
Cambia el tamaño de la ventana a una resolución aleatoria.
Args:
session_id: ID de la sesión del navegador
"""
ctx = mcp.get_context()
app_context: SeleniumMCPServer = ctx.request_context.lifespan_context
return app_context.stealth_features.randomize_viewport(session_id)
@mcp.tool()
def manage_cookies(
session_id: str,
action: str,
cookie_data: Optional[Dict[str, Any]] = None,
cookie_name: Optional[str] = None
) -> Dict[str, Any]:
"""
Gestiona cookies de la sesión.
Args:
session_id: ID de la sesión del navegador
action: Acción a realizar (get_all, add, delete, get)
cookie_data: Datos de la cookie para agregar
cookie_name: Nombre de la cookie para operaciones específicas
"""
ctx = mcp.get_context()
app_context: SeleniumMCPServer = ctx.request_context.lifespan_context
return app_context.stealth_features.manage_cookies(
session_id=session_id,
action=action,
cookie_data=cookie_data,
cookie_name=cookie_name
)
@mcp.tool()
def add_random_delay(
min_seconds: float = 0.5,
max_seconds: float = 2.0
) -> Dict[str, Any]:
"""
Agrega un delay aleatorio entre acciones.
Args:
min_seconds: Tiempo mínimo de delay en segundos
max_seconds: Tiempo máximo de delay en segundos
"""
ctx = mcp.get_context()
app_context: SeleniumMCPServer = ctx.request_context.lifespan_context
return app_context.stealth_features.add_random_delay(min_seconds, max_seconds)
@mcp.tool()
def simulate_human_typing(
session_id: str,
strategy: str,
value: str,
text: str,
min_delay: float = 0.05,
max_delay: float = 0.15,
element_index: int = 0
) -> Dict[str, Any]:
"""
Simula escritura humana con delays aleatorios entre teclas.
Args:
session_id: ID de la sesión del navegador
strategy: Estrategia de localización del elemento
value: Valor del localizador
text: Texto a escribir
min_delay: Delay mínimo entre teclas
max_delay: Delay máximo entre teclas
element_index: Índice del elemento si hay múltiples coincidencias
"""
ctx = mcp.get_context()
app_context: SeleniumMCPServer = ctx.request_context.lifespan_context
return app_context.stealth_features.simulate_human_typing(
session_id=session_id,
strategy=strategy,
value=value,
text=text,
min_delay=min_delay,
max_delay=max_delay,
element_index=element_index
)
@mcp.tool()
def scroll_like_human(
session_id: str,
direction: str = "down",
distance: int = 300,
steps: int = 5
) -> Dict[str, Any]:
"""
Simula scroll humano con movimientos graduales.
Args:
session_id: ID de la sesión del navegador
direction: Dirección del scroll (up, down)
distance: Distancia total a scrollear en píxeles
steps: Número de pasos para dividir el movimiento
"""
ctx = mcp.get_context()
app_context: SeleniumMCPServer = ctx.request_context.lifespan_context
return app_context.stealth_features.scroll_like_human(
session_id=session_id,
direction=direction,
distance=distance,
steps=steps
)
# ===== HERRAMIENTAS DE GESTIÓN DE NAVEGADORES =====
@mcp.tool()
def detect_available_browsers() -> Dict[str, Any]:
"""
Detecta todos los navegadores disponibles en el sistema.
"""
ctx = mcp.get_context()
app_context: SeleniumMCPServer = ctx.request_context.lifespan_context
try:
browsers = app_context.browser_detection.detect_available_browsers()
system_info = app_context.browser_detection.get_system_info()
return {
"success": True,
"browsers": browsers,
"system_info": system_info,
"message": f"Detectados {len(browsers)} navegadores disponibles"
}
except Exception as e:
return {
"success": False,
"error": str(e),
"message": "Error al detectar navegadores"
}
@mcp.tool()
def get_recommended_browser() -> Dict[str, Any]:
"""
Obtiene el navegador recomendado basado en disponibilidad.
"""
ctx = mcp.get_context()
app_context: SeleniumMCPServer = ctx.request_context.lifespan_context
try:
recommended = app_context.browser_detection.get_recommended_browser()
support_info = app_context.browser_detection.check_webdriver_support(recommended)
return {
"success": True,
"recommended_browser": recommended,
"webdriver_support": support_info,
"message": f"Navegador recomendado: {recommended}"
}
except Exception as e:
return {
"success": False,
"error": str(e),
"message": "Error al obtener navegador recomendado"
}
@mcp.tool()
def check_browser_support(browser_name: str) -> Dict[str, Any]:
"""
Verifica el soporte de WebDriver para un navegador específico.
Args:
browser_name: Nombre del navegador a verificar
"""
ctx = mcp.get_context()
app_context: SeleniumMCPServer = ctx.request_context.lifespan_context
try:
support_info = app_context.browser_detection.check_webdriver_support(browser_name)
return {
"success": True,
"browser": browser_name,
"support": support_info,
"message": f"Información de soporte para {browser_name}"
}
except Exception as e:
return {
"success": False,
"error": str(e),
"message": f"Error al verificar soporte para {browser_name}"
}
@mcp.tool()
def list_active_sessions() -> Dict[str, Any]:
"""
Lista todas las sesiones activas con información detallada.
"""
ctx = mcp.get_context()
app_context: SeleniumMCPServer = ctx.request_context.lifespan_context
try:
sessions = app_context.session_manager.list_sessions()
# Agregar información adicional de cada sesión
detailed_sessions = {}
for session_id, session_info in sessions.items():
session_obj = app_context.session_manager.get_session(session_id)
if session_obj:
try:
current_url = session_obj.driver.current_url
title = session_obj.driver.title
except:
current_url = "N/A"
title = "N/A"
detailed_sessions[session_id] = {
**session_info,
"current_url": current_url,
"title": title
}
return {
"success": True,
"sessions": detailed_sessions,
"total_sessions": len(detailed_sessions),
"max_sessions": app_context.config.max_sessions,
"message": f"Listadas {len(detailed_sessions)} sesiones activas"
}
except Exception as e:
return {
"success": False,
"error": str(e),
"message": "Error al listar sesiones activas"
}
@mcp.tool()
def close_all_sessions() -> Dict[str, Any]:
"""
Cierra todas las sesiones activas.
"""
ctx = mcp.get_context()
app_context: SeleniumMCPServer = ctx.request_context.lifespan_context
try:
sessions_before = app_context.session_manager.get_session_count()
app_context.session_manager.close_all_sessions()
return {
"success": True,
"sessions_closed": sessions_before,
"message": f"Cerradas {sessions_before} sesiones"
}
except Exception as e:
return {
"success": False,
"error": str(e),
"message": "Error al cerrar todas las sesiones"
}
@mcp.tool()
def get_session_info(session_id: str) -> Dict[str, Any]:
"""
Obtiene información detallada de una sesión específica.
Args:
session_id: ID de la sesión
"""
ctx = mcp.get_context()
app_context: SeleniumMCPServer = ctx.request_context.lifespan_context
session = app_context.session_manager.get_session(session_id)
if not session:
return {"success": False, "error": "Sesión no encontrada"}
try:
# Información básica de la sesión
session_info = {
"session_id": session_id,
"browser_type": session.browser_type,
"created_at": session.created_at,
"last_activity": session.last_activity,
"age_seconds": time.time() - session.created_at
}
# Información del navegador
try:
browser_info = {
"current_url": session.driver.current_url,
"title": session.driver.title,
"window_size": session.driver.get_window_size(),
"cookies_count": len(session.driver.get_cookies()),
"page_source_length": len(session.driver.page_source)
}
session_info["browser_info"] = browser_info
except Exception as e:
session_info["browser_info"] = {"error": str(e)}
return {
"success": True,
"session": session_info,
"message": f"Información de sesión {session_id}"
}
except Exception as e:
return {
"success": False,
"error": str(e),
"message": f"Error al obtener información de sesión {session_id}"
}
@mcp.tool()
def is_element_empty(
session_id: str,
locator_type: str,
locator_value: str
) -> Dict[str, Any]:
"""
Verifica si un elemento está vacío.
Args:
session_id: ID de la sesión del navegador
locator_type: Tipo de localizador (id, name, class_name, css_selector, xpath, etc.)
locator_value: Valor del localizador
Returns:
Dict con el resultado de la verificación
"""
ctx = mcp.get_context()
app_context: SeleniumMCPServer = ctx.request_context.lifespan_context
return app_context.selenium_tools.is_element_empty(
session_id, locator_type, locator_value
)
@mcp.tool()
def reescribir_HTML(
session_id: str,
new_html: str,
locator_type: str,
locator_value: str
) -> Dict[str, Any]:
"""
Reescribe el HTML de un elemento específico.
Args:
session_id: ID de la sesión del navegador
new_html: Nuevo HTML para reemplazar el elemento
locator_type: Tipo de localizador (id, name, class_name, css_selector, xpath, etc.)
locator_value: Valor del localizador
Returns:
Dict con el resultado de la operación
"""
ctx = mcp.get_context()
app_context: SeleniumMCPServer = ctx.request_context.lifespan_context
return app_context.selenium_tools.reescribir_HTML(
session_id, new_html, locator_type, locator_value
)
@mcp.tool()
def handle_popup(
session_id: str,
popup_locator_type: str,
popup_locator_value: str,
title_id: str,
content_id: str,
take_screenshot: bool = False
) -> Dict[str, Any]:
"""
Maneja la detección y procesamiento de popups.
Args:
session_id: ID de la sesión del navegador
popup_locator_type: Tipo de localizador para el popup (id, name, class_name, etc.)
popup_locator_value: Valor del localizador para el popup
title_id: ID del elemento que contiene el título del popup
content_id: ID del elemento que contiene el contenido del popup
take_screenshot: Si tomar captura de pantalla cuando se detecte el popup
Returns:
Dict con el resultado de la detección del popup
"""
ctx = mcp.get_context()
app_context: SeleniumMCPServer = ctx.request_context.lifespan_context
return app_context.selenium_tools.handle_popup(
session_id, popup_locator_type, popup_locator_value,
title_id, content_id, take_screenshot
)