Comment construire un serveur MCP : Guide étape par étape avec exemples de code

Le Model Context Protocol (MCP) devient la nouvelle norme pour connecter les modèles d’IA à des outils et services réels. La construction d’un serveur MCP vous permet d’exposer des données, des actions et des ressources à un LLM comme Claude via une interface simple et standardisée.

Dans ce guide, vous apprendrez étape par étape comment configurer un serveur MCP de base en Python, définir des ressources et des outils, et le connecter à un client MCP.

Qu’est-ce qu’un serveur MCP

Un serveur MCP agit comme un pont entre un LLM et un système externe comme une base de données, un stockage de fichiers ou une API. Il définit des ressources (données lisibles), des outils (actions) et des instructions de manière à ce que le LLM puisse les utiliser en toute sécurité pendant ses tâches.

Au lieu d’écrire une intégration personnalisée pour chaque modèle ou outil, MCP offre une norme universelle qui fonctionne avec la version 0.1 du protocole (actuelle en avril 2025).

Ce dont vous avez besoin avant de commencer

• Python 3.8 ou ultérieur
• Expérience de base en programmation Python
• SDK MCP pour Python (disponible via pip)
• Un client compatible MCP comme Claude Desktop ou Cursor (optionnel pour les tests)
• Git pour le contrôle de version (recommandé)
• Un éditeur de texte ou IDE (Visual Studio Code recommandé)

Comprendre la structure fondamentale

Dans MCP :

• Serveur : Fournit des ressources et des outils au LLM.
• Client : Connecte le LLM à votre serveur.
• Protocole : Gère la communication entre le client et le serveur.

Vous définirez deux primitives importantes :

• Ressource : Information statique ou dynamique que le LLM peut lire.
• Outil : Une fonction appelable que le LLM peut exécuter.

Le flux de communication fonctionne comme suit :

1. Le LLM (via un client) demande des données ou des actions à votre serveur
2. Votre serveur traite ces demandes et renvoie des réponses standardisées
3. Le LLM peut alors utiliser ces informations dans son raisonnement et ses réponses

1. Configurez votre projet Python

Commencez par créer un répertoire de projet et un environnement virtuel Python.

mkdir my_mcp_server
cd my_mcp_server
python -m venv venv
source venv/bin/activate  # Linux/Mac
venvScriptsactivate     # Windows

Créez une structure de projet de base :

mkdir -p src/resources src/tools tests
touch src/__init__.py src/resources/__init__.py src/tools/__init__.py
touch requirements.txt README.md

Ajoutez ce qui suit à votre fichier requirements.txt :

mcp-server>=0.1.0
pydantic>=2.0.0
pytest>=7.0.0

2. Installez le SDK MCP

Installez le SDK du serveur MCP pour Python et d’autres dépendances :

pip install -r requirements.txt

Si le SDK officiel n’est pas encore publié, vous devrez peut-être l’installer depuis un dépôt GitHub :

pip install git+https://github.com/anthropic/mcp-server-python.git

3. Créez un serveur MCP de base

Créez un fichier nommé src/server.py :

from typing import Dict, Any
from mcp_server import MCPServer
import logging

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

def main() -> None:
    """Initialize and start the MCP server."""
    try:
        server = MCPServer(
            name="MyMCPServer",
            version="0.1.0",
            description="A simple MCP server example"
        )
        
        # Resources and tools will be added here
        
        logger.info("Starting MCP server...")
        server.start()
    except Exception as e:
        logger.error(f"Failed to start MCP server: {e}")
        raise

if __name__ == "__main__":
    main()

Cela configure un serveur MCP de base avec une journalisation appropriée.

4. Définissez une ressource

Les ressources exposent des données que le modèle peut lire. Créons un fichier src/resources/user_profiles.py :

from typing import List, Dict, Any
from pydantic import BaseModel
import logging

logger = logging.getLogger("mcp_server.resources")

class UserProfile(BaseModel):
    """Data model for user profiles."""
    name: str
    role: str
    department: str = "General"
    years_experience: int = 0

def fetch_user_profiles() -> List[Dict[str, Any]]:
    """
    Fetch user profiles from the database.
    
    Returns:
        List[Dict[str, Any]]: A list of user profile dictionaries.
    """
    try:
        # In a real implementation, this would query a database
        # For this example, we'll return mock data
        users = [
            UserProfile(name="Alice", role="Engineer", department="Engineering", years_experience=5),
            UserProfile(name="Bob", role="Product Manager", department="Product", years_experience=3),
            UserProfile(name="Charlie", role="Designer", department="Design", years_experience=7)
        ]
        
        logger.info(f"Successfully fetched {len(users)} user profiles")
        return [user.model_dump() for user in users]
    except Exception as e:
        logger.error(f"Error fetching user profiles: {e}")
        # In production, you might want to return an empty list or raise
        # a specific exception depending on your error handling strategy
        return []

