Account
IT
Inizia gratis

Costruire un MCP server in Python che avvolge una web data API

Un MCP server funzionante in 109 righe di Python: tre tool per dati web, errori che dicono all’agente cosa fare e un budget di crediti per sessione. Testato il 19 ott 2026.

Per costruire un MCP server in Python, installa il pacchetto ufficiale mcp (2.3.0 al 19 ottobre 2026), crea un MCPServer, decora le funzioni async con @mcp.tool() e chiama mcp.run() per stdio. Un server che avvolge una REST API è un livello sottile: i type hint diventano lo schema di input, la docstring diventa le istruzioni che il modello legge, e un ToolError sollevato diventa un errore su cui l’agente può agire.

Qui sotto c’è un server completo di 109 righe che avvolge tre endpoint FastAF (ricerca, news, lettura di pagine), con validazione dell’input, errori leggibili dall’agente, timeout e un budget di crediti per sessione. L’ho eseguito via stdio con il client dell’SDK e registrato in Claude Code il 19 ottobre 2026.

Quale pacchetto e quale import si usano oggi?

Installa mcp da PyPI (pip install mcp httpx). La versione 2.3.0 è uscita il 2 ottobre 2026 e richiede Python 3.10 o successivo (PyPI). Il grande cambiamento rispetto all’era 1.x: FastMCP è stato rinominato MCPServer, e from mcp.server.fastmcp import FastMCP ora solleva ModuleNotFoundError. Il nuovo import è from mcp.server.mcpserver import MCPServer (guida alla migrazione). I tutorial più vecchi mostrano ancora il vecchio import.

Com’è fatto il server completo?

Salvalo come fastaf_web.py. Legge FASTAF_API_KEY e usa per default https://api.fastaf.co/v1; FASTAF_BASE_URL lo sovrascrive.

