De API REST a servidor MCP em Python: passo a passo

Introdução às APIs MCP

Imagine a cena: você passou meses construindo APIs REST sólidas. Seus endpoints funcionam perfeitamente, sua documentação é completa e tudo roda de forma tranquila. Aí, de repente, todo mundo começa a falar sobre agentes de IA e LLMs, e você fica se perguntando: "Como faço para minhas APIs funcionarem com essas aplicações de IA sem começar tudo do zero?"

Se você está concordando com a cabeça, saiba que não está sozinho. Esse é exatamente o desafio que milhares de desenvolvedores estão enfrentando agora.

E aqui vai a boa notícia: você não precisa jogar seu trabalho fora. O Model Context Protocol (MCP) é como um tradutor universal que faz suas APIs REST existentes falarem "IA" fluentemente. Pense nele como um adaptador inteligente adicionado às suas APIs - elas continuam fazendo o que fazem de melhor, mas agora as aplicações de IA conseguem entendê-las e usá-las sem esforço.

Neste guia, vamos construir um servidor MCP passo a passo, começando de forma simples e adicionando recursos conforme avançamos. Você verá exatamente como cada peça se encaixa e, ao final, terá um servidor MCP funcional que envolve sua API REST e se conecta perfeitamente com clientes de IA como o Claude Desktop.

APIs REST vs Servidores MCP

Vamos começar pelo que você já conhece. Uma API REST é um conjunto de endpoints HTTP que aceitam e retornam dados estruturados. Cada endpoint tem um caminho, um método (GET, POST, etc.) e um schema de requisição/resposta. Seus clientes fazem requisições HTTP e recebem respostas JSON de volta.

Um servidor MCP é diferente. Em vez de endpoints HTTP, ele expõe "tools" e "resources" que as aplicações de IA podem usar. Tools são para ações que alteram dados (como criar um usuário), enquanto resources fornecem acesso somente leitura a informações (como obter os detalhes de um usuário).

O insight fundamental é este: sua API REST se torna o motor, e o servidor MCP se torna o tradutor. O servidor MCP recebe requisições das aplicações de IA, as traduz em chamadas à API REST e, em seguida, formata as respostas de um jeito que as aplicações de IA consigam entender.

Veja como os componentes se mapeiam:

Componente da API REST	Equivalente no Servidor MCP	Finalidade
Endpoints POST/PUT/DELETE	Tools	Ações que modificam dados
Endpoints GET	Resources	Acesso somente leitura a dados
Parâmetros de query	Parâmetros de tool	Campos de entrada para tools
Cabeçalhos de autenticação	Variáveis de ambiente	Manipulação segura de credenciais

Pré-requisitos

Antes de começarmos a construir, certifique-se de que você tem:

1. Uma API REST funcional com endpoints documentados (usaremos uma API simples de gerenciamento de usuários como exemplo)
2. Python 3.8+ instalado no seu sistema
3. Conhecimento básico de JSON e conceitos de API
4. Uma chave de API ou método de autenticação para a sua API REST

Para nossos exemplos, vamos assumir que você tem uma API REST com estes endpoints:

• GET /users/{id} - Obter detalhes do usuário
• POST /users - Criar um novo usuário
• PUT /users/{id} - Atualizar informações do usuário
• DELETE /users/{id} - Excluir um usuário

Passo 1: Configurando seu ambiente de desenvolvimento

Vamos começar criando um novo projeto e instalando as dependências necessárias.

# Create a new directory for your MCP server
mkdir my-mcp-server
cd my-mcp-server

# Create a virtual environment
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

# Install the MCP SDK and dependencies
pip install mcp httpx python-dotenv

Crie um arquivo .env para armazenar as credenciais da sua API:

# .env
API_BASE_URL=https://your-api.example.com
API_KEY=your-api-key-here

Passo 2: Criando sua primeira tool MCP

Vamos começar com o servidor MCP mais simples possível. Vamos criar uma única tool que busca informações de usuário na sua API REST.

Crie um arquivo chamado server.py:

import os
import httpx
from mcp.server.fastmcp import FastMCP

# Load environment variables
API_BASE_URL = os.getenv("API_BASE_URL")
API_KEY = os.getenv("API_KEY")

# Initialize the MCP server
mcp = FastMCP("user-api-server")

