Création de serveurs MCP en Python : guide de recherche web et de scraping

Construisez des serveurs MCP pour des assistants IA avec des exemples en Python

Sommaire

Le protocole de contexte du modèle (MCP) révolutionne la manière dont les assistants IA interagissent avec des sources de données externes et des outils. Dans ce guide, nous explorerons comment construire des serveurs MCP en Python, avec des exemples axés sur les capacités de recherche web et de scraping.

MCP robots

Qu’est-ce que le protocole de contexte du modèle ?

Le protocole de contexte du modèle (MCP) est un protocole ouvert introduit par Anthropic pour standardiser la manière dont les assistants IA se connectent aux systèmes externes. Au lieu de construire des intégrations personnalisées pour chaque source de données, le MCP fournit une interface unifiée qui permet :

  • Les assistants IA (comme Claude, ChatGPT ou des applications LLM personnalisées) de découvrir et d’utiliser des outils
  • Les développeurs d’exposer des sources de données, des outils et des prompts via un protocole standardisé
  • Une intégration fluide sans avoir à réinventer la roue pour chaque cas d’utilisation

Le protocole fonctionne sur une architecture client-serveur où :

  • Les clients MCP (assistants IA) découvrent et utilisent des capacités
  • Les serveurs MCP exposent des ressources, des outils et des prompts
  • La communication se fait via JSON-RPC sur stdio ou HTTP/SSE

Si vous souhaitez implémenter des serveurs MCP dans d’autres langages, consultez notre guide sur l’implémentation d’un serveur MCP en Go, qui couvre en détail les spécifications du protocole et la structure des messages.

Pourquoi construire des serveurs MCP en Python ?

Python est un excellent choix pour le développement de serveurs MCP car :

  1. Écosystème riche : Des bibliothèques comme requests, beautifulsoup4, selenium et playwright rendent le scraping web simple
  2. SDK MCP : Le SDK Python officiel (mcp) fournit un support d’implémentation de serveur robuste
  3. Développement rapide : La simplicité de Python permet un prototypage et une itération rapides
  4. Intégration IA/ML : Une intégration facile avec des bibliothèques IA comme langchain, openai et des outils de traitement des données
  5. Soutien de la communauté : Une communauté importante avec une documentation extensive et des exemples

Mise en place de votre environnement de développement

Commencez par créer un environnement virtuel et installez les dépendances nécessaires. L’utilisation d’environnements virtuels est essentielle pour l’isolation des projets Python - si vous avez besoin d’un rappel, consultez notre feuille de triche venv pour des commandes détaillées et des bonnes pratiques.

# Créer et activer l'environnement virtuel
python -m venv mcp-env
source mcp-env/bin/activate  # Sur Windows : mcp-env\Scripts\activate

# Installer le SDK MCP et les bibliothèques de scraping web
pip install mcp requests beautifulsoup4 playwright lxml
playwright install  # Installer les pilotes de navigateur pour Playwright

Alternative moderne : Si vous préférez une résolution et une installation des dépendances plus rapides, envisagez d’utiliser uv - le gestionnaire moderne de packages et d’environnements Python qui peut être significativement plus rapide que pip pour les grands projets.

Construction d’un serveur MCP de base

Commençons par une structure minimale d’un serveur MCP. Si vous êtes nouveau en Python ou si vous avez besoin d’un rappel rapide sur la syntaxe et les modèles courants, notre feuille de triche Python fournit un aperçu complet des fondamentaux de Python.

import asyncio
from mcp.server import Server
from mcp.types import Tool, TextContent
import mcp.server.stdio

# Créer une instance de serveur
app = Server("websearch-scraper")

