rate_limiter

rate_limiter

Rate limiter -- token bucket algorithm for per-agent/per-tool throttling.

Classes

RateLimitConfig dataclass

RateLimitConfig(requests_per_minute: int = 60, burst_size: int = 10, enabled: bool = True)

Configuration for rate limiting.

TokenBucket

TokenBucket(rate: float, capacity: int)

Thread-safe token bucket for rate limiting.

Source code in src/openjarvis/security/rate_limiter.py

def __init__(self, rate: float, capacity: int) -> None:
    self._rate = rate  # tokens per second
    self._capacity = capacity
    self._tokens = float(capacity)
    self._last_refill = time.monotonic()
    self._lock = threading.Lock()

Attributes

available property

available: float

Current available tokens (approximate).

Functions

consume

consume(tokens: int = 1) -> Tuple[bool, float]

Try to consume tokens. Returns (allowed, wait_seconds).

Source code in src/openjarvis/security/rate_limiter.py

def consume(self, tokens: int = 1) -> Tuple[bool, float]:
    """Try to consume tokens. Returns (allowed, wait_seconds)."""
    with self._lock:
        now = time.monotonic()
        elapsed = now - self._last_refill
        self._tokens = min(
            self._capacity,
            self._tokens + elapsed * self._rate,
        )
        self._last_refill = now

        if self._tokens >= tokens:
            self._tokens -= tokens
            return True, 0.0
        else:
            wait = (tokens - self._tokens) / self._rate
            return False, wait

RateLimiter

RateLimiter(config: Optional[RateLimitConfig] = None)

Rate limiter with per-key token buckets.

Keys are typically "agent_id:tool_name" or just "agent_id".

Source code in src/openjarvis/security/rate_limiter.py

def __init__(self, config: Optional[RateLimitConfig] = None) -> None:
    self._config = config or RateLimitConfig()
    self._buckets: Dict[str, TokenBucket] = {}
    self._lock = threading.Lock()

    from openjarvis._rust_bridge import get_rust_module
    _rust = get_rust_module()
    self._rust_impl = _rust.RateLimiter(
        requests_per_minute=self._config.requests_per_minute,
        burst_size=self._config.burst_size,
    )

Functions

check

check(key: str) -> Tuple[bool, float]

Check if request is allowed for key — always via Rust backend.

Source code in src/openjarvis/security/rate_limiter.py

def check(self, key: str) -> Tuple[bool, float]:
    """Check if request is allowed for key — always via Rust backend."""
    if not self._config.enabled:
        return True, 0.0
    return self._rust_impl.check(key)

reset

reset(key: Optional[str] = None) -> None

Reset rate limit state for a key or all keys — always via Rust backend.

Source code in src/openjarvis/security/rate_limiter.py

def reset(self, key: Optional[str] = None) -> None:
    """Reset rate limit state for a key or all keys — always via Rust backend."""
    self._rust_impl.reset(key)
    return
    with self._lock:
        if key:
            self._buckets.pop(key, None)
        else:
            self._buckets.clear()