"""
Agent Registry for auto-discovery and centralized agent management.
The registry automatically discovers and registers all agents across
installed Django apps, providing a centralized access point for agent
management.
"""
import importlib
import inspect
import logging
from typing import Dict, List, Optional, Type
from django.apps import apps
from django.conf import settings
from django_agents.base import BaseAgent
logger = logging.getLogger(__name__)
[docs]
class AgentRegistry:
"""
Centralized registry for managing Django agents.
The registry provides auto-discovery, registration, and lookup
capabilities for all agents in the project.
"""
[docs]
def __init__(self):
"""Initialize the registry."""
self._agents: Dict[str, Dict] = {}
self._auto_discovered = False
[docs]
def register(
self,
agent_class: Type[BaseAgent],
app_name: str,
force: bool = False
) -> None:
"""
Register an agent class.
Args:
agent_class: Agent class to register
app_name: Name of the app containing the agent
force: Force registration even if agent exists
Raises:
ValueError: If agent is already registered and force=False
"""
try:
# Instantiate to get metadata
agent_instance = agent_class()
agent_name = agent_instance.name
if agent_name in self._agents and not force:
raise ValueError(
f"Agent '{agent_name}' is already registered. "
f"Use force=True to override."
)
self._agents[agent_name] = {
'class': agent_class,
'class_name': agent_class.__name__,
'app': app_name,
'module': f'{app_name}.agents',
'instance': agent_instance,
'name': agent_instance.name,
'description': agent_instance.description,
'version': agent_instance.version,
'enabled': agent_instance.enabled,
}
logger.info(f"Registered agent: {agent_name} from {app_name}")
except Exception as e:
logger.error(
f"Failed to register agent {agent_class.__name__} "
f"from {app_name}: {e}"
)
raise
[docs]
def unregister(self, agent_name: str) -> None:
"""
Unregister an agent.
Args:
agent_name: Name of the agent to unregister
Raises:
KeyError: If agent is not registered
"""
if agent_name not in self._agents:
raise KeyError(f"Agent '{agent_name}' is not registered")
del self._agents[agent_name]
logger.info(f"Unregistered agent: {agent_name}")
[docs]
def get_agent(self, agent_name: str) -> Optional[BaseAgent]:
"""
Get an agent instance by name.
Args:
agent_name: Name of the agent
Returns:
Agent instance or None if not found
"""
agent_data = self._agents.get(agent_name)
if agent_data:
# Return a fresh instance
return agent_data['class']()
return None
[docs]
def get_agent_class(self, agent_name: str) -> Optional[Type[BaseAgent]]:
"""
Get an agent class by name.
Args:
agent_name: Name of the agent
Returns:
Agent class or None if not found
"""
agent_data = self._agents.get(agent_name)
if agent_data:
return agent_data['class']
return None
[docs]
def list_agents(
self,
app_name: Optional[str] = None,
enabled_only: bool = False
) -> List[Dict]:
"""
List all registered agents.
Args:
app_name: Filter by app name
enabled_only: Only return enabled agents
Returns:
List of agent metadata dictionaries
"""
agents = list(self._agents.values())
if app_name:
agents = [a for a in agents if a['app'] == app_name]
if enabled_only:
agents = [a for a in agents if a['enabled']]
return agents
[docs]
def get_agents_by_app(self, app_name: str) -> List[Dict]:
"""
Get all agents from a specific app.
Args:
app_name: Name of the app
Returns:
List of agent metadata dictionaries
"""
return [
agent for agent in self._agents.values()
if agent['app'] == app_name
]
[docs]
def is_registered(self, agent_name: str) -> bool:
"""
Check if an agent is registered.
Args:
agent_name: Name of the agent
Returns:
True if registered, False otherwise
"""
return agent_name in self._agents
[docs]
def count(self) -> int:
"""
Get the number of registered agents.
Returns:
Number of registered agents
"""
return len(self._agents)
[docs]
def clear(self) -> None:
"""Clear all registered agents."""
self._agents.clear()
self._auto_discovered = False
logger.info("Cleared all registered agents")
[docs]
def autodiscover(self, force: bool = False) -> None:
"""
Automatically discover and register all agents in installed apps.
Args:
force: Force re-discovery even if already discovered
"""
if self._auto_discovered and not force:
logger.debug("Agents already auto-discovered. Use force=True to re-discover.")
return
logger.info("Starting agent auto-discovery...")
# Clear existing registrations if forcing
if force:
self.clear()
discovered_count = 0
error_count = 0
# Get all installed apps
installed_apps = apps.get_app_configs()
for app_config in installed_apps:
app_name = app_config.name
# Skip Django internal apps if configured
if self._should_skip_app(app_name):
continue
try:
# Try to import agents module
agents_module = importlib.import_module(f'{app_name}.agents')
# Find all agent classes
for name, obj in inspect.getmembers(agents_module):
if (inspect.isclass(obj) and
issubclass(obj, BaseAgent) and
obj is not BaseAgent and
not name.startswith('_')):
try:
self.register(obj, app_name, force=force)
discovered_count += 1
except Exception as e:
logger.warning(
f"Failed to register {name} from {app_name}: {e}"
)
error_count += 1
except (ImportError, ModuleNotFoundError):
# No agents module in this app - that's okay
pass
except Exception as e:
logger.error(f"Error discovering agents in {app_name}: {e}")
error_count += 1
self._auto_discovered = True
logger.info(
f"Auto-discovery complete. Discovered {discovered_count} agents "
f"with {error_count} errors."
)
def _should_skip_app(self, app_name: str) -> bool:
"""
Determine if an app should be skipped during auto-discovery.
Args:
app_name: Name of the app
Returns:
True if app should be skipped
"""
# Get skip list from settings
skip_apps = getattr(
settings,
'DJANGO_AGENTS_SKIP_AUTODISCOVER',
[
'django.contrib.admin',
'django.contrib.auth',
'django.contrib.contenttypes',
'django.contrib.sessions',
'django.contrib.messages',
'django.contrib.staticfiles',
]
)
return app_name in skip_apps
[docs]
def __repr__(self) -> str:
"""String representation."""
return f"<AgentRegistry: {self.count()} agents registered>"
# Global registry instance
registry = AgentRegistry()
[docs]
def autodiscover():
"""
Convenience function to auto-discover agents.
This should be called during Django app initialization.
"""
registry.autodiscover()
[docs]
def get_registry() -> AgentRegistry:
"""
Get the global agent registry instance.
Returns:
Global agent registry
"""
return registry