@mcp.tool()
async def get_user(user_id: str) -> dict:
    """Get user details by ID from the REST API."""

    # Make the REST API call
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"{API_BASE_URL}/users/{user_id}",
            headers={"Authorization": f"Bearer {API_KEY}"}
        )

        if response.status_code == 200:
            return response.json()
        else:
            return {"error": f"Failed to get user: {response.status_code}"}

if __name__ == "__main__":
    mcp.run(transport="stdio")

É isso! Você acabou de criar seu primeiro servidor MCP. Vamos testá-lo:

python server.py

O servidor vai iniciar e aguardar a entrada de dados. Você pode testá-lo usando o MCP Inspector (vamos abordar isso na seção de testes).

Passo 3: Adicionando tratamento de erros e validação

Nossa primeira versão funciona, mas não é muito robusta. Vamos adicionar tratamento de erros e validação de entrada adequados:

async def make_api_request(method: str, endpoint: str, data: dict = None) -> Dict[str, Any]:
    """Helper function to make REST API requests with proper error handling."""

    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json"
    }

    async with httpx.AsyncClient(timeout=30.0) as client:
        try:
            if method == "GET":
                response = await client.get(f"{API_BASE_URL}{endpoint}", headers=headers)
            elif method == "POST":
                response = await client.post(f"{API_BASE_URL}{endpoint}", headers=headers, json=data)
            elif method == "PUT":
                response = await client.put(f"{API_BASE_URL}{endpoint}", headers=headers, json=data)
            elif method == "DELETE":
                response = await client.delete(f"{API_BASE_URL}{endpoint}", headers=headers)

            response.raise_for_status()

            # Handle empty responses (common for DELETE operations)
            if response.status_code == 204 or not response.content:
                return {"success": True, "message": "Operation completed successfully"}

            return response.json()

        except httpx.HTTPStatusError as e:
            return {"error": f"HTTP {e.response.status_code}: {e.response.text}"}
        except httpx.TimeoutException:
            return {"error": "Request timeout - API took too long to respond"}
        except Exception as e:
            return {"error": f"Request failed: {str(e)}"}

Agora nosso servidor trata erros de forma elegante e valida as entradas corretamente.

Passo 4: Adicionando mais tools

Vamos expandir nosso servidor para lidar com a criação e atualização de usuários. Vamos adicionar essas tools uma a uma:

@mcp.tool()
async def create_user(name: str, email: str, role: str = "user") -> dict:
    """Create a new user account.

    Args:
        name: User's full name
        email: User's email address
        role: User role (defaults to 'user')
    """

    # Validate inputs
    if not name or not name.strip():
        return {"error": "Name is required"}

    if not email or "@" not in email:
        return {"error": "Valid email is required"}

    user_data = {
        "name": name.strip(),
        "email": email.lower().strip(),
        "role": role
    }

    result = await make_api_request("POST", "/users", data=user_data)

    if "error" in result:
        return {"error": f"Failed to create user: {result['error']}"}

    return {
        "success": True,
        "message": f"User '{name}' created successfully",
        "user": result
    }

@mcp.tool()
async def update_user(user_id: str, name: str = None, email: str = None, role: str = None) -> dict:
    """Update an existing user's information.

    Args:
        user_id: ID of the user to update
        name: New name (optional)
        email: New email (optional)
        role: New role (optional)
    """

    if not user_id or not user_id.strip():
        return {"error": "User ID is required"}

    if not any([name, email, role]):
        return {"error": "At least one field (name, email, or role) must be provided"}

    # Build update data
    update_data = {}
    if name:
        update_data["name"] = name.strip()
    if email:
        if "@" not in email:
            return {"error": "Valid email is required"}
        update_data["email"] = email.lower().strip()
    if role:
        update_data["role"] = role

    result = await make_api_request("PUT", f"/users/{user_id.strip()}", data=update_data)

    if "error" in result:
        return {"error": f"Failed to update user: {result['error']}"}

    return {
        "success": True,
        "message": f"User {user_id} updated successfully",
        "user": result
    }

Passo 5: Adicionando resources para dados somente leitura

As tools são ótimas para ações, mas às vezes as aplicações de IA só precisam ler dados. É aí que entram os resources. Vamos adicionar um resource para obter informações de usuário:

