Créer un serveur MCP en Python qui encapsule une API de données web

Pour créer un serveur MCP en Python, installez le paquet officiel mcp (2.3.0 au 19 octobre 2026), créez un MCPServer, décorez des fonctions async avec @mcp.tool() et appelez mcp.run() pour stdio. Un serveur qui encapsule une API REST est une couche fine : les annotations de type deviennent le schéma d’entrée, la docstring devient les instructions que lit le modèle, et un ToolError levé devient une erreur sur laquelle l’agent peut agir.

Voici un serveur complet de 109 lignes qui encapsule trois endpoints FastAF (recherche, actualités, lecture de page), avec validation des entrées, erreurs lisibles par l’agent, timeouts et un budget de crédits par session. Je l’ai exécuté en stdio avec le client du SDK et enregistré dans Claude Code le 19 octobre 2026.

Quel paquet et quel import utiliser aujourd’hui ?

Installez mcp depuis PyPI (pip install mcp httpx). La version 2.3.0 est sortie le 2 octobre 2026 et demande Python 3.10 ou plus récent (PyPI). Le grand changement depuis l’ère 1.x : FastMCP a été renommé MCPServer, et from mcp.server.fastmcp import FastMCP lève désormais ModuleNotFoundError. Le nouvel import est from mcp.server.mcpserver import MCPServer (guide de migration). Les tutoriels plus anciens montrent encore l’ancien import.

À quoi ressemble le serveur complet ?

Enregistrez ceci sous fastaf_web.py. Il lit FASTAF_API_KEY et utilise par défaut https://api.fastaf.co/v1 ; FASTAF_BASE_URL le remplace.

"""A small MCP server that wraps three FastAF endpoints. Run: python fastaf_web.py"""
import asyncio
import os
from typing import Annotated, Any

import httpx
from mcp.server.mcpserver import MCPServer
from mcp.server.mcpserver.exceptions import ToolError
from mcp.types import ToolAnnotations
from pydantic import Field

BASE = os.environ.get("FASTAF_BASE_URL", "https://api.fastaf.co/v1")
BUDGET = int(os.environ.get("FASTAF_SESSION_BUDGET", "100"))  # credits per server process
COST = {"serp": 5, "news": 5, "scrape": 1}  # published list costs, used to reserve
spent = 0
lock = asyncio.Lock()

mcp = MCPServer("fastaf-web")
http = httpx.AsyncClient(
    base_url=BASE,
    headers={"Authorization": f"Bearer {os.environ['FASTAF_API_KEY']}"},
    timeout=httpx.Timeout(45.0, connect=5.0),
)
READ_ONLY = ToolAnnotations(read_only_hint=True, open_world_hint=True)


async def call(endpoint: str, **params: Any) -> httpx.Response:
    """GET an endpoint, reserve its cost in the session budget, turn failures into advice for the model."""
    global spent
    cost = COST[endpoint]
    async with lock:  # reserve before the await
        if spent + cost > BUDGET:
            raise ToolError(
                f"Session budget reached ({spent} of {BUDGET} credits used). Do not call "
                "more tools; answer from what you already have and say the budget ran out."
            )
        spent += cost
    for attempt in (1, 2):
        try:
            r = await http.get(f"/{endpoint}", params=params)
        except httpx.TimeoutException:
            raise ToolError("Timed out after 45 s. Retry once, then try a different query.")
        if r.status_code == 429 and attempt == 1:  # one polite retry, as the API asks
            await asyncio.sleep(min(float(r.headers.get("Retry-After", 2)), 10))
            continue
        break
    # API errors are never charged; a timeout keeps the reservation.
    spent += (int(r.headers.get("X-FastAF-Credits-Used", cost)) if r.status_code == 200 else 0) - cost
    if r.status_code == 200:
        return r
    code = r.json()["error"]["code"] if "json" in r.headers.get("content-type", "") else "unknown"
    if r.status_code == 402:
        advice = "The account is out of credits. Stop calling this server and tell the user to top up."
    elif r.status_code == 429:
        advice = "Rate limited even after one retry. Wait a few seconds, then call again, one request at a time."
    elif code == "blocked_target":
        advice = "That URL is blocked on purpose (private address or login page). Use a different public source."
    elif r.status_code == 400:
        advice = "The request was rejected. Check the arguments and try a simpler call."
    else:
        advice = "Temporary failure on the API side, not charged. Retry once."
    raise ToolError(f"{endpoint} failed: HTTP {r.status_code} {code}. {advice}")


