Panel
ES
Empieza gratis

Crea un MCP server en Python que envuelva una API de datos web

Un MCP server funcional en 109 líneas de Python: tres tools de datos web, errores que le dicen al agente qué hacer y un presupuesto de créditos por sesión. Probado el 19 oct 2026.

Para crear un MCP server en Python, instala el paquete oficial mcp (2.3.0 a 19 de octubre de 2026), crea un MCPServer, decora funciones async con @mcp.tool() y llama a mcp.run() para stdio. Un servidor que envuelve una API REST es una capa fina: los type hints se convierten en el esquema de entrada, el docstring en las instrucciones que lee el modelo, y un ToolError lanzado en un error sobre el que el agente puede actuar.

Abajo tienes un servidor completo de 109 líneas que envuelve tres endpoints de FastAF (búsqueda, noticias, lectura de páginas), con validación de entrada, errores legibles para el agente, timeouts y un presupuesto de créditos por sesión. Lo ejecuté por stdio con el cliente del SDK y lo registré en Claude Code el 19 de octubre de 2026.

¿Qué paquete y qué import se usan hoy?

Instala mcp desde PyPI (pip install mcp httpx). La versión 2.3.0 salió el 2 de octubre de 2026 y requiere Python 3.10 o superior (PyPI). El gran cambio respecto a la era 1.x: FastMCP pasó a llamarse MCPServer, y from mcp.server.fastmcp import FastMCP ahora lanza ModuleNotFoundError. El import nuevo es from mcp.server.mcpserver import MCPServer (guía de migración). Los tutoriales antiguos aún muestran el import viejo.

¿Cómo queda el servidor completo?

Guárdalo como fastaf_web.py. Lee FASTAF_API_KEY y usa por defecto https://api.fastaf.co/v1; FASTAF_BASE_URL lo sobrescribe.

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

Tres decisiones cargan con el peso. Cada tool devuelve un dict recortado (el endpoint de noticias incluye miniaturas en base64, que el modelo nunca debería ver). read_page acepta un presupuesto de caracteres e informa de si cortó la página. Y cada docstring indica el coste y cuándo usar la tool en lugar de sus vecinas, porque ese texto es toda la documentación que recibe el modelo.

¿Cómo debe informar una tool de MCP de los errores al modelo?

Lanza ToolError con un mensaje que diga qué hacer a continuación. El SDK lo devuelve con is_error=True, así que el modelo lee tu texto; cualquier otra excepción se trata como un fallo y el modelo solo ve Error executing tool <name> (gestión de errores). Devolver un string de error es la trampa: llega como un resultado exitoso.

La especificación traza la misma línea: los errores de ejecución de una tool llevan “feedback accionable que los modelos de lenguaje pueden usar para corregirse” (especificación de tools); los errores de protocolo son para solicitudes que el modelo no puede arreglar.