@mcp.resource("user://{user_id}")
async def get_user_resource(user_id: str) -> str:
    """Get detailed user information as a resource.

    This provides read-only access to user data in a format
    that's easy for AI applications to understand.
    """

    result = await make_api_request("GET", f"/users/{user_id}")

    if "error" in result:
        return f"Error retrieving user {user_id}: {result['error']}"

    user = result

    # Format the data for AI consumption
    formatted_output = f"""
User Profile for ID {user_id}:
- Name: {user.get('name', 'Not specified')}
- Email: {user.get('email', 'Not specified')}
- Role: {user.get('role', 'Not specified')}
- Status: {user.get('status', 'Active')}
- Created: {user.get('created_at', 'Unknown')}
- Last Updated: {user.get('updated_at', 'Unknown')}
"""

    return formatted_output.strip()

@mcp.resource("users://list")
async def list_users_resource() -> str:
    """Get a list of all users in the system."""

    result = await make_api_request("GET", "/users")

    if "error" in result:
        return f"Error retrieving users: {result['error']}"

    users = result.get("users", []) if isinstance(result, dict) else result

    if not users:
        return "No users found in the system."

    # Format the user list for AI consumption
    formatted_output = f"User Directory ({len(users)} users):\n\n"

    for user in users:
        formatted_output += f"• {user.get('name', 'Unknown')} ({user.get('email', 'No email')})\n"
        formatted_output += f"  ID: {user.get('id', 'Unknown')}, Role: {user.get('role', 'Unknown')}\n\n"

    return formatted_output.strip()

Código completo do servidor MCP

Agora que construímos nosso servidor MCP passo a passo, aqui está o código completo, pronto para produção, que combina todas as peças que discutimos. Você pode copiar este arquivo inteiro e executá-lo imediatamente:

import os
import httpx
from mcp.server.fastmcp import FastMCP
from dotenv import load_dotenv
from typing import Dict, Any, Optional

# Load environment variables from .env file
load_dotenv()

# Load environment variables
API_BASE_URL = os.getenv("API_BASE_URL")
API_KEY = os.getenv("API_KEY")

if not API_BASE_URL or not API_KEY:
    raise ValueError("API_BASE_URL and API_KEY environment variables are required")

# Initialize the MCP server
mcp = FastMCP("user-api-server")

async def make_api_request(method: str, endpoint: str, data: dict = None) -> Dict[str, Any]:
    """Helper function to make REST API requests with proper error handling."""

    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json"
    }

    async with httpx.AsyncClient(timeout=30.0) as client:
        try:
            if method == "GET":
                response = await client.get(f"{API_BASE_URL}{endpoint}", headers=headers)
            elif method == "POST":
                response = await client.post(f"{API_BASE_URL}{endpoint}", headers=headers, json=data)
            elif method == "PUT":
                response = await client.put(f"{API_BASE_URL}{endpoint}", headers=headers, json=data)
            elif method == "DELETE":
                response = await client.delete(f"{API_BASE_URL}{endpoint}", headers=headers)

            response.raise_for_status()

            # Handle empty responses (common for DELETE operations)
            if response.status_code == 204 or not response.content:
                return {"success": True, "message": "Operation completed successfully"}

            return response.json()

        except httpx.HTTPStatusError as e:
            return {"error": f"HTTP {e.response.status_code}: {e.response.text}"}
        except httpx.TimeoutException:
            return {"error": "Request timeout - API took too long to respond"}
        except Exception as e:
            return {"error": f"Request failed: {str(e)}"}

# TOOLS - for actions that modify data
@mcp.tool()
async def get_user(user_id: str) -> dict:
    """Get user details by ID from the REST API.

    Args:
        user_id: The ID of the user to retrieve
    """

    # Validate input
    if not user_id or not user_id.strip():
        return {"error": "User ID is required"}

    result = await make_api_request("GET", f"/users/{user_id.strip()}")

    if "error" in result:
        return {"error": f"Failed to get user: {result['error']}"}

    return {
        "success": True,
        "user": result
    }

@mcp.tool()
async def create_user(name: str, email: str, role: str = "user") -> dict:
    """Create a new user account.

    Args:
        name: User's full name
        email: User's email address
        role: User role (defaults to 'user')
    """

    # Validate inputs
    if not name or not name.strip():
        return {"error": "Name is required"}

    if not email or "@" not in email:
        return {"error": "Valid email is required"}

    user_data = {
        "name": name.strip(),
        "email": email.lower().strip(),
        "role": role
    }

    result = await make_api_request("POST", "/users", data=user_data)

    if "error" in result:
        return {"error": f"Failed to create user: {result['error']}"}

    return {
        "success": True,
        "message": f"User '{name}' created successfully",
        "user": result
    }

