Dashboard
DE
Gratis starten

MCP-Server in Python bauen: eine Web-Daten-API als Agenten-Tools

Ein funktionierender MCP-Server in 109 Zeilen Python: drei Web-Daten-Tools, Fehler, die dem Agenten sagen, was zu tun ist, und ein Credit-Budget pro Session. Getestet am 19. Okt 2026.

Um einen MCP-Server in Python zu bauen, installierst du das offizielle Paket mcp (2.3.0 am 19. Oktober 2026), erzeugst einen MCPServer, dekorierst async-Funktionen mit @mcp.tool() und rufst für stdio mcp.run() auf. Ein Server, der eine REST-API umhüllt, ist eine dünne Schicht: Typ-Hints werden zum Input-Schema, der Docstring wird zur Anweisung, die das Modell liest, und ein ausgelöster ToolError wird zu einem Fehler, mit dem der Agent etwas anfangen kann.

Unten steht ein vollständiger Server mit 109 Zeilen, der drei FastAF-Endpoints umhüllt (Suche, News, Seiten lesen), mit Eingabevalidierung, für Agenten lesbaren Fehlern, Timeouts und einem Credit-Budget pro Session. Ich habe ihn am 19. Oktober 2026 über stdio mit dem Client des SDK laufen lassen und in Claude Code registriert.

Welches Paket und welchen Import nutzt man heute?

Installier mcp von PyPI (pip install mcp httpx). Version 2.3.0 erschien am 2. Oktober 2026 und braucht Python 3.10 oder neuer (PyPI). Die große Änderung gegenüber der 1.x-Zeit: FastMCP wurde in MCPServer umbenannt, und from mcp.server.fastmcp import FastMCP wirft jetzt ModuleNotFoundError. Der neue Import ist from mcp.server.mcpserver import MCPServer (Migrationsguide). Ältere Tutorials zeigen noch den alten Import.

Wie sieht der vollständige Server aus?

Speicher das als fastaf_web.py. Er liest FASTAF_API_KEY und nutzt standardmäßig https://api.fastaf.co/v1; FASTAF_BASE_URL überschreibt das.

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

Drei Entscheidungen tragen das Ganze. Jedes Tool gibt ein gekürztes Dict zurück (der News-Endpoint enthält Base64-Thumbnails, die das Modell nie sehen sollte). read_page nimmt ein Zeichenbudget und meldet, ob die Seite gekappt wurde. Und jeder Docstring nennt die Kosten und sagt, wann man das Tool statt seiner Nachbarn nimmt, denn dieser Text ist die gesamte Dokumentation, die das Modell bekommt.

Wie sollte ein MCP-Tool dem Modell Fehler melden?

Löse ToolError aus, mit einer Meldung, die sagt, was als Nächstes zu tun ist. Das SDK gibt ihn mit is_error=True zurück, das Modell liest also deinen Text; jede andere Exception gilt als Absturz, und das Modell sieht nur Error executing tool <name> (Fehlerbehandlung). Einen Fehlerstring zurückzugeben ist die Falle: Er kommt als erfolgreiches Ergebnis an.

Die Spec zieht dieselbe Linie: Fehler bei der Tool-Ausführung tragen „actionable feedback that language models can use to self-correct” (Tools-Spec); Protokollfehler sind für Requests, die das Modell nicht beheben kann.