El helper call() traduce los errores de FastAF en instrucciones: 402 insufficient_credits significa parar y avisar al usuario (reintentar no puede ayudar); 429 espera una vez el Retry-After y luego le dice al agente que vaya más despacio; 400 blocked_target indica que la URL está bloqueada a propósito, así que busca otra fuente pública. Las solicitudes fallidas nunca se cobran. Pydantic rechaza los argumentos incorrectos (Field(ge=1, le=10), el patrón ^https?://) antes de hacer ninguna solicitud HTTP.

¿Cómo se limitan los créditos por sesión?

Antes de cada llamada, bajo un asyncio.Lock, comprueba spent + cost > BUDGET con el coste de lista (5 para búsqueda y noticias, 1 para un scrape) y suma el coste al contador antes del primer await, para que las llamadas paralelas no pasen las dos. Si no cabe, lanza un ToolError que diga al agente que pare y responda con lo que tiene. Tras la respuesta, cambia la reserva por la cabecera X-FastAF-Credits-Used: un error de la API no se cobra y la devuelve, un timeout del cliente la mantiene. Un servidor stdio vive mientras el cliente lo mantenga, así que un proceso es una sesión.

Dos límites. El contador se reinicia si el cliente reinicia el servidor (guárdalo en un archivo para que persista). Y es una protección por sesión, no un límite de cuenta: el saldo de créditos pertenece a la cuenta y lo comparten todas sus claves. El post del límite de gasto cubre las demás capas.

¿Qué hizo cuando lo ejecuté?

Lancé el servidor desde el cliente del SDK (Client(StdioServerParameters(...))) y llamé a cada tool el 19 de octubre de 2026:

Llamada Resultado Tiempo Créditos
search_web, 3 resultados title, url y snippet ordenados por ranking de 2,5 a 3,2 s (tres ejecuciones) 5
read_page sobre la guía de migración del SDK, max_chars=4000 4.000 de 223.596 caracteres, truncated: true de 0,9 a 1,1 s 1
search_news, 3 resultados title, source, age, url, snippet de 1,8 a 9,0 s (tres ejecuciones) 5
read_page sobre http://192.168.1.1/ blocked_target con consejo menos de 0,1 s 0
results=50, o una URL ftp:// rechazado por la validación, sin llamada HTTP 0,0 s 0
presupuesto fijado en 6, segunda búsqueda rechazada, sin llamada HTTP; una lectura de página de 1 crédito sí pasó 0,0 s 0

Un servidor stub que devolvía 402 y 429 produjo los mensajes esperados, y un 429 seguido de un 200 tuvo éxito al reintentar en 1,03 s (una simulación, no la API real). El lock llegó después de las pruebas reales y se comprobó solo con un stub: con un presupuesto de 6, dos búsquedas paralelas dejan pasar una.

Una ejecución de confirmación de search_web y read_page a través del servidor contra la API de producción costó 6 créditos: 9,6 s la búsqueda, 0,7 s la página. En un primer intento la API respondió 429 dos veces seguidas, el servidor devolvió el consejo de “rate limited” en lugar de fallar, y la repetición pasó.

¿Cómo lo registras en Claude Code y Cursor?

Pon el servidor en un .mcp.json del proyecto. Claude Code expande ${VAR} dentro de env, así que el archivo no contiene ningún secreto (documentación de MCP de Claude Code, consultada el 19 de octubre de 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 por CLI es claude mcp add --transport stdio --env FASTAF_API_KEY=... fastaf-web -- /path/to/venv/bin/python /path/to/fastaf_web.py, que guarda la clave en ~/.claude.json. Con el .mcp.json de arriba, Claude Code 2.1.197 en modo headless llamó a search_news y devolvió los títulos.

Cursor lee .cursor/mcp.json o ~/.cursor/mcp.json con la misma estructura y resuelve ${env:FASTAF_API_KEY} (documentación de Cursor, consultada el 19 de octubre de 2026). No ejecuté Cursor para este post.

¿Conviene construir tu propio servidor o usar el alojado?

Si solo quieres las tools en tu agente, usa el servidor alojado: una URL y una cabecera Bearer, 15 tools, sin código que mantener. El post de configuración del servidor alojado lo explica, y la documentación de MCP lista todas las tools.

Construye el tuyo cuando quieras un conjunto de tools reducido (tres tools le cuestan menos contexto al modelo que quince), tu propia caché o tus propios logs, prompts ajustados a tu dominio, un presupuesto aplicado dentro de tu proceso, o composición con tools internas. El mismo wrapper sirve para cualquier otro endpoint: mira las páginas de la API de búsqueda y de la API de web scraping para los parámetros, la documentación para el contrato completo y los precios para los costes.

En resumen

Instala mcp 2.3.0, usa MCPServer (no FastMCP), lanza ToolError con un consejo para el agente, limita caracteres y créditos dentro del servidor y regístralo con una entrada stdio en .mcp.json. Consigue una clave con 500 créditos gratis y ejecuta el servidor de arriba.