Criando Servidores MCP em Python: Guia de WebSearch e Scraping

Construa servidores MCP para assistentes de IA com exemplos em Python

Conteúdo da página

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

Robôs MCP

O que é o Protocolo de Contexto de Modelo?

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

  • Assistentes de IA (como Claude, ChatGPT ou aplicativos de LLM personalizados) descobrirem e utilizarem ferramentas
  • Desenvolvedores exporem fontes de dados, ferramentas e prompts através de um protocolo padronizado
  • Integração perfeita 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 utilizam capacidades
  • Servidores MCP expõem recursos, ferramentas e prompts
  • A comunicação ocorre via JSON-RPC sobre stdio ou HTTP/SSE

Se você estiver interessado em implementar servidores MCP em outras linguagens, consulte nosso guia sobre implementar um servidor MCP em Go, que cobre as especificações do protocolo e a estrutura de mensagens em detalhes.

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 web direta
  2. SDK MCP: O SDK oficial Python (mcp) oferece suporte robusto à implementação do servidor
  3. Desenvolvimento Rápido: A simplicidade do Python permite prototipagem e iteração rápidas
  4. Integração IA/ML: Integração fácil com bibliotecas de IA como langchain, openai e ferramentas de processamento de dados
  5. Suporte da Comunidade: Grande comunidade 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 o isolamento de projetos Python - se precisar de uma revisão, consulte nosso Resumo de venv para comandos detalhados e melhores práticas.

# Criar e ativar ambiente virtual
python -m venv mcp-env
source mcp-env/bin/activate  # No Windows: mcp-env\Scripts\activate

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

Alternativa Moderna: Se você preferir resolução e instalação de dependências mais rápidas, considere usar o uv - o gerenciador de pacotes e ambientes modernos 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 em Python ou precisa de uma referência rápida para sintaxe e padrões comuns, nosso Resumo de 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

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

@app.list_tools()
async def list_tools() -> list[Tool]:
    """Definir ferramentas disponíveis"""
    return [
        Tool(
            name="search_web",
            description="Pesquisar na web por informações",
            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]:
    """Lidar com a execução da ferramenta"""
    if name == "search_web":
        query = arguments["query"]
        max_results = arguments.get("max_results", 5)
        
        # Implementar 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:
    """Placeholder para implementação real de pesquisa"""
    return f"Encontrados {max_results} resultados para: {query}"

async def main():
    """Executar 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 Funcionalidade de Pesquisa Web

Agora vamos implementar uma ferramenta de pesquisa web real 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]:
    """Pesquisar no DuckDuckGo e analisar 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"Pesquisa falhou: {str(e)}")

def format_search_results(results: list[dict]) -> str:
    """Formatar resultados de pesquisa 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 Web

Vamos adicionar uma ferramenta para raspar e extrair conteúdo de páginas web. Ao raspar conteúdo HTML para uso com LLMs, você também pode querer convertê-lo para o formato Markdown para um melhor processamento. Para este fim, consulte nosso guia abrangente sobre convertendo 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:
    """Raspar conteúdo de uma página 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)
            
            # Aguardar o carregamento do conteúdo
            await page.wait_for_load_state('networkidle')
            
            if selector:
                # Extrair elemento específico
                element = await page.query_selector(selector)
                content = await element.inner_text() if element else "Seletor não encontrado"
            else:
                # Extrair conteúdo principal
                content = await page.inner_text('body')
            
            title = await page.title()
            
            return {
                "title": title,
                "content": content[:5000],  # Limitar o tamanho do conteúdo
                "url": url,
                "success": True
            }
            
        except Exception as e:
            return {
                "error": str(e),
                "url": url,
                "success": False
            }
        finally:
            await browser.close()

# Adicionar ferramenta de raspador ao servidor MCP
@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="search_web",
            description="Pesquisar na web usando 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="Raspar conteúdo de uma página web",
            inputSchema={
                "type": "object",
                "properties": {
                    "url": {"type": "string", "description": "URL a ser raspada"},
                    "selector": {
                        "type": "string",
                        "description": "Seletor CSS opcional para conteúdo específico"
                    }
                },
                "required": ["url"]
            }
        )
    ]

Implementação Completa do Servidor MCP

Aqui está um servidor MCP completo e pronto para produção com capacidades de pesquisa e raspagem:

