Building a Plugin System That Developers Love

When we set out to build Kollab CLI, we had a simple goal: create a terminal AI that adapts to how you work. The solution was a plugin system built around a single philosophy—everything has hooks. This is the story of how we built a plugin system developers actually enjoy using.

The Design Philosophy

Most plugin systems feel like an afterthought—tacked on, clunky, and limited. We wanted something different. Our plugin system needed to satisfy three core principles:

Omnipresent hooks—Every interaction, from keypresses to LLM responses, should be interceptable
Zero boilerplate—Creating a plugin should require minimal code
Safe discovery—Plugins should be discoverable with security validation

The result is a system with 30+ hook points across the entire application lifecycle, from startup to shutdown. Plugins can intercept input, transform LLM requests, process responses, register slash commands, and even contribute widgets to a dynamic dashboard.

Plugin Discovery and Loading

Plugins are discovered automatically from the plugins/ directory. The PluginDiscovery class handles scanning, validation, and module loading with security built in:

class PluginDiscovery:
    def scan_plugin_files(self) -> List[str]:
        """Scan for *_plugin.py files with security validation."""
        for plugin_file in plugins_dir.rglob("*_plugin.py"):
            # Validate each path component
            for part in rel_path.parts:
                if not self._sanitize_plugin_name(part):
                    continue  # Skip invalid plugins

            # Verify plugin location (prevent path traversal)
            if not self._verify_plugin_location(module_path):
                continue

            discovered.append(module_path)

Security is paramount. Every plugin name is validated against a regex pattern, path traversal attempts are blocked, and file permissions are checked. The discovery system supports both file-based plugins (my_plugin.py) and directory-based packages (my_plugin/__init__.py).

Once discovered, the PluginFactory handles instantiation with dependency injection. Every plugin receives access to the event bus, terminal renderer, and configuration manager:

instance = instantiate_plugin_safely(
    plugin_class,
    name="my_plugin",
    event_bus=event_bus,
    renderer=renderer,
    config=config,
)

The KollaborPluginSDK

For developers who want to extend LLM functionality with custom tools, the KollaborPluginSDK provides a streamlined interface. It handles tool registration, validation, and execution:

class KollaborPluginSDK:
    def register_custom_tool(self, tool_definition: Dict[str, Any]) -> bool:
        """Register a custom tool for LLM use."""
        required_fields = ["name", "description", "handler"]
        if not all(field in tool_definition for field in required_fields):
            return False

        self.custom_tools[tool_definition["name"]] = {
            "definition": tool_definition,
            "enabled": tool_definition.get("enabled", True),
            "plugin": tool_definition.get("plugin", "unknown"),
        }
        return True

The SDK also includes a plugin template generator that scaffolds new plugins with all the required methods pre-configured.

Hook Registration

At the heart of the system is the hook registry. Plugins register hooks during initialization with specific event types and priorities. The priority system (SYSTEM=1000, SECURITY=900, PREPROCESSING=500, LLM=100, POSTPROCESSING=50, DISPLAY=10) ensures hooks execute in the right order:

async def register_hooks(self) -> None:
    """Register plugin hooks with the event bus."""
    await self.event_bus.register_hook(
        Hook(
            name="monitor_user_input",
            plugin_name=self.name,
            event_type=EventType.USER_INPUT,
            priority=HookPriority.POSTPROCESSING.value,
            callback=self._log_hook_execution,
            timeout=5,
        )
    )

The EventType enum defines 30+ event types covering the entire application lifecycle:

