Gate a FastMCP Server to Paying Customers sold tiered access: Basic and Pro plans. Selling tiers only matters if the tiers actually differ, and the most common difference is how much a customer may call. This article adds per-plan rate limiting to a FastMCP server: a middleware reads each caller’s plan from their verified token and enforces that plan’s request quota, keyed on the caller’s identity. A Basic customer gets a small quota, a Pro customer a larger one, and over-limit calls are rejected cleanly.

FastMCP ships global and per-client rate limiters, but a quota that varies by plan needs a small custom middleware. You will write one on top of FastMCP’s Middleware base and its RateLimitError, back it with a deterministic sliding-window limiter, and prove the per-plan difference over HTTP. The stack is Mac-native: uv, make, and a launchd service.

What you will build

  • auth.py: API-key auth that tags each caller with a plan (condensed from the license-gating article).
  • ratelimit.py: a sliding-window limiter and a PlanRateLimitMiddleware that applies the caller’s plan quota, keyed on client_id.
  • A server that wires the middleware in and exposes a cheap ping tool to exercise it.
  • A pytest suite that drives the limiter with a fake clock (deterministic, no sleeping), and a launchd deployment.

Prerequisites

  • macOS 13+ with Homebrew (brew.sh).
  • uv 0.5+ — brew install uv; verify uv --version.
  • Xcode Command Line Tools (xcode-select --install) for make.
  • Familiarity with the license-gating article helps, since this reuses the plan-on-a-token idea, but it is self-contained.

Everything runs through uv and make. Auth and rate limiting apply to the HTTP transport.

Step 1: Scaffold and lock down hygiene

Create the workspace and a .gitignore first.

Create the files

mkdir -p rate-limit-fastmcp-server-macos
cd rate-limit-fastmcp-server-macos
touch .gitignore

Add the code: .gitignore

__pycache__/
*.py[cod]
.venv/
.uv/
.pytest_cache/
.ruff_cache/
.mypy_cache/
logs/
*.log
*.pid
licenses.json
.DS_Store

Detailed breakdown

  • Standard uv/macOS hygiene. licenses.json is ignored in case you later swap the inline demo keys for a real key store, as in the license-gating article.

Step 2: Initialize the project with uv

Create the files

uv init --name macmcp --no-workspace
rm -f main.py hello.py
uv add fastmcp
uv add --dev pytest pytest-asyncio

Add the code: pyproject.toml

[project]
name = "macmcp"
version = "0.1.0"
description = "A FastMCP server with per-plan rate limiting"
readme = "README.md"
requires-python = ">=3.12"
dependencies = [
    "fastmcp>=3.4.3",
]

[dependency-groups]
dev = [
    "pytest>=9.1.1",
    "pytest-asyncio>=1.4.0",
]

Detailed breakdown

  • fastmcp provides the Middleware base class and RateLimitError used by the middleware, so no extra dependency is needed.

Step 3: Authenticate and tag the plan

The middleware needs to know each caller’s plan. Authentication resolves an API key to an AccessToken whose claims carry the plan. This is the license-gating idea, condensed to keep the focus on rate limiting.

Create the file

touch auth.py

Add the code: auth.py

"""Minimal API-key auth that tags each caller with a plan.

This is the same idea as the license-gating article, condensed: a key maps to a
customer and a plan. The plan is what the rate-limit middleware reads to pick a
per-caller quota. Keys here are demo values; in production verify against a real
store (see the license-gating article).
"""

from fastmcp.server.auth import AccessToken, TokenVerifier

# demo keys -> customer + plan
API_KEYS = {
    "basic-key-alice": {"client_id": "[email protected]", "plan": "basic"},
    "pro-key-bob": {"client_id": "[email protected]", "plan": "pro"},
}


class KeyVerifier(TokenVerifier):
    """Resolve an API key to an AccessToken carrying the caller's plan."""

    async def verify_token(self, token: str) -> AccessToken | None:
        record = API_KEYS.get(token)
        if record is None:
            return None
        return AccessToken(
            token=token,
            client_id=record["client_id"],
            scopes=["access"],
            claims={"plan": record["plan"]},
        )

