Modèles avancés et déploiement en production

À mesure que vos applications FastMCP évoluent de simples outils de développement en services prêts pour la production, vous devrez prendre en compte les modèles avancés, la scalabilité, la sécurité et les stratégies de déploiement. Cette section couvre le développement et le déploiement de FastMCP de niveau entreprise.

Modèles avancés de serveur

Composition et modularité des serveurs

FastMCP 2.0 prend en charge des modèles sophistiqués de composition de serveurs qui permettent des applications modulaires et faciles à maintenir :

# auth_server.py
from fastmcp import FastMCP
import jwt
import bcrypt

auth_server = FastMCP("Authentication Service")

@auth_server.tool
def authenticate_user(username: str, password: str) -> dict:
    """Authentifier un utilisateur et renvoyer un jeton JWT."""
    # Implémentation pour l'authentification de l'utilisateur
    if verify_credentials(username, password):
        token = jwt.encode({"user": username}, "secret", algorithm="HS256")
        return {"token": token, "user": username, "authenticated": True}
    return {"authenticated": False, "error": "Invalid credentials"}

# data_server.py
from fastmcp import FastMCP
import pandas as pd

data_server = FastMCP("Data Processing Service")

@data_server.tool
def process_dataset(data_path: str, operation: str) -> dict:
    """Traiter un dataset avec des opérations spécifiées."""
    df = pd.read_csv(data_path)

    operations = {
        "summary": lambda df: df.describe().to_dict(),
        "count": lambda df: {"rows": len(df), "columns": len(df.columns)},
        "columns": lambda df: df.columns.tolist()
    }

    if operation not in operations:
        raise ValueError(f"Unknown operation: {operation}")

    return operations[operation](df)

# main_server.py
from fastmcp import FastMCP
from auth_server import auth_server
from data_server import data_server

# Créer le serveur principal et composer les sous-serveurs
main_server = FastMCP(
    name="Enterprise Application",
    instructions="Application d'entreprise complète avec authentification et traitement de données"
)

# Monter les sous-serveurs avec des préfixes
main_server.mount(auth_server, prefix="auth")
main_server.mount(data_server, prefix="data")

# Ajouter des outils au serveur principal
@main_server.tool
def health_check() -> dict:
    """Vérifier la santé de tous les services."""
    return {
        "status": "healthy",
        "services": ["auth", "data", "main"],
        "timestamp": datetime.now().isoformat()
    }

if __name__ == "__main__":
    main_server.run(transport="http", host="0.0.0.0", port=8000)

Serveurs proxy pour l'intégration d'API

Créez des serveurs proxy pour intégrer les API existantes dans l'écosystème MCP :

# api_proxy.py
from fastmcp import FastMCP, Client
import asyncio

# Créer un proxy pour un serveur MCP existant
async def create_api_proxy():
    # Se connecter au serveur MCP distant
    remote_client = Client("https://api.example.com/mcp/sse")

    # Créer un serveur proxy
    proxy_server = FastMCP.as_proxy(
        remote_client,
        name="API Proxy Server",
        instructions="Proxy pour les services API distants"
    )

    # Ajouter des middlewares ou des outils personnalisés au proxy
    @proxy_server.tool
    def local_cache_status() -> dict:
        """Obtenir le statut du cache local."""
        return {"cache_enabled": True, "cache_size": "50MB"}

    return proxy_server

# Utilisation
if __name__ == "__main__":
    proxy = asyncio.run(create_api_proxy())
    proxy.run(transport="http", port=9000)

Génération dynamique d'outils

Générez des outils dynamiquement à partir de configurations ou de spécifications OpenAPI :

# dynamic_tools.py
from fastmcp import FastMCP
import yaml
import requests

mcp = FastMCP("Dynamic API Server")

def create_api_tool(endpoint_config):
    """Crée dynamiquement un outil à partir de la configuration d'un point d'accès API."""
    def api_tool(**kwargs):
        response = requests.request(
            method=endpoint_config["method"],
            url=endpoint_config["url"],
            **kwargs
        )
        return response.json()

    # Définir les métadonnées de la fonction
    api_tool.__name__ = endpoint_config["name"]
    api_tool.__doc__ = endpoint_config["description"]

    return api_tool

# Charger la configuration API
with open("api_config.yaml", "r") as f:
    api_config = yaml.safe_load(f)

# Générer des outils dynamiquement
for endpoint in api_config["endpoints"]:
    tool_func = create_api_tool(endpoint)
    mcp.tool(tool_func)

# Alternative : Générer à partir de la spécification OpenAPI
def load_from_openapi(spec_url: str):
    """Charger les outils à partir de la spécification OpenAPI."""
    # FastMCP a une intégration OpenAPI intégrée
    api_server = FastMCP.from_openapi(spec_url)
    return api_server

