Plugin Tutorial: Build a Rate Limiter¶

This tutorial walks through building a complete plugin from scratch — a rate limiter that restricts how frequently each user can send messages.

What We're Building¶

A plugin that: - Tracks message timestamps per user - Rejects messages if the user sends too fast - Exposes /ratelimit command to check current status - Is configurable via config.yaml

Step 1: Create the file¶

# In your project
touch plugins/rate_limiter_plugin.py

Step 2: Skeleton¶

Start with the minimum:

from simplecontext.plugins.base import BasePlugin

class RateLimiterPlugin(BasePlugin):
    name        = "rate_limiter_plugin"
    version     = "1.0.0"
    description = "Limit message frequency per user."

Step 3: Config and setup¶

from simplecontext.plugins.base import BasePlugin

class RateLimiterPlugin(BasePlugin):
    name        = "rate_limiter_plugin"
    version     = "1.0.0"
    description = "Limit message frequency per user."

    def setup(self):
        # How many messages allowed per window
        self.max_messages = int(self.config.get("max_messages", 10))
        # Window size in seconds
        self.window_secs  = int(self.config.get("window_seconds", 60))
        # Message to send when rate limited
        self.block_msg    = self.config.get(
            "block_message",
            f"⚠️ Slow down! Max {self.max_messages} messages per minute."
        )

Step 4: Track messages¶

Use self.state for persistent storage:

    import time

    def _get_timestamps(self, user_id: str) -> list:
        """Get message timestamps for a user."""
        return self.state.get(f"ts:{user_id}", [])

    def _save_timestamps(self, user_id: str, timestamps: list):
        """Save message timestamps."""
        self.state.set(f"ts:{user_id}", timestamps)

    def _is_rate_limited(self, user_id: str) -> bool:
        """Check if user has exceeded the rate limit."""
        now = time.time()
        window_start = now - self.window_secs

        # Get timestamps within window
        timestamps = self._get_timestamps(user_id)
        recent = [ts for ts in timestamps if ts > window_start]

        # Add current timestamp
        recent.append(now)
        self._save_timestamps(user_id, recent)

        return len(recent) > self.max_messages

Step 5: Hook into the pipeline¶

Use on_before_llm to intercept messages before they reach the LLM:

    def on_before_llm(self, user_id, agent_id, messages):
        """Block the request if user is rate limited."""
        if self._is_rate_limited(user_id):
            self.state.increment("total_blocked")
            # Replace last user message with block message
            for i in range(len(messages) - 1, -1, -1):
                if messages[i]["role"] == "user":
                    messages[i]["content"] = self.block_msg
                    break
        return messages

Step 6: Add an app_command¶

Let users check their current rate limit status:

from simplecontext.plugins.base import BasePlugin, AppCommandContext
import time

class RateLimiterPlugin(BasePlugin):
    # ... (previous code) ...

    app_commands = {
        "ratelimit": {
            "description": "Check your current message rate",
            "usage":       "/ratelimit",
            "handler":     "handle_ratelimit",
        }
    }

    async def handle_ratelimit(self, ctx: AppCommandContext) -> str:
        now          = time.time()
        window_start = now - self.window_secs
        timestamps   = self._get_timestamps(ctx.user_id)
        recent_count = len([ts for ts in timestamps if ts > window_start])
        remaining    = max(0, self.max_messages - recent_count)
        total_blocked = self.state.get("total_blocked", 0)

        return (
            f"📊 *Rate Limit Status*\n\n"
            f"Messages this minute: `{recent_count}/{self.max_messages}`\n"
            f"Remaining: `{remaining}`\n"
            f"Window: `{self.window_secs}s`\n"
            f"Total blocked (all users): `{total_blocked}`"
        )

Step 7: Complete plugin¶

"""
rate_limiter_plugin.py — SimpleContext Plugin
Limit message frequency per user to prevent spam.

Config in config.yaml:
    plugins:
      rate_limiter_plugin:
        enabled: true
        max_messages: 10       # per window
        window_seconds: 60     # window size in seconds
        block_message: "⚠️ Too fast! Please slow down."
"""

import time
from simplecontext.plugins.base import BasePlugin, AppCommandContext


class RateLimiterPlugin(BasePlugin):
    name        = "rate_limiter_plugin"
    version     = "1.0.0"
    description = "Limit message frequency per user."
    depends_on  = []

    app_commands = {
        "ratelimit": {
            "description": "Check your current message rate",
            "usage":       "/ratelimit",
            "handler":     "handle_ratelimit",
        }
    }

    def setup(self):
        self.max_messages = int(self.config.get("max_messages", 10))
        self.window_secs  = int(self.config.get("window_seconds", 60))
        self.block_msg    = self.config.get(
            "block_message",
            f"⚠️ Slow down! Max {self.max_messages} messages per {self.window_secs}s."
        )

    def _get_timestamps(self, user_id: str) -> list:
        return self.state.get(f"ts:{user_id}", [])

    def _save_timestamps(self, user_id: str, timestamps: list):
        self.state.set(f"ts:{user_id}", timestamps)

    def _is_rate_limited(self, user_id: str) -> bool:
        now          = time.time()
        window_start = now - self.window_secs
        timestamps   = self._get_timestamps(user_id)
        recent       = [ts for ts in timestamps if ts > window_start]
        recent.append(now)
        self._save_timestamps(user_id, recent)
        return len(recent) > self.max_messages

    def on_before_llm(self, user_id, agent_id, messages):
        if self._is_rate_limited(user_id):
            self.state.increment("total_blocked")
            for i in range(len(messages) - 1, -1, -1):
                if messages[i]["role"] == "user":
                    messages[i]["content"] = self.block_msg
                    break
        return messages

    async def handle_ratelimit(self, ctx: AppCommandContext) -> str:
        now          = time.time()
        window_start = now - self.window_secs
        timestamps   = self._get_timestamps(ctx.user_id)
        recent_count = len([ts for ts in timestamps if ts > window_start])
        remaining    = max(0, self.max_messages - recent_count)
        blocked      = self.state.get("total_blocked", 0)

        return (
            f"📊 *Rate Limit Status*\n\n"
            f"Messages this window: `{recent_count}/{self.max_messages}`\n"
            f"Remaining: `{remaining}`\n"
            f"Window: `{self.window_secs}s`\n"
            f"Total blocked: `{blocked}`"
        )

Step 8: Test it¶

from simplecontext import SimpleContext
from plugins.rate_limiter_plugin import RateLimiterPlugin

sc = SimpleContext()
sc.use(RateLimiterPlugin(config={
    "max_messages": 3,
    "window_seconds": 10,
}))

# Verify it loaded
plugin = sc._plugins.get("rate_limiter_plugin")
print(plugin)
# → <Plugin name='rate_limiter_plugin' v1.0.0 enabled=True deps=[] commands=['ratelimit']>

# Check app_commands
print(plugin.get_app_commands())
# → {"ratelimit": {"description": "...", "handler": "handle_ratelimit"}}

Step 9: Configure and deploy¶

# config.yaml
plugins:
  rate_limiter_plugin:
    enabled: true
    max_messages: 10
    window_seconds: 60
    block_message: "⚠️ Too many messages! Please wait a moment."

With SimpleContext-Bot, /ratelimit is automatically registered as a Telegram command.

What You Learned¶

Plugin file structure and BasePlugin inheritance
Using self.config for configuration
Using self.state for persistent per-user data
on_before_llm hook for intercepting requests
app_commands and AppCommandContext for bot commands
Testing plugin loading and introspection