Detailed breakdown

  • KeyVerifier.verify_token returns None for an unknown key (which FastMCP turns into a 401) and otherwise an AccessToken whose client_id identifies the caller and whose claims["plan"] records the tier.
  • client_id is the rate-limit key. All of a customer’s requests share one quota because they share one client_id, regardless of how many connections they open.

Step 4: The per-plan rate-limit middleware

The middleware is the heart of the article. It intercepts every tool call, looks up the caller’s plan quota, and either lets the call through or rejects it. The counting logic lives in a sliding-window limiter with an injectable clock, so tests run instantly and deterministically instead of sleeping.

Create the file

touch ratelimit.py

Add the code: ratelimit.py

"""Per-plan rate limiting as a FastMCP middleware.

FastMCP ships global rate limiters, but quotas that differ by plan need a custom
middleware: it reads the caller's plan from the verified access token and applies
that plan's limit, keyed on the caller's `client_id`. The limiter is a sliding
window with an injectable clock so the behavior is deterministic in tests.
"""

import time
from collections import defaultdict, deque

from fastmcp.server.dependencies import get_access_token
from fastmcp.server.middleware import Middleware, MiddlewareContext
from fastmcp.server.middleware.rate_limiting import RateLimitError

# plan -> (max requests, window in seconds)
PLAN_LIMITS: dict[str, tuple[int, int]] = {
    "basic": (5, 60),
    "pro": (30, 60),
}
DEFAULT_LIMIT = (5, 60)


class SlidingWindowLimiter:
    """Track request timestamps per key and allow up to N within a window."""

    def __init__(self, now=time.monotonic):
        self._now = now
        self._hits: dict[str, deque[float]] = defaultdict(deque)

    def check(self, key: str, max_requests: int, window: int) -> tuple[bool, int, float]:
        """Return (allowed, remaining, retry_after_seconds)."""
        t = self._now()
        hits = self._hits[key]
        while hits and hits[0] <= t - window:
            hits.popleft()
        if len(hits) >= max_requests:
            retry_after = window - (t - hits[0])
            return False, 0, max(retry_after, 0.0)
        hits.append(t)
        return True, max_requests - len(hits), 0.0


class PlanRateLimitMiddleware(Middleware):
    """Rate-limit tool calls per caller, using the caller's plan quota."""

    def __init__(self, limiter: SlidingWindowLimiter | None = None):
        self.limiter = limiter or SlidingWindowLimiter()

    async def on_call_tool(self, context: MiddlewareContext, call_next):
        token = get_access_token()
        plan = (token.claims.get("plan") if token else None) or "basic"
        client_id = token.client_id if token else "anonymous"
        max_requests, window = PLAN_LIMITS.get(plan, DEFAULT_LIMIT)

        allowed, remaining, retry_after = self.limiter.check(
            client_id, max_requests, window
        )
        if not allowed:
            raise RateLimitError(
                f"rate limit exceeded for {client_id} ({plan} plan): "
                f"{max_requests} requests / {window}s. Retry in {retry_after:.0f}s."
            )
        return await call_next(context)

Detailed breakdown

  • SlidingWindowLimiter.check keeps a deque of recent request timestamps per key. It first drops timestamps older than the window, then either records the new request and returns allowed, or refuses and reports how long until the oldest request ages out. Injecting now makes the window testable without real time passing.
  • PLAN_LIMITS maps each plan to (max_requests, window_seconds). Basic gets 5 requests per minute; Pro gets 30. DEFAULT_LIMIT covers any unrecognized plan, failing safe to the smallest quota.
  • on_call_tool is the FastMCP middleware hook that wraps every tool call. It reads the caller’s token with get_access_token(), picks the plan’s limit, and checks the window keyed on client_id. On success it calls call_next (running the tool); on failure it raises RateLimitError, which FastMCP returns to the client as a protocol error before the tool runs.
  • Only tool calls are limited. Discovery calls (list_tools, and so on) use other hooks, so listing capabilities does not burn a caller’s quota.

Step 5: Wire the middleware into the server