@app.list_tools()
async def list_tools() -> list[Tool]:
    """Définir les outils disponibles"""
    return [
        Tool(
            name="search_web",
            description="Rechercher des informations sur le web",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "Requête de recherche"
                    },
                    "max_results": {
                        "type": "number",
                        "description": "Nombre maximum de résultats",
                        "default": 5
                    }
                },
                "required": ["query"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    """Gérer l'exécution de l'outil"""
    if name == "search_web":
        query = arguments["query"]
        max_results = arguments.get("max_results", 5)
        
        # Implémenter la logique de recherche ici
        results = await perform_web_search(query, max_results)
        
        return [TextContent(
            type="text",
            text=f"Résultats de recherche pour '{query}':\n\n{results}"
        )]
    
    raise ValueError(f"Outil inconnu : {name}")

async def perform_web_search(query: str, max_results: int) -> str:
    """Remplacement pour l'implémentation réelle de la recherche"""
    return f"Trouvé {max_results} résultats pour : {query}"

async def main():
    """Exécuter le serveur"""
    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
        await app.run(
            read_stream,
            write_stream,
            app.create_initialization_options()
        )

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

Implémentation de la fonctionnalité de recherche web

Implémentons maintenant un outil de recherche web réel en utilisant DuckDuckGo (qui n’exige pas de clés API) :

import asyncio
import requests
from bs4 import BeautifulSoup
from urllib.parse import quote_plus

async def search_duckduckgo(query: str, max_results: int = 5) -> list[dict]:
    """Rechercher sur DuckDuckGo et analyser les résultats"""
    
    url = f"https://html.duckduckgo.com/html/?q={quote_plus(query)}"
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
    }
    
    try:
        response = requests.get(url, headers=headers, timeout=10)
        response.raise_for_status()
        
        soup = BeautifulSoup(response.text, 'html.parser')
        results = []
        
        for result in soup.select('.result')[:max_results]:
            title_elem = result.select_one('.result__title')
            snippet_elem = result.select_one('.result__snippet')
            url_elem = result.select_one('.result__url')
            
            if title_elem and snippet_elem:
                results.append({
                    "title": title_elem.get_text(strip=True),
                    "snippet": snippet_elem.get_text(strip=True),
                    "url": url_elem.get_text(strip=True) if url_elem else "N/A"
                })
        
        return results
        
    except Exception as e:
        raise Exception(f"La recherche a échoué : {str(e)}")

def format_search_results(results: list[dict]) -> str:
    """Formater les résultats de recherche pour l'affichage"""
    if not results:
        return "Aucun résultat trouvé."
    
    formatted = []
    for i, result in enumerate(results, 1):
        formatted.append(
            f"{i}. {result['title']}\n"
            f"   {result['snippet']}\n"
            f"   URL : {result['url']}\n"
        )
    
    return "\n".join(formatted)

Ajout de fonctionnalités de scraping web

Ajoutons un outil pour scraper et extraire du contenu à partir de pages web. Lorsque vous scrapez du contenu HTML pour l’utiliser avec des LLM, vous pouvez également souhaiter le convertir en format Markdown pour un meilleur traitement. Pour cette raison, consultez notre guide complet sur la conversion HTML en Markdown avec Python, qui compare 6 bibliothèques différentes avec des benchmarks et des recommandations pratiques.

from playwright.async_api import async_playwright
import asyncio

async def scrape_webpage(url: str, selector: str = None) -> dict:
    """Scraper du contenu à partir d'une page web à l'aide de Playwright"""
    
    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=True)
        page = await browser.new_page()
        
        try:
            await page.goto(url, timeout=30000)
            
            # Attendre que le contenu soit chargé
            await page.wait_for_load_state('networkidle')
            
            if selector:
                # Extraire un élément spécifique
                element = await page.query_selector(selector)
                content = await element.inner_text() if element else "Sélecteur non trouvé"
            else:
                # Extraire le contenu principal
                content = await page.inner_text('body')
            
            title = await page.title()
            
            return {
                "title": title,
                "content": content[:5000],  # Limiter la longueur du contenu
                "url": url,
                "success": True
            }
            
        except Exception as e:
            return {
                "error": str(e),
                "url": url,
                "success": False
            }
        finally:
            await browser.close()

# Ajouter l'outil de scraper au serveur MCP
@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="search_web",
            description="Rechercher sur le web à l'aide de DuckDuckGo",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "Requête de recherche"},
                    "max_results": {"type": "number", "default": 5}
                },
                "required": ["query"]
            }
        ),
        Tool(
            name="scrape_webpage",
            description="Scraper du contenu à partir d'une page web",
            inputSchema={
                "type": "object",
                "properties": {
                    "url": {"type": "string", "description": "URL à scraper"},
                    "selector": {
                        "type": "string",
                        "description": "Sélecteur CSS optionnel pour un contenu spécifique"
                    }
                },
                "required": ["url"]
            }
        )
    ]

