Construyendo servidores MCP en Python: Guía de búsqueda en la web y raspado

Construya servidores MCP para asistentes de IA con ejemplos en Python

Índice

El Protocolo de Contexto del Modelo (MCP) está revolucionando la forma en que los asistentes de IA interactúan con fuentes de datos externas y herramientas. En esta guía, exploraremos cómo construir servidores MCP en Python, con ejemplos centrados en las capacidades de búsqueda en la web y raspado.

MCP robots

¿Qué es el Protocolo de Contexto del Modelo?

El Protocolo de Contexto del Modelo (MCP) es un protocolo abierto introducido por Anthropic para estandarizar cómo los asistentes de IA se conectan a sistemas externos. En lugar de construir integraciones personalizadas para cada fuente de datos, el MCP proporciona una interfaz unificada que permite:

  • Asistentes de IA (como Claude, ChatGPT o aplicaciones LLM personalizadas) para descubrir y usar herramientas
  • Desarrolladores para exponer fuentes de datos, herramientas y prompts a través de un protocolo estandarizado
  • Integración sin problemas sin reinventar la rueda para cada caso de uso

El protocolo opera en una arquitectura cliente-servidor donde:

  • Clientes MCP (asistentes de IA) descubren y usan capacidades
  • Servidores MCP exponen recursos, herramientas y prompts
  • La comunicación ocurre a través de JSON-RPC sobre stdio o HTTP/SSE

Si estás interesado en implementar servidores MCP en otros idiomas, consulta nuestra guía sobre implementar servidor MCP en Go, que cubre las especificaciones del protocolo y la estructura de mensajes en detalle.

¿Por qué construir servidores MCP en Python?

Python es una excelente opción para el desarrollo de servidores MCP porque:

  1. Ecosistema rico: Bibliotecas como requests, beautifulsoup4, selenium y playwright hacen que el raspado web sea sencillo
  2. SDK MCP: El SDK oficial de Python (mcp) proporciona un soporte de implementación de servidor robusto
  3. Desarrollo rápido: La simplicidad de Python permite prototipado y iteración rápidos
  4. Integración con IA/ML: Integración fácil con bibliotecas de IA como langchain, openai y herramientas de procesamiento de datos
  5. Soporte de la comunidad: Gran comunidad con documentación extensa y ejemplos

Configurando tu entorno de desarrollo

Primero, crea un entorno virtual e instala las dependencias necesarias. El uso de entornos virtuales es esencial para la aislamiento de proyectos en Python - si necesitas un recordatorio, consulta nuestra guía de atajos de venv para comandos detallados y buenas prácticas.

# Crear y activar entorno virtual
python -m venv mcp-env
source mcp-env/bin/activate  # En Windows: mcp-env\Scripts\activate

# Instalar SDK MCP y bibliotecas de raspado web
pip install mcp requests beautifulsoup4 playwright lxml
playwright install  # Instalar controladores de navegador para Playwright

Alternativa moderna: Si prefieres una resolución y instalación de dependencias más rápida, considera usar uv - el nuevo gestor de paquetes y entornos de Python que puede ser significativamente más rápido que pip para proyectos grandes.

Construyendo un servidor MCP básico

Comencemos con una estructura básica de servidor MCP. Si eres nuevo en Python o necesitas una referencia rápida para la sintaxis y patrones comunes, nuestra guía de atajos de Python proporciona una visión general completa de los fundamentos de Python.

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

# Crear instancia del servidor
app = Server("websearch-scraper")

@app.list_tools()
async def list_tools() -> list[Tool]:
    """Definir herramientas disponibles"""
    return [
        Tool(
            name="search_web",
            description="Buscar información en la web",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "Consulta de búsqueda"
                    },
                    "max_results": {
                        "type": "number",
                        "description": "Número máximo de resultados",
                        "default": 5
                    }
                },
                "required": ["query"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    """Manejar la ejecución de herramientas"""
    if name == "search_web":
        query = arguments["query"]
        max_results = arguments.get("max_results", 5)
        
        # Implementar lógica de búsqueda aquí
        results = await perform_web_search(query, max_results)
        
        return [TextContent(
            type="text",
            text=f"Resultados de búsqueda para '{query}':\n\n{results}"
        )]
    
    raise ValueError(f"Herramienta desconocida: {name}")

async def perform_web_search(query: str, max_results: int) -> str:
    """Sustituto para la implementación real de búsqueda"""
    return f"Encontrados {max_results} resultados para: {query}"

async def main():
    """Ejecutar el servidor"""
    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())

Implementando funcionalidad de búsqueda en la web

Ahora implementemos una herramienta real de búsqueda en la web usando DuckDuckGo (que no requiere claves 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]:
    """Buscar en DuckDuckGo y analizar resultados"""
    
    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 búsqueda falló: {str(e)}")

