API REST vers serveur MCP en Python : pas à pas

Introduction aux API MCP

Imaginez la situation : vous avez passé des mois à construire des API REST solides. Vos endpoints fonctionnent parfaitement, votre documentation est complète et tout tourne sans accroc. Puis, soudain, tout le monde se met à parler d'agents IA et de LLM, et vous vous demandez : « Comment faire fonctionner mes API avec ces applications d'IA sans tout recommencer ? »

Si vous hochez la tête, vous n'êtes pas seul. C'est exactement le défi auquel des milliers de développeurs sont confrontés en ce moment.

Voici la bonne nouvelle : vous n'avez pas besoin de jeter votre travail. Le Model Context Protocol (MCP) est comme un traducteur universel qui permet à vos API REST existantes de parler couramment le langage de l'« IA ». Voyez-le comme un adaptateur intelligent ajouté à vos API : elles continuent de faire ce qu'elles font le mieux, mais désormais les applications d'IA peuvent les comprendre et les utiliser sans effort.

Dans ce guide, nous allons construire un serveur MCP étape par étape, en commençant simplement et en ajoutant des fonctionnalités au fur et à mesure. Vous verrez exactement comment chaque élément s'assemble, et à la fin, vous aurez un serveur MCP fonctionnel qui encapsule votre API REST et se connecte de façon transparente aux clients d'IA comme Claude Desktop.

API REST vs serveurs MCP

Commençons par ce que vous connaissez déjà. Une API REST est un ensemble d'endpoints HTTP qui acceptent et renvoient des données structurées. Chaque endpoint possède un chemin, une méthode (GET, POST, etc.) et un schéma de requête/réponse. Vos clients envoient des requêtes HTTP et reçoivent des réponses JSON en retour.