@mcp.tool(annotations=READ_ONLY)
async def search_web(
    query: Annotated[str, Field(min_length=1, max_length=300, description="Plain search query, as typed into Google")],
    results: Annotated[int, Field(ge=1, le=10)] = 5,
) -> dict[str, Any]:
    """Search Google and return ranked results (title, url, snippet). Costs 5 credits per call.
    Use it to find URLs; it does not return page content. Follow up with read_page on the
    one or two most relevant URLs instead of searching again with a near-identical query."""
    r = await call("serp", query=query, results=results)
    hits = [{"rank": o["rank"], "title": o["title"], "url": o["link"], "snippet": o.get("snippet", "")}
            for o in r.json().get("organic_results", [])[:results]]
    return {"results": hits, "session_credits_used": spent}


@mcp.tool(annotations=READ_ONLY)
async def search_news(
    query: Annotated[str, Field(min_length=1, max_length=300, description="Topic or entity to find recent news about")],
    results: Annotated[int, Field(ge=1, le=10)] = 5,
) -> dict[str, Any]:
    """Search Google News for recent articles (title, source, age, url, snippet). Costs 5
    credits per call. Use it for anything that happened in the last weeks; use search_web
    for documentation and evergreen pages."""
    r = await call("news", query=query, results=results)
    items = [{"title": n["title"], "source": n.get("source"), "age": n.get("lastUpdated"),
              "url": n["url"], "snippet": n.get("snippet", "")}
             for n in r.json().get("news_results", [])[:results]]
    return {"results": items, "session_credits_used": spent}


@mcp.tool(annotations=READ_ONLY)
async def read_page(
    url: Annotated[str, Field(pattern=r"^https?://", max_length=2000, description="Public http(s) page URL")],
    max_chars: Annotated[int, Field(ge=1000, le=30000, description="Character budget for the returned text")] = 8000,
) -> dict[str, Any]:
    """Fetch a public web page and return it as markdown. Costs 1 credit. If 'truncated' is true,
    only the first max_chars characters came back: raise max_chars (up to 30000) only if
    the part you need is missing. Cannot read pages behind logins or paywalls."""
    r = await call("scrape", url=url, formats="markdown")
    text = r.text
    return {"url": url, "text": text[:max_chars], "truncated": len(text) > max_chars,
            "total_chars": len(text), "session_credits_used": spent}


if __name__ == "__main__":
    mcp.run()  # stdio; never print() to stdout here, it carries the protocol

Trois choix portent l’essentiel. Chaque outil renvoie un dict allégé (l’endpoint actualités inclut des miniatures en base64 que le modèle ne devrait jamais voir). read_page prend un budget de caractères et signale s’il a coupé la page. Et chaque docstring indique le coût et quand utiliser l’outil plutôt que ses voisins, car ce texte est toute la documentation que reçoit le modèle.

Comment un outil MCP doit-il signaler une erreur au modèle ?

Levez ToolError avec un message qui dit quoi faire ensuite. Le SDK le renvoie avec is_error=True, donc le modèle lit votre texte ; toute autre exception est traitée comme un plantage et le modèle ne voit que Error executing tool <name> (gestion des erreurs). Renvoyer une chaîne d’erreur est le piège : elle arrive comme un résultat réussi.

La spécification trace la même ligne : les erreurs d’exécution d’un outil portent un « retour exploitable que les modèles de langage peuvent utiliser pour se corriger » (spécification des outils) ; les erreurs de protocole sont réservées aux requêtes que le modèle ne peut pas corriger.