def format_search_results(results: list[dict]) -> str:
    """Formatear resultados de búsqueda para su visualización"""
    if not results:
        return "No se encontraron resultados."
    
    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)

Añadiendo capacidades de raspado web

Añadamos una herramienta para raspar y extraer contenido de páginas web. Cuando se raspa contenido HTML para usarlo con LLMs, también puede que desees convertirlo a formato Markdown para un mejor procesamiento. Para este propósito, consulta nuestra guía completa sobre convertir HTML a Markdown con Python, que compara 6 bibliotecas diferentes con benchmarks y recomendaciones prácticas.

from playwright.async_api import async_playwright
import asyncio

async def scrape_webpage(url: str, selector: str = None) -> dict:
    """Raspar contenido de una página web usando 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)
            
            # Esperar a que el contenido se cargue
            await page.wait_for_load_state('networkidle')
            
            if selector:
                # Extraer elemento específico
                element = await page.query_selector(selector)
                content = await element.inner_text() if element else "Selector no encontrado"
            else:
                # Extraer contenido principal
                content = await page.inner_text('body')
            
            title = await page.title()
            
            return {
                "title": title,
                "content": content[:5000],  # Limitar longitud del contenido
                "url": url,
                "success": True
            }
            
        except Exception as e:
            return {
                "error": str(e),
                "url": url,
                "success": False
            }
        finally:
            await browser.close()

# Añadir herramienta de raspado al servidor MCP
@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="search_web",
            description="Buscar en la web usando DuckDuckGo",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "Consulta de búsqueda"},
                    "max_results": {"type": "number", "default": 5}
                },
                "required": ["query"]
            }
        ),
        Tool(
            name="scrape_webpage",
            description="Raspar contenido de una página web",
            inputSchema={
                "type": "object",
                "properties": {
                    "url": {"type": "string", "description": "URL a raspar"},
                    "selector": {
                        "type": "string",
                        "description": "Selector CSS opcional para contenido específico"
                    }
                },
                "required": ["url"]
            }
        )
    ]

Implementación completa del servidor MCP

Aquí tienes una implementación completa y lista para producción del servidor MCP con capacidades de búsqueda y raspado:

#!/usr/bin/env python3
"""
Servidor MCP para búsqueda en la web y raspado
Proporciona herramientas para buscar en la web y extraer contenido de páginas
"""

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

# Configurar registro
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("websearch-scraper")

# Crear servidor
app = Server("websearch-scraper")

# Implementación de búsqueda
async def search_web(query: str, max_results: int = 5) -> str:
    """Buscar en DuckDuckGo y devolver resultados formateados"""
    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 ""
                })
        
        # Formatear resultados
        if not results:
            return "No se encontraron resultados."
        
        formatted = [f"Encontrados {len(results)} resultados para '{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 búsqueda falló: {e}")
        return f"Error de búsqueda: {str(e)}"

# Implementación de raspado
async def scrape_page(url: str, selector: str = None) -> str:
    """Raspar contenido de una página web usando 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 "Selector no encontrado"
            else:
                content = await page.inner_text('body')
            
            # Limitar longitud del contenido
            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"El raspado falló: {e}")
            return f"Error de raspado: {str(e)}"
        finally:
            await browser.close()