Implémentation complète du serveur MCP

Voici une implémentation complète, prête pour la production, d’un serveur MCP avec des capacités de recherche et de scraping :

#!/usr/bin/env python3
"""
Serveur MCP pour la recherche web et le scraping
Fournit des outils pour la recherche web et l'extraction de contenu à partir de pages
"""

import asyncio
import logging
from typing import Any
import requests
from bs4 import BeautifulSoup
from urllib.parse import quote_plus
from playwright.async_api import async_playwright

from mcp.server import Server
from mcp.types import Tool, TextContent, ImageContent, EmbeddedResource
import mcp.server.stdio

# Configurer le journalisation
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("websearch-scraper")

# Créer le serveur
app = Server("websearch-scraper")

# Implémentation de la recherche
async def search_web(query: str, max_results: int = 5) -> str:
    """Rechercher sur DuckDuckGo et retourner les résultats formatés"""
    url = f"https://html.duckduckgo.com/html/?q={quote_plus(query)}"
    headers = {"User-Agent": "Mozilla/5.0"}
    
    try:
        response = requests.get(url, headers=headers, timeout=10)
        soup = BeautifulSoup(response.text, 'html.parser')
        results = []
        
        for result in soup.select('.result')[:max_results]:
            title = result.select_one('.result__title')
            snippet = result.select_one('.result__snippet')
            link = result.select_one('.result__url')
            
            if title and snippet:
                results.append({
                    "title": title.get_text(strip=True),
                    "snippet": snippet.get_text(strip=True),
                    "url": link.get_text(strip=True) if link else ""
                })
        
        # Formater les résultats
        if not results:
            return "Aucun résultat trouvé."
        
        formatted = [f"Trouvé {len(results)} résultats pour '{query}':\n"]
        for i, r in enumerate(results, 1):
            formatted.append(f"\n{i}. **{r['title']}**")
            formatted.append(f"   {r['snippet']}")
            formatted.append(f"   {r['url']}")
        
        return "\n".join(formatted)
        
    except Exception as e:
        logger.error(f"La recherche a échoué : {e}")
        return f"Erreur de recherche : {str(e)}"

# Implémentation du scraper
async def scrape_page(url: str, selector: str = None) -> str:
    """Scraper le contenu d'une page web à l'aide de Playwright"""
    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=True)
        page = await browser.new_page()
        
        try:
            await page.goto(url, timeout=30000)
            await page.wait_for_load_state('networkidle')
            
            title = await page.title()
            
            if selector:
                element = await page.query_selector(selector)
                content = await element.inner_text() if element else "Sélecteur non trouvé"
            else:
                content = await page.inner_text('body')
            
            # Limiter la longueur du contenu
            content = content[:8000] + "..." if len(content) > 8000 else content
            
            result = f"**{title}**\n\nURL : {url}\n\n{content}"
            return result
            
        except Exception as e:
            logger.error(f"Le scraping a échoué : {e}")
            return f"Erreur de scraping : {str(e)}"
        finally:
            await browser.close()

