MCP Server Rate Limits: How to Protect Your Tools From Runaway Agents

An MCP server gives an agent hands, and agents use their hands quickly. A model that is polite in a demo will call your search tool forty times in a row when it is confused, fan out ten parallel calls when it is eager, and retry instantly when it is told to wait. Your database, your upstream API quota, and your cloud bill feel all of it. This post shows how to put limits around an MCP server that stop the damage, tell the model what happened in words it can act on, and stay out of the way of legitimate work.

Why MCP servers need their own limits

• Model provider limits do not cover your tools. They cap tokens and requests to the model. A single model request budget can trigger hundreds of calls to your server.
• Agents are bursty by nature. Parallel tool calls, retries after confusion, and loops all arrive as sudden spikes.
• Calls are not equal. One tool returns a cached record. Another runs a ten-second query, starts a job, or spends money on a paid API.
• Many clients share one server. A remote server may serve many users, each with several sessions and sub-agents, and one runaway should not starve the rest.
• Your upstream has limits too. If your tool wraps an API with its own quota, the agent can burn that quota for every user in minutes.
• A hijacked agent is a fast attacker. If prompt injection makes an agent dump data through your tools, quotas bound how much can leave. Our security playbook covers the wider threat model.

Where each limit belongs

Rate limiting works best in layers, with each layer doing the job it is suited for.

The gateway protects your infrastructure from anything. The in-server limiter understands tools, identities, and costs. Stdio servers have no HTTP layer at all, so the in-server limiter is the only protection available, and it matters just as much there: a local server often fronts a paid or rate-limited API.

Choose the right algorithm

A good MCP server combines them: a per-user token bucket, a per-user-per-tool bucket for heavy tools, a global bucket as a backstop, a concurrency cap, a daily quota, and a penalty box per upstream.

The limiter

This is a framework-agnostic limiter you can wrap around any tool handler. It checks, in order, the upstream penalty box, the concurrency cap, the daily quota, and then the identity, tool, and global buckets. If a later layer refuses, tokens already taken from earlier layers are given back, so a refused call never uses up budget. State for idle users is pruned so the limiter cannot grow without bound.

import functools, math, threading, time
from dataclasses import dataclass
from typing import Callable, Optional


class TokenBucket:
    def __init__(self, capacity: float, per_second: float, now: float):
        self.capacity, self.rate = float(capacity), float(per_second)
        self.tokens, self.t = float(capacity), now

    def take(self, cost: float, now: float):
        self.tokens = min(self.capacity, self.tokens + (now - self.t) * self.rate)
        self.t = now
        if cost > self.capacity:
            return False, math.inf                 # can never succeed as requested
        if self.tokens >= cost:
            self.tokens -= cost
            return True, 0.0
        return False, (cost - self.tokens) / self.rate

    def refund(self, cost: float) -> None:
        self.tokens = min(self.capacity, self.tokens + cost)


@dataclass
class Policy:
    burst: float
    per_second: float
    max_concurrent: int = 4
    daily_quota: Optional[float] = None            # cost units per identity per UTC day


@dataclass
class Decision:
    allowed: bool
    scope: str = ""                                # upstream | concurrency | daily | identity | tool | global
    retry_after: float = 0.0
    reason: str = ""


class UpstreamRateLimited(Exception):
    def __init__(self, retry_after: float):
        super().__init__(f"upstream asked for {retry_after}s")
        self.retry_after = retry_after