"""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

Tre scelte reggono il peso. Ogni tool restituisce un dizionario ripulito (l’endpoint news include miniature in base64, che il modello non dovrebbe mai vedere). read_page accetta un budget di caratteri e segnala se ha tagliato la pagina. E ogni docstring dichiara il costo e quando usare il tool al posto dei suoi vicini, perché quel testo è tutta la documentazione che il modello riceve.

Come deve un tool MCP segnalare gli errori al modello?

Solleva ToolError con un messaggio che dice cosa fare dopo. L’SDK lo restituisce con is_error=True, quindi il modello legge il tuo testo; qualsiasi altra eccezione è trattata come un crash e il modello vede soltanto Error executing tool <name> (gestione degli errori). Restituire una stringa di errore è la trappola: arriva come risultato riuscito.

La specifica traccia la stessa linea: gli errori di esecuzione dei tool portano “actionable feedback that language models can use to self-correct” (specifica dei tool); gli errori di protocollo sono per richieste che il modello non può correggere.

L’helper call() mappa gli errori di FastAF in istruzioni: 402 insufficient_credits significa fermati e dillo all’utente (riprovare non aiuta); 429 attende Retry-After una volta, poi dice all’agente di rallentare; 400 blocked_target dice che l’URL è bloccato di proposito, quindi cerca un’altra fonte pubblica. Le richieste fallite non vengono mai addebitate. Pydantic rifiuta gli argomenti sbagliati (Field(ge=1, le=10), il pattern ^https?://) prima che parta qualsiasi richiesta HTTP.

Come si mette un tetto ai crediti per sessione?

Prima di ogni chiamata, sotto un asyncio.Lock, controlla spent + cost > BUDGET con il costo di listino (5 per ricerca e news, 1 per uno scrape) e somma il costo al contatore prima del primo await, così le chiamate parallele non passano entrambe. Se non ci sta, solleva un ToolError che dice all’agente di fermarsi e rispondere con quello che ha. Dopo la risposta, sostituisci la prenotazione con l’header X-FastAF-Credits-Used: un errore dell’API non viene addebitato e la restituisce, un timeout lato client la tiene. Un server stdio vive finché il client lo tiene in vita, quindi un processo è una sessione.

Due limiti. Il contatore si azzera se il client riavvia il server (scrivilo su file per renderlo persistente). Ed è una protezione per sessione, non un limite dell’account: il saldo crediti appartiene all’account ed è condiviso da tutte le sue chiavi. Il post sul tetto di spesa tratta gli altri livelli.

Cosa ha fatto quando l’ho eseguito?

Ho avviato il server dal client dell’SDK (Client(StdioServerParameters(...))) e chiamato ogni tool il 19 ottobre 2026:

Chiamata Risultato Tempo Crediti
search_web, 3 risultati titolo, url, snippet in classifica da 2,5 a 3,2 s (tre esecuzioni) 5
read_page sulla guida alla migrazione dell’SDK, max_chars=4000 4.000 di 223.596 caratteri, truncated: true da 0,9 a 1,1 s 1
search_news, 3 risultati titolo, fonte, età, url, snippet da 1,8 a 9,0 s (tre esecuzioni) 5
read_page su http://192.168.1.1/ blocked_target con consiglio meno di 0,1 s 0
results=50, oppure URL ftp:// rifiutato dalla validazione, nessuna chiamata HTTP 0,0 s 0
budget impostato a 6, seconda ricerca rifiutata, nessuna chiamata HTTP; una lettura di pagina da 1 credito è passata comunque 0,0 s 0

Uno stub che restituiva 402 e 429 ha prodotto i messaggi attesi, e un 429 seguito da un 200 è riuscito al nuovo tentativo in 1,03 s (una simulazione, non l’API reale). Il lock è arrivato dopo le prove reali ed è stato verificato solo con uno stub: con un budget di 6, due ricerche parallele ne lasciano passare una.

Un’esecuzione di conferma di search_web e read_page attraverso il server sull’API di produzione è costata 6 crediti: 9,6 s per la ricerca, 0,7 s per la pagina. Al primo tentativo l’API ha risposto 429 due volte di fila, il server ha restituito il consiglio “rate limited” invece di andare in crash, e la riesecuzione è passata.

Come lo registri in Claude Code e Cursor?

Metti il server in un .mcp.json di progetto. Claude Code espande ${VAR} dentro env, quindi il file non contiene nessun segreto (documentazione MCP di Claude Code, controllata il 19 ottobre 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 forma da CLI è claude mcp add --transport stdio --env FASTAF_API_KEY=... fastaf-web -- /path/to/venv/bin/python /path/to/fastaf_web.py, che salva la chiave in ~/.claude.json. Con il .mcp.json qui sopra, Claude Code 2.1.197 in modalità headless ha chiamato search_news e restituito i titoli.

Cursor legge .cursor/mcp.json o ~/.cursor/mcp.json con la stessa forma e risolve ${env:FASTAF_API_KEY} (documentazione di Cursor, controllata il 19 ottobre 2026). Per questo post non ho eseguito Cursor.

Conviene costruirsi il proprio server o usare quello ospitato?

Se vuoi solo i tool nel tuo agente, usa il server ospitato: un URL e un header Bearer, 15 tool, nessun codice da mantenere. Lo tratta il post sulla configurazione ospitata, e la documentazione MCP elenca ogni tool.

Costruiscine uno tuo quando vuoi un set di tool ristretto (tre tool costano al modello meno contesto di quindici), una tua cache o un tuo logging, prompt tarati sul tuo dominio, un budget imposto nel tuo processo, o la composizione con tool interni. Lo stesso wrapper va bene per qualsiasi altro endpoint: vedi le pagine della search API e della web scraping API per i parametri, la documentazione per il contratto completo, e i prezzi per i costi.

La versione breve

Installa mcp 2.3.0, usa MCPServer (non FastMCP), solleva ToolError con un consiglio per l’agente, metti un tetto a caratteri e crediti dentro il server, e registralo con una voce stdio in .mcp.json. Prendi una chiave con 500 crediti gratuiti e fai girare il server qui sopra.