@mcp.tool()
async def update_user(user_id: str, name: Optional[str] = None, email: Optional[str] = None, role: Optional[str] = None) -> dict:
    """Update an existing user's information.

    Args:
        user_id: ID of the user to update
        name: New name (optional)
        email: New email (optional)
        role: New role (optional)
    """

    if not user_id or not user_id.strip():
        return {"error": "User ID is required"}

    if not any([name, email, role]):
        return {"error": "At least one field (name, email, or role) must be provided"}

    # Build update data
    update_data = {}
    if name:
        update_data["name"] = name.strip()
    if email:
        if "@" not in email:
            return {"error": "Valid email is required"}
        update_data["email"] = email.lower().strip()
    if role:
        update_data["role"] = role

    result = await make_api_request("PUT", f"/users/{user_id.strip()}", data=update_data)

    if "error" in result:
        return {"error": f"Failed to update user: {result['error']}"}

    return {
        "success": True,
        "message": f"User {user_id} updated successfully",
        "user": result
    }

@mcp.tool()
async def delete_user(user_id: str) -> dict:
    """Delete a user account.

    Args:
        user_id: ID of the user to delete
    """

    if not user_id or not user_id.strip():
        return {"error": "User ID is required"}

    result = await make_api_request("DELETE", f"/users/{user_id.strip()}")

    if "error" in result:
        return {"error": f"Failed to delete user: {result['error']}"}

    return {
        "success": True,
        "message": f"User {user_id} deleted successfully"
    }

# RESOURCES - for read-only data access optimized for AI consumption
@mcp.resource("user://{user_id}")
async def get_user_resource(user_id: str) -> str:
    """Get detailed user information as a resource.

    This provides read-only access to user data in a format
    that's easy for AI applications to understand.
    """

    result = await make_api_request("GET", f"/users/{user_id}")

    if "error" in result:
        return f"Error retrieving user {user_id}: {result['error']}"

    user = result

    # Format the data for AI consumption
    formatted_output = f"""
User Profile for ID {user_id}:
- Name: {user.get('name', 'Not specified')}
- Email: {user.get('email', 'Not specified')}
- Role: {user.get('role', 'Not specified')}
- Status: {user.get('status', 'Active')}
- Created: {user.get('created_at', 'Unknown')}
- Last Updated: {user.get('updated_at', 'Unknown')}
"""

    return formatted_output.strip()

@mcp.resource("users://list")
async def list_users_resource() -> str:
    """Get a list of all users in the system."""

    result = await make_api_request("GET", "/users")

    if "error" in result:
        return f"Error retrieving users: {result['error']}"

    users = result.get("users", []) if isinstance(result, dict) else result

    if not users:
        return "No users found in the system."

    # Format the user list for AI consumption
    formatted_output = f"User Directory ({len(users)} users):\n\n"

    for user in users:
        formatted_output += f"• {user.get('name', 'Unknown')} ({user.get('email', 'No email')})\n"
        formatted_output += f"  ID: {user.get('id', 'Unknown')}, Role: {user.get('role', 'Unknown')}\n\n"

    return formatted_output.strip()

if __name__ == "__main__":
    mcp.run(transport="stdio")

Salve isto como server.py, certifique-se de que seu arquivo .env esteja configurado com as credenciais da sua API e execute:

python server.py

Passo 6: Testando seu servidor MCP

Antes de conectar aos clientes de IA, vamos testar nosso servidor usando o MCP Inspector. Primeiro, instale-o:

npm install -g @modelcontextprotocol/inspector

Em seguida, teste seu servidor:

mcp-inspector python server.py

Isso abre uma interface web onde você pode:

• Ver todas as suas tools e resources
• Testar cada tool com diferentes entradas
• Visualizar as respostas
• Depurar quaisquer problemas

Experimente chamar a tool get_user com um ID de usuário, ou a tool create_user com os parâmetros de nome e e-mail. Você deverá ver as chamadas reais à API REST sendo feitas e as respostas formatadas para consumo pela IA.

Passo 7: Conectando ao Claude Desktop

Agora a parte empolgante - conectar seu servidor MCP ao Claude Desktop para que você possa realmente usá-lo com uma aplicação de IA.