# Utilisation
# openapi_server = load_from_openapi("https://api.example.com/openapi.json")
# mcp.mount(openapi_server, prefix="api")

Authentification prête pour la production

Authentification basée sur JWT

# secure_server.py
from fastmcp import FastMCP
from fastmcp.auth import JWTAuthProvider
import os

# Configurer l'authentification JWT
auth_provider = JWTAuthProvider(
    secret_key=os.getenv("JWT_SECRET_KEY"),
    algorithm="HS256",
    token_header="Authorization"
)

# Créer un serveur authentifié
mcp = FastMCP(
    "Secure Production Server",
    auth_provider=auth_provider
)

@mcp.tool(require_auth=True)
def get_sensitive_data(user_context) -> dict:
    """Obtenir des données sensibles - nécessite une authentification."""
    return {
        "user_id": user_context.user_id,
        "data": "sensitive information",
        "access_level": user_context.permissions
    }

@mcp.tool  # Outil public, aucune authentification requise
def get_public_info() -> dict:
    """Obtenir des informations publiques."""
    return {"status": "public", "version": "1.0.0"}

Authentification par clé API

# api_key_server.py
from fastmcp import FastMCP
from fastmcp.auth import APIKeyAuthProvider

# Configurer l'authentification par clé API
auth_provider = APIKeyAuthProvider(
    api_keys={
        "key123": {"name": "Client A", "permissions": ["read", "write"]},
        "key456": {"name": "Client B", "permissions": ["read"]}
    },
    header_name="X-API-Key"
)

mcp = FastMCP("Serveur protégé par clé API", auth_provider=auth_provider)

@mcp.tool(require_permissions=["write"])
def write_data(data: dict, auth_context) -> dict:
    """Écrire des données - nécessite l'autorisation d'écriture."""
    # Implémentation
    return {"written": True, "client": auth_context.client_name}

Stratégies de déploiement

Déploiement Docker

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

# Installer les dépendances
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copier l'application
COPY . .

# Créer un utilisateur non root
RUN useradd -m -u 1000 mcpuser && chown -R mcpuser:mcpuser /app
USER mcpuser

# Vérification de santé
HEALTHCHECK --interval=30s --timeout=3s --start-period=5s --retries=3 \
    CMD curl -f http://localhost:8000/health || exit 1

# Lancer le serveur
EXPOSE 8000
CMD ["python", "server.py"]

# docker-compose.yml
version: '3.8'

services:
  fastmcp-server:
    build: .
    ports:
      - "8000:8000"
    environment:
      - FASTMCP_LOG_LEVEL=INFO
      - JWT_SECRET_KEY=${JWT_SECRET_KEY}
      - DATABASE_URL=${DATABASE_URL}
    volumes:
      - ./logs:/app/logs
    restart: unless-stopped
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
      interval: 30s
      timeout: 10s
      retries: 3

  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    volumes:
      - redis_data:/data
    restart: unless-stopped

  nginx:
    image: nginx:alpine
    ports:
      - "80:80"
      - "443:443"
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf
      - ./ssl:/etc/nginx/ssl
    depends_on:
      - fastmcp-server
    restart: unless-stopped

volumes:
  redis_data:

Déploiement Kubernetes

# k8s-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: fastmcp-server
  labels:
    app: fastmcp-server
spec:
  replicas: 3
  selector:
    matchLabels:
      app: fastmcp-server
  template:
    metadata:
      labels:
        app: fastmcp-server
    spec:
      containers:
      - name: fastmcp-server
        image: your-registry/fastmcp-server:latest
        ports:
        - containerPort: 8000
        env:
        - name: FASTMCP_LOG_LEVEL
          value: "INFO"
        - name: JWT_SECRET_KEY
          valueFrom:
            secretKeyRef:
              name: fastmcp-secrets
              key: jwt-secret
        resources:
          requests:
            memory: "256Mi"
            cpu: "250m"
          limits:
            memory: "512Mi"
            cpu: "500m"
        livenessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 30
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /ready
            port: 8000
          initialDelaySeconds: 5
          periodSeconds: 5

---
apiVersion: v1
kind: Service
metadata:
  name: fastmcp-service
spec:
  selector:
    app: fastmcp-server
  ports:
  - protocol: TCP
    port: 80
    targetPort: 8000
  type: LoadBalancer

---
apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
  name: fastmcp-ingress
  annotations:
    kubernetes.io/ingress.class: nginx
    cert-manager.io/cluster-issuer: letsencrypt-prod
spec:
  tls:
  - hosts:
    - api.yourdomain.com
    secretName: fastmcp-tls
  rules:
  - host: api.yourdomain.com
    http:
      paths:
      - path: /
        pathType: Prefix
        backend:
          service:
            name: fastmcp-service
            port:
              number: 80

