Construindo Servidores MCP em Python: Guia de Pesquisa na Web e Raspagem

Construa servidores MCP para assistentes de IA com exemplos em Python

Conteúdo da página

O Protocolo de Contexto do Modelo (MCP) está revolucionando a forma como os assistentes de IA interagem com fontes de dados externas e ferramentas. Neste guia, exploraremos como construir servidores MCP em Python, com exemplos focados nas capacidades de busca na web e raspagem.

Robôs MCP

O que é o Protocolo de Contexto do Modelo?

O Protocolo de Contexto do Modelo (MCP) é um protocolo aberto introduzido pela Anthropic para padronizar como os assistentes de IA se conectam a sistemas externos. Em vez de construir integrações personalizadas para cada fonte de dados, o MCP fornece uma interface unificada que permite:

  • Assistentes de IA (como o Claude, o ChatGPT ou aplicações personalizadas de LLM) descobrir e usar ferramentas
  • Desenvolvedores expor fontes de dados, ferramentas e prompts por meio de um protocolo padronizado
  • Integração sem esforço sem reinventar a roda para cada caso de uso

O protocolo opera em uma arquitetura cliente-servidor onde:

  • Clientes MCP (assistentes de IA) descobrem e usam capacidades
  • Servidores MCP expõem recursos, ferramentas e prompts
  • A comunicação ocorre via JSON-RPC por meio de stdio ou HTTP/SSE

Se você estiver interessado em implementar servidores MCP em outros idiomas, consulte nosso guia sobre implementação de servidor MCP em Go, que aborda detalhadamente as especificações do protocolo e a estrutura de mensagem.

Por que construir servidores MCP em Python?

O Python é uma excelente escolha para o desenvolvimento de servidores MCP porque:

  1. Ecossistema rico: Bibliotecas como requests, beautifulsoup4, selenium e playwright tornam a raspagem da web direta
  2. SDK MCP: O SDK oficial do Python (mcp) fornece suporte robusto para a implementação do servidor
  3. Desenvolvimento rápido: A simplicidade do Python permite prototipagem e iteração rápidas
  4. Integração com IA/ML: Facilidade de integração com bibliotecas de IA como langchain, openai e ferramentas de processamento de dados
  5. Suporte da comunidade: Comunidade grande com documentação extensa e exemplos

Configurando seu ambiente de desenvolvimento

Primeiro, crie um ambiente virtual e instale as dependências necessárias. O uso de ambientes virtuais é essencial para a isolamento de projetos em Python - se você precisar de um reforço, consulte nossa folha de dicas venv para comandos detalhados e melhores práticas.

# Crie e ative o ambiente virtual
python -m venv mcp-env
source mcp-env/bin/activate  # No Windows: mcp-env\Scripts\activate

# Instale o SDK MCP e bibliotecas de raspagem da web
pip install mcp requests beautifulsoup4 playwright lxml
playwright install  # Instale os drivers do navegador para o Playwright

Alternativa moderna: Se você preferir uma resolução e instalação mais rápidas das dependências, considere usar uv - o novo gerenciador de pacotes e ambientes do Python que pode ser significativamente mais rápido que o pip para projetos grandes.

Construindo um servidor MCP básico

Vamos começar com uma estrutura mínima de servidor MCP. Se você é novo no Python ou precisa de uma referência rápida para sintaxe e padrões comuns, nossa folha de dicas do Python fornece uma visão abrangente dos fundamentos do Python.

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

# Crie a instância do servidor
app = Server("websearch-scraper")

@app.list_tools()
async def list_tools() -> list[Tool]:
    """Defina as ferramentas disponíveis"""
    return [
        Tool(
            name="search_web",
            description="Procure informações na web",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "Consulta de pesquisa"
                    },
                    "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]:
    """Lide com a execução da ferramenta"""
    if name == "search_web":
        query = arguments["query"]
        max_results = arguments.get("max_results", 5)
        
        # Implemente a lógica de pesquisa aqui
        results = await perform_web_search(query, max_results)
        
        return [TextContent(
            type="text",
            text=f"Resultados da pesquisa para '{query}':\n\n{results}"
        )]
    
    raise ValueError(f"Ferramenta desconhecida: {name}")

async def perform_web_search(query: str, max_results: int) -> str:
    """Place holder para a implementação real de pesquisa"""
    return f"Encontrado {max_results} resultados para: {query}"

async def main():
    """Execute o 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 a funcionalidade de busca na web

Agora vamos implementar uma ferramenta real de busca na web usando o DuckDuckGo (que não requer chaves de 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]:
    """Busque no DuckDuckGo e analise os 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"A busca falhou: {str(e)}")