Instalando o Claude Desktop

Primeiro, baixe e instale o Claude Desktop caso ainda não tenha feito isso.

Configurando seu servidor MCP

O Claude Desktop procura as configurações do servidor MCP em um arquivo específico. A localização depende do seu sistema operacional:

No macOS:

~/Library/Application Support/Claude/claude_desktop_config.json

No Windows:

%APPDATA%\Claude\claude_desktop_config.json

No Linux:

~/.config/Claude/claude_desktop_config.json

Crie este arquivo caso ele não exista e adicione a configuração do seu servidor MCP:

{
  "mcpServers": {
    "user-api-server": {
      "command": "python",
      "args": ["/path/to/your/server.py"],
      "env": {
        "API_BASE_URL": "https://your-api.example.com",
        "API_KEY": "your-api-key-here"
      }
    }
  }
}

Observações importantes:

• Substitua /path/to/your/server.py pelo caminho absoluto real do seu arquivo de servidor
• Substitua as variáveis de ambiente pela URL e pela chave reais da sua API
• Certifique-se de que o Python esteja no PATH do seu sistema, ou use o caminho completo para o seu executável do Python

Testando a conexão

1. Reinicie o Claude Desktop após salvar o arquivo de configuração
2. Abra uma nova conversa no Claude Desktop
3. Procure pelo indicador MCP - você deverá ver um pequeno ícone de ferramenta ou indicador mostrando que seu servidor MCP está conectado
4. Teste suas tools pedindo ao Claude para interagir com sua API:

Experimente estes exemplos de prompts:

• "Você pode obter informações sobre o usuário de ID 123?"
• "Crie um novo usuário chamado John Doe com o e-mail john@example.com"
• "Atualize o usuário 456 para mudar o papel dele para admin"

O Claude agora deverá ser capaz de chamar suas tools MCP e trabalhar com os dados da sua API REST!

Resolvendo problemas de conexão

Se o Claude Desktop não conectar ao seu servidor:

1. Verifique os logs - o Claude Desktop geralmente mostra erros de conexão no console de desenvolvedor
2. Confira os caminhos dos arquivos - certifique-se de que todos os caminhos na configuração sejam absolutos e corretos
3. Teste seu servidor de forma independente - execute python server.py diretamente para garantir que ele inicie sem erros
4. Verifique as variáveis de ambiente - garanta que as credenciais da sua API estejam definidas corretamente
5. Reinicie o Claude Desktop - alterações de configuração exigem uma reinicialização

Passo 8: Recursos avançados e boas práticas

Adicionando tratamento de autenticação

Para uso em produção, você vai querer um tratamento de autenticação mais sofisticado:

import os
from datetime import datetime, timedelta

class APIAuthenticator:
    def __init__(self):
        self.api_key = os.getenv("API_KEY")
        self.token = None
        self.token_expires = None

    async def get_headers(self):
        """Get authentication headers, refreshing token if needed."""

        # For simple API key auth
        if self.api_key:
            return {"Authorization": f"Bearer {self.api_key}"}

        # For OAuth or token-based auth (implement token refresh logic here)
        if self.token and self.token_expires > datetime.now():
            return {"Authorization": f"Bearer {self.token}"}

        # Refresh token logic would go here
        await self.refresh_token()
        return {"Authorization": f"Bearer {self.token}"}

    async def refresh_token(self):
        """Implement your token refresh logic here."""
        pass

# Use the authenticator in your API requests
auth = APIAuthenticator()

async def make_authenticated_request(method: str, endpoint: str, data: dict = None):
    headers = await auth.get_headers()
    headers["Content-Type"] = "application/json"

    # Rest of your request logic...

Adicionando logging e monitoramento

Adicione logging abrangente para ajudar na depuração e no monitoramento:

import logging

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

async def make_api_request(method: str, endpoint: str, data: dict = None) -> Dict[str, Any]:
    """Enhanced API request function with logging."""

    logger.info(f"Making {method} request to {endpoint}")

    try:
        # Your existing request logic...

        logger.info(f"Request successful: {method} {endpoint}")
        return response.json()

    except Exception as e:
        logger.error(f"Request failed: {method} {endpoint} - {str(e)}")
        return {"error": f"Request failed: {str(e)}"}

Lidando com limites de taxa (rate limits)

Adicione lógica inteligente de limitação de taxa e de nova tentativa (retry):