Surveillance et observabilité

Journalisation et métriques

# monitored_server.py
from fastmcp import FastMCP, Context
import logging
import time
from prometheus_client import Counter, Histogram, start_http_server

# Configurer les métriques
TOOL_CALLS = Counter('mcp_tool_calls_total', 'Nombre total d\'appels d\'outils', ['tool_name', 'status'])
TOOL_DURATION = Histogram('mcp_tool_duration_seconds', 'Temps d\'exécution de l\'outil', ['tool_name'])

# Configurer la journalisation
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('fastmcp.log'),
        logging.StreamHandler()
    ]
)

logger = logging.getLogger(__name__)

mcp = FastMCP("Serveur surveillé")

def monitor_tool(func):
    """Décorateur pour ajouter la surveillance aux outils."""
    def wrapper(*args, **kwargs):
        start_time = time.time()
        tool_name = func.__name__

        try:
            logger.info(f"Démarrage de l'outil : {tool_name}")
            result = func(*args, **kwargs)
            TOOL_CALLS.labels(tool_name=tool_name, status='success').inc()
            logger.info(f"Outil {tool_name} terminé avec succès")
            return result
        except Exception as e:
            TOOL_CALLS.labels(tool_name=tool_name, status='error').inc()
            logger.error(f"L'outil {tool_name} a échoué : {str(e)}")
            raise
        finally:
            duration = time.time() - start_time
            TOOL_DURATION.labels(tool_name=tool_name).observe(duration)
            logger.info(f"L'outil {tool_name} a pris {duration:.2f}s")

    return wrapper

@mcp.tool
@monitor_tool
def process_data(data: list) -> dict:
    """Traiter les données avec surveillance."""
    # Logique de traitement
    return {"processed": len(data), "status": "success"}

# Démarrer le serveur de métriques Prometheus
start_http_server(9090)

if __name__ == "__main__":
    mcp.run(transport="http", port=8000)

Vérifications de santé et disjoncteurs

# resilient_server.py
from fastmcp import FastMCP, Context
import asyncio
from circuit_breaker import CircuitBreaker
import aioredis

mcp = FastMCP("Serveur résilient")

# Disjoncteur pour les appels API externes
api_circuit_breaker = CircuitBreaker(
    failure_threshold=5,
    recovery_timeout=30,
    expected_exception=Exception
)

@mcp.tool
async def call_external_api(endpoint: str, ctx: Context) -> dict:
    """Appeler une API externe avec protection par disjoncteur."""
    @api_circuit_breaker
    async def make_api_call():
        # Implémentation de l'appel API réel
        async with aiohttp.ClientSession() as session:
            async with session.get(endpoint) as response:
                return await response.json()

    try:
        return await make_api_call()
    except Exception as e:
        await ctx.error(f"Échec de l'appel API : {str(e)}")
        return {"error": "Service temporairement indisponible"}

@mcp.resource("health://status")
async def health_check() -> dict:
    """Vérification de santé complète."""
    health_status = {
        "status": "healthy",
        "timestamp": datetime.now().isoformat(),
        "checks": {}
    }

    # Vérifier la connexion à la base de données
    try:
        # Logique de vérification de santé de la base de données
        health_status["checks"]["database"] = "healthy"
    except Exception:
        health_status["checks"]["database"] = "unhealthy"
        health_status["status"] = "unhealthy"

    # Vérifier la connexion Redis
    try:
        redis = aioredis.from_url("redis://localhost")
        await redis.ping()
        health_status["checks"]["redis"] = "healthy"
    except Exception:
        health_status["checks"]["redis"] = "unhealthy"
        health_status["status"] = "degraded"

    return health_status

Bonnes pratiques de sécurité

Validation et assainissement des entrées

# secure_tools.py
from fastmcp import FastMCP
from pydantic import BaseModel, validator
import re
import html

mcp = FastMCP("Serveur sécurisé")

class UserInput(BaseModel):
    username: str
    email: str
    age: int

    @validator('username')
    def validate_username(cls, v):
        if not re.match(r'^[a-zA-Z0-9_]+$', v):
            raise ValueError('Le nom d\'utilisateur ne peut contenir que des lettres, des chiffres et des underscores')
        if len(v) < 3 or len(v) > 20:
            raise ValueError('Le nom d\'utilisateur doit avoir entre 3 et 20 caractères')
        return v

    @validator('email')
    def validate_email(cls, v):
        email_pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
        if not re.match(email_pattern, v):
            raise ValueError('Format d\'e-mail invalide')
        return v

    @validator('age')
    def validate_age(cls, v):
        if v < 0 or v > 150:
            raise ValueError('L\'âge doit être compris entre 0 et 150')
        return v