# Definiciones de herramientas MCP
@app.list_tools()
async def list_tools() -> list[Tool]:
    """Listar herramientas MCP disponibles"""
    return [
        Tool(
            name="search_web",
            description="Buscar en la web usando DuckDuckGo. Devuelve títulos, fragmentos y URLs.",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "La consulta de búsqueda"
                    },
                    "max_results": {
                        "type": "number",
                        "description": "Número máximo de resultados (por defecto: 5)",
                        "default": 5
                    }
                },
                "required": ["query"]
            }
        ),
        Tool(
            name="scrape_webpage",
            description="Extraer contenido de una página web. Puede apuntar a elementos específicos con selectores CSS.",
            inputSchema={
                "type": "object",
                "properties": {
                    "url": {
                        "type": "string",
                        "description": "La URL a raspar"
                    },
                    "selector": {
                        "type": "string",
                        "description": "Selector CSS opcional para extraer contenido específico"
                    }
                },
                "required": ["url"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: Any) -> list[TextContent]:
    """Manejar la ejecución de herramientas"""
    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"Herramienta desconocida: {name}")
    
    except Exception as e:
        logger.error(f"La ejecución de la herramienta falló: {e}")
        return [TextContent(
            type="text",
            text=f"Error al ejecutar {name}: {str(e)}"
        )]

async def main():
    """Ejecutar el servidor MCP"""
    logger.info("Iniciando servidor 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())

Configurando tu servidor MCP

Para usar tu servidor MCP con Claude Desktop u otros clientes MCP, crea un archivo de configuración:

Para Claude Desktop (claude_desktop_config.json):

{
  "mcpServers": {
    "websearch-scraper": {
      "command": "python",
      "args": [
        "/ruta/a/tu/mcp_server.py"
      ],
      "env": {}
    }
  }
}

Ubicación:

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

Pruebas de tu servidor MCP

Crea un script de prueba para verificar la funcionalidad:

import asyncio
import json
import sys
from io import StringIO

async def test_mcp_server():
    """Probar el servidor MCP localmente"""
    
    # Probar búsqueda
    print("Probando búsqueda en la web...")
    results = await search_web("tutorial de MCP en Python", 3)
    print(results)
    
    # Probar raspador
    print("\n\nProbando raspador de páginas web...")
    content = await scrape_page("https://example.com")
    print(content[:500])

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

Características avanzadas y buenas prácticas

1. Limitación de velocidad

Implementa la limitación de velocidad para evitar sobrecargar los servidores objetivo:

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("Límite de velocidad excedido")
        
        self.requests[key].append(now)

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

2. Caché

Añade caché para mejorar el rendimiento:

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. Manejo de errores

Implementa un manejo de errores robusto:

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

4. Validación de entrada

Valida las entradas del usuario antes del procesamiento:

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

Consideraciones de implementación

Usando transporte SSE para implementaciones web

Para implementaciones web, usa el transporte SSE (Server-Sent Events). Si estás considerando una implementación sin servidor, podrías estar interesado en comparar el rendimiento de AWS Lambda entre JavaScript, Python y Golang para tomar una decisión informada sobre tu entorno de ejecución:

import mcp.server.sse

async def main_sse():
    """Ejecutar servidor con transporte 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()

Implementación en AWS Lambda

Los servidores MCP también pueden implementarse como funciones AWS Lambda, especialmente cuando se usa el transporte SSE. Para guías completas sobre la implementación en Lambda:

Implementación en Docker

Crea un Dockerfile para despliegue en contenedores:

FROM python:3.11-slim

WORKDIR /app

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

# Instalar dependencias de Python
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
RUN playwright install chromium
RUN playwright install-deps

# Copiar aplicación
COPY mcp_server.py .

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

Optimización del rendimiento

Operaciones asincrónicas

Usa asyncio para operaciones concurrentes:

async def search_multiple_queries(queries: list[str]) -> list[str]:
    """Buscar múltiples consultas concurrentemente"""
    tasks = [search_web(query) for query in queries]
    results = await asyncio.gather(*tasks)
    return results

Reutilización de conexiones

Reutiliza conexiones para un mejor rendimiento:

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

Mejores prácticas de seguridad

  1. Sanitización de entradas: Siempre valide y sane las entradas de los usuarios
  2. Lista blanca de URLs: Considere implementar una lista blanca de URLs para el web scraping
  3. Control de tiempos de espera: Establezca tiempos de espera adecuados para prevenir el agotamiento de recursos
  4. Límites de contenido: Límite el tamaño del contenido scrapeado
  5. Autenticación: Implemente autenticación para despliegues en producción
  6. HTTPS: Use HTTPS para el transporte de SSE en producción

Trabajo con diferentes proveedores de LLM

Aunque MCP fue desarrollado por Anthropic para Claude, el protocolo está diseñado para funcionar con cualquier LLM. Si está construyendo servidores MCP que interactúan con múltiples proveedores de IA y necesita salidas estructuradas, querrá revisar nuestro comparativa de salidas estructuradas entre proveedores de LLM populares incluyendo OpenAI, Gemini, Anthropic, Mistral y AWS Bedrock.

Enlaces útiles y recursos

Recursos relacionados

MCP y implementación del protocolo

Desarrollo en Python

Web scraping y procesamiento de contenido

Recursos para despliegue sin servidor

Integración con LLM

Conclusión

Construir servidores MCP en Python abre poderosas posibilidades para ampliar asistentes de IA con herramientas personalizadas y fuentes de datos. Las capacidades de búsqueda web y scraping demostradas aquí son solo el comienzo: puede ampliar esta base para integrar bases de datos, APIs, sistemas de archivos y casi cualquier sistema externo.

El Protocolo de Contexto de Modelo aún está evolucionando, pero su enfoque estandarizado para la integración de herramientas de IA lo hace una tecnología emocionante para desarrolladores que construyen la próxima generación de aplicaciones impulsadas por IA. Ya sea que esté creando herramientas internas para su organización o construyendo servidores MCP públicos para la comunidad, Python proporciona una excelente base para el desarrollo y despliegue rápidos.