Attach the verifier and register the middleware. Add a cheap ping tool to exercise the limit and a whoami tool to confirm identity.

Create the file

touch server.py

Add the code: server.py

"""A FastMCP server with per-plan rate limiting.

Callers authenticate with an API key that carries their plan. A middleware
enforces a per-plan request quota keyed on the caller's identity, so a Basic
customer and a Pro customer get different limits. Auth and rate limiting apply
to the HTTP transport.
"""

import os

from fastmcp import FastMCP
from fastmcp.server.dependencies import get_access_token

from auth import KeyVerifier
from ratelimit import PlanRateLimitMiddleware

mcp = FastMCP("macmcp", auth=KeyVerifier())
mcp.add_middleware(PlanRateLimitMiddleware())


@mcp.tool
def ping() -> str:
    """A cheap tool to exercise the rate limit."""
    return "pong"


@mcp.tool
def whoami() -> dict:
    """Return the caller's identity and plan."""
    token = get_access_token()
    return {"client_id": token.client_id, "plan": token.claims.get("plan")}


def main() -> None:
    if os.environ.get("MCP_TRANSPORT", "stdio").lower() == "http":
        mcp.run(
            transport="http",
            host=os.environ.get("MCP_HOST", "127.0.0.1"),
            port=int(os.environ.get("MCP_PORT", "8000")),
        )
    else:
        mcp.run()  # stdio (unauthenticated; local dev only)


if __name__ == "__main__":
    main()

Detailed breakdown

  • mcp.add_middleware(PlanRateLimitMiddleware()) registers the middleware for the whole server. You can also pass middleware=[...] to the FastMCP constructor; add_middleware is handy when the list is built conditionally.
  • ping is deliberately trivial so a client can call it repeatedly and watch the quota deplete. whoami confirms which plan the server sees for a key.
  • Because rate limiting only takes effect once a request has an authenticated identity, it lives behind the same HTTP transport as auth.

Step 6: Test the limiter with a fake clock

The window logic is the security boundary, so test it directly with a controllable clock. No time.sleep, so the suite is instant and reliable.

Create the files

mkdir -p tests
touch tests/__init__.py tests/test_ratelimit.py pytest.ini

Add the code: pytest.ini

[pytest]
asyncio_mode = auto
filterwarnings =
    ignore::DeprecationWarning

Add the code: tests/test_ratelimit.py

"""Test the sliding-window limiter and per-plan quotas with a fake clock."""

import pytest

from auth import KeyVerifier
from ratelimit import PLAN_LIMITS, SlidingWindowLimiter


class FakeClock:
    def __init__(self):
        self.t = 0.0

    def __call__(self) -> float:
        return self.t

    def advance(self, seconds: float):
        self.t += seconds


def test_allows_up_to_limit_then_blocks():
    clock = FakeClock()
    limiter = SlidingWindowLimiter(now=clock)
    for _ in range(5):
        allowed, _, _ = limiter.check("alice", max_requests=5, window=60)
        assert allowed
    allowed, remaining, retry = limiter.check("alice", 5, 60)
    assert not allowed and remaining == 0 and retry > 0


def test_window_slides_and_frees_capacity():
    clock = FakeClock()
    limiter = SlidingWindowLimiter(now=clock)
    for _ in range(5):
        limiter.check("alice", 5, 60)
    assert not limiter.check("alice", 5, 60)[0]
    clock.advance(61)  # the whole window has passed
    assert limiter.check("alice", 5, 60)[0]


def test_clients_are_independent():
    clock = FakeClock()
    limiter = SlidingWindowLimiter(now=clock)
    for _ in range(5):
        limiter.check("alice", 5, 60)
    assert not limiter.check("alice", 5, 60)[0]
    assert limiter.check("bob", 5, 60)[0]  # bob has his own window


def test_remaining_counts_down():
    clock = FakeClock()
    limiter = SlidingWindowLimiter(now=clock)
    _, remaining1, _ = limiter.check("alice", 3, 60)
    _, remaining2, _ = limiter.check("alice", 3, 60)
    assert remaining1 == 2 and remaining2 == 1


