ec33ac1936
Based on claude-code-tools TmuxCLIController, this refactor: - Added DockerTmuxController class for robust tmux session management - Implements send_keys() with configurable delay_enter - Implements capture_pane() for output retrieval - Implements wait_for_prompt() for pattern-based completion detection - Implements wait_for_idle() for content-hash-based idle detection - Implements wait_for_shell_prompt() for shell prompt detection Also includes workflow improvements: - Pre-task git snapshot before agent execution - Post-task commit protocol in agent guidelines Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
452 lines
16 KiB
Python
452 lines
16 KiB
Python
#!/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)
|