Un serveur MCP est différent. Au lieu d'endpoints HTTP, il expose des « outils » (tools) et des « ressources » (resources) que les applications d'IA peuvent utiliser. Les outils servent aux actions qui modifient des données (comme créer un utilisateur), tandis que les ressources fournissent un accès en lecture seule aux informations (comme obtenir les détails d'un utilisateur).

L'idée clé est la suivante : votre API REST devient le moteur, et le serveur MCP devient le traducteur. Le serveur MCP reçoit les requêtes des applications d'IA, les traduit en appels d'API REST, puis met en forme les réponses de manière à ce que les applications d'IA puissent les comprendre.

Voici comment les composants se correspondent :

Composant d'API REST	Équivalent serveur MCP	Objectif
Endpoints POST/PUT/DELETE	Tools	Actions qui modifient des données
Endpoints GET	Resources	Accès aux données en lecture seule
Paramètres de requête	Paramètres d'outil	Champs d'entrée pour les outils
En-têtes d'authentification	Variables d'environnement	Gestion sécurisée des identifiants

Prérequis

Avant de commencer à construire, assurez-vous d'avoir :

1. Une API REST fonctionnelle avec des endpoints documentés (nous utiliserons une simple API de gestion d'utilisateurs comme exemple)
2. Python 3.8+ installé sur votre système
3. Des connaissances de base en JSON et sur les concepts d'API
4. Une clé d'API ou une méthode d'authentification pour votre API REST

Pour nos exemples, nous supposerons que vous disposez d'une API REST avec ces endpoints :

• GET /users/{id} - Obtenir les détails d'un utilisateur
• POST /users - Créer un nouvel utilisateur
• PUT /users/{id} - Mettre à jour les informations d'un utilisateur
• DELETE /users/{id} - Supprimer un utilisateur

Étape 1 : Configuration de votre environnement de développement

Commençons par créer un nouveau projet et installer les dépendances nécessaires.

# 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

Créez un fichier .env pour stocker les identifiants de votre API :

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

Étape 2 : Création de votre premier outil MCP

Commençons par le serveur MCP le plus simple possible. Nous allons créer un unique outil qui récupère les informations d'un utilisateur depuis votre API REST.

Créez un fichier nommé 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")

Et voilà ! Vous venez de créer votre premier serveur MCP. Testons-le :

python server.py

Le serveur va démarrer et attendre une entrée. Vous pouvez le tester à l'aide du MCP Inspector (nous aborderons cela dans la section consacrée aux tests).

Étape 3 : Ajout de la gestion des erreurs et de la validation

Notre première version fonctionne, mais elle n'est pas très robuste. Ajoutons une gestion des erreurs et une validation des entrées appropriées :

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

Désormais, notre serveur gère les erreurs avec élégance et valide correctement les entrées.

Étape 4 : Ajout d'outils supplémentaires

Étendons notre serveur pour gérer la création et la mise à jour d'utilisateurs. Nous allons ajouter ces outils un par un :

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

Étape 5 : Ajout de ressources pour les données en lecture seule

Les outils sont parfaits pour les actions, mais parfois les applications d'IA ont simplement besoin de lire des données. C'est là qu'interviennent les ressources. Ajoutons une ressource pour récupérer les informations d'un utilisateur :

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

Code complet du serveur MCP

Maintenant que nous avons construit notre serveur MCP étape par étape, voici le code complet, prêt pour la production, qui combine toutes les pièces que nous avons abordées. Vous pouvez copier ce fichier entier et l'exécuter immédiatement :

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

Enregistrez ce fichier sous le nom server.py, assurez-vous que votre fichier .env est configuré avec les identifiants de votre API, puis exécutez :

python server.py

Étape 6 : Test de votre serveur MCP

Avant de nous connecter aux clients d'IA, testons notre serveur à l'aide du MCP Inspector. D'abord, installez-le :

npm install -g @modelcontextprotocol/inspector

Puis testez votre serveur :

mcp-inspector python server.py

Cela ouvre une interface web où vous pouvez :

• Voir tous vos outils et ressources
• Tester chaque outil avec différentes entrées
• Consulter les réponses
• Déboguer les éventuels problèmes

Essayez d'appeler l'outil get_user avec un identifiant d'utilisateur, ou l'outil create_user avec des paramètres de nom et d'e-mail. Vous devriez voir les véritables appels d'API REST effectués ainsi que les réponses mises en forme pour être consommées par l'IA.

Étape 7 : Connexion à Claude Desktop

Passons maintenant à la partie passionnante : connecter votre serveur MCP à Claude Desktop pour pouvoir réellement l'utiliser avec une application d'IA.

Installation de Claude Desktop

Tout d'abord, téléchargez et installez Claude Desktop si ce n'est pas déjà fait.

Configuration de votre serveur MCP

Claude Desktop recherche les configurations de serveurs MCP dans un fichier spécifique. Son emplacement dépend de votre système d'exploitation :

Sur macOS :

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

Sur Windows :

%APPDATA%\Claude\claude_desktop_config.json

Sur Linux :

~/.config/Claude/claude_desktop_config.json

Créez ce fichier s'il n'existe pas, et ajoutez la configuration de votre serveur 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"
      }
    }
  }
}

Remarques importantes :

• Remplacez /path/to/your/server.py par le chemin absolu réel de votre fichier serveur
• Remplacez les variables d'environnement par l'URL et la clé réelles de votre API
• Assurez-vous que Python figure dans le PATH de votre système, ou utilisez le chemin complet vers votre exécutable Python

Test de la connexion

1. Redémarrez Claude Desktop après avoir enregistré le fichier de configuration
2. Ouvrez une nouvelle conversation dans Claude Desktop
3. Repérez l'indicateur MCP - vous devriez voir une petite icône ou un indicateur d'outil montrant que votre serveur MCP est connecté
4. Testez vos outils en demandant à Claude d'interagir avec votre API :

Essayez ces exemples de requêtes :

• « Peux-tu obtenir des informations sur l'utilisateur avec l'ID 123 ? »
• « Crée un nouvel utilisateur nommé John Doe avec l'e-mail john@example.com »
• « Mets à jour l'utilisateur 456 pour changer son rôle en admin »

Claude devrait désormais être capable d'appeler vos outils MCP et de travailler avec les données de votre API REST !

Dépannage des problèmes de connexion

Si Claude Desktop ne se connecte pas à votre serveur :