def test_pro_plan_allows_more_than_basic():
    assert PLAN_LIMITS["pro"][0] > PLAN_LIMITS["basic"][0]


async def test_key_verifier_tags_plan():
    v = KeyVerifier()
    basic = await v.verify_token("basic-key-alice")
    assert basic.claims["plan"] == "basic"
    pro = await v.verify_token("pro-key-bob")
    assert pro.claims["plan"] == "pro"
    assert await v.verify_token("nope") is None

Detailed breakdown

  • FakeClock replaces time.monotonic, so advance(61) moves past a one-minute window instantly. The window tests never sleep.
  • The five limiter tests cover the contract: allow up to the limit then block, free capacity once the window slides, keep clients independent, count down remaining, and confirm Pro’s quota exceeds Basic’s.
  • test_key_verifier_tags_plan proves the token carries the plan the middleware relies on.

Run them:

uv run pytest -v

All tests pass, with no server running.

Step 7: The Makefile

Wrap development, a hammer target that fires repeated calls, and the launchd deployment. Plain make prints help.

Create the file

touch Makefile

Add the code: Makefile

.DEFAULT_GOAL := help

# --- Deploy configuration (launchd user agent) --------------------------------
LABEL   := com.macmcp.ratelimit
PLIST   := $(HOME)/Library/LaunchAgents/$(LABEL).plist
UV      := $(shell command -v uv)
DIR     := $(shell pwd)
DOMAIN  := gui/$(shell id -u)

# `make hammer` arguments
KEY ?= basic-key-alice
N   ?= 8

.PHONY: help install serve test hammer deploy undeploy status logs clean

help:  ## Show this help screen
	@grep -E '^[a-zA-Z_-]+:.*?## .*$$' $(MAKEFILE_LIST) \
		| awk 'BEGIN {FS = ":.*?## "}; {printf "  \033[36m%-10s\033[0m %s\n", $$1, $$2}'

install:  ## Sync runtime and dev dependencies
	uv sync

serve:  ## Run the HTTP server in the foreground (Ctrl-C to stop)
	MCP_TRANSPORT=http uv run python server.py

test:  ## Run the test suite
	uv run pytest -v

hammer:  ## Fire N calls with a key: make hammer KEY=pro-key-bob N=8
	@KEY=$(KEY) N=$(N) uv run python scripts/hammer.py

deploy:  ## Render the launchd plist and start the service
	@mkdir -p logs
	@sed -e 's#@UV@#$(UV)#g' \
	     -e 's#@DIR@#$(DIR)#g' \
	     -e 's#@PATH@#$(dir $(UV)):/usr/bin:/bin#g' \
	     deploy/$(LABEL).plist.template > $(PLIST)
	launchctl bootout $(DOMAIN) $(PLIST) 2>/dev/null || true
	launchctl bootstrap $(DOMAIN) $(PLIST)
	@echo "Deployed $(LABEL). Check: make status"

undeploy:  ## Stop the service and remove the launchd plist
	launchctl bootout $(DOMAIN) $(PLIST) 2>/dev/null || true
	rm -f $(PLIST)
	@echo "Removed $(LABEL)."

status:  ## Show whether the service is registered and running
	@launchctl print $(DOMAIN)/$(LABEL) 2>/dev/null \
		| grep -E 'state =|pid =' || echo "$(LABEL) is not loaded."

logs:  ## Tail the service log files
	@tail -n 40 -f logs/server.out.log logs/server.err.log

clean:  ## Remove caches and logs
	rm -rf .pytest_cache logs __pycache__ tests/__pycache__

Detailed breakdown

  • hammer fires N calls with a given KEY and reports how many the plan allowed. Override KEY to switch plans and N to change the burst. The launchd lifecycle matches the companion deploy article.

Step 8: Deploy and watch the quotas differ

Add the launchd plist and the hammer client, then run the demonstration.

Create the files

mkdir -p deploy scripts
touch deploy/com.macmcp.ratelimit.plist.template scripts/hammer.py

Add the code: deploy/com.macmcp.ratelimit.plist.template

