De API REST a servidor MCP en Python: paso a paso

Introducción a las APIs de MCP

Imagina esto: has pasado meses construyendo APIs REST sólidas. Tus endpoints funcionan perfectamente, tu documentación es exhaustiva y todo funciona sin problemas. Entonces, de repente, todo el mundo habla de agentes de IA y LLMs, y te preguntas: "¿Cómo hago que mis APIs funcionen con estas aplicaciones de IA sin empezar desde cero?"

Si asientes con la cabeza, no estás solo. Este es exactamente el reto al que se enfrentan miles de desarrolladores en este momento.

Aquí está la buena noticia: no necesitas tirar tu trabajo a la basura. El Model Context Protocol (MCP) es como un traductor universal que hace que tus APIs REST existentes hablen "IA" con fluidez. Piénsalo como añadir un adaptador inteligente a tus APIs: siguen haciendo lo que mejor saben hacer, pero ahora las aplicaciones de IA pueden entenderlas y usarlas sin esfuerzo.

En esta guía, recorreremos paso a paso la construcción de un servidor MCP, empezando de forma sencilla y añadiendo funciones a medida que avanzamos. Verás exactamente cómo encaja cada pieza y, al final, tendrás un servidor MCP funcional que envuelve tu API REST y se conecta sin problemas con clientes de IA como Claude Desktop.

APIs REST frente a servidores MCP

Empecemos con lo que ya conoces. Una API REST es un conjunto de endpoints HTTP que aceptan y devuelven datos estructurados. Cada endpoint tiene una ruta, un método (GET, POST, etc.) y un esquema de petición/respuesta. Tus clientes hacen peticiones HTTP y reciben respuestas JSON de vuelta.

Un servidor MCP es diferente. En lugar de endpoints HTTP, expone "tools" (herramientas) y "resources" (recursos) que las aplicaciones de IA pueden usar. Las herramientas son para acciones que modifican datos (como crear un usuario), mientras que los recursos ofrecen acceso de solo lectura a la información (como obtener los detalles de un usuario).

La idea clave es esta: tu API REST se convierte en el motor y el servidor MCP se convierte en el traductor. El servidor MCP recibe peticiones de las aplicaciones de IA, las traduce en llamadas a la API REST y luego formatea las respuestas de una manera que las aplicaciones de IA puedan entender.

Así es como se corresponden los componentes:

Componente de la API REST	Equivalente en el servidor MCP	Propósito
Endpoints POST/PUT/DELETE	Tools	Acciones que modifican datos
Endpoints GET	Resources	Acceso a datos de solo lectura
Parámetros de consulta	Parámetros de tool	Campos de entrada para las herramientas
Cabeceras de autenticación	Variables de entorno	Manejo seguro de credenciales

Requisitos previos

Antes de empezar a construir, asegúrate de tener:

1. Una API REST funcional con endpoints documentados (usaremos una sencilla API de gestión de usuarios como ejemplo)
2. Python 3.8+ instalado en tu sistema
3. Conocimientos básicos de JSON y de conceptos de API
4. Una clave de API o método de autenticación para tu API REST

Para nuestros ejemplos, asumiremos que tienes una API REST con estos endpoints:

• GET /users/{id} - Obtener los detalles de un usuario
• POST /users - Crear un nuevo usuario
• PUT /users/{id} - Actualizar la información de un usuario
• DELETE /users/{id} - Eliminar un usuario

Paso 1: Configurar tu entorno de desarrollo

Empecemos creando un nuevo proyecto e instalando las dependencias necesarias.

# 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

Crea un archivo .env para almacenar las credenciales de tu API:

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

Paso 2: Crear tu primera tool de MCP

Empecemos con el servidor MCP más sencillo posible. Crearemos una única herramienta que obtenga la información de un usuario desde tu API REST.

Crea un archivo llamado 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")

¡Eso es todo! Acabas de crear tu primer servidor MCP. Vamos a probarlo:

python server.py

El servidor se iniciará y esperará entrada. Puedes probarlo usando el MCP Inspector (lo veremos en la sección de pruebas).

Paso 3: Añadir manejo de errores y validación

Nuestra primera versión funciona, pero no es muy robusta. Añadamos un manejo de errores y una validación de entradas adecuados:

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)}"}

Ahora nuestro servidor maneja los errores con elegancia y valida las entradas correctamente.

Paso 4: Añadir más tools

Ampliemos nuestro servidor para gestionar la creación y actualización de usuarios. Añadiremos estas herramientas una por una:

@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
    }

Paso 5: Añadir resources para datos de solo lectura

Las herramientas son estupendas para las acciones, pero a veces las aplicaciones de IA solo necesitan leer datos. Ahí es donde entran los recursos. Añadamos un recurso para obtener la información de un usuario:

@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 del servidor MCP

Ahora que hemos construido nuestro servidor MCP paso a paso, aquí está el código completo, listo para producción, que combina todas las piezas que hemos comentado. Puedes copiar este archivo completo y ejecutarlo de inmediato:

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

Guárdalo como server.py, asegúrate de que tu archivo .env está configurado con las credenciales de tu API y ejecuta:

python server.py

Paso 6: Probar tu servidor MCP

Antes de conectarlo a clientes de IA, probemos nuestro servidor usando el MCP Inspector. Primero, instálalo:

npm install -g @modelcontextprotocol/inspector

Luego prueba tu servidor:

mcp-inspector python server.py

Esto abre una interfaz web donde puedes:

• Ver todas tus herramientas y recursos
• Probar cada herramienta con diferentes entradas
• Visualizar las respuestas
• Depurar cualquier problema

Prueba a llamar a la herramienta get_user con un ID de usuario, o a la herramienta create_user con los parámetros de nombre y correo electrónico. Deberías ver cómo se realizan las llamadas reales a la API REST y las respuestas formateadas para el consumo por parte de la IA.