class EventType(Enum):
    # User input events
    USER_INPUT_PRE = "user_input_pre"
    USER_INPUT = "user_input"
    USER_INPUT_POST = "user_input_post"

    # Key press events
    KEY_PRESS_PRE = "key_press_pre"
    KEY_PRESS = "key_press"
    KEY_PRESS_POST = "key_press_post"

    # LLM events
    LLM_REQUEST_PRE = "llm_request_pre"
    LLM_REQUEST = "llm_request"
    LLM_RESPONSE_PRE = "llm_response_pre"
    LLM_RESPONSE = "llm_response"
    LLM_THINKING = "llm_thinking"

    # Tool events
    TOOL_CALL_PRE = "tool_call_pre"
    TOOL_CALL = "tool_call"
    TOOL_CALL_POST = "tool_call_post"

    # System events
    SYSTEM_STARTUP = "system_startup"
    SYSTEM_READY = "system_ready"
    SYSTEM_SHUTDOWN = "system_shutdown"

Slash Commands Made Simple

Plugins can register slash commands that appear in the interactive command menu. The CommandDefinition class makes this straightforward:

save_command = CommandDefinition(
    name="save",
    description="Save conversation to file or clipboard",
    handler=self._handle_save_command,
    plugin_name=self.name,
    aliases=["export", "transcript"],
    mode=CommandMode.INSTANT,
    category=CommandCategory.CONVERSATION,
    subcommands=[
        SubcommandInfo("transcript", "[clipboard|both|local]", "Plain text format"),
        SubcommandInfo("markdown", "[clipboard|both|local]", "Markdown format"),
        SubcommandInfo("jsonl", "[clipboard|both|local]", "JSON lines format"),
    ],
)
self.command_registry.register_command(save_command)

Commands support multiple interaction modes (INSTANT, MODAL, STATUS_TAKEOVER), aliases, categories, and subcommands that appear dynamically in the menu.

Real-World Plugin Examples

Tmux Plugin: Terminal Session Management

The TmuxPlugin demonstrates the full power of the system. It registers slash commands for managing tmux sessions, provides a live modal view with keyboard passthrough, and contributes a status widget:

class TmuxPlugin:
    async def _handle_view_session(self, args: List[str]) -> CommandResult:
        """View a tmux session live in alt buffer."""
        # Content generator for live modal
        async def get_tmux_content() -> List[str]:
            header = f"[Session: {self._current_session}]"
            content = self._capture_tmux_pane(self._current_session)
            return [header, "─" * len(header)] + content

        # Input callback for keyboard passthrough
        async def handle_input(key_press) -> bool:
            if key_press.name == "Escape":
                return True  # Exit modal

            # Forward keys to tmux
            if key_press.char:
                self._send_keys_to_tmux(self._current_session, key_press.char)
            return False

        # Emit live modal trigger event
        await self.event_bus.emit_with_hooks(
            EventType.LIVE_MODAL_TRIGGER,
            {
                "content_generator": get_tmux_content,
                "config": LiveModalConfig(refresh_rate=2.0),
                "input_callback": handle_input,
            },
        )

Save Conversation Plugin

The SaveConversationPlugin shows how to add export functionality with multiple format support:

class SaveConversationPlugin:
    async def _handle_save_command(self, command) -> str:
        """Handle the /save command with format selection."""
        args = command.args if hasattr(command, "args") else []

        # Parse format: transcript, markdown, jsonl, raw
        save_format = args[0] if args else "transcript"
        save_to = args[1] if len(args) >= 2 else "file"

        # Get conversation from LLM service
        conversation_history = self.llm_service.conversation_history
        messages = [{"role": msg.role, "content": msg.content} for msg in conversation_history]

        # Format and save
        formatted_content = self._format_conversation(messages, save_format)
        if save_to in ["file", "both"]:
            saved_path = self._save_to_file(formatted_content, output_dir, save_format)
        if save_to in ["clipboard", "both"]:
            self._copy_to_clipboard(formatted_content)

        return f"Conversation saved to {saved_path}"

Hook Monitoring Plugin: The Showcase

The HookMonitoringPlugin is our demonstration plugin that showcases all ecosystem features. It monitors hook execution performance, discovers other plugins, registers monitoring services via the SDK, and demonstrates cross-plugin communication:

