Weather MCP (Python: High-Level and Low-Level servers)

Minimal MCP server exposing a weather.today(city) tool, a weather://today/{city} resource and a weather.ensureCity prompt. Includes both a FastMCP-based server (High-Level) and a Low-Level server to demonstrate Nevermined Payments integration.

About this demo

This repository is a reference/demo project used to test and validate the Model Context Protocol (MCP) integration inside Nevermined's Python SDK payments-py. It showcases how to protect MCP tools, resources and prompts with the paywall, both in a High‑Level (FastMCP) server and a Low‑Level JSON‑RPC server. It is intended for examples, local experimentation and integration tests, not as production‑ready code.

Requirements

Python >= 3.10
Poetry (recommended) or pip

Install

poetry install
# or
pip install -e .

Run the server (High-Level: FastMCP over HTTP)

poetry run uvicorn weather_mcp_py.app:create_app \
  --factory --host 127.0.0.1 --port 8000

Environment (server):

export NVM_SERVER_API_KEY=...     # Server key (builder/agent owner)
export NVM_AGENT_ID=weather-agent # Logical agent id used in validation (or your real ID)
export NVM_ENV=staging_sandbox    # optional (staging_sandbox | production)
poetry run uvicorn weather_mcp_py.app:create_app --factory --host 127.0.0.1 --port 8000

Run the server (Low-Level)

export MCP_SERVER_MODE=low
poetry run uvicorn weather_mcp_py.app:create_app \
  --factory --host 127.0.0.1 --port 8000

High-Level endpoint: POST /mcp (JSON‑RPC; stateful HTTP via headers)
Low-Level endpoint: POST /mcp-low (raw JSON‑RPC; path is arbitrary, client default uses /mcp-low)
Health: GET /health

Client demo (High-Level)

# Defaults: MCP_BASE_URL=http://localhost:8000, city "Madrid"
poetry run weather-mcp

# Custom base URL and city
MCP_BASE_URL=http://localhost:8000 MCP_CITY=Paris poetry run weather-mcp

Client environment:

export MCP_BASE_URL=http://localhost:8000
export NVM_API_KEY=...            # Subscriber key
export NVM_PLAN_ID=...            # Plan that grants access
export NVM_AGENT_ID=...           # Agent id associated to the plan
poetry run weather-mcp

Client demo (Low-Level)

# Defaults: MCP_LOW_ENDPOINT=http://localhost:8000/mcp-low, city "Madrid"
poetry run weather-mcp-low

# Custom endpoint and city
MCP_LOW_ENDPOINT=http://localhost:8000/mcp-low MCP_CITY=Paris poetry run weather-mcp-low

Tests

poetry run pytest -q

Nevermined auth

The client obtains an access token with its NVM_API_KEY and sends it as Authorization: Bearer .... The server requires Authorization and performs validation via the paywall. If unauthorized, the server responds with a custom JSON‑RPC error -32003.

Server env (recap):

export NVM_SERVER_API_KEY=...     # Server key (builder/agent owner)
export NVM_AGENT_ID=weather-agent # Logical agent id used in validation (or your real ID)
export NVM_ENV=staging_sandbox    # optional

Client env (recap):

export NVM_API_KEY=...            # Subscriber key
export NVM_PLAN_ID=...            # Plan that grants access
export NVM_AGENT_ID=...           # Agent id associated to the plan

MCP Inspector (over HTTP)

If you have Node.js available, you can use the MCP Inspector to connect to the High-Level server:

npx @modelcontextprotocol/inspector connect http://localhost:8000/mcp

Note: Inspector requests typically do not include Authorization headers; use the Python clients above for auth tests.

Endpoints

High-Level (FastMCP):
- POST /mcp — JSON‑RPC requests (initialize handled here; server-side session via headers)
- GET /mcp — SSE stream for server notifications (if supported by the client/tooling)
- GET /health — simple health check
Low-Level:
- POST /mcp-low — Minimal JSON‑RPC with Authorization header passthrough
- GET /health — simple health check

Context and Authorization: FastMCP vs Low‑Level

FastMCP (High‑Level): the paywall obtains the context automatically because it is configured with FastMCP's getContext. You do not need to pass extra to the handler.

# FastMCP: build context automatically via FastMCP.get_context
from mcp.server.fastmcp import FastMCP
from payments_py.payments import Payments

fastmcp = FastMCP(name="weather-mcp", json_response=True)
payments = Payments({"nvm_api_key": NVM_SERVER_API_KEY, "environment": NVM_ENV})

payments.mcp.configure({
    "agentId": NVM_AGENT_ID,
    "serverName": "weather-mcp",
    "getContext": fastmcp.get_context,  # Paywall builds extra from this
})

protected_tool = payments.mcp.with_paywall(
    weather_tool_handler,
    {"kind": "tool", "name": "weather.today", "credits": weather_tool_credits_calculator},
)

# When calling, do NOT pass extra; paywall resolves it from FastMCP context
res = await protected_tool({"city": city})

Low‑Level: you must pass extra explicitly to the handler. Typically you capture the request and the Authorization header in middleware or directly in the ASGI app and build extra with build_extra_from_http_headers.

# Low-Level: build extra from HTTP headers and pass it to the handler
from payments_py.mcp import build_extra_from_http_headers

