"""Rate limiting for MCP server."""
from collections import defaultdict
from datetime import datetime, timedelta
from typing import Dict, List
class RateLimiter:
"""Simple rate limiter with sliding window."""
def __init__(self, max_requests: int = 100, window_seconds: int = 60):
self.max_requests = max_requests
self.window = timedelta(seconds=window_seconds)
self.requests: Dict[str, List[datetime]] = defaultdict(list)
def is_allowed(self, client_id: str = "default") -> bool:
"""Check if request is allowed for client."""
now = datetime.now()
# Remove old requests outside the window
self.requests[client_id] = [
ts for ts in self.requests[client_id]
if now - ts < self.window
]
# Check if under limit
if len(self.requests[client_id]) < self.max_requests:
self.requests[client_id].append(now)
return True
return False
def get_stats(self, client_id: str = "default") -> dict:
"""Get rate limit statistics for client."""
now = datetime.now()
active_requests = [
ts for ts in self.requests[client_id]
if now - ts < self.window
]
return {
"client_id": client_id,
"requests_in_window": len(active_requests),
"max_requests": self.max_requests,
"window_seconds": int(self.window.total_seconds()),
"remaining": max(0, self.max_requests - len(active_requests))
}