Der Helfer call() übersetzt die FastAF-Fehler in Anweisungen: 402 insufficient_credits heißt anhalten und es dem Nutzer sagen (ein Retry kann nicht helfen); 429 wartet einmal auf Retry-After und sagt dem Agenten dann, er solle langsamer werden; 400 blocked_target besagt, dass die URL absichtlich gesperrt ist, such also eine andere öffentliche Quelle. Fehlgeschlagene Requests werden nie berechnet. Pydantic lehnt falsche Argumente (Field(ge=1, le=10), das Muster ^https?://) ab, bevor ein HTTP-Request abgeschickt wird.

Wie begrenzt man Credits pro Session?

Prüf vor jedem Aufruf unter einem asyncio.Lock spent + cost > BUDGET mit den Listenkosten (5 für Suche und News, 1 für einen Scrape) und addiere die Kosten vor dem ersten await zum Zähler, damit parallele Aufrufe nicht beide durchkommen. Passt es nicht, löse einen ToolError aus, der dem Agenten sagt, er solle aufhören und mit dem antworten, was er hat. Nach der Antwort tausche die Reservierung gegen den Header X-FastAF-Credits-Used: Ein API-Fehler wird nicht berechnet und gibt sie zurück, ein Timeout beim Client behält sie. Ein stdio-Server lebt so lange, wie der Client ihn behält, ein Prozess ist also eine Session.

Zwei Grenzen. Der Zähler setzt sich zurück, wenn der Client den Server neu startet (schreib ihn in eine Datei, um ihn zu behalten). Und es ist eine Absicherung pro Session, kein Limit des Accounts: Das Credit-Guthaben gehört zum Account und wird von allen seinen Keys geteilt. Der Beitrag zum Ausgabenlimit behandelt die anderen Ebenen.

Was hat er beim Ausführen getan?

Ich habe den Server aus dem SDK-Client gestartet (Client(StdioServerParameters(...))) und am 19. Oktober 2026 jedes Tool aufgerufen:

Aufruf Ergebnis Zeit Credits
search_web, 3 Ergebnisse Rang, Titel, URL, Snippet 2,5 bis 3,2 s (drei Läufe) 5
read_page auf den SDK-Migrationsguide, max_chars=4000 4.000 von 223.596 Zeichen, truncated: true 0,9 bis 1,1 s 1
search_news, 3 Ergebnisse Titel, Quelle, Alter, URL, Snippet 1,8 bis 9,0 s (drei Läufe) 5
read_page auf http://192.168.1.1/ blocked_target mit Hinweis unter 0,1 s 0
results=50 oder ftp://-URL von der Validierung abgelehnt, kein HTTP-Aufruf 0,0 s 0
Budget auf 6 gesetzt, zweite Suche abgelehnt, kein HTTP-Aufruf; ein Seitenabruf für 1 Credit ging trotzdem durch 0,0 s 0

Ein Stub-Server, der 402 und 429 lieferte, erzeugte die erwarteten Meldungen, und ein 429 gefolgt von 200 klappte beim Retry in 1,03 s (eine Simulation, nicht die Live-API). Der Lock kam nach den Live-Läufen dazu und wurde nur gegen einen Stub geprüft: Bei einem Budget von 6 lassen zwei parallele Suchen eine durch.

Ein Bestätigungslauf von search_web und read_page durch den Server gegen die Produktions-API kostete 6 Credits: 9,6 s für die Suche, 0,7 s für die Seite. Beim ersten Versuch antwortete die API zweimal hintereinander mit 429, der Server gab den Hinweis „rate limited” zurück, statt abzustürzen, und der erneute Lauf ging durch.

Wie registrierst du ihn in Claude Code und Cursor?

Trag den Server in eine .mcp.json im Projekt ein. Claude Code expandiert ${VAR} in env, die Datei enthält also kein Secret (MCP-Docs von Claude Code, geprüft am 19. Oktober 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"
      }
    }
  }
}

Die CLI-Form ist claude mcp add --transport stdio --env FASTAF_API_KEY=... fastaf-web -- /path/to/venv/bin/python /path/to/fastaf_web.py, die den Key in ~/.claude.json speichert. Mit der .mcp.json oben rief Claude Code 2.1.197 im Headless-Modus search_news auf und lieferte die Titel.

Cursor liest .cursor/mcp.json oder ~/.cursor/mcp.json in derselben Form und löst ${env:FASTAF_API_KEY} auf (Cursor-Docs, geprüft am 19. Oktober 2026). Cursor habe ich für diesen Beitrag nicht ausgeführt.

Baust du besser einen eigenen Server oder nutzt den gehosteten?

Wenn du nur die Tools in deinem Agenten willst, nimm den gehosteten Server: eine URL und ein Bearer-Header, 15 Tools, kein Code zu pflegen. Der Beitrag zum gehosteten Setup behandelt das, und die MCP-Docs listen jedes Tool auf.

Bau deinen eigenen, wenn du ein schmales Toolset willst (drei Tools kosten das Modell weniger Kontext als fünfzehn), eigenes Caching oder Logging, auf deine Domäne abgestimmte Prompts, ein in deinem Prozess durchgesetztes Budget oder die Kombination mit internen Tools. Derselbe Wrapper passt zu jedem anderen Endpoint: Parameter stehen auf den Seiten zur Such-API und zur Web-Scraping-API, den vollständigen Vertrag findest du in den Docs, Kosten in den Preisen.

Die Kurzfassung

Installier mcp 2.3.0, nimm MCPServer (nicht FastMCP), löse ToolError mit Hinweisen für den Agenten aus, begrenze Zeichen und Credits im Server und registrier ihn mit einem stdio-Eintrag in .mcp.json. Hol dir einen Key mit 500 kostenlosen Credits und lass den Server von oben laufen.