Skip to main content
The gateway now ships in the praisonai-bot package. praisonai serve gateway still works exactly as documented here; for a standalone install see praisonai-bot Migration.
from praisonaiagents import Agent

agent = Agent(name="assistant", instructions="Be helpful.")
# Gateway rate limiting queues or rejects excess requests before the agent runs
agent.start("Status update?")
Rate limiting caps how many requests a given identity can make within a time window, returning a retry_after_seconds hint when they exceed it. 
from praisonaiagents.gateway import SlidingWindowRateLimitPolicy

policy = SlidingWindowRateLimitPolicy(max_requests=100, window_seconds=60)
The user or bot sends high-volume traffic; gateway rate limiting protects agents and downstream APIs.

​
How It Works

​
Quick Start

1

Create a sliding-window policy

from praisonaiagents.gateway import SlidingWindowRateLimitPolicy

policy = SlidingWindowRateLimitPolicy(
    max_requests=100,     # 100 requests per window
    window_seconds=60,    # 60-second rolling window
    lockout_seconds=300,  # 5-minute cooldown after exceeding limit
)
2

Inject into the gateway

from praisonai.bots import BotOS
from praisonaiagents import Agent
from praisonaiagents.gateway import SlidingWindowRateLimitPolicy

policy = SlidingWindowRateLimitPolicy(max_requests=60, window_seconds=60)

agent = Agent(name="SupportBot", instructions="Help users.")
bots = BotOS(agent=agent, platforms=["telegram"], rate_limit_policy=policy)
bots.run()

​
Wire Response

A limited decision maps to ConnectErrorCode.RATE_LIMITED on the handshake wire: 
{
  "type": "hello_error",
  "code": "rate_limited",
  "retry_after_seconds": 42.0
}
Clients read retry_after_seconds and back off before reconnecting. This keeps retry storms from amplifying the load that triggered limiting in the first place.

​
SlidingWindowRateLimitPolicy Reference

ParameterTypeDefaultDescription
max_requestsint0Maximum requests per window. 0 disables limiting — every request is allowed (legacy always-allow behaviour).
window_secondsfloat60.0Rolling window duration in seconds. Must be > 0.
lockout_secondsfloat0.0Cooldown after exceeding the ceiling. 0 means no cooldown — the key is denied until the window expires naturally.
from praisonaiagents.gateway import SlidingWindowRateLimitPolicy

# Disabled (legacy always-allow)
SlidingWindowRateLimitPolicy(max_requests=0)

# 100 req/min, no cooldown
SlidingWindowRateLimitPolicy(max_requests=100, window_seconds=60)

# 5 req/min with 5-minute lockout for abusive clients
SlidingWindowRateLimitPolicy(max_requests=5, window_seconds=60, lockout_seconds=300)

​
RateLimitDecision Fields

from praisonaiagents.gateway import RateLimitDecision
RateLimitDecision is a frozen dataclass — immutable by design so it can be safely returned across threads.
FieldTypeDescription
allowedboolWhether the request may proceed
retry_after_secondsfloat | NoneBackoff hint when allowed=False; None when allowed
RateLimitDecision is frozen (@dataclass(frozen=True)). Never try to mutate it — create a new instance instead.

​
Bring Your Own Limiter

Implement RateLimitPolicyProtocol for per-tenant, distributed, or cost-based limiting: 
from praisonaiagents.gateway import RateLimitPolicyProtocol, RateLimitDecision

class TenantRateLimitPolicy:
    """Per-tenant limiter — example sketch, no real Redis dependency."""

    def __init__(self, limits: dict):
        # e.g. {"tenant_a": 1000, "tenant_b": 100}
        self._limits = limits
        self._counters: dict = {}

    def check(self, *, identity: str, scope: str, now: float) -> RateLimitDecision:
        limit = self._limits.get(identity, 50)   # default 50 req/min
        count = self._counters.get(identity, 0)
        if count >= limit:
            return RateLimitDecision(allowed=False, retry_after_seconds=60.0)
        self._counters[identity] = count + 1
        return RateLimitDecision(allowed=True)

policy = TenantRateLimitPolicy(limits={"premium": 1000, "free": 60})
RateLimitPolicy is a backward-compatible alias for RateLimitPolicyProtocol. Both names import from praisonaiagents.gateway.

​
The Gateway Policy Family

Rate limiting completes the gateway’s symmetric policy-protocol family:
ProtocolWhat it decides
SendPolicyProtocolWhether to send an outbound frame now
GatewayIdlePolicyProtocolWhether the gateway is idle enough to scale to zero
GatewayDrainPolicyProtocolTiming and order of graceful drain
GatewayConcurrencyPolicyProtocolBound on concurrent in-flight turns
RateLimitPolicyProtocol (new)Bound on inbound request rate per identity/scope
All five follow the same shape: pure, stateless decision over typed facts, with a frozen decision dataclass returned.

​
Best Practices

SlidingWindowRateLimitPolicy is dependency-free and covers the common case. Only inject a custom policy when you need per-tenant fairness, shared state (Redis), or cost-based limits.
A 5-minute lockout after 5 exceeded requests in a minute prevents a script from hammering the gateway every 60 s. The retry_after_seconds hint in the response lets well-behaved clients back off cleanly.
Clients respect retry_after_seconds. A hint of 0 invites immediate retry; a hint matching your actual window (window_seconds) gives the client a correct backoff. Use lockout_seconds when you want a longer penalty than the window naturally provides.

Gateway

WebSocket control plane — full YAML and Python reference

Reliability Preset

One-switch production posture (includes admission control)

Admission Control

Bound concurrent turns with a fair queue and overflow policy

Gateway Security

Operator authentication and scope-based access control