#!/usr/bin/env python3 """ Plugin Skill Loader - Load and integrate plugin capabilities into Luzia skill system Bridges Claude marketplace plugins with Luzia's skill matching and task dispatch system. Features: 1. Load plugin capabilities as Luzia skills 2. Create skill metadata from plugin definitions 3. Integrate with responsive dispatcher 4. Cache skills for performance 5. Track plugin skill usage 6. Provide plugin-to-skill mapping """ import json from pathlib import Path from typing import Dict, List, Optional, Any, Set from datetime import datetime import logging from dataclasses import dataclass from plugin_marketplace import ( PluginMarketplaceRegistry, PluginCapabilityMatcher, MarketplacePlugin, PluginCapability ) logger = logging.getLogger(__name__) @dataclass class PluginSkill: """Skill derived from a plugin capability""" skill_id: str name: str description: str plugin_id: str plugin_name: str capability_name: str category: str tags: List[str] trust_level: str keywords: List[str] metadata: Dict[str, Any] def to_dict(self): return { 'skill_id': self.skill_id, 'name': self.name, 'description': self.description, 'plugin_id': self.plugin_id, 'plugin_name': self.plugin_name, 'capability_name': self.capability_name, 'category': self.category, 'tags': self.tags, 'trust_level': self.trust_level, 'keywords': self.keywords, 'metadata': self.metadata } class PluginSkillLoader: """ Loads plugin capabilities as Luzia skills Converts plugin marketplace definitions into executable skills that can be matched to tasks and integrated into the responsive dispatcher. """ def __init__(self, registry: Optional[PluginMarketplaceRegistry] = None, cache_dir: Optional[Path] = None): """Initialize skill loader Args: registry: Plugin marketplace registry (created if not provided) cache_dir: Directory for caching skills """ self.registry = registry or PluginMarketplaceRegistry() self.cache_dir = cache_dir or Path("/tmp/.luzia-plugin-skills") self.cache_dir.mkdir(parents=True, exist_ok=True) self.skills: Dict[str, PluginSkill] = {} self.skill_index: Dict[str, Set[str]] = {} # keyword -> skill_ids self.category_index: Dict[str, Set[str]] = {} # category -> skill_ids self.plugin_skill_map: Dict[str, List[str]] = {} # plugin_id -> skill_ids self.matcher = PluginCapabilityMatcher(self.registry) self.load_cache() def load_cache(self) -> None: """Load cached skills from disk""" cache_file = self.cache_dir / "skills.json" if cache_file.exists(): try: data = json.loads(cache_file.read_text()) self._rebuild_from_dict(data) logger.info(f"Loaded {len(self.skills)} plugin skills from cache") except Exception as e: logger.warning(f"Failed to load skill cache: {e}") def generate_skills_from_plugins(self) -> Dict[str, PluginSkill]: """Generate skills from all registered plugins Returns: Dict of generated skills """ skills = {} plugins = self.registry.list_plugins() for plugin in plugins: plugin_skills = self._plugin_to_skills(plugin) skills.update(plugin_skills) self.plugin_skill_map[plugin.id] = list(plugin_skills.keys()) self.skills = skills self._rebuild_indices() self.save_cache() logger.info(f"Generated {len(skills)} skills from {len(plugins)} plugins") return skills def _plugin_to_skills(self, plugin: MarketplacePlugin) -> Dict[str, PluginSkill]: """Convert plugin capabilities to skills Args: plugin: Plugin to convert Returns: Dict of skills keyed by skill_id """ skills = {} for capability in plugin.capabilities: skill_id = f"{plugin.id}:{capability.name}" skill = PluginSkill( skill_id=skill_id, name=f"{capability.name} ({plugin.name})", description=capability.description, plugin_id=plugin.id, plugin_name=plugin.name, capability_name=capability.name, category=capability.category, tags=capability.tags, trust_level=plugin.trust_level, keywords=self._extract_keywords(capability), metadata={ 'plugin_url': plugin.url, 'plugin_vendor': plugin.vendor, 'plugin_version': plugin.version, 'requires_auth': getattr(capability, 'requires_auth', False) } ) skills[skill_id] = skill return skills def _extract_keywords(self, capability: PluginCapability) -> List[str]: """Extract keywords from capability for matching Args: capability: Plugin capability Returns: List of keywords """ keywords = list(capability.tags) + [capability.category] # Add derived keywords from description description_lower = capability.description.lower() keyword_patterns = { 'security': ['secure', 'vulnerability', 'threat', 'exploit'], 'performance': ['speed', 'optimization', 'benchmark', 'latency'], 'analysis': ['analyze', 'inspect', 'examine', 'review'], 'code': ['code', 'coding', 'programming', 'developer'], 'integration': ['integrate', 'api', 'connect', 'interface'] } for keyword, patterns in keyword_patterns.items(): for pattern in patterns: if pattern in description_lower: if keyword not in keywords: keywords.append(keyword) break return keywords def _rebuild_indices(self) -> None: """Rebuild keyword and category indices""" self.skill_index = {} self.category_index = {} for skill_id, skill in self.skills.items(): # Index by keywords for keyword in skill.keywords: if keyword not in self.skill_index: self.skill_index[keyword] = set() self.skill_index[keyword].add(skill_id) # Index by category if skill.category not in self.category_index: self.category_index[skill.category] = set() self.category_index[skill.category].add(skill_id) def find_skills_for_task(self, task_description: str, min_relevance: float = 0.5) -> List[Dict[str, Any]]: """Find relevant plugin skills for a task Args: task_description: Description of the task min_relevance: Minimum relevance score (0-1) Returns: List of matched skills with relevance info """ if not self.skills: self.generate_skills_from_plugins() matched_plugins = self.matcher.match_plugins(task_description, min_relevance) matched_skills = [] for plugin_match in matched_plugins: plugin_id = plugin_match['id'] if plugin_id in self.plugin_skill_map: for skill_id in self.plugin_skill_map[plugin_id]: if skill_id in self.skills: skill = self.skills[skill_id] matched_skills.append({ 'skill_id': skill.skill_id, 'name': skill.name, 'description': skill.description, 'category': skill.category, 'plugin_id': skill.plugin_id, 'plugin_name': skill.plugin_name, 'relevance_score': plugin_match['relevance_score'], 'tags': skill.tags, 'trust_level': skill.trust_level }) # Sort by relevance matched_skills.sort(key=lambda x: x['relevance_score'], reverse=True) return matched_skills def get_skill(self, skill_id: str) -> Optional[PluginSkill]: """Get a skill by ID Args: skill_id: ID of the skill Returns: Skill or None if not found """ return self.skills.get(skill_id) def list_skills(self, category: Optional[str] = None, plugin_id: Optional[str] = None) -> List[PluginSkill]: """List available skills with optional filtering Args: category: Optional category filter plugin_id: Optional plugin filter Returns: List of skills """ skills = list(self.skills.values()) if category: skills = [s for s in skills if s.category == category] if plugin_id: skills = [s for s in skills if s.plugin_id == plugin_id] return skills def save_cache(self) -> None: """Save skills to cache""" cache_file = self.cache_dir / "skills.json" data = { 'timestamp': datetime.now().isoformat(), 'skill_count': len(self.skills), 'skills': { skill_id: skill.to_dict() for skill_id, skill in self.skills.items() }, 'plugin_skill_map': { pid: list(sids) for pid, sids in self.plugin_skill_map.items() } } cache_file.write_text(json.dumps(data, indent=2)) logger.info(f"Saved {len(self.skills)} plugin skills to cache") def _rebuild_from_dict(self, data: Dict[str, Any]) -> None: """Rebuild skills from cached data""" for skill_id, skill_data in data.get('skills', {}).items(): skill = PluginSkill( skill_id=skill_data['skill_id'], name=skill_data['name'], description=skill_data['description'], plugin_id=skill_data['plugin_id'], plugin_name=skill_data['plugin_name'], capability_name=skill_data['capability_name'], category=skill_data['category'], tags=skill_data['tags'], trust_level=skill_data['trust_level'], keywords=skill_data['keywords'], metadata=skill_data.get('metadata', {}) ) self.skills[skill_id] = skill self.plugin_skill_map = { pid: list(sids) for pid, sids in data.get('plugin_skill_map', {}).items() } self._rebuild_indices() def export_for_dispatcher(self) -> Dict[str, Any]: """Export skills in format suitable for responsive dispatcher Returns: Dict with dispatcher-compatible skill definitions """ return { 'source': 'plugin-marketplace', 'timestamp': datetime.now().isoformat(), 'skill_count': len(self.skills), 'skills': { skill_id: { 'name': skill.name, 'description': skill.description, 'category': skill.category, 'keywords': skill.keywords, 'tags': skill.tags, 'plugin_id': skill.plugin_id, 'trust_level': skill.trust_level, 'metadata': skill.metadata } for skill_id, skill in self.skills.items() }, 'categories': list(self.category_index.keys()), 'plugin_count': len(self.plugin_skill_map) } def export_for_knowledge_graph(self) -> Dict[str, Any]: """Export skills for knowledge graph ingestion Returns: Dict suitable for knowledge graph storage """ skills_by_category = {} for skill in self.skills.values(): if skill.category not in skills_by_category: skills_by_category[skill.category] = [] skills_by_category[skill.category].append(skill.to_dict()) return { 'source': 'plugin-marketplace-skills', 'timestamp': datetime.now().isoformat(), 'total_skills': len(self.skills), 'skills_by_category': skills_by_category, 'plugins_used': len(self.plugin_skill_map), 'trust_distribution': self._get_trust_distribution() } def _get_trust_distribution(self) -> Dict[str, int]: """Get distribution of trust levels in skills""" distribution: Dict[str, int] = {} for skill in self.skills.values(): trust = skill.trust_level distribution[trust] = distribution.get(trust, 0) + 1 return distribution # Convenience functions def get_plugin_skill_loader(registry: Optional[PluginMarketplaceRegistry] = None, cache_dir: Optional[Path] = None) -> PluginSkillLoader: """Get or create plugin skill loader""" return PluginSkillLoader(registry, cache_dir) def generate_all_skills() -> Dict[str, PluginSkill]: """Generate all plugin skills""" loader = get_plugin_skill_loader() return loader.generate_skills_from_plugins()