import asyncio
from typing import Optional

class RateLimitHandler:
    def __init__(self, max_retries: int = 3, base_delay: float = 1.0):
        self.max_retries = max_retries
        self.base_delay = base_delay

    async def make_request_with_retry(self, request_func, *args, **kwargs):
        """Make a request with exponential backoff retry logic."""

        for attempt in range(self.max_retries + 1):
            try:
                response = await request_func(*args, **kwargs)

                # Check for rate limiting
                if hasattr(response, 'status_code') and response.status_code == 429:
                    if attempt < self.max_retries:
                        delay = self.base_delay * (2 ** attempt)
                        logger.warning(f"Rate limited, retrying in {delay} seconds...")
                        await asyncio.sleep(delay)
                        continue

                return response

            except Exception as e:
                if attempt < self.max_retries:
                    delay = self.base_delay * (2 ** attempt)
                    logger.warning(f"Request failed, retrying in {delay} seconds: {str(e)}")
                    await asyncio.sleep(delay)
                    continue
                raise e

        raise Exception("Max retries exceeded")

Código completo pronto para produção: todos os recursos avançados incluídos

Aqui está o servidor MCP completo, pronto para produção, que integra todos os recursos avançados do Passo 8. Esta versão inclui tratamento de autenticação, logging abrangente e limitação de taxa inteligente:

import os
import httpx
import asyncio
import logging
from mcp.server.fastmcp import FastMCP
from dotenv import load_dotenv
from typing import Dict, Any, Optional
from datetime import datetime, timedelta

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

# Load environment variables from .env file
load_dotenv()

# Load environment variables
API_BASE_URL = os.getenv("API_BASE_URL")
API_KEY = os.getenv("API_KEY")

if not API_BASE_URL or not API_KEY:
    raise ValueError("API_BASE_URL and API_KEY environment variables are required")

# Initialize the MCP server
mcp = FastMCP("user-api-server")

class APIAuthenticator:
    """Handle API authentication with token refresh capabilities."""

    def __init__(self):
        self.api_key = os.getenv("API_KEY")
        self.token = None
        self.token_expires = None

    async def get_headers(self):
        """Get authentication headers, refreshing token if needed."""

        # For simple API key auth
        if self.api_key:
            return {"Authorization": f"Bearer {self.api_key}"}

        # For OAuth or token-based auth (implement token refresh logic here)
        if self.token and self.token_expires > datetime.now():
            return {"Authorization": f"Bearer {self.token}"}

        # Refresh token logic would go here
        await self.refresh_token()
        return {"Authorization": f"Bearer {self.token}"}

    async def refresh_token(self):
        """Implement your token refresh logic here."""
        # This is where you'd implement OAuth token refresh
        # For now, we'll just use the API key
        pass

class RateLimitHandler:
    """Handle rate limiting with exponential backoff retry logic."""

    def __init__(self, max_retries: int = 3, base_delay: float = 1.0):
        self.max_retries = max_retries
        self.base_delay = base_delay

    async def make_request_with_retry(self, request_func, *args, **kwargs):
        """Make a request with exponential backoff retry logic."""

        for attempt in range(self.max_retries + 1):
            try:
                response = await request_func(*args, **kwargs)

                # Check for rate limiting
                if hasattr(response, 'status_code') and response.status_code == 429:
                    if attempt < self.max_retries:
                        delay = self.base_delay * (2 ** attempt)
                        logger.warning(f"Rate limited, retrying in {delay} seconds...")
                        await asyncio.sleep(delay)
                        continue

                return response

            except Exception as e:
                if attempt < self.max_retries:
                    delay = self.base_delay * (2 ** attempt)
                    logger.warning(f"Request failed, retrying in {delay} seconds: {str(e)}")
                    await asyncio.sleep(delay)
                    continue
                raise e

        raise Exception("Max retries exceeded")

# Initialize authentication and rate limiting
auth = APIAuthenticator()
rate_limiter = RateLimitHandler()