<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN"
  "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
<plist version="1.0">
<dict>
    <key>Label</key>
    <string>com.macmcp.ratelimit</string>

    <key>ProgramArguments</key>
    <array>
        <string>@UV@</string>
        <string>run</string>
        <string>python</string>
        <string>server.py</string>
    </array>

    <key>WorkingDirectory</key>
    <string>@DIR@</string>

    <key>EnvironmentVariables</key>
    <dict>
        <key>MCP_TRANSPORT</key>
        <string>http</string>
        <key>MCP_HOST</key>
        <string>127.0.0.1</string>
        <key>MCP_PORT</key>
        <string>8000</string>
        <key>PATH</key>
        <string>@PATH@</string>
    </dict>

    <key>RunAtLoad</key>
    <true/>

    <key>KeepAlive</key>
    <true/>

    <key>StandardOutPath</key>
    <string>@DIR@/logs/server.out.log</string>

    <key>StandardErrorPath</key>
    <string>@DIR@/logs/server.err.log</string>
</dict>
</plist>

Add the code: scripts/hammer.py

"""Hammer the ping tool with a key and report how many calls the plan allows.

    KEY=basic-key-alice N=8 uv run python scripts/hammer.py
"""

import asyncio
import os

from fastmcp import Client
from fastmcp.client.auth import BearerAuth

URL = os.environ.get("MCP_URL", "http://127.0.0.1:8000/mcp/")
KEY = os.environ.get("KEY", "basic-key-alice")
N = int(os.environ.get("N", "8"))


async def main() -> None:
    ok = blocked = 0
    async with Client(URL, auth=BearerAuth(KEY)) as client:
        for _ in range(N):
            try:
                await client.call_tool("ping", {})
                ok += 1
            except Exception:  # noqa: BLE001 - RateLimitError surfaces here
                blocked += 1
    print(f"key {KEY}: {ok} allowed, {blocked} blocked out of {N} ping calls")


if __name__ == "__main__":
    asyncio.run(main())

Detailed breakdown

  • hammer.py authenticates with a bearer key and calls ping N times, counting successes and rejections. Every RateLimitError after the quota shows up as a blocked call.

Deploy and run it:

make deploy
make status                 # state = running, a live pid

A Basic key exhausts its quota; a Pro key does not:

make hammer KEY=basic-key-alice N=8
# key basic-key-alice: 5 allowed, 3 blocked out of 8 ping calls

make hammer KEY=pro-key-bob N=8
# key pro-key-bob: 8 allowed, 0 blocked out of 8 ping calls

The two callers are independent, so Alice hitting her limit never affects Bob. Tear down when done:

make undeploy

Troubleshooting

  • Every call is blocked immediately. The quota is per minute; if you re-run hammer within the same window, earlier calls still count. Wait for the window to slide or lower N.
  • A Pro key is limited like Basic. The token’s plan claim is not pro. Check whoami for that key, and confirm API_KEYS maps it to "plan": "pro".
  • Limits reset when the server restarts. The limiter keeps counts in memory. That is fine for one instance; back it with Redis (shared, atomic counters) if you run several instances behind a load balancer.
  • 401 Unauthorized. The key is unknown. Use basic-key-alice or pro-key-bob, or add your own to API_KEYS.

Recap

You turned plan tiers into enforced quotas:

  • A custom FastMCP middleware reads each caller’s plan from the verified token and applies that plan’s limit, keyed on client_id, rejecting over-limit calls with RateLimitError.
  • A sliding-window limiter with an injectable clock makes the counting deterministic and instantly testable.
  • The demonstration is real: a Basic key gets 5 calls a minute and a Pro key gets 30, with callers isolated from each other.
  • uv, make, and launchd package and deploy it.

Next improvements

  • Back the limiter with Redis so the quota is shared across multiple server instances.
  • Return the remaining quota and a Retry-After hint to the client on rejection.
  • Add a second dimension (a monthly quota alongside the per-minute rate) for usage-based billing.
  • Load PLAN_LIMITS from configuration so quotas change without a code deploy.