# Définitions des outils MCP
@app.list_tools()
async def list_tools() -> list[Tool]:
    """Lister les outils MCP disponibles"""
    return [
        Tool(
            name="search_web",
            description="Rechercher sur le web à l'aide de DuckDuckGo. Retourne les titres, les extraits et les URLs.",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "La requête de recherche"
                    },
                    "max_results": {
                        "type": "number",
                        "description": "Nombre maximum de résultats (par défaut : 5)",
                        "default": 5
                    }
                },
                "required": ["query"]
            }
        ),
        Tool(
            name="scrape_webpage",
            description="Extraire du contenu à partir d'une page web. Peut cibler des éléments spécifiques avec des sélecteurs CSS.",
            inputSchema={
                "type": "object",
                "properties": {
                    "url": {
                        "type": "string",
                        "description": "L'URL à scraper"
                    },
                    "selector": {
                        "type": "string",
                        "description": "Sélecteur CSS optionnel pour extraire un contenu spécifique"
                    }
                },
                "required": ["url"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: Any) -> list[TextContent]:
    """Gérer l'exécution de l'outil"""
    try:
        if name == "search_web":
            query = arguments["query"]
            max_results = arguments.get("max_results", 5)
            result = await search_web(query, max_results)
            return [TextContent(type="text", text=result)]
        
        elif name == "scrape_webpage":
            url = arguments["url"]
            selector = arguments.get("selector")
            result = await scrape_page(url, selector)
            return [TextContent(type="text", text=result)]
        
        else:
            raise ValueError(f"Outil inconnu : {name}")
    
    except Exception as e:
        logger.error(f"Exécution de l'outil a échoué : {e}")
        return [TextContent(
            type="text",
            text=f"Erreur lors de l'exécution de {name} : {str(e)}"
        )]

async def main():
    """Exécuter le serveur MCP"""
    logger.info("Démarrage du serveur MCP WebSearch-Scraper")
    
    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
        await app.run(
            read_stream,
            write_stream,
            app.create_initialization_options()
        )

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

Configuration de votre serveur MCP

Pour utiliser votre serveur MCP avec Claude Desktop ou d’autres clients MCP, créez un fichier de configuration :

Pour Claude Desktop (claude_desktop_config.json) :

{
  "mcpServers": {
    "websearch-scraper": {
      "command": "python",
      "args": [
        "/chemin/vers/votre/mcp_server.py"
      ],
      "env": {}
    }
  }
}

Emplacement :

  • macOS : ~/Library/Application Support/Claude/claude_desktop_config.json
  • Windows : %APPDATA%\Claude\claude_desktop_config.json
  • Linux : ~/.config/Claude/claude_desktop_config.json

Test de votre serveur MCP

Créez un script de test pour vérifier la fonctionnalité :

import asyncio
import json
import sys
from io import StringIO

async def test_mcp_server():
    """Tester localement le serveur MCP"""
    
    # Tester la recherche
    print("Test de la recherche web...")
    results = await search_web("Tutoriel Python MCP", 3)
    print(results)
    
    # Tester le scraper
    print("\n\nTest du scraper de page web...")
    content = await scrape_page("https://example.com")
    print(content[:500])

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

Fonctionnalités avancées et bonnes pratiques

1. Limitation de débit

Implémentez une limitation de débit pour éviter de surcharger les serveurs cibles :

import time
from collections import defaultdict

class RateLimiter:
    def __init__(self, max_requests: int = 10, time_window: int = 60):
        self.max_requests = max_requests
        self.time_window = time_window
        self.requests = defaultdict(list)
    
    async def acquire(self, key: str):
        now = time.time()
        self.requests[key] = [
            t for t in self.requests[key] 
            if now - t < self.time_window
        ]
        
        if len(self.requests[key]) >= self.max_requests:
            raise Exception("Limite de débit dépassée")
        
        self.requests[key].append(now)

limiter = RateLimiter(max_requests=10, time_window=60)

2. Mise en cache

Ajoutez un cache pour améliorer les performances :

from functools import lru_cache
import hashlib

@lru_cache(maxsize=100)
async def cached_search(query: str, max_results: int):
    return await search_web(query, max_results)

3. Gestion des erreurs

Implémentez une gestion d’erreurs robuste :

from enum import Enum

class ErrorType(Enum):
    NETWORK_ERROR = "network_error"
    PARSE_ERROR = "parse_error"
    RATE_LIMIT = "rate_limit_exceeded"
    INVALID_INPUT = "invalid_input"

def handle_error(error: Exception, error_type: ErrorType) -> str:
    logger.error(f"{error_type.value} : {str(error)}")
    return f"Erreur ({error_type.value}) : {str(error)}"

4. Validation des entrées

Validez les entrées utilisateur avant le traitement :

from urllib.parse import urlparse

def validate_url(url: str) -> bool:
    try:
        result = urlparse(url)
        return all([result.scheme, result.netloc])
    except:
        return False

def validate_query(query: str) -> bool:
    return len(query.strip()) > 0 and len(query) < 500

Considérations pour le déploiement

Utilisation du transport SSE pour le déploiement web

Pour les déploiements web, utilisez le transport SSE (Server-Sent Events). Si vous envisagez un déploiement serverless, vous pourriez être intéressé par la comparaison des performances d’AWS Lambda entre JavaScript, Python et Golang pour prendre une décision éclairée sur votre runtime :

import mcp.server.sse

async def main_sse():
    """Exécuter le serveur avec le transport SSE"""
    from starlette.applications import Starlette
    from starlette.routing import Mount
    
    sse = mcp.server.sse.SseServerTransport("/messages")
    
    starlette_app = Starlette(
        routes=[
            Mount("/mcp", app=sse.get_server())
        ]
    )
    
    import uvicorn
    await uvicorn.Server(
        config=uvicorn.Config(starlette_app, host="0.0.0.0", port=8000)
    ).serve()

Déploiement sur AWS Lambda

Les serveurs MCP peuvent également être déployés comme fonctions AWS Lambda, particulièrement lors de l’utilisation du transport SSE. Pour des guides complets sur le déploiement Lambda :

Déploiement Docker

Créez un Dockerfile pour le déploiement conteneurisé :

FROM python:3.11-slim

WORKDIR /app

# Installer les dépendances système
RUN apt-get update && apt-get install -y \
    wget \
    && rm -rf /var/lib/apt/lists/*

# Installer les dépendances Python
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
RUN playwright install chromium
RUN playwright install-deps

# Copier l'application
COPY mcp_server.py .

CMD ["python", "mcp_server.py"]

Optimisation des performances

Opérations asynchrones

Utilisez asyncio pour les opérations concurrentes :

async def search_multiple_queries(queries: list[str]) -> list[str]:
    """Rechercher plusieurs requêtes de manière concurrente"""
    tasks = [search_web(query) for query in queries]
    results = await asyncio.gather(*tasks)
    return results

Réutilisation des connexions

Réutilisez les connexions pour une meilleure performance :

import aiohttp

session = None

async def get_session():
    global session
    if session is None:
        session = aiohttp.ClientSession()
    return session

async def fetch_url(url: str) -> str:
    session = await get_session()
    async with session.get(url) as response:
        return await response.text()

Bonnes pratiques de sécurité

  1. Nettoyage des entrées : Validez toujours et nettoyez les entrées utilisateur
  2. Liste blanche des URL : Pensez à implémenter une liste blanche des URL pour le web scraping
  3. Contrôles de délai : Définissez des délais appropriés pour éviter l’épuisement des ressources
  4. Limites de contenu : Limitez la taille du contenu scrapé
  5. Authentification : Implémentez une authentification pour les déploiements en production
  6. HTTPS : Utilisez HTTPS pour le transport SSE en production

Travail avec différents fournisseurs LLM

Bien que MCP ait été développé par Anthropic pour Claude, le protocole a été conçu pour fonctionner avec tout LLM. Si vous construisez des serveurs MCP qui interagissent avec plusieurs fournisseurs d’IA et avez besoin de sorties structurées, vous devriez consulter notre comparaison des sorties structurées parmi les fournisseurs LLM populaires comprenant OpenAI, Gemini, Anthropic, Mistral et AWS Bedrock.

Liens utiles et ressources

Ressources liées

MCP et implémentation du protocole

Développement en Python

Web scraping et traitement du contenu

Ressources de déploiement serverless

Intégration LLM

Conclusion

La création de serveurs MCP en Python ouvre des possibilités puissantes pour étendre les assistants IA avec des outils et des sources de données personnalisés. Les capacités de recherche web et de scraping démontrées ici ne sont qu’un début — vous pouvez étendre cette base pour intégrer des bases de données, des API, des systèmes de fichiers et presque tout système externe.

Le Model Context Protocol évolue encore, mais son approche standardisée pour l’intégration des outils IA en fait une technologie passionnante pour les développeurs qui construisent la prochaine génération d’applications alimentées par l’IA. Que vous créiez des outils internes pour votre organisation ou que vous construisez des serveurs MCP publics pour la communauté, Python offre une excellente base pour le développement et le déploiement rapides.