def format_search_results(results: list[dict]) -> str:
    """Formate os resultados da busca para exibição"""
    if not results:
        return "Nenhum resultado encontrado."
    
    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)

Adicionando capacidades de raspagem da web

Vamos adicionar uma ferramenta para raspar e extrair conteúdo de páginas da web. Ao raspar conteúdo HTML para uso com LLMs, você também pode querer convertê-lo para o formato Markdown para um processamento melhor. Para esse propósito, consulte nosso guia abrangente sobre conversão de HTML para Markdown com Python, que compara 6 bibliotecas diferentes com benchmarks e recomendações práticas.

from playwright.async_api import async_playwright
import asyncio

async def scrape_webpage(url: str, selector: str = None) -> dict:
    """Raspe conteúdo de uma página da web usando o 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)
            
            # Aguarde o conteúdo carregar
            await page.wait_for_load_state('networkidle')
            
            if selector:
                # Extraia um elemento específico
                element = await page.query_selector(selector)
                content = await element.inner_text() if element else "Seletor não encontrado"
            else:
                # Extraia o conteúdo principal
                content = await page.inner_text('body')
            
            title = await page.title()
            
            return {
                "title": title,
                "content": content[:5000],  # Limite o comprimento do conteúdo
                "url": url,
                "success": True
            }
            
        except Exception as e:
            return {
                "error": str(e),
                "url": url,
                "success": False
            }
        finally:
            await browser.close()

# Adicione a ferramenta de raspagem ao servidor MCP
@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="search_web",
            description="Busque na web usando o DuckDuckGo",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "Consulta de pesquisa"},
                    "max_results": {"type": "number", "default": 5}
                },
                "required": ["query"]
            }
        ),
        Tool(
            name="scrape_webpage",
            description="Raspe conteúdo de uma página da web",
            inputSchema={
                "type": "object",
                "properties": {
                    "url": {"type": "string", "description": "URL para raspar"},
                    "selector": {
                        "type": "string",
                        "description": "Seletor CSS opcional para conteúdo específico"
                    }
                },
                "required": ["url"]
            }
        )
    ]

Implementação completa do servidor MCP

Aqui está uma implementação completa e pronta para produção do servidor MCP com capacidades de busca e raspagem:

#!/usr/bin/env python3
"""
Servidor MCP para busca na web e raspagem
Fornece ferramentas para buscar na web e extrair conteúdo 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

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

# Crie o servidor
app = Server("websearch-scraper")