Paso 7: Conectar con Claude Desktop

Ahora viene la parte emocionante: conectar tu servidor MCP a Claude Desktop para que puedas usarlo realmente con una aplicación de IA.

Instalar Claude Desktop

Primero, descarga e instala Claude Desktop si aún no lo has hecho.

Configurar tu servidor MCP

Claude Desktop busca las configuraciones de los servidores MCP en un archivo específico. La ubicación depende de tu sistema operativo:

En macOS:

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

En Windows:

%APPDATA%\Claude\claude_desktop_config.json

En Linux:

~/.config/Claude/claude_desktop_config.json

Crea este archivo si no existe y añade la configuración de tu 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"
      }
    }
  }
}

Notas importantes:

• Reemplaza /path/to/your/server.py con la ruta absoluta real a tu archivo de servidor
• Reemplaza las variables de entorno con la URL y la clave reales de tu API
• Asegúrate de que Python está en el PATH de tu sistema, o usa la ruta completa a tu ejecutable de Python

Probar la conexión

1. Reinicia Claude Desktop después de guardar el archivo de configuración
2. Abre una nueva conversación en Claude Desktop
3. Busca el indicador de MCP: deberías ver un pequeño icono o indicador de herramienta que muestra que tu servidor MCP está conectado
4. Prueba tus herramientas pidiéndole a Claude que interactúe con tu API:

Prueba estos ejemplos de prompts:

• "¿Puedes obtener información sobre el usuario con ID 123?"
• "Crea un nuevo usuario llamado John Doe con el correo electrónico john@example.com"
• "Actualiza el usuario 456 para cambiar su rol a admin"

¡Claude ahora debería ser capaz de llamar a tus herramientas MCP y trabajar con los datos de tu API REST!

Solución de problemas de conexión

Si Claude Desktop no se conecta a tu servidor:

1. Revisa los logs: Claude Desktop suele mostrar los errores de conexión en su consola de desarrollador
2. Verifica las rutas de los archivos: asegúrate de que todas las rutas en la configuración son absolutas y correctas
3. Prueba tu servidor de forma independiente: ejecuta python server.py directamente para asegurarte de que arranca sin errores
4. Comprueba las variables de entorno: asegúrate de que las credenciales de tu API están configuradas correctamente
5. Reinicia Claude Desktop: los cambios de configuración requieren un reinicio

Paso 8: Funciones avanzadas y buenas prácticas

Añadir manejo de autenticación

Para uso en producción, querrás un manejo de autenticación más 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...

Añadir registro y monitorización

Añade un registro exhaustivo para facilitar la depuración y la monitorización:

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)}"}

Gestionar los límites de tasa

Añade una lógica inteligente de límite de tasa y de reintentos:

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 listo para producción: todas las funciones avanzadas incluidas

Aquí está el servidor MCP completo, listo para producción, que integra todas las funciones avanzadas del Paso 8. Esta versión incluye manejo de autenticación, registro exhaustivo y límite de tasa 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")

Guárdalo como server_production.py y ejecútalo con:

python server_production.py

Errores comunes y cómo evitarlos

Problemas de discrepancia de esquemas

Problema: Los esquemas de tus tools de MCP no coinciden con lo que espera tu API REST.

Solución: Prueba siempre tus esquemas contra llamadas reales a la API. Usa herramientas como Postman para verificar primero el comportamiento de tu API y luego asegúrate de que tus esquemas de MCP coinciden exactamente.

Complejidad de la autenticación

Problema: Los flujos de OAuth y la renovación de tokens pueden volverse complejos en servidores MCP de larga duración.

Solución: Construye una autenticación modular que gestione la renovación de tokens automáticamente. Prueba tu lógica de autenticación por separado de tus tools de MCP.

Suposiciones sobre el rendimiento

Problema: Las aplicaciones de IA pueden generar patrones de tráfico muy diferentes a los de los clientes tradicionales.

Solución: Implementa un límite de tasa adecuado, agrupación de conexiones (connection pooling) y monitorización. Realiza pruebas con patrones de uso de IA realistas.

Calidad de los mensajes de error

Problema: Los mensajes de error técnicos de tu API REST no resultan útiles para las aplicaciones de IA ni para los usuarios finales.

Solución: Transforma los errores técnicos en explicaciones comprensibles para las personas que las aplicaciones de IA puedan usar para ayudar a los usuarios a entender y resolver los problemas.

Conclusiones

Acabas de construir un servidor MCP completo que conecta tu API REST con las aplicaciones de IA. Empezaste con una herramienta sencilla, añadiste manejo de errores, ampliaste la funcionalidad y lo conectaste a Claude Desktop. ¡Tus APIs están ahora listas para la IA!

El enfoque paso a paso que hemos usado aquí, empezando de forma sencilla y ampliando la funcionalidad, es la clave para el desarrollo exitoso de servidores MCP. Puedes aplicar estos mismos patrones a cualquier API REST, ya sea una sencilla interfaz CRUD o un complejo sistema empresarial.

Recuerda los principios clave:

• Empieza sencillo con una sola herramienta y ve ampliando gradualmente
• Maneja los errores con elegancia y proporciona retroalimentación significativa
• Prueba a fondo usando el MCP Inspector antes de conectarte a los clientes de IA
• Piensa desde la perspectiva de la IA al diseñar las interfaces de las herramientas

Tu API REST, mejorada con las capacidades de MCP, ya no es solo una fuente de datos: es un participante activo en flujos de trabajo impulsados por IA. El puente que has construido entre las APIs tradicionales y las aplicaciones de IA abre posibilidades que apenas estamos empezando a explorar.

Referencias

[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