Après avoir déployé cette stack sur trois postes de trading algorithmique en production, je peux affirmer que le couplage MCP + WebSocket Binance transforme radicalement la façon dont un LLM accède aux marchés. Cet article condense les leçons apprises, les pièges évités et les optimisations qui m'ont permis de descendre sous la barre des 200 ms de latence E2E, pour un coût mensuel réduit d'un facteur 6 par rapport à l'API directe.

1. Architecture cible et flux de données

Le pattern retenu sépare strictement trois préoccupations : acquisition (WebSocket), exposition (MCP Server), consommation (Claude Desktop). Cette séparation permet de scaler chaque brique indépendamment et de réutiliser le même serveur MCP pour d'autres clients MCP-compatibles (Cursor, Continue, Cline).

┌─────────────────┐    wss://stream.binance.com    ┌──────────────────┐
│  Binance Spot   │ ───────────────────────────▶   │  MCP Server      │
│  WebSocket API  │ ◀───────────────────────────   │  (FastMCP + uv)  │
└─────────────────┘    orderbook@200ms depth=20    └──────────────────┘
                                                           │ stdio/jsonrpc
                                                           ▼
                                                  ┌──────────────────┐
                                                  │  Claude Desktop  │
                                                  │  (via HolySheep) │
                                                  └──────────────────┘

2. Prérequis techniques

3. Étape 1 — Initialisation du projet

mkdir binance-mcp && cd binance-mcp
uv init --python 3.11
uv add "mcp[cli]" websockets orjson pydantic-settings httpx
uv add --dev pytest-asyncio ruff mypy

4. Étape 2 — Client WebSocket Binance résilient

Le code ci-dessous est celui qui tourne en production. Il gère la reconnexion exponentielle, le heartbeat applicatif et la déduplication des messages.

# binance_client.py
from __future__ import annotations
import asyncio, time, logging
from collections.abc import AsyncIterator
import websockets, orjson

LOG = logging.getLogger("binance.ws")

class BinanceStream:
    """Client WebSocket production-grade pour Binance Spot."""

    BASE = "wss://stream.binance.com:9443/stream"
    PING_INTERVAL = 180  # secondes (max 24h avant drop)
    MAX_BACKOFF = 30

    def __init__(self, symbols: list[str], depth: int = 20):
        self.streams = (
            [f"{s.lower()}@depth{depth}@100ms" for s in symbols]
          + [f"{s.lower()}@ticker" for s in symbols]
        )
        self._seq: dict[str, int] = {}
        self._lock = asyncio.Lock()
        self._stats = {"msgs": 0, "drops": 0, "started": time.monotonic()}

    async def _connect(self):
        url = f"{self.BASE}?streams=" + "/".join(self.streams)
        return await websockets.connect(
            url, ping_interval=self.PING_INTERVAL,
            max_size=2**20, compression=None,
        )

    async def stream(self) -> AsyncIterator[dict]:
        backoff = 1
        while True:
            try:
                async with await self._connect() as ws:
                    LOG.info("WS connected", extra={"streams": len(self.streams)})
                    backoff = 1
                    async for raw in ws:
                        msg = orjson.loads(raw)
                        if not await self._dedupe(msg):
                            yield msg.get("data", msg)
                            self._stats["msgs"] += 1
            except (websockets.ConnectionClosed, OSError) as e:
                self._stats["drops"] += 1
                LOG.warning("WS drop: %s — retry in %ss", e, backoff)
                await asyncio.sleep(backoff)
                backoff = min(backoff * 2, self.MAX_BACKOFF)

    async def _dedupe(self, msg: dict) -> bool:
        d = msg.get("data", msg)
        key = d.get("s") or d.get("stream", "")
        seq = d.get("u") or d.get("E")
        if seq is None:
            return False
        async with self._lock:
            if self._seq.get(key) == seq:
                return True
            self._seq[key] = seq
            return False

    def stats(self) -> dict:
        elapsed = time.monotonic() - self._stats["started"]
        return {
            **self._stats,
            "elapsed_s": round(elapsed, 1),
            "msg_per_s": round(self._stats["msgs"] / max(elapsed, 1), 1),
        }

5. Étape 3 — Serveur MCP avec outils de trading

FastMCP simplifie l'enregistrement des outils via décorateurs. J'expose quatre outils : top-of-book, profondeur, chandelles et ticker 24 h. Le cache court (_TTL_MS) absorbe la rafale d'appels LLM sans marteler Binance.

# server.py
from __future__ import annotations
import asyncio, time
from mcp.server.fastmcp import FastMCP
from binance_client import BinanceStream

mcp = FastMCP("binance-realtime")
_cache: dict[str, tuple[float, dict]] = {}
_TTL_MS = 250  # absorbe le rate limit LLM sans servir de donnée périmée

async def _get_fresh(symbol: str) -> dict:
    cached = _cache.get(symbol)
    if cached and (time.time() - cached[0]) * 1000 < _TTL_MS:
        return cached[1]
    raise RuntimeError(f"No fresh data for {symbol} within {_TTL_MS}ms")

@mcp.tool()
async def get_ticker(symbol: str) -> dict:
    """Ticker 24h (last, bid, ask, volume) pour symbol (ex: BTCUSDT)."""
    d = await _get_fresh(symbol.upper())
    return {
        "symbol": d["s"], "last": float(d["c"]),
        "bid": float(d["b"]),  "ask": float(d["a"]),
        "vol_24h": float(d["v"]), "change_pct": float(d["P"]),
    }

@mcp.tool()
async def get_order_book(symbol: str, limit: int = 20) -> dict:
    """Profondeur (bids/asks) jusqu'à limit niveaux."""
    if not 1 <= limit <= 100:
        raise ValueError("limit must

Ressources connexes
Articles connexes