22. Plugin Architecture

Plugins let users extend your framework without modifying core code. Built well, they enable customization. Built poorly, they create maintenance nightmares.

Plugin Interface Definition

Define clear contracts:

from abc import ABC, abstractmethod
from typing import Any, Protocol

class AgentPlugin(Protocol):
    """Protocol that all plugins must implement."""
    
    @property
    def name(self) -> str:
        """Unique identifier for this plugin."""
        ...
    
    @property
    def version(self) -> str:
        """Semantic version string."""
        ...
    
    async def initialize(self, config: dict[str, Any]) -> None:
        """Called once when plugin is loaded."""
        ...
    
    async def process(self, input_data: Any) -> Any:
        """Process input and return output."""
        ...

Plugin Lifecycle

class PluginManager:
    def __init__(self):
        self._plugins: dict[str, AgentPlugin] = {}
        self._hooks: dict[str, list[callable]] = {
            "pre_process": [],
            "post_process": [],
        }
    
    def register(self, plugin: AgentPlugin):
        if plugin.name in self._plugins:
            raise ValueError(f"Plugin {plugin.name} already registered")
        
        self._plugins[plugin.name] = plugin
        # Run plugin initialization
        asyncio.create_task(plugin.initialize({}))
    
    def register_hook(self, hook_name: str, callback: callable):
        if hook_name not in self._hooks:
            raise ValueError(f"Unknown hook: {hook_name}")
        self._hooks[hook_name].append(callback)
    
    async def run_hooks(self, hook_name: str, *args, **kwargs):
        for callback in self._hooks.get(hook_name, []):
            await callback(*args, **kwargs)

Sandboxing Plugins

Plugins run with the same privileges as your framework unless sandboxed:

import sys
import tracemalloc

class SandboxedPlugin:
    """Wrapper that monitors plugin resource usage."""
    
    def __init__(self, plugin: AgentPlugin, memory_limit_mb: int = 512):
        self.plugin = plugin
        self.memory_limit = memory_limit_mb * 1024 * 1024
    
    async def process(self, input_data: Any) -> Any:
        tracemalloc.start()
        try:
            result = await self.plugin.process(input_data)
            
            current, peak = tracemalloc.get_traced_memory()
            if peak > self.memory_limit:
                raise RuntimeError(
                    f"Plugin {self.plugin.name} exceeded memory limit: "
                    f"{peak / 1024 / 1024:.1f}MB > {self.memory_limit / 1024 / 1024}MB"
                )
            return result
        finally:
            tracemalloc.stop()

Plugin Loading

import importlib

def load_plugin_from_module(module_path: str) -> AgentPlugin:
    """Load a plugin from a dotted module path."""
    module = importlib.import_module(module_path)
    
    # Find the plugin class (convention: class name ends with Plugin)
    for name in dir(module):
        obj = getattr(module, name)
        if isinstance(obj, type) and name.endswith('Plugin'):
            return obj()
    
    raise ValueError(f"No Plugin class found in {module_path}")