browser_mcp_server.py

#!/usr/bin/env python3
"""
Browser MCP Server - Automatisation web pour PageChat
Permet aux IA d'interagir avec des pages web, capturer des écrans, extraire du contenu
"""

import asyncio
import base64
import json
import logging
from typing import Optional, Dict, Any, List
from dataclasses import dataclass
import mcp.server.stdio
import mcp.types as types
from mcp.server import NotificationOptions, Server
from mcp.server.models import InitializationOptions
import mcp.server.stdio
from playwright.async_api import async_playwright, Browser, Page, BrowserContext
from bs4 import BeautifulSoup
import re
from urllib.parse import urljoin, urlparse

# Configuration du logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("browser-mcp")

@dataclass
class BrowserState:
    """État global du navigateur"""
    browser: Optional[Browser] = None
    context: Optional[BrowserContext] = None
    page: Optional[Page] = None
    playwright = None

# Instance globale
browser_state = BrowserState()

# Serveur MCP
server = Server("browser-mcp")

@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
    """Liste des outils disponibles"""
    return [
        types.Tool(
            name="navigate_to_url",
            description="Naviguer vers une URL spécifique",
            inputSchema={
                "type": "object",
                "properties": {
                    "url": {"type": "string", "description": "URL à visiter"},
                    "wait_for": {"type": "string", "description": "Sélecteur CSS à attendre (optionnel)"}
                },
                "required": ["url"]
            },
        ),
        types.Tool(
            name="take_screenshot",
            description="Capturer une capture d'écran de la page actuelle",
            inputSchema={
                "type": "object", 
                "properties": {
                    "full_page": {"type": "boolean", "description": "Capture complète de la page", "default": True},
                    "element_selector": {"type": "string", "description": "Sélecteur CSS d'un élément spécifique"}
                }
            },
        ),
        types.Tool(
            name="extract_text",
            description="Extraire le texte visible de la page ou d'un élément",
            inputSchema={
                "type": "object",
                "properties": {
                    "selector": {"type": "string", "description": "Sélecteur CSS (optionnel, toute la page si vide)"},
                    "clean": {"type": "boolean", "description": "Nettoyer le texte extrait", "default": True}
                }
            },
        ),
        types.Tool(
            name="click_element",
            description="Cliquer sur un élément de la page",
            inputSchema={
                "type": "object",
                "properties": {
                    "selector": {"type": "string", "description": "Sélecteur CSS de l'élément à cliquer"}
                },
                "required": ["selector"]
            },
        ),
        types.Tool(
            name="fill_input",
            description="Remplir un champ de saisie",
            inputSchema={
                "type": "object",
                "properties": {
                    "selector": {"type": "string", "description": "Sélecteur CSS du champ"},
                    "value": {"type": "string", "description": "Valeur à saisir"}
                },
                "required": ["selector", "value"]
            },
        ),
        types.Tool(
            name="get_page_info",
            description="Obtenir les informations de base de la page (titre, URL, description)",
            inputSchema={"type": "object"},
        ),
        types.Tool(
            name="wait_for_element",
            description="Attendre qu'un élément apparaisse sur la page",
            inputSchema={
                "type": "object",
                "properties": {
                    "selector": {"type": "string", "description": "Sélecteur CSS à attendre"},
                    "timeout": {"type": "number", "description": "Timeout en millisecondes", "default": 10000}
                },
                "required": ["selector"]
            },
        ),
        types.Tool(
            name="search_text",
            description="Rechercher du texte sur la page actuelle",
            inputSchema={
                "type": "object",
                "properties": {
                    "text": {"type": "string", "description": "Texte à rechercher"},
                    "case_sensitive": {"type": "boolean", "description": "Recherche sensible à la casse", "default": False}
                },
                "required": ["text"]
            },
        ),
        types.Tool(
            name="extract_links",
            description="Extraire tous les liens de la page",
            inputSchema={
                "type": "object",
                "properties": {
                    "filter_domain": {"type": "string", "description": "Filtrer par domaine (optionnel)"}
                }
            },
        ),
        types.Tool(
            name="get_element_info",
            description="Obtenir des informations sur un élément spécifique",
            inputSchema={
                "type": "object",
                "properties": {
                    "selector": {"type": "string", "description": "Sélecteur CSS de l'élément"}
                },
                "required": ["selector"]
            },
        )
    ]

