Gateway Rate Limit

Bound inbound turns per identity/scope with a config-driven sliding window, or inject a custom limiter

from praisonaiagents import Agent

agent = Agent(name="assistant", instructions="Be helpful.")
# Per-client gateway rate limits protect this agent from traffic spikes
agent.start("Quick question")

from praisonai_bot.bots import BotOS
from praisonaiagents import Agent

agent = Agent(name="Support", instructions="Help users")
bot = BotOS(agent=agent, platforms=["telegram"], per_user_per_minute=20)
bot.run()

One line protects every connected channel — Telegram, Discord, Slack, HTTP, WebSocket.

Gateway rate limiting completes the gateway’s policy-protocol family: bound how often an identity can start turns in a given scope with the built-in sliding window, or inject your own limiter.

The user hits the gateway repeatedly; per-client rate limits cap requests and return clear backoff signals.

Quick Start

1. Simplest: one YAML key

   Cap how many turns one user may start per minute — unlimited by default (0).

   # gateway.yaml  (or botos.yaml under the same gateway: block)
   gateway:
     per_user_per_minute: 20

   The equivalent Python one-liner:

   from praisonai_bot.bots import BotOS
   from praisonaiagents import Agent
   
   agent = Agent(name="Support", instructions="Help users")
   bot = BotOS(agent=agent, platforms=["telegram"], per_user_per_minute=20)
   bot.run()

   Over the budget, the user sees: “Rate limit reached — please try again in Ns.” The turn is refused before any agent work starts. A fresh conversation does not reset the budget — the gateway charges a stable identity (session id on bot channels, authenticated token or peer IP on HTTP / WebSocket).

   Value	Behaviour
   0 (default)	Unlimited — no-op
   20	20 turns per minute per identity
   -1	ValueError at startup
   0.5 or true	ValueError at startup

Choosing which path

For a cooldown penalty, Redis-backed state, or per-tenant limits, use the policy surface:

1. Simple sliding window

   Five turns per minute per identity. max_requests=0 (default) disables the gate entirely — safe for dev environments.

   from praisonaiagents import Agent
   from praisonaiagents.gateway import SlidingWindowRateLimitPolicy
   from praisonai.bots import BotOS
   
   agent = Agent(name="Support", instructions="Help users")
   
   rate_limit = SlidingWindowRateLimitPolicy(
       max_requests=5,
       window_seconds=60,
   )
   
   bot = BotOS(
       agent=agent,
       platforms=["telegram"],
       rate_limit_policy=rate_limit,
   )
   bot.start()

2. With lockout cooldown

   After hitting the ceiling, lock out the identity for an additional cooldown period.

   from praisonaiagents import Agent
   from praisonaiagents.gateway import SlidingWindowRateLimitPolicy
   from praisonai.bots import BotOS
   
   agent = Agent(name="Support", instructions="Help users")
   
   rate_limit = SlidingWindowRateLimitPolicy(
       max_requests=5,
       window_seconds=60,
       lockout_seconds=300,   # 5-minute penalty after abuse
   )
   
   bot = BotOS(
       agent=agent,
       platforms=["telegram"],
       rate_limit_policy=rate_limit,
   )
   bot.start()

3. YAML configuration

   Configure via gateway.yaml. For a plain per-minute cap, gateway.per_user_per_minute: N is all you need; use the rate_limit block when you need a custom window or a lockout:

   gateway:
     rate_limit:
       max_requests: 5
       window_seconds: 60
       lockout_seconds: 0    # 0 = no lockout, deny until window rolls over

4. Custom limiter

   Implement RateLimitPolicyProtocol for Redis-backed, per-tenant, or cost-based limits:

   import redis
   from praisonaiagents.gateway import RateLimitDecision, RateLimitPolicyProtocol
   from praisonai.bots import BotOS
   from praisonaiagents import Agent
   
   redis_client = redis.Redis(host="localhost", port=6379, db=0)
   
   class RedisRateLimitPolicy:
       def check(self, *, identity: str, scope: str, now: float) -> RateLimitDecision:
           key = f"rl:{scope}:{identity}"
           count = redis_client.incr(key)
           if count == 1:
               redis_client.expire(key, 60)
           if count > 10:
               return RateLimitDecision(allowed=False, retry_after_seconds=60.0)
           return RateLimitDecision(allowed=True)
   
   assert isinstance(RedisRateLimitPolicy(), RateLimitPolicyProtocol)
   
   agent = Agent(name="Support", instructions="Help users")
   bot = BotOS(agent=agent, platforms=["telegram"], rate_limit_policy=RedisRateLimitPolicy())
   bot.start()

How It Works

Configuration Options

SlidingWindowRateLimitPolicy accepts three knobs:

from praisonaiagents.gateway import SlidingWindowRateLimitPolicy

policy = SlidingWindowRateLimitPolicy(
    max_requests=10,
    window_seconds=60.0,
    lockout_seconds=0.0,
)

print(policy.enabled)   # True when max_requests > 0

Decision Table

Per (scope, identity) key:

Choosing Lockout vs. No Lockout

Stable identity

gateway.per_user_per_minute charges each turn to a caller identity the client cannot reset.

Rotating a bearer token, cycling x-session-id, or opening a new WebSocket conversation does not mint a fresh budget — the caller keeps hitting the same token- or IP-keyed budget.

Building a Custom Limiter

Any object with a check(*, identity, scope, now) method satisfies RateLimitPolicyProtocol. The signature must be keyword-only.

from praisonaiagents.gateway import RateLimitDecision, RateLimitPolicyProtocol

class TenantRateLimitPolicy:
    def __init__(self, limits: dict[str, int]):
        self._limits = limits
        self._counts: dict[tuple, int] = {}

    def check(self, *, identity: str, scope: str, now: float) -> RateLimitDecision:
        limit = self._limits.get(identity, 5)
        key = (scope, identity)
        count = self._counts.get(key, 0) + 1
        self._counts[key] = count
        if count > limit:
            return RateLimitDecision(allowed=False, retry_after_seconds=60.0)
        return RateLimitDecision(allowed=True)

policy = TenantRateLimitPolicy(limits={"premium-tenant": 100, "free-tenant": 10})

assert isinstance(policy, RateLimitPolicyProtocol)

Key requirements:

• Keyword-only args — check(*, identity, scope, now) — positional calls will raise TypeError.
• Return a closed RateLimitDecision — never None.
• @runtime_checkable — isinstance(policy, RateLimitPolicyProtocol) works at runtime.

State Ownership

Best Practices

Related