# Implementação de busca
async def search_web(query: str, max_results: int = 5) -> str:
    """Busque no DuckDuckGo e retorne resultados formatados"""
    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 ""
                })
        
        # Formate os resultados
        if not results:
            return "Nenhum resultado encontrado."
        
        formatted = [f"Encontrado {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"A busca falhou: {e}")
        return f"Erro de busca: {str(e)}"

# Implementação de raspagem
async def scrape_page(url: str, selector: str = None) -> str:
    """Raspe o conteúdo da página usando o 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 "Seletor não encontrado"
            else:
                content = await page.inner_text('body')
            
            # Limite o comprimento do conteúdo
            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"Raspagem falhou: {e}")
            return f"Erro de raspagem: {str(e)}"
        finally:
            await browser.close()

# Definições de ferramentas MCP
@app.list_tools()
async def list_tools() -> list[Tool]:
    """Liste as ferramentas MCP disponíveis"""
    return [
        Tool(
            name="search_web",
            description="Busque na web usando o DuckDuckGo. Retorna títulos, snippets e URLs.",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "A consulta de busca"
                    },
                    "max_results": {
                        "type": "number",
                        "description": "Número máximo de resultados (padrão: 5)",
                        "default": 5
                    }
                },
                "required": ["query"]
            }
        ),
        Tool(
            name="scrape_webpage",
            description="Extraia conteúdo de uma página da web. Pode alvo elementos específicos com seletores CSS.",
            inputSchema={
                "type": "object",
                "properties": {
                    "url": {
                        "type": "string",
                        "description": "A URL para raspar"
                    },
                    "selector": {
                        "type": "string",
                        "description": "Seletor CSS opcional para extrair conteúdo específico"
                    }
                },
                "required": ["url"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: Any) -> list[TextContent]:
    """Lide com a execução da ferramenta"""
    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"Ferramenta desconhecida: {name}")
    
    except Exception as e:
        logger.error(f"Execução da ferramenta falhou: {e}")
        return [TextContent(
            type="text",
            text=f"Erro ao executar {name}: {str(e)}"
        )]

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

Para usar seu servidor MCP com o Claude Desktop ou outros clientes MCP, crie um arquivo de configuração:

Para o Claude Desktop (claude_desktop_config.json):

{
  "mcpServers": {
    "websearch-scraper": {
      "command": "python",
      "args": [
        "/caminho/para/seu/mcp_server.py"
      ],
      "env": {}
    }
  }
}

Localização:

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

Testando seu servidor MCP

Crie um script de teste para verificar a funcionalidade:

import asyncio
import json
import sys
from io import StringIO

async def test_mcp_server():
    """Teste o servidor MCP localmente"""
    
    # Teste de busca
    print("Testando busca na web...")
    results = await search_web("Tutorial de MCP em Python", 3)
    print(results)
    
    # Teste de raspagem
    print("\n\nTestando raspagem de páginas da web...")
    content = await scrape_page("https://example.com")
    print(content[:500])

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

Funcionalidades avançadas e melhores práticas

1. Limitação de taxa

Implemente a limitação de taxa para evitar sobrecarregar servidores-alvo:

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 taxa excedido")
        
        self.requests[key].append(now)

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

2. Caching

Adicione caching para melhorar o desempenho:

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. Tratamento de erros

Implemente tratamento de erros 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"Erro ({error_type.value}): {str(error)}"

4. Validação de entrada

Valide entradas do usuário antes do processamento:

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

Considerações sobre implantação

Usando transporte SSE para implantação na web

Para implantações baseadas na web, use o transporte SSE (Server-Sent Events). Se você estiver considerando implantação sem servidor, talvez esteja interessado em comparar desempenho de Lambda do AWS em JavaScript, Python e Golang para tomar uma decisão informada sobre seu tempo de execução:

import mcp.server.sse

async def main_sse():
    """Execute o servidor com 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()

Implantação em AWS Lambda

Servidores MCP também podem ser implantados como funções AWS Lambda, especialmente ao usar o transporte SSE. Para guias abrangentes sobre implantação em Lambda:

Implantação com Docker

Crie um Dockerfile para implantação em contêiner:

FROM python:3.11-slim

WORKDIR /app

# Instale dependências do sistema
RUN apt-get update && apt-get install -y \
    wget \
    && rm -rf /var/lib/apt/lists/*

# Instale dependências do Python
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
RUN playwright install chromium
RUN playwright install-deps

# Copie a aplicação
COPY mcp_server.py .

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

Otimização de desempenho

Operações assíncronas

Use asyncio para operações concorrentes:

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

Pooling de conexões

Reutilize conexões para melhor desempenho:

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

Boas práticas de segurança

  1. Sanitização de entrada: Sempre valide e sanitize as entradas do usuário
  2. Lista de URLs permitidas: Considere a implementação de uma lista de URLs permitidas para scraping
  3. Controle de tempo limite: Defina limites de tempo apropriados para evitar o esgotamento de recursos
  4. Limites de conteúdo: Limite o tamanho do conteúdo extraído
  5. Autenticação: Implemente autenticação para implantações em produção
  6. HTTPS: Use HTTPS para o transporte de SSE em produção

Trabalhando com diferentes provedores de LLM

Embora o MCP tenha sido desenvolvido pela Anthropic para o Claude, o protocolo foi projetado para funcionar com qualquer LLM. Se você está construindo servidores MCP que interagem com múltiplos provedores de IA e precisa de saídas estruturadas, você deverá revisar nossa comparação de saída estruturada entre os principais provedores de LLM incluindo OpenAI, Gemini, Anthropic, Mistral e AWS Bedrock.

Recursos relacionados

MCP e implementação do protocolo

Desenvolvimento em Python

Web scraping e processamento de conteúdo

Recursos de implantação sem servidor

Integração com LLM

Conclusão

Construir servidores MCP em Python abre possibilidades poderosas para estender assistentes de IA com ferramentas personalizadas e fontes de dados. As capacidades de busca na web e scraping demonstradas aqui são apenas o começo — você pode estender essa base para integrar bancos de dados, APIs, sistemas de arquivos e praticamente qualquer sistema externo.

O Protocolo de Contexto do Modelo ainda está em evolução, mas sua abordagem padronizada para a integração de ferramentas de IA torna-o uma tecnologia empolgante para desenvolvedores que estão construindo a próxima geração de aplicações com IA. Seja você criando ferramentas internas para sua organização ou construindo servidores MCP públicos para a comunidade, o Python oferece uma excelente base para o desenvolvimento e implantação rápidos.