Build an MCP server in Python that wraps a web data API

To build an MCP server in Python, install the official mcp package (2.3.0 as of 19 October 2026), create an MCPServer, decorate async functions with @mcp.tool(), and call mcp.run() for stdio. A server that wraps a REST API is a thin layer: type hints become the input schema, the docstring becomes the instructions the model reads, and a raised ToolError becomes an error the agent can act on.

Below is a complete 109-line server wrapping three FastAF endpoints (search, news, page reading), with input validation, agent-readable errors, timeouts and a per-session credit budget. I ran it over stdio with the SDK’s client and registered it in Claude Code on 19 October 2026.

Which package and import do you use today?

Install mcp from PyPI (pip install mcp httpx). Version 2.3.0 was released on 2 October 2026 and needs Python 3.10 or newer (PyPI). The big change from the 1.x era: FastMCP was renamed MCPServer, and from mcp.server.fastmcp import FastMCP now raises ModuleNotFoundError. The new import is from mcp.server.mcpserver import MCPServer (migration guide). Older tutorials still show the old import.

What does the complete server look like?

Save this as fastaf_web.py. It reads FASTAF_API_KEY and defaults to https://api.fastaf.co/v1; FASTAF_BASE_URL overrides it.

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

Three choices carry the weight. Each tool returns a trimmed dict (the news endpoint includes base64 thumbnails, which the model should never see). read_page takes a character budget and reports whether it cut the page. And every docstring states the cost and when to use the tool instead of its neighbours, because that text is all the documentation the model gets.

How should an MCP tool report errors to the model?

Raise ToolError with a message that says what to do next. The SDK returns it with is_error=True, so the model reads your text; any other exception is treated as a crash and the model sees only Error executing tool <name> (handling errors). Returning an error string is the trap: it arrives as a successful result.

The spec draws the same line: tool execution errors carry “actionable feedback that language models can use to self-correct” (tools spec); protocol errors are for requests the model cannot fix.

The call() helper maps the FastAF errors to instructions: 402 insufficient_credits means stop and tell the user (retrying cannot help); 429 waits for Retry-After once, then tells the agent to slow down; 400 blocked_target says the URL is blocked on purpose, so find another public source. Failed requests are never charged. Pydantic rejects bad arguments (Field(ge=1, le=10), the ^https?:// pattern) before any HTTP request is made.

How do you cap credits per session?

Before each call, under an asyncio.Lock, check spent + cost > BUDGET with the list cost (5 for search and news, 1 for a scrape) and add the cost to the counter before the first await, so parallel calls cannot both pass. If it does not fit, raise a ToolError telling the agent to stop and answer from what it has. After the response, swap the reservation for the X-FastAF-Credits-Used header: an API error is never charged and gives it back, a client timeout keeps it. A stdio server lives as long as the client keeps it, so one process is one session.

Two limits. The counter resets if the client restarts the server (write it to a file to persist it). And it is a per-session guard, not an account limit: the credit balance belongs to the account and is shared by all its keys. The spending cap post covers the other layers.

What did it do when I ran it?

I launched the server from the SDK client (Client(StdioServerParameters(...))) and called each tool on 19 October 2026:

Call	Result	Time	Credits
`search_web`, 3 results	ranked title, url, snippet	2.5 to 3.2 s (three runs)	5
`read_page` on the SDK migration guide, `max_chars=4000`	4,000 of 223,596 characters, `truncated: true`	0.9 to 1.1 s	1
`search_news`, 3 results	title, source, age, url, snippet	1.8 to 9.0 s (three runs)	5
`read_page` on `http://192.168.1.1/`	`blocked_target` with advice	under 0.1 s	0
`results=50`, or `ftp://` URL	rejected by validation, no HTTP call	0.0 s	0
budget set to 6, second search	refused, no HTTP call; a 1-credit page read still passed	0.0 s	0

A stub server returning 402 and 429 produced the expected messages, and a 429 followed by 200 succeeded on retry in 1.03 s (a simulation, not the live API). The lock came after the live runs and was checked against a stub only: with a budget of 6, two parallel searches let one through.

A confirmation run of search_web and read_page through the server on the production API cost 6 credits: 9.6 s for the search, 0.7 s for the page. On a first attempt the API answered 429 twice in a row, the server returned the “rate limited” advice instead of crashing, and the rerun passed.

How do you register it in Claude Code and Cursor?

Put the server in a project .mcp.json. Claude Code expands ${VAR} inside env, so the file holds no secret (Claude Code MCP docs, checked 19 October 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"
      }
    }
  }
}

The CLI form is claude mcp add --transport stdio --env FASTAF_API_KEY=... fastaf-web -- /path/to/venv/bin/python /path/to/fastaf_web.py, which stores the key in ~/.claude.json. With the .mcp.json above, Claude Code 2.1.197 in headless mode called search_news and returned the titles.

Cursor reads .cursor/mcp.json or ~/.cursor/mcp.json with the same shape and resolves ${env:FASTAF_API_KEY} (Cursor docs, checked 19 October 2026). I did not run Cursor for this post.

Should you build your own server or use the hosted one?

If you just want the tools in your agent, use the hosted server: one URL and a Bearer header, 15 tools, no code to maintain. The hosted setup post covers it, and the MCP docs list every tool.

Build your own when you want a narrow toolset (three tools cost the model less context than fifteen), your own caching or logging, prompts tuned to your domain, a budget enforced in your process, or composition with internal tools. The same wrapper fits any other endpoint: see the search API and web scraping API pages for parameters, the docs for the full contract, and pricing for costs.

The short version

Install mcp 2.3.0, use MCPServer (not FastMCP), raise ToolError with advice for the agent, cap characters and credits inside the server, and register it with a stdio entry in .mcp.json. Get a key with 500 free credits and run the server above.