Rate Limiting

rate_limiting

Rate limiting utilities for OAuth endpoints.

AsyncRedisClient

Bases: Protocol

Minimal async Redis interface used for rate-limit storage.

SlidingWindowRateLimiter

SlidingWindowRateLimiter(
    requests_per_window: int,
    window_seconds: int,
    redis: AsyncRedisClient | None = None,
)

Sliding-window rate limiter for OAuth endpoints.

Tracks requests per client within a sliding time window.

When a Redis client is provided, state is stored in a Redis sorted set and is shared across all replicas and survives pod restarts. When redis is None the limiter falls back to an in-process defaultdict (suitable for local development and single-replica deployments).

Redis key format: mcp_auth:ratelimit:<client_id>:<window_seconds>

Initialize the rate limiter.

Parameters:

Name	Type	Description	Default
requests_per_window	int	Maximum number of requests allowed in the window	required
window_seconds	int	Size of the time window in seconds	required
redis	AsyncRedisClient | None	Optional async Redis client for shared, persistent storage. When None, falls back to in-process in-memory storage.	None

Source code in mcp_authflow/rate_limiting.py

def __init__(
    self,
    requests_per_window: int,
    window_seconds: int,
    redis: AsyncRedisClient | None = None,
):
    """Initialize the rate limiter.

    Args:
        requests_per_window: Maximum number of requests allowed in the window
        window_seconds: Size of the time window in seconds
        redis: Optional async Redis client for shared, persistent storage.
               When None, falls back to in-process in-memory storage.
    """
    self.requests_per_window = requests_per_window
    self.window_seconds = window_seconds
    self._redis = redis
    self._clients: dict[str, list[float]] = defaultdict(list)
    self._last_sweep = 0.0

is_allowed async

is_allowed(client_id: str) -> bool

Check if the client is allowed to make a request.

Records the request if allowed.

Parameters:

Name	Type	Description	Default
client_id	str	OAuth client identifier	required

Returns:

Type	Description
bool	True if the request is allowed, False if rate limited

Source code in mcp_authflow/rate_limiting.py

async def is_allowed(self, client_id: str) -> bool:
    """Check if the client is allowed to make a request.

    Records the request if allowed.

    Args:
        client_id: OAuth client identifier

    Returns:
        True if the request is allowed, False if rate limited
    """
    if self._redis is not None:
        return await self._is_allowed_redis(client_id)
    return self._is_allowed_memory(client_id)

get_retry_after async

get_retry_after(client_id: str) -> int

Get the number of seconds until the client can retry.

Parameters:

Name	Type	Description	Default
client_id	str	OAuth client identifier	required

Returns:

Type	Description
int	Number of seconds to wait before retrying (minimum 1), or 0 if
int	no requests have been recorded for this client.

Source code in mcp_authflow/rate_limiting.py

async def get_retry_after(self, client_id: str) -> int:
    """Get the number of seconds until the client can retry.

    Args:
        client_id: OAuth client identifier

    Returns:
        Number of seconds to wait before retrying (minimum 1), or 0 if
        no requests have been recorded for this client.
    """
    if self._redis is not None:
        return await self._get_retry_after_redis(client_id)
    return self._get_retry_after_memory(client_id)