async def ensure_browser():
    """S'assurer que le navigateur est initialisé"""
    if not browser_state.browser:
        browser_state.playwright = await async_playwright().start()
        
        # Configuration pour HF Spaces (environnement conteneurisé)
        launch_options = {
            'headless': True,
            'args': [
                '--no-sandbox', 
                '--disable-dev-shm-usage',
                '--disable-gpu',
                '--disable-dev-tools',
                '--no-first-run',
                '--disable-extensions',
                '--disable-default-apps',
                '--disable-background-timer-throttling',
                '--disable-backgrounding-occluded-windows',
                '--disable-renderer-backgrounding',
                '--single-process'  # Important pour les conteneurs limités
            ]
        }
        
        # Détecter si on est dans HF Spaces
        import os
        if os.environ.get('SPACE_ID') or os.environ.get('HUGGINGFACE_HUB_CACHE'):
            # Configuration spéciale pour HF Spaces
            launch_options['args'].extend([
                '--disable-features=VizDisplayCompositor',
                '--run-all-compositor-stages-before-draw',
                '--memory-pressure-off'
            ])
            logger.info("🚀 Configuration HF Spaces détectée")
        
        try:
            browser_state.browser = await browser_state.playwright.chromium.launch(**launch_options)
        except Exception as e:
            if "Executable doesn't exist" in str(e):
                logger.error("❌ Navigateurs Playwright non installés. Tentative de réinstallation...")
                try:
                    # Tentative de réinstallation automatique
                    import subprocess
                    import sys
                    result = subprocess.run([sys.executable, "-m", "playwright", "install", "chromium"], 
                                          capture_output=True, text=True, timeout=120)
                    if result.returncode == 0:
                        logger.info("✅ Réinstallation Playwright réussie")
                        # Nouvelle tentative
                        browser_state.browser = await browser_state.playwright.chromium.launch(**launch_options)
                    else:
                        logger.error(f"❌ Échec réinstallation: {result.stderr}")
                        raise Exception("Impossible d'installer les navigateurs Playwright")
                except subprocess.TimeoutExpired:
                    logger.error("❌ Timeout lors de l'installation Playwright")
                    raise Exception("Timeout installation Playwright")
                except Exception as install_error:
                    logger.error(f"❌ Erreur installation: {install_error}")
                    raise
            else:
                raise
        
        browser_state.context = await browser_state.browser.new_context(
            viewport={'width': 1280, 'height': 720},
            user_agent='Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36'
        )
        browser_state.page = await browser_state.context.new_page()
        logger.info("✅ Navigateur initialisé avec succès")
        
        # Configurer les timeouts pour les environnements cloud
        browser_state.page.set_default_timeout(30000)  # 30 secondes
        browser_state.page.set_default_navigation_timeout(45000)  # 45 secondes