async def asgi_app(scope, receive, send):
    # 1) Collect headers into a dict
    headers_list = scope.get("headers", [])  # List[Tuple[bytes, bytes]]
    headers = {k.decode(): v.decode() for k, v in headers_list}

    # 2) Build extra (contains requestInfo + auth token extraction helpers)
    extra = build_extra_from_http_headers(headers)

    # 3) Route and call a tool handler passing extra as second arg
    result = await protected_tool({"city": "Madrid"}, extra)
    # ... return JSON-RPC result ...

In this app, the Low‑Level variant already implements this pattern inside the ASGI (server_lowlevel.py): it builds extra from the headers and passes it as the second parameter to the handler (handler(args, extra)).

Additional note (resources in FastMCP): when invoking a protected resource from FastMCP, you can pass None as extra and the paywall will resolve the context via the configured getContext.

Acceptance checklist

List Tools shows weather.today
Calling weather.today with { "city": "Madrid" } returns a text summary and a resource_link to weather://today/Madrid
Reading that resource returns JSON with the TodayWeather fields

Notes

Inspector requests do not include Authorization; prefer the demo clients when testing paywall.

Tutorial: Protecting an MCP server with Nevermined (Paywall + Credits Burn)

This guide shows how to protect your MCP tools with Nevermined so that only subscribed users can access them, and how to burn credits after each call. The Python SDK mirrors the TypeScript flow but with Python naming.

1) Initialize Nevermined in your MCP server

import os
from payments_py.payments import Payments

payments = Payments({
    "nvm_api_key": os.environ["NVM_SERVER_API_KEY"],
    "environment": os.environ.get("NVM_ENV", "staging_sandbox"),
})

# Configure paywall defaults once
payments.mcp.configure({
    "agentId": os.environ["NVM_AGENT_ID"],
    "serverName": "weather-mcp",
})

2) Wrap your handlers with the paywall (works in both servers)

# Your original tool handler
async def my_handler(args):
    return {"content": [{"type": "text", "text": "Hello World"}]}

# Protect it with paywall (single call). Burn 1 credit per call
protected_handler = payments.mcp.with_paywall(
    my_handler,
    {"kind": "tool", "name": "my.namespace.tool", "credits": 1},
)

# Low-Level server
server.registerTool("my.namespace.tool", {"title": "My Tool"}, protected_handler)

# High-Level (FastMCP): call the protected handler from your decorated tool
@fastmcp.tool(name="my.namespace.tool", title="My Tool")
async def _tool(arg: str | None = None) -> str:
    res = await protected_handler({"arg": arg or ""})
    # Extract text content
    for c in (res.get("content") or []):
        if isinstance(c, dict) and c.get("type") == "text" and isinstance(c.get("text"), str):
            return c["text"]
    return str(res)

What the paywall does:

Extracts Authorization from the MCP HTTP headers automatically.
Validates access with Nevermined.
If unauthorized, responds with a JSON‑RPC error -32003 (and suggests plans when possible).
Runs your handler.
Burns credits after the call based on the credits option.

3) Client side

Use the Nevermined client to obtain an access token and pass it as Authorization to your MCP transport.

from payments_py.payments import Payments

subs_payments = Payments({
    "nvm_api_key": os.environ["NVM_API_KEY"],
    "environment": os.environ.get("NVM_ENV", "staging_sandbox"),
})

creds = subs_payments.agents.get_agent_access_token(
    os.environ["NVM_PLAN_ID"], os.environ["NVM_AGENT_ID"],
)
access_token = creds.get("accessToken")

# Send Authorization: Bearer {access_token} in your HTTP client

4) Error semantics

Missing token → JSON‑RPC -32003 (“Authorization required”).
Invalid/not subscribed → JSON‑RPC -32003 (“Payment required”, optionally with plan suggestions).
Network/other errors → JSON‑RPC -32002.

5) Advanced

Customize credits to a function that receives a context { args, result, request } and returns an int.
Use payments.mcp.with_paywall to protect tools, resources, and prompts.

Example: dynamic credits for tool calls (e.g., random 1..10 credits per call):

import random

def dynamic_credits(_ctx):
    return 1 + int(random.random() * 10)

protected = payments.mcp.with_paywall(
    my_handler,
    {"kind": "tool", "name": "my.namespace.tool", "credits": dynamic_credits},
)

Example: burn 1 credit for resource reads (manual control, without the paywall wrapper):

from urllib.parse import urlparse

async def resource_handler(uri, variables, extra):
    headers = (extra or {}).get("requestInfo", {}).get("headers", {})
    raw = headers.get("authorization") or headers.get("Authorization")
    if not raw:
        raise {"code": -32003, "message": "Authorization required"}
    token = raw[7:].strip() if raw.startswith("Bearer ") else raw

    logical_url = f"mcp://weather-mcp/resources/weather-today?city={variables.get('city', [''])[0]}"
    agent_id = os.environ["NVM_AGENT_ID"]

    start = payments.requests.start_processing_request(agent_id, token, logical_url, "GET")
    if not (start or {}).get("balance", {}).get("isSubscriber"):
        raise {"code": -32003, "message": "Payment required"}

    # ... build the JSON body for the resource ...
    body = {"ok": True}

    payments.requests.redeem_credits_from_request(start["agentRequestId"], token, 1)
    return {"contents": [{"uri": uri.geturl(), "mimeType": "application/json", "text": json.dumps(body)}]}

In most cases, prefer the with_paywall wrapper which authenticates and redeems for you and supports streaming (async iterables).

Weather MCP Python