async def make_api_request(method: str, endpoint: str, data: dict = None) -> Dict[str, Any]:
    """Enhanced API request function with authentication, logging, and rate limiting."""

    logger.info(f"Making {method} request to {API_BASE_URL}{endpoint}")

    async def _make_request():
        headers = await auth.get_headers()
        headers["Content-Type"] = "application/json"

        async with httpx.AsyncClient(timeout=30.0) as client:
            if method == "GET":
                return await client.get(f"{API_BASE_URL}{endpoint}", headers=headers)
            elif method == "POST":
                return await client.post(f"{API_BASE_URL}{endpoint}", headers=headers, json=data)
            elif method == "PUT":
                return await client.put(f"{API_BASE_URL}{endpoint}", headers=headers, json=data)
            elif method == "DELETE":
                return await client.delete(f"{API_BASE_URL}{endpoint}", headers=headers)

    try:
        response = await rate_limiter.make_request_with_retry(_make_request)
        response.raise_for_status()

        # Handle empty responses (common for DELETE operations)
        if response.status_code == 204 or not response.content:
            logger.info(f"Request successful: {method} {endpoint}")
            return {"success": True, "message": "Operation completed successfully"}

        logger.info(f"Request successful: {method} {endpoint}")
        return response.json()

    except httpx.HTTPStatusError as e:
        logger.error(f"HTTP error: {e.response.status_code} for {method} {endpoint}")
        return {"error": f"HTTP {e.response.status_code}: {e.response.text}"}
    except httpx.TimeoutException:
        logger.error(f"Timeout error for {method} {endpoint}")
        return {"error": "Request timeout - API took too long to respond"}
    except Exception as e:
        logger.error(f"Unexpected error for {method} {endpoint}: {str(e)}")
        return {"error": f"Request failed: {str(e)}"}

# TOOLS - for actions that modify data
@mcp.tool()
async def get_user(user_id: str) -> dict:
    """Get user details by ID from the REST API.

    Args:
        user_id: The ID of the user to retrieve
    """

    # Validate input
    if not user_id or not user_id.strip():
        return {"error": "User ID is required"}

    result = await make_api_request("GET", f"/users/{user_id.strip()}")

    if "error" in result:
        return {"error": f"Failed to get user: {result['error']}"}

    return {
        "success": True,
        "user": result
    }

@mcp.tool()
async def create_user(name: str, email: str, role: str = "user") -> dict:
    """Create a new user account.

    Args:
        name: User's full name
        email: User's email address
        role: User role (defaults to 'user')
    """

    # Validate inputs
    if not name or not name.strip():
        return {"error": "Name is required"}

    if not email or "@" not in email:
        return {"error": "Valid email is required"}

    user_data = {
        "name": name.strip(),
        "email": email.lower().strip(),
        "role": role
    }

    result = await make_api_request("POST", "/users", data=user_data)

    if "error" in result:
        return {"error": f"Failed to create user: {result['error']}"}

    return {
        "success": True,
        "message": f"User '{name}' created successfully",
        "user": result
    }

@mcp.tool()
async def update_user(user_id: str, name: Optional[str] = None, email: Optional[str] = None, role: Optional[str] = None) -> dict:
    """Update an existing user's information.

    Args:
        user_id: ID of the user to update
        name: New name (optional)
        email: New email (optional)
        role: New role (optional)
    """

    if not user_id or not user_id.strip():
        return {"error": "User ID is required"}

    if not any([name, email, role]):
        return {"error": "At least one field (name, email, or role) must be provided"}

    # Build update data
    update_data = {}
    if name:
        update_data["name"] = name.strip()
    if email:
        if "@" not in email:
            return {"error": "Valid email is required"}
        update_data["email"] = email.lower().strip()
    if role:
        update_data["role"] = role

    result = await make_api_request("PUT", f"/users/{user_id.strip()}", data=update_data)

    if "error" in result:
        return {"error": f"Failed to update user: {result['error']}"}

    return {
        "success": True,
        "message": f"User {user_id} updated successfully",
        "user": result
    }

@mcp.tool()
async def delete_user(user_id: str) -> dict:
    """Delete a user account.

    Args:
        user_id: ID of the user to delete
    """

    if not user_id or not user_id.strip():
        return {"error": "User ID is required"}

    result = await make_api_request("DELETE", f"/users/{user_id.strip()}")

    if "error" in result:
        return {"error": f"Failed to delete user: {result['error']}"}

    return {
        "success": True,
        "message": f"User {user_id} deleted successfully"
    }