@server.call_tool()
async def handle_call_tool(name: str, arguments: dict) -> list[types.TextContent]:
    """Gestionnaire des appels d'outils"""
    try:
        await ensure_browser()
        
        if name == "navigate_to_url":
            url = arguments["url"]
            wait_for = arguments.get("wait_for")
            
            logger.info(f"Navigation vers: {url}")
            await browser_state.page.goto(url, wait_until="domcontentloaded")
            
            if wait_for:
                await browser_state.page.wait_for_selector(wait_for, timeout=10000)
            
            title = await browser_state.page.title()
            return [types.TextContent(
                type="text", 
                text=f"✅ Navigation réussie vers {url}\nTitre: {title}"
            )]
            
        elif name == "take_screenshot":
            full_page = arguments.get("full_page", True)
            element_selector = arguments.get("element_selector")
            
            if element_selector:
                element = await browser_state.page.query_selector(element_selector)
                if element:
                    screenshot_bytes = await element.screenshot()
                else:
                    return [types.TextContent(type="text", text="❌ Élément non trouvé")]
            else:
                screenshot_bytes = await browser_state.page.screenshot(full_page=full_page)
            
            # Encoder en base64
            screenshot_b64 = base64.b64encode(screenshot_bytes).decode()
            
            return [types.TextContent(
                type="text",
                text=f"📸 Capture d'écran prise (taille: {len(screenshot_bytes)} bytes)\nBase64: data:image/png;base64,{screenshot_b64[:100]}..."
            )]
            
        elif name == "extract_text":
            selector = arguments.get("selector")
            clean = arguments.get("clean", True)
            
            if selector:
                elements = await browser_state.page.query_selector_all(selector)
                texts = []
                for element in elements:
                    text = await element.inner_text()
                    if clean:
                        text = re.sub(r'\s+', ' ', text.strip())
                    texts.append(text)
                result = "\n".join(texts)
            else:
                result = await browser_state.page.inner_text("body")
                if clean:
                    result = re.sub(r'\s+', ' ', result.strip())
            
            return [types.TextContent(
                type="text",
                text=f"📝 Texte extrait:\n{result[:2000]}..." if len(result) > 2000 else f"📝 Texte extrait:\n{result}"
            )]
            
        elif name == "click_element":
            selector = arguments["selector"]
            
            try:
                await browser_state.page.click(selector, timeout=5000)
                return [types.TextContent(type="text", text=f"👆 Clic effectué sur: {selector}")]
            except Exception as e:
                return [types.TextContent(type="text", text=f"❌ Erreur lors du clic: {str(e)}")]
                
        elif name == "fill_input":
            selector = arguments["selector"]
            value = arguments["value"]
            
            try:
                await browser_state.page.fill(selector, value)
                return [types.TextContent(type="text", text=f"✏️ Champ rempli: {selector} = '{value}'")]
            except Exception as e:
                return [types.TextContent(type="text", text=f"❌ Erreur lors de la saisie: {str(e)}")]
                
        elif name == "get_page_info":
            title = await browser_state.page.title()
            url = browser_state.page.url
            
            # Essayer d'extraire la description
            try:
                description_element = await browser_state.page.query_selector('meta[name="description"]')
                description = await description_element.get_attribute("content") if description_element else "Non disponible"
            except:
                description = "Non disponible"
            
            return [types.TextContent(
                type="text",
                text=f"📄 Informations de la page:\n• Titre: {title}\n• URL: {url}\n• Description: {description}"
            )]
            
        elif name == "wait_for_element":
            selector = arguments["selector"]
            timeout = arguments.get("timeout", 10000)
            
            try:
                await browser_state.page.wait_for_selector(selector, timeout=timeout)
                return [types.TextContent(type="text", text=f"✅ Élément trouvé: {selector}")]
            except Exception as e:
                return [types.TextContent(type="text", text=f"⏰ Timeout: élément non trouvé dans les {timeout}ms")]
                
        elif name == "search_text":
            text = arguments["text"]
            case_sensitive = arguments.get("case_sensitive", False)
            
            page_text = await browser_state.page.inner_text("body")
            if not case_sensitive:
                found = text.lower() in page_text.lower()
            else:
                found = text in page_text
                
            return [types.TextContent(
                type="text",
                text=f"🔍 Recherche de '{text}': {'✅ Trouvé' if found else '❌ Non trouvé'}"
            )]
            
        elif name == "extract_links":
            filter_domain = arguments.get("filter_domain")
            
            links = await browser_state.page.evaluate('''
                () => {
                    const links = Array.from(document.querySelectorAll('a[href]'));
                    return links.map(link => ({
                        text: link.innerText.trim(),
                        href: link.href,
                        title: link.title || ''
                    }));
                }
            ''')
            
            if filter_domain:
                links = [link for link in links if filter_domain in link['href']]
            
            result = f"🔗 {len(links)} liens trouvés:\n"
            for i, link in enumerate(links[:20]):  # Limiter à 20 liens
                result += f"{i+1}. {link['text'][:50]} -> {link['href']}\n"
            
            if len(links) > 20:
                result += f"... et {len(links) - 20} autres liens"
                
            return [types.TextContent(type="text", text=result)]
            
        elif name == "get_element_info":
            selector = arguments["selector"]
            
            try:
                element = await browser_state.page.query_selector(selector)
                if element:
                    info = await element.evaluate('''
                        (el) => ({
                            tagName: el.tagName,
                            text: el.innerText?.substring(0, 200),
                            href: el.href,
                            src: el.src,
                            id: el.id,
                            className: el.className,
                            visible: window.getComputedStyle(el).display !== 'none'
                        })
                    ''')
                    
                    result = f"ℹ️ Informations sur l'élément '{selector}':\n"
                    for key, value in info.items():
                        if value:
                            result += f"• {key}: {value}\n"
                    
                    return [types.TextContent(type="text", text=result)]
                else:
                    return [types.TextContent(type="text", text=f"❌ Élément non trouvé: {selector}")]
            except Exception as e:
                return [types.TextContent(type="text", text=f"❌ Erreur: {str(e)}")]
        
        else:
            return [types.TextContent(type="text", text=f"❌ Outil inconnu: {name}")]
            
    except Exception as e:
        logger.error(f"Erreur dans {name}: {str(e)}")
        return [types.TextContent(type="text", text=f"❌ Erreur: {str(e)}")]

async def cleanup():
    """Nettoyer les ressources"""
    if browser_state.browser:
        await browser_state.browser.close()
    if browser_state.playwright:
        await browser_state.playwright.stop()

async def main():
    """Point d'entrée principal"""
    try:
        # Configuration du serveur
        async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
            await server.run(
                read_stream,
                write_stream,
                InitializationOptions(
                    server_name="browser-mcp",
                    server_version="1.0.0",
                    capabilities=server.get_capabilities(
                        notification_options=NotificationOptions(),
                        experimental_capabilities={},
                    ),
                ),
            )
    except KeyboardInterrupt:
        logger.info("Arrêt du serveur")
    finally:
        await cleanup()

if __name__ == "__main__":
    asyncio.run(main())