Maintenant, mettez à jour src/server.py pour inclure cette ressource :

from typing import Dict, Any
from mcp_server import MCPServer, Resource
import logging
from src.resources.user_profiles import fetch_user_profiles

# ... existing code ...

def main() -> None:
    """Initialize and start the MCP server."""
    try:
        server = MCPServer(
            name="MyMCPServer",
            version="0.1.0",
            description="A simple MCP server example"
        )
        
        # Add the user profiles resource
        user_profiles = Resource(
            name="user_profiles",
            description="List of user profiles from the company database.",
            fetch_fn=fetch_user_profiles
        )
        
        server.add_resource(user_profiles)
        
        logger.info("Starting MCP server...")
        server.start()
    except Exception as e:
        logger.error(f"Failed to start MCP server: {e}")
        raise

if __name__ == "__main__":
    main()

Le LLM peut maintenant interroger user_profiles via le client MCP.

5. Définissez un outil

Les outils permettent au LLM d’exécuter une action. Créez un fichier src/tools/user_management.py :

from typing import Dict, Any, Optional
from pydantic import BaseModel, Field, ValidationError
import logging

logger = logging.getLogger("mcp_server.tools")

class CreateUserRequest(BaseModel):
    """Validation model for user creation requests."""
    name: str = Field(..., min_length=2, description="User's full name")
    role: str = Field(..., min_length=2, description="User's job role")
    department: Optional[str] = Field("General", description="User's department")
    years_experience: Optional[int] = Field(0, ge=0, description="Years of professional experience")

def create_user_profile(request_data: Dict[str, Any]) -> Dict[str, Any]:
    """
    Create a new user profile in the database.
    
    Args:
        request_data (Dict[str, Any]): User data containing name, role, etc.
        
    Returns:
        Dict[str, Any]: Response with status and user info
    """
    try:
        # Validate the input data
        user_data = CreateUserRequest(**request_data)
        
        # In a real implementation, this would insert into a database
        # For this example, we'll just log the action
        logger.info(f"Creating new user: {user_data.name} - {user_data.role} in {user_data.department}")
        
        # Return success response with created user data
        return {
            "status": "success",
            "message": f"User {user_data.name} created successfully",
            "user": user_data.model_dump()
        }
    except ValidationError as e:
        # Handle validation errors
        logger.error(f"Validation error: {e}")
        return {
            "status": "error",
            "message": "Invalid user data provided",
            "details": str(e)
        }
    except Exception as e:
        # Handle other errors
        logger.error(f"Error creating user: {e}")
        return {
            "status": "error",
            "message": "Failed to create user",
            "details": str(e)
        }

Maintenant, mettez à jour src/server.py pour inclure cet outil :

from typing import Dict, Any
from mcp_server import MCPServer, Resource, Tool
import logging
from src.resources.user_profiles import fetch_user_profiles
from src.tools.user_management import create_user_profile

# ... existing code ...

def main() -> None:
    """Initialize and start the MCP server."""
    try:
        server = MCPServer(
            name="MyMCPServer",
            version="0.1.0",
            description="A simple MCP server example"
        )
        
        # Add the user profiles resource
        user_profiles = Resource(
            name="user_profiles",
            description="List of user profiles from the company database.",
            fetch_fn=fetch_user_profiles
        )
        
        # Add the create user tool
        create_user = Tool(
            name="create_user_profile",
            description="Create a new user profile in the database.",
            parameters={
                "name": {"type": "string", "description": "User's full name"},
                "role": {"type": "string", "description": "User's job role"},
                "department": {"type": "string", "description": "User's department (optional)"},
                "years_experience": {"type": "integer", "description": "Years of experience (optional)"}
            },
            execute_fn=create_user_profile
        )
        
        server.add_resource(user_profiles)
        server.add_tool(create_user)
        
        logger.info("Starting MCP server...")
        server.start()
    except Exception as e:
        logger.error(f"Failed to start MCP server: {e}")
        raise

6. Gestion des erreurs et validation

Créez un fichier src/utils/validation.py pour centraliser votre logique de validation :

