""" Agent Registry Centralized registry for agent discovery, registration, and management. """ import importlib import logging import os from pathlib import Path from typing import Any, Dict, List, Optional, Type from .agent import BaseAgent class AgentRegistry: """ Registry for managing and discovering agents. Features: - Agent registration and discovery - Category-based organization - Dynamic loading from filesystem - Agent search and filtering - Version management Example: ```python registry = AgentRegistry() # Register an agent registry.register(MyAgent()) # Get an agent agent = registry.get_agent('my-agent') # List all agents all_agents = registry.list_agents() # List by category devops_agents = registry.list_agents(category='devops') # Search agents results = registry.search('deploy') ``` """ def __init__(self, auto_discover: bool = True): """Initialize the registry.""" self.logger = logging.getLogger("agent.registry") self._agents: Dict[str, BaseAgent] = {} self._agents_by_category: Dict[str, List[str]] = {} self._agent_classes: Dict[str, Type[BaseAgent]] = {} if auto_discover: self.discover_agents() def register( self, agent: BaseAgent, override: bool = False ) -> None: """ Register an agent. Args: agent: Agent instance to register override: Allow overriding existing agent Raises: ValueError: If agent already exists and override=False """ agent_name = agent.metadata.name if agent_name in self._agents and not override: raise ValueError( f"Agent '{agent_name}' already registered. " f"Use override=True to replace." ) self._agents[agent_name] = agent # Add to category index category = agent.metadata.category if category not in self._agents_by_category: self._agents_by_category[category] = [] if agent_name not in self._agents_by_category[category]: self._agents_by_category[category].append(agent_name) self.logger.info(f"Registered agent: {agent_name}") def register_class( self, agent_class: Type[BaseAgent], name: Optional[str] = None ) -> None: """ Register an agent class (lazy instantiation). Args: agent_class: Agent class to register name: Optional name override """ agent_name = name or agent_class.__name__ self._agent_classes[agent_name] = agent_class self.logger.info(f"Registered agent class: {agent_name}") def unregister(self, agent_name: str) -> bool: """ Unregister an agent. Args: agent_name: Name of agent to remove Returns: True if removed, False if not found """ if agent_name in self._agents: agent = self._agents[agent_name] category = agent.metadata.category del self._agents[agent_name] # Remove from category index if category in self._agents_by_category: self._agents_by_category[category].remove(agent_name) self.logger.info(f"Unregistered agent: {agent_name}") return True return False def get_agent( self, agent_name: str, create_new: bool = False ) -> Optional[BaseAgent]: """ Get an agent by name. Args: agent_name: Name of the agent create_new: Create a new instance if it's a registered class Returns: Agent instance or None if not found """ # Check instances first if agent_name in self._agents: return self._agents[agent_name] # Check classes if create_new and agent_name in self._agent_classes: agent_class = self._agent_classes[agent_name] agent = agent_class() self.register(agent) return agent return None def list_agents( self, category: Optional[str] = None ) -> List[BaseAgent]: """ List all registered agents. Args: category: Filter by category (optional) Returns: List of agent instances """ if category: agent_names = self._agents_by_category.get(category, []) return [self._agents[name] for name in agent_names] return list(self._agents.values()) def list_categories(self) -> List[str]: """Get list of all agent categories.""" return list(self._agents_by_category.keys()) def search( self, query: str, search_in: List[str] = None ) -> List[BaseAgent]: """ Search for agents. Args: query: Search query search_in: Fields to search in (name, description, tags) Returns: List of matching agents """ if search_in is None: search_in = ['name', 'description', 'tags'] query_lower = query.lower() results = [] for agent in self._agents.values(): matched = False if 'name' in search_in: if query_lower in agent.metadata.name.lower(): matched = True if 'description' in search_in: if query_lower in agent.metadata.description.lower(): matched = True if 'tags' in search_in: for tag in agent.metadata.tags: if query_lower in tag.lower(): matched = True break if matched: results.append(agent) return results def discover_agents( self, base_path: Optional[Path] = None ) -> int: """ Discover and register agents from filesystem. Args: base_path: Base directory to search (defaults to ./categories) Returns: Number of agents discovered """ if base_path is None: # Get path relative to this file current_dir = Path(__file__).parent.parent base_path = current_dir / "categories" if not base_path.exists(): self.logger.warning( f"Agent directory not found: {base_path}" ) return 0 discovered = 0 # Scan all category directories for category_dir in base_path.iterdir(): if not category_dir.is_dir(): continue category_name = category_dir.name # Scan all Python files in the category for agent_file in category_dir.glob("*.py"): if agent_file.name.startswith("_"): continue try: # Import the module module_name = ( f"agents.categories.{category_name}." f"{agent_file.stem}" ) module = importlib.import_module(module_name) # Find agent classes in the module for attr_name in dir(module): attr = getattr(module, attr_name) # Check if it's a BaseAgent subclass if ( isinstance(attr, type) and issubclass(attr, BaseAgent) and attr is not BaseAgent ): # Instantiate and register try: agent = attr() self.register(agent) discovered += 1 except Exception as e: self.logger.error( f"Failed to instantiate {attr_name}: " f"{str(e)}" ) except Exception as e: self.logger.error( f"Failed to load {agent_file}: {str(e)}" ) self.logger.info(f"Discovered {discovered} agents") return discovered def get_stats(self) -> Dict[str, Any]: """Get registry statistics.""" return { 'total_agents': len(self._agents), 'total_categories': len(self._agents_by_category), 'agents_by_category': { cat: len(agents) for cat, agents in self._agents_by_category.items() }, 'categories': list(self._agents_by_category.keys()) } def export_manifest(self) -> Dict[str, Any]: """Export agent manifest.""" return { 'agents': [ agent.get_info() for agent in self._agents.values() ], 'categories': self.list_categories(), 'stats': self.get_stats() }