Auth, Rate Limiting et Cache Intelligent

Pourquoi sécuriser un serveur MCP

Un serveur MCP finance en production fait face à trois menaces :

La solution est une triple couche : authentification → rate limiting → cache. Chaque requête passe par les trois dans cet ordre.

Authentification par Bearer token

Le pattern le plus simple et le plus adapté pour MCP : un Bearer token envoyé dans le header HTTP. Le client MCP le transmet automatiquement si configuré :

# Côté client — connexion avec token
claude mcp add finance-prod \
  --transport sse \
  --url https://mcp.example.com/sse \
  --header "Authorization: Bearer sk-mcp-abc123..."

Implémentation côté serveur

# auth.py — Middleware d'authentification
import os
import hashlib
import hmac
from fastapi import Request, HTTPException

# Tokens autorisés (en prod: base de données ou vault)
VALID_TOKENS = {
    os.environ.get("MCP_SERVICE_TOKEN", ""): {
        "client": "scanner-bot",
        "tier": "unlimited",
        "rate_limit": 0  # 0 = pas de limite
    },
}

async def verify_token(request: Request) -> dict:
    """Vérifie le Bearer token et retourne les infos client."""
    auth = request.headers.get("Authorization", "")
    if not auth.startswith("Bearer "):
        raise HTTPException(401, "Missing Bearer token")

    token = auth[7:]
    token_hash = hashlib.sha256(token.encode()).hexdigest()

    client_info = VALID_TOKENS.get(token)
    if not client_info:
        raise HTTPException(403, "Invalid token")

    return client_info

Tiers d'accès

Rate limiting : protéger les APIs upstream

Le rate limiting protège deux choses : vos APIs upstream (Yahoo a une limite de ~2000 req/heure) et votre serveur lui-même contre les boucles accidentelles. On utilise l'algorithme token bucket :

# rate_limiter.py — Token bucket en mémoire
import time
from collections import defaultdict

class RateLimiter:
    def __init__(self):
        self.buckets: dict[str, dict] = defaultdict(
            lambda: {"tokens": 60, "last_refill": time.monotonic()}
        )

    def check(self, client_id: str, max_tokens: int, refill_rate: float) -> bool:
        """Retourne True si la requête est autorisée."""
        now = time.monotonic()
        bucket = self.buckets[client_id]

        # Recharger les tokens
        elapsed = now - bucket["last_refill"]
        bucket["tokens"] = min(
            max_tokens,
            bucket["tokens"] + elapsed * refill_rate
        )
        bucket["last_refill"] = now

        # Consommer un token
        if bucket["tokens"] >= 1:
            bucket["tokens"] -= 1
            return True
        return False

    def retry_after(self, client_id: str, refill_rate: float) -> float:
        """Secondes à attendre avant le prochain token."""
        bucket = self.buckets[client_id]
        return max(0, (1 - bucket["tokens"]) / refill_rate)

rate_limiter = RateLimiter()

Deux niveaux de rate limiting

# Rate limiting upstream — partagé entre tous les clients
yahoo_limiter = RateLimiter()  # 30/min pour Yahoo

async def get_quote_with_limits(ticker: str, client_id: str):
    # 1. Vérifier le rate limit client
    if not rate_limiter.check(client_id, max_tokens=60, refill_rate=1.0):
        wait = rate_limiter.retry_after(client_id, 1.0)
        return {"error": "rate_limited", "retry_after_s": round(wait, 1)}

    # 2. Vérifier le rate limit upstream
    if not yahoo_limiter.check("yahoo", max_tokens=30, refill_rate=0.5):
        # Pas de slot Yahoo → servir depuis le cache ou attendre
        cached = cache.get(f"quote:{ticker}")
        if cached:
            cached["_from_cache"] = True
            return cached
        wait = yahoo_limiter.retry_after("yahoo", 0.5)
        return {"error": "upstream_rate_limited", "retry_after_s": round(wait)}

    # 3. Appeler Yahoo
    result = await provider.get_quote(ticker)
    cache.set(f"quote:{ticker}", result, ttl=15)
    return result

Cache TTL intelligent par type de donnée

Le cache est le composant qui fait la plus grosse différence en production. Sans cache, un scanner qui analyse 100 tickers fait 100+ appels Yahoo. Avec un cache TTL adapté, il en fait 10-20.