1. Vérifiez les journaux - Claude Desktop affiche généralement les erreurs de connexion dans sa console de développement
2. Vérifiez les chemins de fichiers - Assurez-vous que tous les chemins de la configuration sont absolus et corrects
3. Testez votre serveur de façon indépendante - Exécutez python server.py directement pour vous assurer qu'il démarre sans erreur
4. Vérifiez les variables d'environnement - Assurez-vous que les identifiants de votre API sont correctement définis
5. Redémarrez Claude Desktop - Les changements de configuration nécessitent un redémarrage

Étape 8 : Fonctionnalités avancées et bonnes pratiques

Ajout de la gestion de l'authentification

Pour une utilisation en production, vous aurez besoin d'une gestion de l'authentification plus sophistiquée :

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...

Ajout de la journalisation et de la surveillance

Ajoutez une journalisation complète pour faciliter le débogage et la surveillance :

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

Gestion des limites de débit

Ajoutez une gestion intelligente des limites de débit et une logique de nouvelle tentative :

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

Code complet prêt pour la production : toutes les fonctionnalités avancées incluses

Voici le serveur MCP complet et prêt pour la production qui intègre toutes les fonctionnalités avancées de l'Étape 8. Cette version comprend la gestion de l'authentification, une journalisation complète et une limitation de débit intelligente :

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

Enregistrez ce fichier sous le nom server_production.py et exécutez-le avec :

python server_production.py

Pièges courants et comment les éviter

Problèmes d'incompatibilité de schéma

Problème : les schémas de vos outils MCP ne correspondent pas à ce qu'attend votre API REST.

Solution : testez toujours vos schémas par rapport à de véritables appels d'API. Utilisez des outils comme Postman pour vérifier d'abord le comportement de votre API, puis assurez-vous que vos schémas MCP correspondent exactement.

Complexité de l'authentification

Problème : les flux OAuth et le rafraîchissement des jetons peuvent devenir complexes dans les serveurs MCP à longue durée d'exécution.

Solution : construisez une authentification modulaire qui gère automatiquement le rafraîchissement des jetons. Testez votre logique d'authentification séparément de vos outils MCP.

Hypothèses sur les performances

Problème : les applications d'IA peuvent générer des schémas de trafic très différents de ceux des clients traditionnels.

Solution : mettez en place une limitation de débit, un pooling de connexions et une surveillance appropriés. Testez avec des schémas d'utilisation d'IA réalistes.

Qualité des messages d'erreur

Problème : les messages d'erreur techniques de votre API REST ne sont pas utiles pour les applications d'IA ni pour les utilisateurs finaux.

Solution : transformez les erreurs techniques en explications compréhensibles que les applications d'IA peuvent utiliser pour aider les utilisateurs à comprendre et à résoudre les problèmes.

Conclusions

Vous venez de construire un serveur MCP complet qui fait le pont entre votre API REST et les applications d'IA. Vous avez commencé avec un simple outil, ajouté la gestion des erreurs, étendu les fonctionnalités et connecté le tout à Claude Desktop. Vos API sont désormais prêtes pour l'IA !

L'approche pas à pas que nous avons utilisée ici - commencer simplement et développer les fonctionnalités progressivement - est la clé d'un développement de serveur MCP réussi. Vous pouvez appliquer ces mêmes schémas à n'importe quelle API REST, qu'il s'agisse d'une simple interface CRUD ou d'un système d'entreprise complexe.

Rappelez-vous les principes clés :

• Commencez simplement avec un seul outil et développez progressivement
• Gérez les erreurs avec élégance et fournissez un retour d'information pertinent
• Testez minutieusement à l'aide du MCP Inspector avant de vous connecter aux clients d'IA
• Pensez du point de vue de l'IA lors de la conception des interfaces d'outils

Votre API REST, enrichie de capacités MCP, n'est plus seulement une source de données : c'est désormais un participant actif aux workflows pilotés par l'IA. Le pont que vous avez construit entre les API traditionnelles et les applications d'IA ouvre des possibilités que nous commençons à peine à explorer.

Références

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

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

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

[4] Téléchargement de Claude Desktop. https://claude.ai/download