# RESOURCES - for read-only data access optimized for AI consumption
@mcp.resource("user://{user_id}")
async def get_user_resource(user_id: str) -> str:
    """Get detailed user information as a resource.

    This provides read-only access to user data in a format
    that's easy for AI applications to understand.
    """

    result = await make_api_request("GET", f"/users/{user_id}")

    if "error" in result:
        return f"Error retrieving user {user_id}: {result['error']}"

    user = result

    # Format the data for AI consumption
    formatted_output = f"""
User Profile for ID {user_id}:
- Name: {user.get('name', 'Not specified')}
- Email: {user.get('email', 'Not specified')}
- Role: {user.get('role', 'Not specified')}
- Status: {user.get('status', 'Active')}
- Created: {user.get('created_at', 'Unknown')}
- Last Updated: {user.get('updated_at', 'Unknown')}
"""

    return formatted_output.strip()

@mcp.resource("users://list")
async def list_users_resource() -> str:
    """Get a list of all users in the system."""

    result = await make_api_request("GET", "/users")

    if "error" in result:
        return f"Error retrieving users: {result['error']}"

    users = result.get("users", []) if isinstance(result, dict) else result

    if not users:
        return "No users found in the system."

    # Format the user list for AI consumption
    formatted_output = f"User Directory ({len(users)} users):\n\n"

    for user in users:
        formatted_output += f"• {user.get('name', 'Unknown')} ({user.get('email', 'No email')})\n"
        formatted_output += f"  ID: {user.get('id', 'Unknown')}, Role: {user.get('role', 'Unknown')}\n\n"

    return formatted_output.strip()

if __name__ == "__main__":
    logger.info("Starting production-ready MCP server with advanced features...")
    mcp.run(transport="stdio")

Salve isto como server_production.py e execute com:

python server_production.py

Armadilhas comuns e como evitá-las

Problemas de incompatibilidade de schema

Problema: Os schemas das suas tools MCP não correspondem ao que sua API REST espera.

Solução: Sempre teste seus schemas contra chamadas reais à API. Use ferramentas como o Postman para verificar o comportamento da sua API primeiro e, em seguida, garanta que seus schemas MCP correspondam exatamente.

Complexidade da autenticação

Problema: Os fluxos de OAuth e a atualização de tokens podem ficar complexos em servidores MCP de longa duração.

Solução: Construa uma autenticação modular que trate a atualização de tokens automaticamente. Teste sua lógica de autenticação separadamente das suas tools MCP.

Suposições sobre desempenho

Problema: As aplicações de IA podem gerar padrões de tráfego muito diferentes dos clientes tradicionais.

Solução: Implemente limitação de taxa adequada, pool de conexões e monitoramento. Teste com padrões de uso de IA realistas.

Qualidade das mensagens de erro

Problema: As mensagens de erro técnicas da sua API REST não são úteis para as aplicações de IA nem para os usuários finais.

Solução: Transforme os erros técnicos em explicações amigáveis que as aplicações de IA possam usar para ajudar os usuários a entender e corrigir problemas.

Conclusões

Você acabou de construir um servidor MCP completo que faz a ponte entre sua API REST e as aplicações de IA. Você começou com uma tool simples, adicionou tratamento de erros, expandiu a funcionalidade e a conectou ao Claude Desktop. Suas APIs agora estão prontas para IA!

A abordagem passo a passo que usamos aqui - começar de forma simples e ir acrescentando funcionalidades - é a chave para o desenvolvimento bem-sucedido de servidores MCP. Você pode aplicar esses mesmos padrões a qualquer API REST, seja ela uma interface CRUD simples ou um sistema empresarial complexo.

Lembre-se dos princípios fundamentais:

• Comece de forma simples com uma tool e evolua gradualmente
• Trate os erros de forma elegante e forneça um feedback significativo
• Teste minuciosamente usando o MCP Inspector antes de conectar aos clientes de IA
• Pense sob a perspectiva da IA ao projetar as interfaces das tools

Sua API REST, aprimorada com recursos MCP, não é mais apenas uma fonte de dados - ela é uma participante ativa em fluxos de trabalho movidos por IA. A ponte que você construiu entre APIs tradicionais e aplicações de IA abre possibilidades que estamos apenas começando a explorar.

Referências

[1] Anthropic. "Introducing the Model Context Protocol." https://www.anthropic.com/news/model-context-protocol

[2] Model Context Protocol Official Documentation. https://modelcontextprotocol.io/

[3] Model Context Protocol Python SDK. https://modelcontextprotocol.io/quickstart/server

[4] Claude Desktop Download. https://claude.ai/download