# cache.py — Cache en mémoire avec TTL
import time
from typing import Any, Optional

class TTLCache:
    def __init__(self, max_size: int = 10_000):
        self.store: dict[str, tuple[Any, float]] = {}
        self.max_size = max_size
        self.hits = 0
        self.misses = 0

    def get(self, key: str) -> Optional[Any]:
        if key in self.store:
            value, expires_at = self.store[key]
            if time.monotonic() < expires_at:
                self.hits += 1
                return value
            # Expiré — supprimer
            del self.store[key]
        self.misses += 1
        return None

    def set(self, key: str, value: Any, ttl: int):
        # Eviction LRU si plein
        if len(self.store) >= self.max_size:
            oldest = min(self.store, key=lambda k: self.store[k][1])
            del self.store[oldest]
        self.store[key] = (value, time.monotonic() + ttl)

    @property
    def hit_rate(self) -> float:
        total = self.hits + self.misses
        return self.hits / total if total > 0 else 0.0

# TTL par défaut par type de donnée
TTL_MAP = {
    "quote": 15,
    "bars_daily": 43200,   # 12h
    "technicals": 3600,    # 1h
    "fundamentals": 86400, # 24h
    "earnings": 86400,     # 24h
    "overview": 30,
    "screener": 3600,      # 1h
}

cache = TTLCache()

Intégrer le cache dans les outils

async def handle_query_data(provider, args: dict) -> dict:
    ticker = args["ticker"].upper()
    types = args["types"]
    result = {"ticker": ticker}

    for data_type in types:
        cache_key = f"{data_type}:{ticker}"
        cached = cache.get(cache_key)

        if cached:
            result[data_type] = cached
            result.setdefault("_cached", []).append(data_type)
            continue

        # Cache miss → appel provider
        data = await fetch_data(provider, ticker, data_type)
        cache.set(cache_key, data, ttl=TTL_MAP.get(data_type, 60))
        result[data_type] = data

    return result

Assembler la middleware chain

On compose les trois couches en middleware FastAPI. Chaque requête traverse auth → rate limit → handler :

# server.py — version sécurisée
from fastapi import FastAPI, Request, Depends
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI(title="MCP Finance Server")

# CORS — nécessaire si le client est un navigateur
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # En prod: restreindre
    allow_methods=["GET", "POST"],
    allow_headers=["Authorization", "Content-Type"],
)

# Middleware de logging + rate limiting
@app.middleware("http")
async def security_middleware(request: Request, call_next):
    # Skip health check
    if request.url.path == "/health":
        return await call_next(request)

    # 1. Auth
    try:
        client_info = await verify_token(request)
    except HTTPException as e:
        return JSONResponse(
            status_code=e.status_code,
            content={"error": e.detail}
        )

    # 2. Rate limit
    limit = client_info.get("rate_limit", 60)
    if limit > 0:
        if not rate_limiter.check(
            client_info["client"],
            max_tokens=limit,
            refill_rate=limit / 60
        ):
            wait = rate_limiter.retry_after(client_info["client"], limit / 60)
            return JSONResponse(
                status_code=429,
                content={"error": "rate_limited", "retry_after_s": round(wait, 1)},
                headers={"Retry-After": str(int(wait) + 1)}
            )

    # 3. Injecter les infos client dans la requête
    request.state.client = client_info

    # 4. Passer au handler
    response = await call_next(request)

    # 5. Headers de diagnostic
    response.headers["X-Cache-Hit-Rate"] = f"{cache.hit_rate:.1%}"
    return response

Monitoring et métriques

Un serveur en production sans monitoring est un serveur qui va casser en silence. Les métriques essentielles :

# Endpoint de métriques
@app.get("/metrics")
async def metrics():
    return {
        "cache": {
            "hit_rate": f"{cache.hit_rate:.1%}",
            "size": len(cache.store),
            "hits": cache.hits,
            "misses": cache.misses,
        },
        "rate_limits": {
            client: {
                "tokens_remaining": bucket["tokens"],
            }
            for client, bucket in rate_limiter.buckets.items()
        },
        "uptime_s": time.monotonic() - start_time,
        "tools_count": len(TOOLS),
        "total_requests": request_counter,
    }