Costruire Server MCP in Python: Guida per Ricerca Web e Scraping

Costruisci server MCP per assistenti AI con esempi in Python

Indice

Il Model Context Protocol (MCP) sta rivoluzionando il modo in cui gli assistenti AI interagiscono con le fonti di dati esterne e gli strumenti. In questa guida, esploreremo come costruire server MCP in Python, con esempi focalizzati sulle capacità di ricerca web e scraping.

MCP robot

Cosa è il Model Context Protocol?

Il Model Context Protocol (MCP) è un protocollo aperto introdotto da Anthropic per standardizzare il modo in cui gli assistenti AI si connettono a sistemi esterni. Invece di costruire integrazioni personalizzate per ogni fonte di dati, l’MCP fornisce un’interfaccia unificata che consente:

  • Assistenti AI (come Claude, ChatGPT o applicazioni LLM personalizzate) di scoprire e utilizzare strumenti
  • Sviluppatori di esporre fonti di dati, strumenti e prompt attraverso un protocollo standardizzato
  • Integrazione senza sforzi senza reinventare la ruota per ogni caso d’uso

Il protocollo opera su un’architettura client-server dove:

  • Client MCP (assistenti AI) scoprono e utilizzano capacità
  • Server MCP espongono risorse, strumenti e prompt
  • La comunicazione avviene tramite JSON-RPC su stdio o HTTP/SSE

Se sei interessato all’implementazione di server MCP in altri linguaggi, consulta la nostra guida su implementare un server MCP in Go, che copre in dettaglio le specifiche del protocollo e la struttura dei messaggi.

Perché costruire server MCP in Python?

Python è un’ottima scelta per lo sviluppo di server MCP perché:

  1. Ecosistema ricco: Librerie come requests, beautifulsoup4, selenium e playwright rendono semplice lo scraping web
  2. SDK MCP: L’SDK Python ufficiale (mcp) fornisce un supporto robusto per l’implementazione del server
  3. Sviluppo rapido: La semplicità di Python permette prototipi e iterazioni rapide
  4. Integrazione AI/ML: Facile integrazione con librerie AI come langchain, openai e strumenti di elaborazione dati
  5. Supporto della comunità: Grande comunità con documentazione estesa e esempi

Configurazione dell’ambiente di sviluppo

Per prima cosa, crea un ambiente virtuale e installa le dipendenze necessarie. L’uso degli ambienti virtuali è essenziale per l’isolamento dei progetti Python - se hai bisogno di un ripasso, consulta la nostra guida venv per comandi dettagliati e buone pratiche.

# Crea e attiva l'ambiente virtuale
python -m venv mcp-env
source mcp-env/bin/activate  # Su Windows: mcp-env\Scripts\activate

# Installa l'SDK MCP e le librerie per lo scraping web
pip install mcp requests beautifulsoup4 playwright lxml
playwright install  # Installa i driver del browser per Playwright

Alternativa moderna: Se preferisci una risoluzione più rapida delle dipendenze e un’installazione più veloce, considera l’uso di uv - il nuovo gestore di pacchetti e ambienti Python che può essere significativamente più veloce di pip per i progetti di grandi dimensioni.

Costruzione di un server MCP di base

Iniziamo con una struttura di base per un server MCP. Se sei nuovo di Python o hai bisogno di un riferimento rapido per la sintassi e i pattern comuni, la nostra guida Python fornisce un’overview completa dei fondamenti di Python.

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

# Crea un'istanza del server
app = Server("websearch-scraper")

@app.list_tools()
async def list_tools() -> list[Tool]:
    """Definisci gli strumenti disponibili"""
    return [
        Tool(
            name="search_web",
            description="Cerca informazioni sul web",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "Query di ricerca"
                    },
                    "max_results": {
                        "type": "number",
                        "description": "Numero massimo di risultati",
                        "default": 5
                    }
                },
                "required": ["query"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    """Gestisci l'esecuzione degli strumenti"""
    if name == "search_web":
        query = arguments["query"]
        max_results = arguments.get("max_results", 5)
        
        # Implementa la logica di ricerca qui
        results = await perform_web_search(query, max_results)
        
        return [TextContent(
            type="text",
            text=f"Risultati della ricerca per '{query}':\n\n{results}"
        )]
    
    raise ValueError(f"Strumento sconosciuto: {name}")

async def perform_web_search(query: str, max_results: int) -> str:
    """Placeholder per l'implementazione effettiva della ricerca"""
    return f"Trovati {max_results} risultati per: {query}"

async def main():
    """Avvia il server"""
    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())

Implementazione della funzionalità di ricerca web

Ora implementiamo uno strumento reale per la ricerca web utilizzando DuckDuckGo (che non richiede API key):

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]:
    """Cerca su DuckDuckGo e analizza i risultati"""
    
    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"Ricerca fallita: {str(e)}")

