Du venv au premier outil fonctionnel : on construit un serveur MCP complet avec FastAPI et le SDK Python officiel. Routing JSON-RPC, transport SSE, tool registration, et test en direct avec Claude Code.
Avant d'écrire une seule ligne, assurons-nous que l'environnement est prêt. On reste volontairement minimaliste : Python, quelques packages, et un éditeur.
| Composant | Version min. | Rôle |
|---|---|---|
| Python | 3.10+ | Le runtime — asyncio natif, type hints complets |
| FastAPI | 0.100+ | Framework web async — routing, validation, docs auto |
| uvicorn | 0.23+ | Serveur ASGI — sert l'app FastAPI |
| mcp | 1.0+ | Le SDK Python officiel du Model Context Protocol |
| httpx | 0.25+ | Client HTTP async — pour appeler les APIs financières |
# Installation en une commande
python -m venv .venv
source .venv/bin/activate
pip install fastapi uvicorn mcp httpx pydantic
FastAPI est le choix naturel pour un serveur MCP Python. Il est async-first (essentiel pour gérer des appels d'API financières en parallèle), génère automatiquement la doc OpenAPI, et la validation Pydantic s'aligne avec les JSON Schemas de MCP. Les alternatives (Flask, Django) sont possibles mais moins ergonomiques pour ce cas d'usage.
Un serveur MCP finance bien organisé suit une structure en modules. On commence avec le strict minimum, et chaque partie de la série ajoutera des couches.
mcp-finance-server/ ├── server.py # Point d'entrée : FastAPI app + MCP setup ├── tools/ # Chaque outil = un module │ ├── __init__.py │ ├── query_data.py # QueryData — prix, technicals, fondamentaux │ ├── screener.py # RunScreener — scanning par filtres │ └── market.py # GetMarketOverview — indices, VIX, sectors ├── providers/ # Sources de données (Yahoo, SEC, FRED...) │ ├── __init__.py │ ├── yahoo.py │ └── sec.py ├── cache.py # Cache intelligent (TTL par type de donnée) ├── config.py # Configuration (API keys, TTLs, limits) ├── requirements.txt └── Dockerfile
Pour cette partie, on travaille uniquement avec server.py et un outil basique. Les modules tools/, providers/ et cache.py viendront dans les parties 3 et 4.
FastAPI + MCP. Le seul fichier obligatoire. ~200 lignes pour un serveur complet.
Un fichier par outil. Séparation propre : logique métier isolée du transport MCP.
Sources de données interchangeables. Yahoo today, Bloomberg demain — même interface.
Cache TTL par type. Quotes = 15s, fondamentaux = 24h. On ne surcharge pas les APIs.
Le package mcp (Python SDK officiel) fournit les briques de base pour construire un serveur. Il gère le protocole JSON-RPC, la sérialisation, la découverte des outils, et les transports (stdio, SSE).
| Composant | Import | Rôle |
|---|---|---|
| Server | mcp.server.Server | Le cœur — enregistre les outils, dispatch les appels |
| Tool | mcp.types.Tool | Définition d'un outil (nom, description, schéma) |
| SseServerTransport | mcp.server.sse | Transport HTTP/SSE pour les connexions distantes |
| Types | mcp.types | TextContent, ImageContent, CallToolResult, etc. |
# Import minimal pour un serveur MCP from mcp.server import Server from mcp.server.sse import SseServerTransport from mcp.types import ( Tool, TextContent, CallToolResult, ListToolsResult, ) # Créer le serveur server = Server("finance-mcp") # Le serveur expose 2 handlers obligatoires : # 1. list_tools → retourne la liste des outils disponibles # 2. call_tool → exécute un outil et retourne le résultat
initializetools/list → le Server retourne tous les outils avec leurs schémastools/call avec les argumentsCallToolResultTout le protocole MCP utilise JSON-RPC 2.0. Chaque message a un id, une method, et des params. Le SDK gère tout ça — vous n'aurez jamais à parser du JSON-RPC manuellement. Mais comprendre que c'est JSON-RPC aide à debugger : les erreurs sont des objets {"code": -32000, "message": "..."} standard.
Voici le squelette complet d'un serveur MCP finance. Chaque section est expliquée après le code :
#!/usr/bin/env python3 """MCP Finance Server — Squelette minimal.""" import json import logging from contextlib import asynccontextmanager from fastapi import FastAPI, Request from fastapi.responses import StreamingResponse import uvicorn from mcp.server import Server from mcp.server.sse import SseServerTransport from mcp.types import ( Tool, TextContent, CallToolResult, ListToolsResult ) logging.basicConfig(level=logging.INFO) logger = logging.getLogger("mcp-finance") # ────────────────────────────────────────────── # 1. Créer le serveur MCP # ────────────────────────────────────────────── mcp_server = Server("finance-mcp") # ────────────────────────────────────────────── # 2. Définir les outils disponibles # ────────────────────────────────────────────── TOOLS = [ Tool( name="GetQuote", description="""Get current price data for a stock ticker. Returns: last price, change (%), volume, market cap, 52-week range. Use this when the user asks about current stock prices.""", inputSchema={ "type": "object", "properties": { "ticker": { "type": "string", "description": "Stock ticker (e.g. AAPL, MSFT, TSLA)" } }, "required": ["ticker"] } ), ] @mcp_server.list_tools() async def list_tools() -> list[Tool]: return TOOLS # ────────────────────────────────────────────── # 3. Implémenter le dispatch # ────────────────────────────────────────────── @mcp_server.call_tool() async def call_tool(name: str, arguments: dict) -> CallToolResult: if name == "GetQuote": result = await get_quote(arguments["ticker"]) return CallToolResult( content=[TextContent( type="text", text=json.dumps(result, indent=2) )] ) raise ValueError(f"Unknown tool: {name}") # ────────────────────────────────────────────── # 4. Logique métier de l'outil # ────────────────────────────────────────────── async def get_quote(ticker: str) -> dict: # Pour l'instant, données simulées # Partie 3 → Yahoo Finance API réelle return { "ticker": ticker.upper(), "price": 189.84, "change_pct": 1.23, "volume": 54_230_100, "market_cap": "2.95T", "week_52_range": "164.08 - 199.62", "source": "simulated" } # ────────────────────────────────────────────── # 5. Transport SSE + FastAPI # ────────────────────────────────────────────── sse_transport = SseServerTransport("/mcp") app = FastAPI(title="MCP Finance Server") @app.get("/sse") async def sse_endpoint(request: Request): async with sse_transport.connect_sse( request.scope, request.receive, request._send ) as streams: await mcp_server.run( streams[0], streams[1], mcp_server.create_initialization_options() ) @app.post("/mcp") async def mcp_endpoint(request: Request): await sse_transport.handle_post_message( request.scope, request.receive, request._send ) # Health check @app.get("/health") async def health(): return {"status": "ok", "tools": len(TOOLS)} if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=8080)
Ce squelette n'a ni authentification, ni rate limiting, ni cache. C'est volontaire : on les ajoute en Partie 4. En l'état, c'est un serveur de développement — ne l'exposez pas sur Internet.
| Bloc | Lignes | Ce qu'il fait |
|---|---|---|
| 1. Création | 1 | Server("finance-mcp") — instancie le serveur avec un nom d'identification |
| 2. Tools | ~20 | Déclare chaque outil avec son nom, sa description, et son schéma JSON |
| 3. Dispatch | ~15 | Route les appels tools/call vers la bonne fonction |
| 4. Logique | ~10 | La fonction métier — c'est ici qu'on appelle les APIs financières |
| 5. Transport | ~20 | Monte SSE sur FastAPI — 2 routes (/sse pour la connexion, /mcp pour les appels) |
L'outil GetQuote ci-dessus fonctionne, mais il est simpliste. Voici les principes pour écrire un outil qui sera bien utilisé par le LLM :
C'est la description que le LLM lit pour décider d'appeler votre outil. Elle doit être :
Le inputSchema est un JSON Schema standard. Le LLM le lit pour savoir quels paramètres passer. Soyez précis :
# ❌ Mauvais schéma — trop vague { "type": "object", "properties": { "query": {"type": "string"} } } # ✅ Bon schéma — précis et documenté { "type": "object", "properties": { "ticker": { "type": "string", "description": "Stock ticker symbol (e.g. AAPL, MSFT)", "pattern": "^[A-Z]{1,5}$" }, "types": { "type": "array", "items": { "type": "string", "enum": ["quote", "technicals", "fundamentals"] }, "description": "Data types to retrieve" } }, "required": ["ticker", "types"] }
Retournez du JSON structuré avec des clés claires. Le LLM parse le résultat pour construire sa réponse :
# ❌ Mauvais résultat — texte libre "AAPL is at $189.84, up 1.23%" # ✅ Bon résultat — JSON structuré { "ticker": "AAPL", "price": 189.84, "change_pct": 1.23, "volume": 54230100, "market_cap_b": 2950, "as_of": "2026-06-20T16:00:00Z", "source": "yahoo_finance" }
Deux champs que tout outil finance doit retourner : source (d'où viennent les données) et as_of (quand). Le LLM peut alors dire « D'après Yahoo Finance à 16h00... » au lieu de présenter les données comme des vérités absolues. C'est aussi ce qui alimente la vérification anti-hallucination.
Le transport SSE (Server-Sent Events) est le mode recommandé pour un serveur MCP distant. Il utilise deux endpoints HTTP :
| Endpoint | Méthode | Rôle |
|---|---|---|
/sse | GET | Connexion SSE longue durée. Le serveur pousse les réponses via ce canal. |
/mcp | POST | Le client envoie les requêtes JSON-RPC (initialize, tools/list, tools/call) ici. |
Le flux est asymétrique : le client POST les requêtes, et le serveur répond via SSE. C'est différent d'un WebSocket où les deux côtés envoient sur le même canal.
# Séquence HTTP réelle : # 1. Client ouvre une connexion SSE GET /sse HTTP/1.1 Accept: text/event-stream # 2. Serveur répond avec un stream SSE + un endpoint POST HTTP/1.1 200 OK Content-Type: text/event-stream event: endpoint data: /mcp?session_id=abc123 # 3. Client envoie les requêtes MCP via POST POST /mcp?session_id=abc123 HTTP/1.1 Content-Type: application/json {"jsonrpc":"2.0","method":"tools/list","id":1} # 4. Serveur pousse la réponse via SSE event: message data: {"jsonrpc":"2.0","result":{"tools":[...]},"id":1}
MCP introduit progressivement le transport « Streamable HTTP » qui simplifie le modèle : un seul endpoint POST qui peut optionnellement upgrader vers SSE. Pour cette série, on reste sur SSE car :
Le test le plus rapide : connecter votre serveur à Claude Code en 2 commandes.
# Terminal 1 : lancer le serveur cd mcp-finance-server source .venv/bin/activate python server.py # → INFO: Started server process # → INFO: Uvicorn running on http://0.0.0.0:8080
# Terminal 2 : ajouter le serveur MCP à Claude Code claude mcp add finance-local --transport sse --url http://localhost:8080/sse # Vérifier la connexion claude mcp list # → finance-local sse http://localhost:8080/sse connected
# Dans Claude Code, posez la question : > Quel est le cours d'Apple ? # Claude va automatiquement : # 1. Voir que l'outil GetQuote est disponible # 2. Appeler GetQuote(ticker="AAPL") # 3. Recevoir le JSON et formuler une réponse # Réponse attendue : # "Le cours d'Apple (AAPL) est actuellement à $189.84, # en hausse de +1.23%. Le volume est de 54.2M actions. # Note : ces données sont simulées (source: simulated)."
Parce que le résultat inclut "source": "simulated", Claude signalera automatiquement que les données ne sont pas réelles. C'est exactement le comportement qu'on veut : transparence sur la fraîcheur et la fiabilité des données. En Partie 3, quand on branchera Yahoo Finance, la source deviendra "yahoo_finance" et Claude présentera les données comme fiables.
Pour tester sans Claude Code, utilisez curl :
# Test du health check curl http://localhost:8080/health # → {"status":"ok","tools":1} # Test SSE (Ctrl+C pour arrêter) curl -N http://localhost:8080/sse # → event: endpoint # → data: /mcp?session_id=... # Test d'appel direct (avec le session_id du SSE) curl -X POST http://localhost:8080/mcp?session_id=XXX \ -H "Content-Type: application/json" \ -d '{ "jsonrpc": "2.0", "method": "tools/call", "params": { "name": "GetQuote", "arguments": {"ticker": "AAPL"} }, "id": 1 }'
/health est un bonus utile mais pas requise par le protocole.source et as_of pour la traçabilitéclaude mcp add — pas besoin de client custom