from typing import Dict, Any, List, Optional, Type
from pydantic import BaseModel, ValidationError
import logging

logger = logging.getLogger("mcp_server.validation")

def validate_request(
    data: Dict[str, Any],
    model_class: Type[BaseModel]
) -> tuple[Optional[BaseModel], Optional[Dict[str, Any]]]:
    """
    Validate request data against a Pydantic model.
    
    Args:
        data: The input data to validate
        model_class: The Pydantic model class to use for validation
        
    Returns:
        tuple: (validated_model, error_dict)
            - If valid: (model instance, None)
            - If invalid: (None, error dictionary)
    """
    try:
        validated_data = model_class(**data)
        return validated_data, None
    except ValidationError as e:
        errors = e.errors()
        error_dict = {
            "status": "error",
            "message": "Validation failed",
            "errors": errors
        }
        logger.error(f"Validation error: {errors}")
        return None, error_dict

Cette fonction utilitaire peut être utilisée dans tous vos outils pour valider les données d’entrée de manière cohérente.

7. Exécutez et testez le serveur MCP

Créez un script de test simple test_server.py pour vérifier que votre serveur fonctionne :

import requests
import json
import time
import subprocess
import sys
from pathlib import Path

def test_server():
    """Simple test to verify the MCP server is running correctly."""
    # Start the server in a separate process
    server_process = subprocess.Popen([sys.executable, "src/server.py"])
    
    try:
        # Wait for server to start
        time.sleep(2)
        
        # Test the server using the MCP client
        # In a real test, you would use the MCP client SDK
        # For this example, we'll simulate a client using HTTP requests
        
        # Assuming the server is running on localhost:8000
        base_url = "http://localhost:8000"
        
        # Test fetching resources
        response = requests.get(f"{base_url}/resources/user_profiles")
        assert response.status_code == 200
        data = response.json()
        print("Resource response:", json.dumps(data, indent=2))
        
        # Test executing a tool
        tool_data = {
            "name": "Test User",
            "role": "Tester",
            "department": "QA"
        }
        response = requests.post(
            f"{base_url}/tools/create_user_profile",
            json=tool_data
        )
        assert response.status_code == 200
        data = response.json()
        print("Tool response:", json.dumps(data, indent=2))
        
        print("All tests passed!")
        
    finally:
        # Clean up: terminate the server process
        server_process.terminate()
        server_process.wait()

if __name__ == "__main__":
    test_server()

Exécutez votre serveur :

python src/server.py

Dans un terminal séparé, vous pourriez voir une sortie comme celle-ci lorsque le serveur est en cours d’exécution :

2025-04-28 10:15:23 - mcp_server - INFO - Starting MCP server...
2025-04-28 10:15:23 - mcp_server - INFO - Server listening on 0.0.0.0:8000
2025-04-28 10:15:30 - mcp_server.resources - INFO - Successfully fetched 3 user profiles
2025-04-28 10:15:35 - mcp_server.tools - INFO - Creating new user: Test User - Tester in QA

Ensuite, configurez votre client MCP (comme Claude Desktop) pour se connecter à votre serveur MCP local en fournissant l’URL du serveur ou la commande pour démarrer le serveur.

Considérations de sécurité

Lors du déploiement d’un serveur MCP, tenez compte de ces meilleures pratiques de sécurité :

1. Authentification : Implémentez des clés API ou OAuth pour authentifier les clients.

def authenticate_request(request):
    api_key = request.headers.get("X-API-Key")
    if not api_key or api_key != os.environ.get("MCP_API_KEY"):
        raise ValueError("Invalid API key")

2. Validation des entrées : Validez toujours toutes les entrées à l’aide des modèles Pydantic.
3. Limitation de débit : Implémentez une limitation de débit pour prévenir les abus.
4. HTTPS : Utilisez toujours HTTPS en production.
5. Actions restreintes : Définissez des limites claires pour ce que les outils peuvent faire.

Optimisation des performances

1. Mise en cache : Mettez en cache les récupérations de ressources coûteuses :

from functools import lru_cache

@lru_cache(maxsize=128, ttl=300)  # Cache for 5 minutes
def fetch_user_profiles():
    # Expensive database query
    pass

2. Traitement asynchrone : Utilisez async pour les opérations liées aux E/S :

async def fetch_user_profiles():
    async with aiohttp.ClientSession() as session:
        async with session.get("https://api.example.com/users") as response:
            data = await response.json()
            return data

3. Pooling de connexions : Utilisez des pools de connexions pour l’accès à la base de données.

Déploiement