class HookMonitoringPlugin:
    def _create_all_hooks(self) -> List[Hook]:
        """Create comprehensive hooks for all event types."""
        hooks = []
        timeout = self.config.get("plugins.hook_monitoring.hook_timeout", 5)

        # Register for every event type
        hooks.extend([
            Hook(name="monitor_user_input_pre", ...),
            Hook(name="monitor_key_press", ...),
            Hook(name="monitor_llm_request", ...),
            Hook(name="monitor_tool_call", ...),
            # ... 30+ hooks total
        ])
        return hooks

    async def _register_monitoring_services(self) -> None:
        """Register services for other plugins to use."""
        self.sdk.register_custom_tool({
            "name": "monitor_performance",
            "description": "Monitor plugin performance",
            "handler": self._provide_performance_monitoring,
        })

Configuration and Widgets

Each plugin can define its default configuration and contribute widgets to the configuration modal. The get_config_widgets() method returns a widget definition:

@staticmethod
def get_config_widgets() -> Dict[str, Any]:
    return {
        "title": "Hook Monitoring Plugin",
        "widgets": [
            {
                "type": "checkbox",
                "label": "Debug Logging",
                "config_path": "plugins.hook_monitoring.debug_logging",
                "help": "Enable detailed debug logging for hooks",
            },
            {
                "type": "slider",
                "label": "Hook Timeout",
                "config_path": "plugins.hook_monitoring.hook_timeout",
                "min_value": 1,
                "max_value": 30,
                "step": 1,
            },
        ],
    }

Supported widget types include checkbox, slider, text input, and dropdown. The configuration system supports dot notation for nested access (config.get("plugins.myplugin.enabled")).

Building Your First Plugin

Creating a plugin is straightforward. Here's a minimal example that logs every keypress:

"""My first Kollabor plugin."""

from typing import Any, Dict
from core.events import EventType, Hook, HookPriority
import logging

logger = logging.getLogger(__name__)

class MyFirstPlugin:
    def __init__(self, name: str, event_bus, renderer, config):
        self.name = name
        self.event_bus = event_bus
        self.renderer = renderer
        self.config = config

    async def initialize(self, event_bus, config, **kwargs):
        """Initialize the plugin."""
        self.command_registry = kwargs.get("command_registry")
        self._register_commands()

    async def register_hooks(self):
        """Register hooks."""
        await self.event_bus.register_hook(
            Hook(
                name="log_keypress",
                plugin_name=self.name,
                event_type=EventType.KEY_PRESS,
                priority=HookPriority.POSTPROCESSING.value,
                callback=self._log_keypress,
            )
        )

    async def _log_keypress(self, data: dict, event) -> dict:
        """Log each keypress."""
        logger.info(f"Key pressed: {data.get('key')}")
        return data

    async def shutdown(self):
        """Cleanup."""
        logger.info("MyFirstPlugin shut down")

    @staticmethod
    def get_default_config() -> Dict[str, Any]:
        return {"plugins": {"my_first": {"enabled": True}}}

Save this as plugins/my_first_plugin.py and it will be automatically discovered on startup.

Lessons Learned

Building this system taught us a few things about plugin architecture:

Async everywhere—Plugin hooks are async by default. This prevents blocking the main event loop and makes the UI responsive even when plugins do heavy work.
Graceful degradation—A single plugin failure shouldn't crash the app. All hook execution is wrapped in error handlers that log failures without stopping the chain.
Dependency injection—Plugins receive their dependencies through __init__ rather than importing them directly. This makes testing easier and enables mocking during development.
Priority matters—The hook priority system is critical. Security hooks need to run before user input hooks, and display hooks need to run after data processing hooks.

What's Next

We're continuously expanding the plugin system. Future plans include:

A plugin marketplace for sharing community plugins
Sandboxed plugin execution for untrusted plugins
Hot-reload for faster plugin development
Plugin composition—plugins that depend on other plugins

The full plugin documentation is available on GitHub. If you build something cool, let us know!