#!/usr/bin/env python3
"""
Servidor MCP para Pesquisa e Raspagem Web
Fornece ferramentas para pesquisar 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

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

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

# Implementação de pesquisa
async def search_web(query: str, max_results: int = 5) -> str:
    """Pesquisar no DuckDuckGo e retornar 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 ""
                })
        
        # Formatar resultados
        if not results:
            return "Nenhum resultado encontrado."
        
        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"Pesquisa falhou: {e}")
        return f"Erro na pesquisa: {str(e)}"

# Implementação de raspador
async def scrape_page(url: str, selector: str = None) -> str:
    """Raspar conteúdo da página 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)
            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')
            
            # Limitar tamanho 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 na raspagem: {str(e)}"
        finally:
            await browser.close()

# Definições de Ferramentas MCP
@app.list_tools()
async def list_tools() -> list[Tool]:
    """Listar ferramentas MCP disponíveis"""
    return [
        Tool(
            name="search_web",
            description="Pesquisar na web usando DuckDuckGo. Retorna títulos, snippets e URLs.",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "A consulta de pesquisa"
                    },
                    "max_results": {
                        "type": "number",
                        "description": "Número máximo de resultados (padrão: 5)",
                        "default": 5
                    }
                },
                "required": ["query"]
            }
        ),
        Tool(
            name="scrape_webpage",
            description="Extrair conteúdo de uma página web. Pode mirar em elementos específicos com seletores CSS.",
            inputSchema={
                "type": "object",
                "properties": {
                    "url": {
                        "type": "string",
                        "description": "A URL a ser raspada"
                    },
                    "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]:
    """Lidar 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():
    """Executar o 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 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():
    """Testar servidor MCP localmente"""
    
    # Testar pesquisa
    print("Testando pesquisa web...")
    results = await search_web("Tutorial Python MCP", 3)
    print(results)
    
    # Testar raspador
    print("\n\nTestando raspador de páginas web...")
    content = await scrape_page("https://example.com")
    print(content[:500])

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

Recursos Avançados e Melhores Práticas

1. Limitação de Taxa

Implemente limitação de taxa para evitar sobrecarregar os 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. Cache

Adicione cache 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 as 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 de Implantação

Usando Transporte SSE para Implantação Web

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

import mcp.server.sse

async def main_sse():
    """Executar 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 do AWS Lambda

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

Implantação Docker

Crie um Dockerfile para implantação em container:

FROM python:3.11-slim

WORKDIR /app

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

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

# Copiar 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]:
    """Pesquisar várias consultas simultaneamente"""
    tasks = [search_web(query) for query in queries]
    results = await asyncio.gather(*tasks)
    return results

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

Melhores Práticas de Segurança

  1. Sanitização de Entrada: Sempre valide e sanitize entradas do usuário
  2. Lista de Permissões de URL: Considere implementar uma lista de permissões de URL para raspagem
  3. Controles de Tempo Limite: Defina tempos limite apropriados para evitar esgotamento de recursos
  4. Limites de Conteúdo: Limite o tamanho do conteúdo raspado
  5. Autenticação: Implemente autenticação para implantações em produção
  6. HTTPS: Use HTTPS para transporte 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ê estiver construindo servidores MCP que interagem com múltiplos provedores de IA e precisar de saídas estruturadas, você deverá revisar nossa comparação de saída estruturada entre provedores populares de LLM, incluindo OpenAI, Gemini, Anthropic, Mistral e AWS Bedrock.

Recursos Relacionados

Implementação de MCP e Protocolo

Desenvolvimento Python

Raspagem Web e Processamento de Conteúdo

Recursos de Implantação Serverless

Integração 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 pesquisa e raspagem web demonstradas aqui são apenas o início — você pode estender esta base para integrar bancos de dados, APIs, sistemas de arquivos e virtualmente qualquer sistema externo.

O Protocolo de Contexto de Modelo ainda está em evolução, mas sua abordagem padronizada para integração de ferramentas de IA o torna uma tecnologia emocionante para desenvolvedores que constroem a próxima geração de aplicativos alimentados por IA. Seja criando ferramentas internas para sua organização ou construindo servidores MCP públicos para a comunidade, o Python oferece uma excelente base para desenvolvimento e implantação rápidos.

Assinar

Receba novos artigos sobre sistemas, infraestrutura e engenharia de IA.