#!/usr/bin/env python3 """ Claude Plugin Marketplace Integration Provides support for Claude's official plugin marketplace as a trusted source for Luzia skills and capabilities. Features: 1. Register Claude official plugins as trusted skill sources 2. Load plugin metadata and capabilities 3. Match plugins to task requirements 4. Cache plugin information for performance 5. Integrate with shared knowledge graph 6. Provide plugin-based skill discovery Official Marketplace: - https://marketplace.claude.ai/plugins - Trusted vendor: Anthropic - Plugin format: Plugin metadata + OpenAPI/tool definitions """ import json import urllib.request import urllib.error from pathlib import Path from typing import Dict, List, Optional, Any, Tuple from datetime import datetime, timedelta from dataclasses import dataclass, asdict import hashlib import ssl import logging # Set up logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) @dataclass class PluginCapability: """Single capability provided by a plugin""" name: str description: str category: str # code, analysis, research, integration, etc. tags: List[str] requires_auth: bool = False supported_models: List[str] = None def to_dict(self): return { 'name': self.name, 'description': self.description, 'category': self.category, 'tags': self.tags, 'requires_auth': self.requires_auth, 'supported_models': self.supported_models or ['all'] } @dataclass class MarketplacePlugin: """Plugin definition from Claude marketplace""" id: str name: str description: str vendor: str # Should be 'anthropic' for official plugins version: str url: str # marketplace URL capabilities: List[PluginCapability] trust_level: str # 'trusted', 'verified', 'community' last_updated: str metadata: Dict[str, Any] def to_dict(self): return { 'id': self.id, 'name': self.name, 'description': self.description, 'vendor': self.vendor, 'version': self.version, 'url': self.url, 'capabilities': [c.to_dict() for c in self.capabilities], 'trust_level': self.trust_level, 'last_updated': self.last_updated, 'metadata': self.metadata } class PluginMarketplaceRegistry: """ Registry of trusted plugins from Claude marketplace Serves as the canonical list of available plugins and their capabilities. """ # Official Claude plugins with verified capabilities OFFICIAL_PLUGINS = { 'code-simplifier': { 'id': 'code-simplifier', 'name': 'Code Simplifier', 'description': 'Simplifies and optimizes code for readability and performance', 'vendor': 'anthropic', 'version': '1.0.0', 'url': 'https://marketplace.claude.ai/plugins/code-simplifier', 'capabilities': [ { 'name': 'simplify_code', 'description': 'Analyze and simplify code for better readability', 'category': 'code-analysis', 'tags': ['refactor', 'optimization', 'readability'] }, { 'name': 'detect_complexity', 'description': 'Identify overly complex code patterns', 'category': 'code-analysis', 'tags': ['complexity', 'metrics'] }, { 'name': 'suggest_improvements', 'description': 'Suggest code improvements and best practices', 'category': 'code-analysis', 'tags': ['suggestions', 'best-practices'] } ], 'trust_level': 'trusted', 'tags': ['code-quality', 'refactoring', 'optimization'] }, 'code-reviewer': { 'id': 'code-reviewer', 'name': 'Code Reviewer', 'description': 'Comprehensive code review with security and performance analysis', 'vendor': 'anthropic', 'version': '1.0.0', 'url': 'https://marketplace.claude.ai/plugins/code-reviewer', 'capabilities': [ { 'name': 'security_review', 'description': 'Identify security vulnerabilities in code', 'category': 'security', 'tags': ['security', 'vulnerabilities', 'owasp'] }, { 'name': 'performance_review', 'description': 'Analyze code for performance bottlenecks', 'category': 'performance', 'tags': ['performance', 'optimization', 'benchmarking'] }, { 'name': 'best_practices_review', 'description': 'Check code against best practices and patterns', 'category': 'code-quality', 'tags': ['patterns', 'best-practices', 'standards'] } ], 'trust_level': 'trusted', 'tags': ['code-review', 'security', 'quality'] }, 'api-integration': { 'id': 'api-integration', 'name': 'API Integration Helper', 'description': 'Helps integrate third-party APIs and services', 'vendor': 'anthropic', 'version': '1.0.0', 'url': 'https://marketplace.claude.ai/plugins/api-integration', 'capabilities': [ { 'name': 'generate_api_client', 'description': 'Generate API client code from specifications', 'category': 'integration', 'tags': ['api', 'client-generation', 'openapi'] }, { 'name': 'validate_api_spec', 'description': 'Validate OpenAPI/Swagger specifications', 'category': 'validation', 'tags': ['validation', 'openapi', 'swagger'] } ], 'trust_level': 'trusted', 'tags': ['integration', 'api', 'client-generation'] } } def __init__(self, cache_dir: Optional[Path] = None): """Initialize plugin registry Args: cache_dir: Directory for caching plugin metadata """ self.cache_dir = cache_dir or Path("/tmp/.luzia-plugins") self.cache_dir.mkdir(parents=True, exist_ok=True) self.plugins: Dict[str, MarketplacePlugin] = {} self.plugin_index: Dict[str, List[str]] = {} # capability -> [plugin_ids] self.load_official_plugins() def load_official_plugins(self) -> None: """Load official Claude plugins from registry""" for plugin_id, plugin_data in self.OFFICIAL_PLUGINS.items(): capabilities = [ PluginCapability( name=cap['name'], description=cap['description'], category=cap['category'], tags=cap.get('tags', []) ) for cap in plugin_data.get('capabilities', []) ] plugin = MarketplacePlugin( id=plugin_data['id'], name=plugin_data['name'], description=plugin_data['description'], vendor=plugin_data['vendor'], version=plugin_data['version'], url=plugin_data['url'], capabilities=capabilities, trust_level=plugin_data['trust_level'], last_updated=datetime.now().isoformat(), metadata={'tags': plugin_data.get('tags', [])} ) self.register_plugin(plugin) logger.info(f"Loaded official plugin: {plugin.name}") def register_plugin(self, plugin: MarketplacePlugin) -> None: """Register a plugin and index its capabilities Args: plugin: Plugin to register """ self.plugins[plugin.id] = plugin # Index capabilities for quick lookup for capability in plugin.capabilities: if capability.name not in self.plugin_index: self.plugin_index[capability.name] = [] self.plugin_index[capability.name].append(plugin.id) # Also index by category category_key = f"category:{capability.category}" if category_key not in self.plugin_index: self.plugin_index[category_key] = [] if plugin.id not in self.plugin_index[category_key]: self.plugin_index[category_key].append(plugin.id) def find_plugins_for_task(self, task_description: str, task_keywords: List[str]) -> List[Tuple[str, float]]: """Find relevant plugins for a task Args: task_description: Description of the task task_keywords: Keywords extracted from task Returns: List of (plugin_id, relevance_score) tuples sorted by relevance """ scores: Dict[str, float] = {} for keyword in task_keywords: # Check direct capability matches if keyword in self.plugin_index: for plugin_id in self.plugin_index[keyword]: scores[plugin_id] = scores.get(plugin_id, 0) + 1.0 # Check category matches category_key = f"category:{keyword}" if category_key in self.plugin_index: for plugin_id in self.plugin_index[category_key]: scores[plugin_id] = scores.get(plugin_id, 0) + 0.7 # Check tags and description matches for plugin_id, plugin in self.plugins.items(): plugin_tags = plugin.metadata.get('tags', []) for tag in plugin_tags: if tag in task_keywords: scores[plugin_id] = scores.get(plugin_id, 0) + 0.5 # Sort by relevance score ranked = sorted(scores.items(), key=lambda x: x[1], reverse=True) return ranked def get_plugin(self, plugin_id: str) -> Optional[MarketplacePlugin]: """Get a plugin by ID Args: plugin_id: ID of the plugin Returns: Plugin or None if not found """ return self.plugins.get(plugin_id) def list_plugins(self, category: Optional[str] = None) -> List[MarketplacePlugin]: """List available plugins Args: category: Optional category to filter by Returns: List of plugins """ if category: plugin_ids = self.plugin_index.get(f"category:{category}", []) return [self.plugins[pid] for pid in plugin_ids if pid in self.plugins] return list(self.plugins.values()) def save_to_cache(self) -> None: """Save plugin registry to cache""" cache_file = self.cache_dir / "registry.json" data = { 'timestamp': datetime.now().isoformat(), 'plugins': { pid: plugin.to_dict() for pid, plugin in self.plugins.items() }, 'index': self.plugin_index } cache_file.write_text(json.dumps(data, indent=2)) logger.info(f"Saved plugin registry to {cache_file}") def export_for_knowledge_graph(self) -> Dict[str, Any]: """Export plugin registry for knowledge graph ingestion Returns: Dict suitable for knowledge graph storage """ return { 'source': 'claude-marketplace', 'timestamp': datetime.now().isoformat(), 'plugin_count': len(self.plugins), 'plugins': { pid: { 'name': plugin.name, 'description': plugin.description, 'capabilities': [c.to_dict() for c in plugin.capabilities], 'trust_level': plugin.trust_level, 'metadata': plugin.metadata } for pid, plugin in self.plugins.items() }, 'categories': self._extract_categories(), 'trust_distribution': self._get_trust_distribution() } def _extract_categories(self) -> Dict[str, int]: """Extract unique categories and counts""" categories: Dict[str, int] = {} for plugin in self.plugins.values(): for capability in plugin.capabilities: cat = capability.category categories[cat] = categories.get(cat, 0) + 1 return categories def _get_trust_distribution(self) -> Dict[str, int]: """Get distribution of trust levels""" distribution: Dict[str, int] = {} for plugin in self.plugins.values(): trust = plugin.trust_level distribution[trust] = distribution.get(trust, 0) + 1 return distribution class PluginCapabilityMatcher: """ Matches plugin capabilities to task requirements Uses skill matching to find optimal plugins for execution """ def __init__(self, registry: PluginMarketplaceRegistry): """Initialize matcher Args: registry: Plugin marketplace registry """ self.registry = registry def extract_task_keywords(self, task_description: str) -> List[str]: """Extract relevant keywords from task description Args: task_description: Description of the task Returns: List of extracted keywords """ keywords = [] # Common task keywords mapping keyword_map = { 'review': ['code-review', 'security', 'performance', 'quality'], 'simplif': ['refactor', 'optimization', 'readability'], 'analyze': ['analysis', 'metrics', 'inspection'], 'integrat': ['integration', 'api', 'client'], 'securit': ['security', 'vulnerability', 'owasp'], 'perform': ['performance', 'optimization', 'benchmark'], 'refactor': ['refactor', 'optimization', 'best-practices'], 'test': ['testing', 'validation', 'qa'], 'document': ['documentation', 'api-docs', 'spec'] } # Extract matching keywords task_lower = task_description.lower() for keyword_pattern, matches in keyword_map.items(): if keyword_pattern in task_lower: keywords.extend(matches) return list(set(keywords)) # Remove duplicates def match_plugins(self, task_description: str, min_relevance: float = 0.5) -> List[Dict[str, Any]]: """Match plugins to task requirements Args: task_description: Description of the task min_relevance: Minimum relevance score (0-1) Returns: List of matched plugins with scores and capabilities """ keywords = self.extract_task_keywords(task_description) plugin_scores = self.registry.find_plugins_for_task(task_description, keywords) results = [] for plugin_id, score in plugin_scores: if score > min_relevance: plugin = self.registry.get_plugin(plugin_id) if plugin: results.append({ 'id': plugin.id, 'name': plugin.name, 'description': plugin.description, 'relevance_score': score, 'capabilities': [c.to_dict() for c in plugin.capabilities], 'trust_level': plugin.trust_level }) return results # Convenience functions for CLI usage def get_marketplace_registry(cache_dir: Optional[Path] = None) -> PluginMarketplaceRegistry: """Get or create plugin marketplace registry""" return PluginMarketplaceRegistry(cache_dir) def find_plugins_for_task(task: str, registry: Optional[PluginMarketplaceRegistry] = None) -> List[Dict]: """Quick function to find plugins for a task""" if registry is None: registry = get_marketplace_registry() matcher = PluginCapabilityMatcher(registry) return matcher.match_plugins(task)