scraper_server.py

#!/usr/bin/env python3
"""
Serveur MCP de scraping moderne – utilise Crawl4AI ou ScrapeGraphAI
Extrait le contenu épuré d'une URL au format Markdown ou JSON.
"""
import os, json, logging, sys
from mcp.server import Server, NotificationOptions
from mcp.server.models import InitializationCapabilities
from mcp.server.stdio import stdio_server

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("mcp-scraper")

server = Server("scraper-server")

@server.list_tools()
async def list_tools():
    return [
        {
            "name": "scrape_url",
            "description": "Scrape une page web et retourne le contenu épuré (Markdown/JSON).",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "url": {"type": "string", "description": "URL à scraper"},
                    "format": {"type": "string", "description": "Format de sortie: markdown ou json", "default": "markdown"}
                },
                "required": ["url"]
            }
        }
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "scrape_url":
        url = arguments["url"]
        output_format = arguments.get("format", "markdown")
        try:
            # Utilise Crawl4AI (pip install crawl4ai)
            from crawl4ai import WebCrawler
            crawler = WebCrawler()
            result = await crawler.crawl(url)
            if output_format == "json":
                return json.dumps({"success": True, "content": result.json})
            else:
                return json.dumps({"success": True, "content": result.markdown[:5000]})
        except ImportError:
            # Fallback : utilise ScrapeGraphAI ou un simple requests+bs4
            import requests
            from bs4 import BeautifulSoup
            resp = requests.get(url, timeout=10)
            soup = BeautifulSoup(resp.text, 'html.parser')
            text = soup.get_text()[:5000]
            return json.dumps({"success": True, "content": text})
    return json.dumps({"error": "Outil inconnu"})

async def main():
    async with stdio_server() as (read_stream, write_stream):
        await server.run(read_stream, write_stream, InitializationCapabilities())

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