class Limiter:
    def __init__(self, per_identity: Policy, per_tool: Optional[dict] = None,
                 global_policy: Optional[Policy] = None,
                 clock: Callable[[], float] = time.monotonic,
                 wall: Callable[[], float] = time.time, idle_ttl: float = 900.0):
        self.per_identity, self.per_tool = per_identity, per_tool or {}
        self.global_policy = global_policy
        self.clock, self.wall, self.idle_ttl = clock, wall, idle_ttl
        self.lock = threading.Lock()
        self.buckets: dict = {}                    # key -> [bucket, last_seen]
        self.active: dict = {}                     # identity -> calls in flight
        self.daily: dict = {}                      # (identity, day) -> units used
        self.penalty_until: dict = {}              # upstream -> time it may be called again
        self._last_prune = clock()

    # -- internals -------------------------------------------------------
    def _bucket(self, key, pol: Policy, now: float) -> TokenBucket:
        entry = self.buckets.get(key)
        if entry is None:
            entry = self.buckets[key] = [TokenBucket(pol.burst, pol.per_second, now), now]
        entry[1] = now
        return entry[0]

    def _prune(self, now: float) -> None:
        if now - self._last_prune < 60:
            return
        self._last_prune = now
        for k in [k for k, (_, seen) in self.buckets.items() if now - seen > self.idle_ttl]:
            del self.buckets[k]
        today = int(self.wall() // 86400)
        for k in [k for k in self.daily if k[1] != today]:
            del self.daily[k]
        for k in [k for k, n in self.active.items() if n <= 0]:
            del self.active[k]

    # -- public API ------------------------------------------------------
    def penalize(self, upstream: str, seconds: float) -> None:
        with self.lock:
            self.penalty_until[upstream] = max(self.penalty_until.get(upstream, 0.0),
                                               self.clock() + seconds)

    def acquire(self, identity: str, tool: str, cost: float = 1.0,
                upstream: Optional[str] = None) -> Decision:
        now = self.clock()
        with self.lock:
            self._prune(now)
            if upstream and self.penalty_until.get(upstream, 0.0) > now:
                return Decision(False, "upstream", self.penalty_until[upstream] - now,
                                f"{upstream} asked us to slow down")
            if self.active.get(identity, 0) >= self.per_identity.max_concurrent:
                return Decision(False, "concurrency", 1.0, "too many calls already in flight")
            day, quota = int(self.wall() // 86400), self.per_identity.daily_quota
            used = self.daily.get((identity, day), 0.0)
            if quota is not None and used + cost > quota:
                return Decision(False, "daily", 86400 - (self.wall() % 86400), "daily quota used up")

            taken = []
            layers = (("identity", ("i", identity), self.per_identity),
                      ("tool", ("t", identity, tool), self.per_tool.get(tool)),
                      ("global", ("g",), self.global_policy))
            for scope, key, pol in layers:
                if pol is None:
                    continue
                ok, wait = self._bucket(key, pol, now).take(cost, now)
                if not ok:
                    for k in taken:                # a later layer said no: give tokens back
                        self.buckets[k][0].refund(cost)
                    if wait == math.inf:
                        return Decision(False, scope, 0.0, "this call costs more than the limit allows")
                    return Decision(False, scope, wait, "rate limit reached")
                taken.append(key)

            self.active[identity] = self.active.get(identity, 0) + 1
            if quota is not None:
                self.daily[(identity, day)] = used + cost
            return Decision(True)

    def release(self, identity: str) -> None:
        with self.lock:
            self.active[identity] = max(0, self.active.get(identity, 0) - 1)


def refusal(tool: str, d: Decision) -> dict:
    # A normal tool result flagged as an error: the model sees it and can adapt.
    if d.retry_after > 0:
        hint = f"Wait about {math.ceil(d.retry_after)} seconds before calling it again. Do not retry sooner."
    else:
        hint = "This call cannot succeed as requested; make it smaller."
    text = (f"Rate limit reached for '{tool}' ({d.scope}: {d.reason}). {hint} "
            "Meanwhile use what you already have, or tell the user what is pending.")
    return {"content": [{"type": "text", "text": text}], "isError": True}


def limited(limiter: Limiter, tool: str, cost: float = 1.0, upstream: Optional[str] = None):
    def deco(fn):
        @functools.wraps(fn)
        async def wrapper(identity: str, *args, **kwargs):
            d = limiter.acquire(identity, tool, cost, upstream)
            if not d.allowed:
                return refusal(tool, d)
            try:
                return await fn(*args, **kwargs)
            except UpstreamRateLimited as e:      # share the penalty with every caller
                if upstream:
                    limiter.penalize(upstream, e.retry_after)
                return refusal(tool, Decision(False, "upstream", e.retry_after, f"{upstream} is rate limiting us"))
            finally:
                limiter.release(identity)
        return wrapper
    return deco

The design choices matter more than the details:

• Identity, not IP. Key limits on the authenticated user or tenant. Sub-agents and parallel sessions for one user share that user's budget.
• Monotonic clock for rates, wall clock for days. System time can jump, so buckets use a monotonic source, and only the daily quota looks at the calendar.
• Refusals are cheap and fast. The limiter never sleeps and never queues. A refused call returns immediately with a precise wait time.
• An oversized call is a different answer. If one call costs more than the bucket can ever hold, waiting will not help, so the message says to make the call smaller.
• Penalties are shared. When the upstream says "slow down," every caller hears it, instead of each one hitting the API and extending the penalty.

Tell the model what happened

How you refuse decides what the agent does next. A protocol-level failure may be handled by the client as a transport problem, and the model never sees it, so the agent retries blindly. Return a normal tool result with isError set to true, and write the message for the model:

• Say which limit was hit and why, in a few words.
• Give a number. "Wait about 12 seconds" is acted on. "Try again later" is not.
• Say what not to do. "Do not retry sooner" prevents the immediate-retry spiral.
• Offer a way forward: use information already gathered, try a different tool, or tell the user what is pending.

The refusal function above does exactly this. Reserve HTTP 429 with Retry-After for the gateway layer, where the caller is a client library or a load balancer rather than a model.

Wire it into your tools

Decorate each tool with its name, a cost, and the upstream it depends on. Give heavy tools their own bucket and a higher cost, and cap page sizes inside the tool as well, since a limit on call count does nothing about a call that asks for a million rows.

limiter = Limiter(
    per_identity=Policy(burst=30, per_second=2, max_concurrent=4, daily_quota=2000),
    per_tool={
        "search_issues": Policy(burst=10, per_second=1),
        "export_report": Policy(burst=2, per_second=0.05),     # heavy: one every 20 s
    },
    global_policy=Policy(burst=200, per_second=50),
)


@limited(limiter, "search_issues", cost=1, upstream="github")
async def search_issues(query: str, limit: int = 20) -> dict:
    limit = min(limit, 50)                         # cap page size inside the tool as well
    ...                                            # call the upstream API; raise UpstreamRateLimited(n) on 429


@limited(limiter, "export_report", cost=20)
async def export_report(report_id: str) -> dict:
    ...

Choosing the numbers

Start from measurement, not guesses. Look at how much legitimate users and agents actually call each tool, then set limits with headroom.

These numbers are illustrations, not recommendations for your workload. A workable method: set the sustained rate to about twice the 95th percentile of legitimate use, set the burst to the largest normal burst for one task, and size concurrency to what your slowest dependency can handle. Then run in log-only mode first, recording who would have been limited without refusing anyone. Enforce once the false positives are understood, and tighten gradually.

Design tools that are hard to abuse

• Cache repeated reads for a short time. A looping agent then gets cheap answers instead of fresh load, and cache hits can cost zero units.
• Collapse identical in-flight calls so ten parallel requests for the same data do one piece of work.
• Cap and paginate. Enforce a maximum page size and return a cursor with an explicit "no more results" marker.
• Make heavy work asynchronous. Return a job ID, and make the status tool cheap, with a next_poll_after_seconds field so the model does not poll in a tight loop.
• Return less. Smaller outputs reduce load, token cost, and the data exposed if an agent is hijacked.
• Fail fast on bad input before spending any budget on the real work.

Remote servers, replicas, and fairness

• Several replicas multiply your limits. The limiter above is per process. With four replicas, a user can get four times the intended rate. Use a shared store with atomic operations, such as Redis with a script, route each user to one replica, or divide limits by the replica count as an approximation.
• Multi-tenant fairness. Add a per-tenant budget above the per-user one so one customer's many users cannot starve other customers, and keep the global bucket as the last line of defense.
• Authentication failures need their own tight limit, enforced at the gateway before any work is done.
• Surface limits in HTTP headers at the gateway. Retry-After is standard; the RateLimit family of headers follows an IETF draft, so check current support before relying on it.

Observe it

A limiter you cannot see will either block good traffic or miss bad traffic. Log every refusal with the scope, tool, a hashed identity, the cost, and the retry time. Chart these:

• Refused calls per tool and per scope (identity, tool, global, concurrency, daily, upstream).
• Bucket utilization, so you can see load approaching the limit before it bites.
• The 95th percentile of retry times handed out.
• Refusal rate per identity. One user who is refused constantly is looping or abusing the server. A tool refused for everyone means the limit is too tight or the tool is too slow.

Alert on sustained global-scope refusals and on a single identity exceeding a refusal rate threshold. When an agent is the cause, the fix usually lies in its loop guard or its tool errors, which we cover in the reliable agents playbook. For protocol-level problems with MCP itself, see the MCP protocol errors guide.

Test the limiter

The tests use a fake clock, so they are instant and deterministic. They cover bursts and refill, identity isolation, refunds, concurrency, quotas, cost weights, the upstream penalty box, pruning, and the exact messages the model receives.

import pytest


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

    def __call__(self):
        return self.t


def make(clock, **kw):
    kw.setdefault("per_identity", Policy(burst=3, per_second=1, max_concurrent=2))
    return Limiter(clock=clock, wall=lambda: 1_000_000 + clock.t, **kw)


def test_bucket_allows_a_burst_then_refills():
    clk = Clock()
    b = TokenBucket(3, 1, clk())
    assert all(b.take(1, clk())[0] for _ in range(3))
    ok, wait = b.take(1, clk())
    assert not ok and wait == pytest.approx(1.0)
    clk.t = 1.0
    assert b.take(1, clk())[0]
    assert b.take(10, clk()) == (False, math.inf)


def test_identities_do_not_share_budgets():
    clk = Clock()
    lim = make(clk, per_identity=Policy(burst=2, per_second=0.1, max_concurrent=99))
    for _ in range(2):
        d = lim.acquire("alice", "t"); lim.release("alice")
        assert d.allowed
    assert lim.acquire("alice", "t").scope == "identity"
    assert lim.acquire("bob", "t").allowed


def test_failed_layer_refunds_earlier_layers():
    clk = Clock()
    lim = make(clk, per_identity=Policy(burst=5, per_second=0.1, max_concurrent=99),
               global_policy=Policy(burst=1, per_second=0.1))
    assert lim.acquire("alice", "t").allowed
    d = lim.acquire("bob", "t")
    assert d.scope == "global"
    assert lim.buckets[("i", "bob")][0].tokens == 5     # nothing was consumed


def test_concurrency_cap_and_release():
    clk = Clock()
    lim = make(clk, per_identity=Policy(burst=99, per_second=99, max_concurrent=2))
    assert lim.acquire("a", "t").allowed and lim.acquire("a", "t").allowed
    assert lim.acquire("a", "t").scope == "concurrency"
    lim.release("a")
    assert lim.acquire("a", "t").allowed


def test_daily_quota():
    clk = Clock()
    lim = make(clk, per_identity=Policy(burst=99, per_second=99, max_concurrent=99, daily_quota=3))
    for _ in range(3):
        assert lim.acquire("a", "t").allowed
    d = lim.acquire("a", "t")
    assert d.scope == "daily" and 0 < d.retry_after <= 86400


def test_cost_weighting_and_oversized_calls():
    clk = Clock()
    lim = make(clk, per_identity=Policy(burst=10, per_second=1, max_concurrent=99))
    assert lim.acquire("a", "export", cost=8).allowed
    d = lim.acquire("a", "export", cost=8)
    assert not d.allowed and d.retry_after == pytest.approx(6.0)
    assert lim.acquire("a", "export", cost=11).reason.startswith("this call costs more")


def test_upstream_penalty_box_is_shared():
    clk = Clock()
    lim = make(clk)
    lim.penalize("github", 30)
    d = lim.acquire("anyone", "t", upstream="github")
    assert d.scope == "upstream" and d.retry_after == pytest.approx(30)
    clk.t = 31
    assert lim.acquire("anyone", "t", upstream="github").allowed


def test_idle_state_is_pruned():
    clk = Clock()
    lim = make(clk, idle_ttl=100)
    for i in range(50):
        lim.acquire(f"user{i}", "t"); lim.release(f"user{i}")
    assert len(lim.buckets) == 50
    clk.t = 500
    lim.acquire("fresh", "t")
    assert len(lim.buckets) == 1


@pytest.mark.asyncio
async def test_decorator_refuses_in_a_way_the_model_can_read():
    clk = Clock()
    lim = make(clk, per_identity=Policy(burst=1, per_second=0.1, max_concurrent=9))

    @limited(lim, "ping")
    async def ping():
        return {"content": [{"type": "text", "text": "pong"}]}

    assert (await ping("a"))["content"][0]["text"] == "pong"
    out = await ping("a")
    assert out["isError"] is True and "Wait about" in out["content"][0]["text"]


@pytest.mark.asyncio
async def test_upstream_429_penalizes_everyone_and_releases_the_slot():
    clk = Clock()
    lim = make(clk)

    @limited(lim, "search", upstream="github")
    async def search():
        raise UpstreamRateLimited(20)

    out = await search("a")
    assert out["isError"] and lim.active["a"] == 0
    assert lim.acquire("b", "search", upstream="github").scope == "upstream"

Pitfalls to avoid

• Keying on IP address. Shared networks punish innocent users and attackers rotate addresses.
• Queueing instead of refusing. Long waits hold connections, trigger timeouts, and exhaust workers.
• Refusals the model cannot read. A bare error code makes the agent retry immediately.
• One global limit for everything. A cheap lookup and a heavy export need different budgets.
• Forgetting multiple replicas, so the real limit is a multiple of the configured one.
• Unbounded limiter state. Idle identities must expire.
• Ignoring upstream 429s. Hammering a rate-limited API extends the penalty for everyone.
• Enforcing on day one. Observe first, then enforce.
• Treating limits as security. They bound damage; they do not replace authentication, authorization, and least privilege.

Frequently asked questions

Keep reading