def format_search_results(results: list[dict]) -> str:
    """Formatta i risultati della ricerca per la visualizzazione"""
    if not results:
        return "Nessun risultato trovato."
    
    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)

Aggiunta delle capacità di scraping web

Aggiungiamo uno strumento per scaricare e estrarre contenuti da pagine web. Quando si effettua lo scraping di contenuti HTML per l’uso con gli LLM, potresti anche volerli convertire in formato Markdown per un’elaborazione migliore. Per questo scopo, consulta la nostra guida completa su convertire HTML in Markdown con Python, che confronta 6 librerie diverse con benchmark e raccomandazioni pratiche.

from playwright.async_api import async_playwright
import asyncio

async def scrape_webpage(url: str, selector: str = None) -> dict:
    """Scrapa il contenuto da una pagina web utilizzando 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)
            
            # Attendi che il contenuto venga caricato
            await page.wait_for_load_state('networkidle')
            
            if selector:
                # Estrai un elemento specifico
                element = await page.query_selector(selector)
                content = await element.inner_text() if element else "Selettore non trovato"
            else:
                # Estrai il contenuto principale
                content = await page.inner_text('body')
            
            title = await page.title()
            
            return {
                "title": title,
                "content": content[:5000],  # Limita la lunghezza del contenuto
                "url": url,
                "success": True
            }
            
        except Exception as e:
            return {
                "error": str(e),
                "url": url,
                "success": False
            }
        finally:
            await browser.close()

# Aggiungi lo strumento scraper al server MCP
@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="search_web",
            description="Cerca sul web utilizzando DuckDuckGo",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "Query di ricerca"},
                    "max_results": {"type": "number", "default": 5}
                },
                "required": ["query"]
            }
        ),
        Tool(
            name="scrape_webpage",
            description="Scrapa il contenuto da una pagina web",
            inputSchema={
                "type": "object",
                "properties": {
                    "url": {"type": "string", "description": "URL da scrapare"},
                    "selector": {
                        "type": "string",
                        "description": "Selettore CSS opzionale per il contenuto specifico"
                    }
                },
                "required": ["url"]
            }
        )
    ]

Implementazione completa del server MCP

Ecco un server MCP completo e pronto per la produzione con capacità di ricerca e scraping:

#!/usr/bin/env python3
"""
Server MCP per la ricerca web e lo scraping
Fornisce strumenti per cercare sul web e estrarre contenuti da pagine
"""

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

# Configura il logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("websearch-scraper")

# Crea il server
app = Server("websearch-scraper")

# Implementazione della ricerca
async def search_web(query: str, max_results: int = 5) -> str:
    """Cerca su DuckDuckGo e restituisce i risultati formattati"""
    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 ""
                })
        
        # Formatta i risultati
        if not results:
            return "Nessun risultato trovato."
        
        formatted = [f"Trovati {len(results)} risultati per '{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"Ricerca fallita: {e}")
        return f"Errore di ricerca: {str(e)}"

# Implementazione dello scraper
async def scrape_page(url: str, selector: str = None) -> str:
    """Scrapa il contenuto di una pagina web utilizzando 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 "Selettore non trovato"
            else:
                content = await page.inner_text('body')
            
            # Limita la lunghezza del contenuto
            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"Scraping fallito: {e}")
            return f"Errore di scraping: {str(e)}"
        finally:
            await browser.close()