Développement local

Pour le développement local, exécutez :

python src/server.py

Déploiement Docker

Créez un fichier Dockerfile :

FROM python:3.10-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8000

CMD ["python", "src/server.py"]

Construisez et exécutez :

docker build -t mcp-server .
docker run -p 8000:8000 mcp-server

Déploiement cloud (AWS)

1. Créez une instance EC2 ou utilisez AWS App Runner
2. Déployez votre conteneur Docker
3. Configurez un Application Load Balancer
4. Configurez des groupes de sécurité pour restreindre l’accès

Test de votre serveur MCP

Créez un fichier de test tests/test_resources.py :

import pytest
from src.resources.user_profiles import fetch_user_profiles

def test_fetch_user_profiles():
    """Test that user profiles can be fetched successfully."""
    profiles = fetch_user_profiles()
    
    # Check structure
    assert isinstance(profiles, list)
    assert len(profiles) > 0
    
    # Check content
    first_profile = profiles[0]
    assert "name" in first_profile
    assert "role" in first_profile
    assert isinstance(first_profile["name"], str)

Exécutez les tests avec :

pytest

Erreurs courantes et dépannage

Problème Solution Exemple Impossible de se connecter au serveur MCP Vérifiez que votre serveur est en cours d’exécution et que le port est correct netstat -tulpn | grep 8000 Le LLM ne trouve pas les ressources Vérifiez que les champs name et description sont correctement définis Vérifiez votre initialisation de Resource Erreurs dans l’exécution des outils Validez que les paramètres d’entrée correspondent aux types attendus Utilisez Pydantic pour la validation Le client ne peut pas analyser la sortie Assurez-vous que vos fonctions renvoient des données sérialisables en JSON Utilisez .model_dump() au lieu d’objets personnalisés Le serveur plante au démarrage Vérifiez vos importations et variables d’environnement Définissez DEBUG=True pour une journalisation détaillée Délai d’expiration de l’outil Ajoutez une gestion des délais d’attente pour les appels API externes Utilisez asyncio.wait_for() avec un délai Échecs d’authentification Vérifiez les clés API et les permissions Vérifiez les en-têtes de requête Erreurs d’analyse XML/JSON Utilisez des en-têtes de type de contenu appropriés Définissez Content-Type: application/json

Prochaines étapes

Après avoir construit votre serveur MCP de base, envisagez ces extensions avancées :

1. Intégration de base de données : Connectez-vous à PostgreSQL, MongoDB ou d’autres bases de données.
2. Opérations sur les fichiers : Ajoutez des outils pour la lecture, l’écriture et la transformation de fichiers.
3. API externes : Intégrez des services populaires comme GitHub, Slack ou Google Drive.
4. Webhooks : Permettez au LLM de déclencher des événements dans d’autres systèmes.
5. Ressources en streaming : Prenez en charge le streaming de grands ensembles de données.
6. Actions contextuelles : Ajoutez des outils qui comprennent le contexte actuel du LLM.

Exemple : Ajout d’une connexion à une base de données

import psycopg2
from contextlib import contextmanager

@contextmanager
def get_db_connection():
    """Create a database connection context manager."""
    conn = None
    try:
        conn = psycopg2.connect(
            host=os.environ.get("DB_HOST"),
            database=os.environ.get("DB_NAME"),
            user=os.environ.get("DB_USER"),
            password=os.environ.get("DB_PASSWORD")
        )
        yield conn
    finally:
        if conn is not None:
            conn.close()

def fetch_user_profiles_from_db():
    """Fetch user profiles from a PostgreSQL database."""
    with get_db_connection() as conn:
        with conn.cursor() as cur:
            cur.execute("SELECT name, role, department FROM users")
            columns = [desc[0] for desc in cur.description]
            return [dict(zip(columns, row)) for row in cur.fetchall()]

Conclusion

La construction d’un serveur MCP simple en Python ouvre la porte à rendre les LLM beaucoup plus puissants. En exposant des données et des actions via un protocole propre et standardisé, vous facilitez l’interaction sécurisée et significative des systèmes d’IA avec des services externes.

Commencez petit, concentrez-vous sur une ressource et un outil, et vous pourrez vous développer au fil du temps vers des cas d’utilisation plus avancés comme les bases de données, le stockage cloud ou les API internes.

L’écosystème MCP se développe rapidement, et l’implémentation de ces normes dès maintenant positionnera vos applications pour bénéficier des améliorations tant du protocole que des LLM qui l’utilisent.

FAQ