22. Plugin Development SDK

OpenCLaw supports extensibility through plugins. The plugin SDK provides interfaces for extending core functionality while maintaining system stability and security.

Plugin Architecture

Plugins register capabilities during initialization and communicate with the core through defined interfaces.

# plugin_sdk.py
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Dict, Any, List, Optional
import importlib.util
import os

@dataclass
class PluginMetadata:
    name: str
    version: str
    author: str
    description: str
    dependencies: List[str] = None

class PluginInterface(ABC):
    @abstractmethod
    def get_metadata(self) -> PluginMetadata:
        pass
    
    @abstractmethod
    def initialize(self, context: PluginContext) -> bool:
        pass
    
    @abstractmethod
    def shutdown(self):
        pass

class PluginContext:
    def __init__(self, core_api: 'CoreAPI'):
        self.api = core_api
        self.config = {}
        self.logger = core_api.get_logger()

class CoreAPI:
    def send_message(self, message: str, channel: str = 'default') -> str:
        pass
    
    def register_capability(self, plugin_name: str, capability: Capability):
        pass
    
    def get_logger(self):
        pass

class PluginManager:
    def __init__(self, plugin_directory: str, core_api: CoreAPI):
        self.directory = plugin_directory
        self.api = core_api
        self.loaded_plugins: Dict[str, PluginInterface] = {}
    
    def discover_plugins(self) -> List[str]:
        plugin_specs = []
        for filename in os.listdir(self.directory):
            if filename.endswith('.py') and not filename.startswith('_'):
                plugin_specs.append(os.path.join(self.directory, filename))
        return plugin_specs
    
    def load_plugin(self, plugin_path: str) -> Optional[PluginInterface]:
        spec = importlib.util.spec_from_file_location("plugin", plugin_path)
        module = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(module)
        
        plugin_class = getattr(module, 'Plugin')
        plugin_instance = plugin_class()
        
        context = PluginContext(self.api)
        if plugin_instance.initialize(context):
            metadata = plugin_instance.get_metadata()
            self.loaded_plugins[metadata.name] = plugin_instance
            return plugin_instance
        
        return None
    
    def unload_plugin(self, plugin_name: str):
        plugin = self.loaded_plugins.get(plugin_name)
        if plugin:
            plugin.shutdown()
            del self.loaded_plugins[plugin_name]

Capability Registration

Plugins declare capabilities that integrate with OpenCLaw's processing pipeline.

from enum import Enum

class CapabilityType(Enum):
    MESSAGE_PROCESSOR = "message_processor"
    CONTEXT_ENHANCER = "context_enhancer"
    ACTION_HANDLER = "action_handler"
    OUTPUT_FORMATTER = "output_formatter"

@dataclass
class Capability:
    name: str
    ctype: CapabilityType
    priority: int
    handler: Any

class MessageProcessorPlugin(PluginInterface):
    def process_message(self, message: str, context: Dict) -> Dict:
        pass

class CapabilityRegistry:
    def __init__(self):
        self.capabilities: Dict[CapabilityType, List[Capability]] = {
            ct: [] for ct in CapabilityType
        }
    
    def register(self, capability: Capability):
        self.capabilities[capability.ctype].append(capability)
        self.capabilities[capability.ctype].sort(key=lambda c: c.priority)
    
    def get_capabilities(self, ctype: CapabilityType) -> List[Capability]:
        return self.capabilities.get(ctype, [])
    
    def unregister(self, capability_name: str):
        for capabilities in self.capabilities.values():
            capabilities[:] = [c for c in capabilities if c.name != capability_name]

Plugin Security

Plugins execute with elevated access and require sandboxing. The SDK provides capability限制了 that prevent plugins from accessing unauthorized resources.

class PluginSandbox:
    def __init__(self):
        self.allowed_modules = ['json', 'datetime', 'math', 're']
        self.denied_paths = ['/etc', '/root', '/home']
    
    def validate_plugin(self, plugin_code: str) -> bool:
        if 'import os' in plugin_code:
            return False
        if 'subprocess' in plugin_code:
            return False
        if 'open(' in plugin_code:
            return False
        return True