# Definizioni degli strumenti MCP
@app.list_tools()
async def list_tools() -> list[Tool]:
    """Elenca gli strumenti MCP disponibili"""
    return [
        Tool(
            name="search_web",
            description="Cerca sul web utilizzando DuckDuckGo. Restituisce titoli, snippet e URL.",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "La query di ricerca"
                    },
                    "max_results": {
                        "type": "number",
                        "description": "Numero massimo di risultati (default: 5)",
                        "default": 5
                    }
                },
                "required": ["query"]
            }
        ),
        Tool(
            name="scrape_webpage",
            description="Estrai il contenuto da una pagina web. Può mirare a elementi specifici con selezionatori CSS.",
            inputSchema={
                "type": "object",
                "properties": {
                    "url": {
                        "type": "string",
                        "description": "L'URL da scrapare"
                    },
                    "selector": {
                        "type": "string",
                        "description": "Selettore CSS opzionale per l'estrazione di contenuti specifici"
                    }
                },
                "required": ["url"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: Any) -> list[TextContent]:
    """Gestisci l'esecuzione degli strumenti"""
    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"Strumento sconosciuto: {name}")
    
    except Exception as e:
        logger.error(f"Fallimento nell'esecuzione dello strumento: {e}")
        return [TextContent(
            type="text",
            text=f"Errore nell'esecuzione di {name}: {str(e)}"
        )]

async def main():
    """Avvia il server MCP"""
    logger.info("Avviando il server 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())

Configurazione del server MCP

Per utilizzare il tuo server MCP con Claude Desktop o altri client MCP, crea un file di configurazione:

Per Claude Desktop (claude_desktop_config.json):

{
  "mcpServers": {
    "websearch-scraper": {
      "command": "python",
      "args": [
        "/path/to/your/mcp_server.py"
      ],
      "env": {}
    }
  }
}

Posizione:

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

Test del server MCP

Crea uno script di test per verificare la funzionalità:

import asyncio
import json
import sys
from io import StringIO

async def test_mcp_server():
    """Testa il server MCP localmente"""
    
    # Testa la ricerca
    print("Test della ricerca web...")
    results = await search_web("Tutorial MCP Python", 3)
    print(results)
    
    # Testa lo scraper
    print("\n\nTest dello scraper di pagine web...")
    content = await scrape_page("https://example.com")
    print(content[:500])

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

Funzionalità avanzate e buone pratiche

1. Limitazione del tasso di richieste

Implementa la limitazione del tasso per evitare di sovraccaricare i server target:

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 di tasso superato")
        
        self.requests[key].append(now)

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

2. Caching

Aggiungi il caching per migliorare le prestazioni:

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. Gestione degli errori

Implementa una gestione robusta degli errori:

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"Errore ({error_type.value}): {str(error)}"

4. Validazione degli input

Valida gli input degli utenti prima del processing:

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

Considerazioni per la distribuzione

Utilizzo del trasporto SSE per la distribuzione web

Per le distribuzioni web, utilizza il trasporto SSE (Server-Sent Events). Se stai considerando una distribuzione serverless, potresti essere interessato a confrontare le prestazioni di AWS Lambda tra JavaScript, Python e Golang per prendere una decisione informata sul tuo runtime:

import mcp.server.sse

async def main_sse():
    """Avvia il server con il trasporto 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()

Distribuzione su AWS Lambda

I server MCP possono essere distribuiti anche come funzioni AWS Lambda, specialmente quando si utilizza il trasporto SSE. Per guide complete sulla distribuzione su Lambda:

Distribuzione con Docker

Crea un Dockerfile per la distribuzione containerizzata:

FROM python:3.11-slim

WORKDIR /app

# Installa le dipendenze del sistema
RUN apt-get update && apt-get install -y \
    wget \
    && rm -rf /var/lib/apt/lists/*

# Installa le dipendenze Python
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
RUN playwright install chromium
RUN playwright install-deps

# Copia l'applicazione
COPY mcp_server.py .

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

Ottimizzazione delle prestazioni

Operazioni asincrone

Utilizza asyncio per operazioni concorrenti:

async def search_multiple_queries(queries: list[str]) -> list[str]:
    """Cerca multiple query contemporaneamente"""
    tasks = [search_web(query) for query in queries]
    results = await asyncio.gather(*tasks)
    return results

Pooling delle connessioni

Riusa le connessioni per migliorare le prestazioni:

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()

Pratiche di sicurezza

  1. Sanitizzazione degli input: Validare sempre e sanitizzare gli input degli utenti
  2. Elenco bianco degli URL: Considerare l’implementazione di un elenco bianco degli URL per lo scraping
  3. Controllo dei timeout: Impostare timeout appropriati per prevenire l’esaurimento delle risorse
  4. Limiti del contenuto: Limitare la dimensione del contenuto estratto
  5. Autenticazione: Implementare l’autenticazione per le distribuzioni in produzione
  6. HTTPS: Utilizzare HTTPS per il trasporto SSE in produzione

Lavorare con diversi fornitori di LLM

Sebbene MCP sia stato sviluppato da Anthropic per Claude, il protocollo è stato progettato per funzionare con qualsiasi LLM. Se stai costruendo server MCP che interagiscono con diversi fornitori di AI e hai bisogno di output strutturati, vorrai rivedere il nostro confronto degli output strutturati tra i principali fornitori di LLM tra cui OpenAI, Gemini, Anthropic, Mistral e AWS Bedrock.

Risorse correlate

MCP e implementazione del protocollo

Sviluppo in Python

Scraping web e elaborazione del contenuto

Risorse per il deployment serverless

Integrazione con LLM

Conclusione

Costruire server MCP in Python apre potenti possibilità per estendere gli assistenti AI con strumenti personalizzati e fonti di dati. Le capacità di ricerca web e scraping dimostrate qui sono solo l’inizio: puoi estendere questa base per integrare database, API, sistemi di file e quasi qualsiasi sistema esterno.

Il Model Context Protocol è ancora in evoluzione, ma il suo approccio standardizzato all’integrazione degli strumenti AI lo rende una tecnologia eccitante per gli sviluppatori che costruiscono la prossima generazione di applicazioni alimentate da AI. Che tu stia creando strumenti interni per l’organizzazione o costruendo server MCP pubblici per la comunità, Python fornisce una solida base per uno sviluppo e un deployment rapidi.