Le helper call() traduit les erreurs FastAF en instructions : 402 insufficient_credits veut dire s’arrêter et prévenir l’utilisateur (réessayer ne sert à rien) ; 429 attend Retry-After une fois, puis dit à l’agent de ralentir ; 400 blocked_target indique que l’URL est bloquée volontairement, il faut donc trouver une autre source publique. Les requêtes échouées ne sont jamais facturées. Pydantic rejette les mauvais arguments (Field(ge=1, le=10), le motif ^https?://) avant tout appel HTTP.

Comment plafonner les crédits par session ?

Avant chaque appel, sous un asyncio.Lock, vérifiez spent + cost > BUDGET avec le coût public (5 pour la recherche et les actualités, 1 pour un scrape) et ajoutez le coût au compteur avant le premier await, pour que deux appels parallèles ne passent pas tous les deux. Si cela ne tient pas, levez un ToolError qui dit à l’agent de s’arrêter et de répondre avec ce qu’il a. Après la réponse, remplacez la réservation par l’en-tête X-FastAF-Credits-Used : une erreur de l’API n’est pas facturée et la rend, un timeout côté client la garde. Un serveur stdio vit aussi longtemps que le client le garde, donc un processus égale une session.

Deux limites. Le compteur repart à zéro si le client relance le serveur (écrivez-le dans un fichier pour le conserver). Et c’est une protection par session, pas une limite de compte : le solde de crédits appartient au compte et est partagé par toutes ses clés. L’article sur le plafond de dépense couvre les autres couches.

Qu’a-t-il fait quand je l’ai exécuté ?

J’ai lancé le serveur depuis le client du SDK (Client(StdioServerParameters(...))) et appelé chaque outil le 19 octobre 2026 :

Appel	Résultat	Durée	Crédits
`search_web`, 3 résultats	titre, url, extrait classés	2,5 à 3,2 s (trois runs)	5
`read_page` sur le guide de migration du SDK, `max_chars=4000`	4 000 sur 223 596 caractères, `truncated: true`	0,9 à 1,1 s	1
`search_news`, 3 résultats	titre, source, âge, url, extrait	1,8 à 9,0 s (trois runs)	5
`read_page` sur `http://192.168.1.1/`	`blocked_target` avec conseil	moins de 0,1 s	0
`results=50`, ou URL `ftp://`	rejeté par la validation, aucun appel HTTP	0,0 s	0
budget fixé à 6, deuxième recherche	refusée, aucun appel HTTP ; une lecture de page à 1 crédit est passée	0,0 s	0

Un serveur stub renvoyant 402 et 429 a produit les messages attendus, et un 429 suivi d’un 200 a réussi au nouvel essai en 1,03 s (une simulation, pas l’API réelle). Le verrou est arrivé après les essais réels et n’a été vérifié qu’avec un stub : avec un budget de 6, deux recherches parallèles en laissent passer une.

Une exécution de confirmation de search_web et read_page via le serveur sur l’API de production a coûté 6 crédits : 9,6 s pour la recherche, 0,7 s pour la page. Au premier essai, l’API a répondu 429 deux fois de suite, le serveur a renvoyé le conseil « rate limited » au lieu de planter, et la relance est passée.

Comment l’enregistrer dans Claude Code et Cursor ?

Mettez le serveur dans un .mcp.json de projet. Claude Code développe ${VAR} dans env, donc le fichier ne contient aucun secret (documentation MCP de Claude Code, vérifiée le 19 octobre 2026) :

{
  "mcpServers": {
    "fastaf-web": {
      "type": "stdio",
      "command": "/path/to/venv/bin/python",
      "args": ["/path/to/fastaf_web.py"],
      "env": {
        "FASTAF_API_KEY": "${FASTAF_API_KEY}",
        "FASTAF_SESSION_BUDGET": "50"
      }
    }
  }
}

La forme CLI est claude mcp add --transport stdio --env FASTAF_API_KEY=... fastaf-web -- /path/to/venv/bin/python /path/to/fastaf_web.py, qui stocke la clé dans ~/.claude.json. Avec le .mcp.json ci-dessus, Claude Code 2.1.197 en mode headless a appelé search_news et renvoyé les titres.

Cursor lit .cursor/mcp.json ou ~/.cursor/mcp.json avec la même forme et résout ${env:FASTAF_API_KEY} (documentation Cursor, vérifiée le 19 octobre 2026). Je n’ai pas lancé Cursor pour cet article.

Faut-il construire son propre serveur ou utiliser celui qui est hébergé ?

Si vous voulez simplement les outils dans votre agent, utilisez le serveur hébergé : une URL et un en-tête Bearer, 15 outils, aucun code à maintenir. L’article sur la configuration hébergée le couvre, et la documentation MCP liste chaque outil.

Construisez le vôtre quand vous voulez un jeu d’outils restreint (trois outils coûtent moins de contexte au modèle que quinze), votre propre cache ou journalisation, des prompts adaptés à votre domaine, un budget imposé dans votre processus, ou une composition avec des outils internes. Le même wrapper convient à n’importe quel autre endpoint : voir les pages API de recherche et API de web scraping pour les paramètres, la documentation pour le contrat complet, et les tarifs pour les coûts.

En bref

Installez mcp 2.3.0, utilisez MCPServer (pas FastMCP), levez ToolError avec des conseils pour l’agent, plafonnez les caractères et les crédits à l’intérieur du serveur, et enregistrez-le avec une entrée stdio dans .mcp.json. Obtenez une clé avec 500 crédits gratuits et lancez le serveur ci-dessus.