@mcp.tool
def create_user(user_data: UserInput) -> dict:
    """Crée un utilisateur avec une entrée validée."""
    # L'entrée est automatiquement validée par Pydantic
    return {
        "user_id": generate_user_id(),
        "username": user_data.username,
        "email": user_data.email,
        "status": "created"
    }

@mcp.tool
def sanitize_html_content(content: str) -> str:
    """Assainir le contenu HTML pour prévenir les attaques XSS."""
    # Assainissement HTML basique
    sanitized = html.escape(content)
    return sanitized

Limitation de débit et étranglement

# rate_limited_server.py
from fastmcp import FastMCP
from fastmcp.middleware import RateLimitMiddleware
import asyncio

# Configurer la limitation de débit
rate_limiter = RateLimitMiddleware(
    requests_per_minute=60,
    burst_size=10,
    storage_backend="redis",  # ou "memory"
    redis_url="redis://localhost:6379"
)

mcp = FastMCP("Serveur à débit limité", middleware=[rate_limiter])

@mcp.tool
def expensive_operation(data: str) -> dict:
    """Une opération coûteuse qui devrait être limitée."""
    # Simuler un traitement coûteux
    time.sleep(2)
    return {"processed": True, "data_length": len(data)}

Optimisation des performances

Opérations asynchrones et pool de connexions

# optimized_server.py
from fastmcp import FastMCP, Context
import asyncio
import aiohttp
import asyncpg
from contextlib import asynccontextmanager

class DatabasePool:
    def __init__(self):
        self.pool = None

    async def create_pool(self):
        self.pool = await asyncpg.create_pool(
            "postgresql://user:pass@localhost/db",
            min_size=10,
            max_size=20
        )

    async def close_pool(self):
        if self.pool:
            await self.pool.close()

db_pool = DatabasePool()

@asynccontextmanager
async def lifespan():
    # Démarrage
    await db_pool.create_pool()
    yield
    # Arrêt
    await db_pool.close_pool()

mcp = FastMCP("Serveur optimisé", lifespan=lifespan)

@mcp.tool
async def query_database(query: str) -> dict:
    """Exécuter une requête de base de données en utilisant le pool de connexions."""
    async with db_pool.pool.acquire() as connection:
        result = await connection.fetch(query)
        return {"rows": [dict(row) for row in result]}

@mcp.tool
async def batch_http_requests(urls: list[str], ctx: Context) -> list[dict]:
    """Effectuer plusieurs requêtes HTTP simultanément."""
    await ctx.info(f"Effectuer {len(urls)} requêtes concurrentes...")

    async with aiohttp.ClientSession() as session:
        tasks = []
        for url in urls:
            task = session.get(url)
            tasks.append(task)

        responses = await asyncio.gather(*tasks, return_exceptions=True)

        results = []
        for i, response in enumerate(responses):
            if isinstance(response, Exception):
                results.append({"url": urls[i], "error": str(response)})
            else:
                results.append({
                    "url": urls[i],
                    "status": response.status,
                    "data": await response.text()
                })

        return results

Conclusion

FastMCP 2.0 offre une plateforme complète pour construire des serveurs MCP prêts pour la production, capables de s'adapter, des simples outils de développement aux applications d'entreprise. Points clés à retenir :

Bonnes pratiques de développement

Utiliser une composition de serveurs modulaire pour un code maintenable
Implémenter une gestion complète des erreurs et une validation
Ajouter une journalisation et une surveillance appropriées dès le début
Tester minutieusement vos outils et ressources

Prête pour la production

Mettre en œuvre une authentification et une autorisation appropriées
Utiliser une configuration basée sur l'environnement
Ajouter des vérifications de santé et des disjoncteurs
Surveiller les performances et l'utilisation des ressources

Stratégies de déploiement

Conteneuriser vos applications pour la cohérence
Utiliser des plateformes d'orchestration comme Kubernetes pour l'échelle
Implémenter des pipelines CI/CD appropriés
Planifier la haute disponibilité et la reprise après sinistre

Considérations de sécurité

Valider et assainir toutes les entrées
Mettre en œuvre la limitation de débit et l'étranglement
Utiliser HTTPS et une authentification appropriée
Audits de sécurité et mises à jour réguliers

FastMCP permet de construire des applications sophistiquées basées sur l'IA qui s'intègrent parfaitement aux flux de travail de développement modernes et à l'infrastructure de production. Que vous construisiez des outils de développement, des services de traitement de données ou des applications d'entreprise complexes, FastMCP fournit les bases dont vous avez besoin pour réussir.

Bonne construction avec FastMCP ! 🚀