"""
Project Manager
Discovers and manages project instances from environment variables.
Handles plugin lifecycle and tool registration.
"""
import logging
import os
import re
from typing import Any
from plugins import BasePlugin, registry
logger = logging.getLogger(__name__)
class ProjectManager:
"""
Manage multiple project instances.
Projects are discovered from environment variables:
- {PLUGIN_TYPE}_{PROJECT_ID}_{CONFIG_KEY}
Example:
WORDPRESS_SITE1_URL=https://example.com
WORDPRESS_SITE1_USERNAME=admin
WORDPRESS_SITE1_APP_PASSWORD=xxxx
WORDPRESS_SITE2_URL=https://other.com
...
"""
def __init__(self):
"""Initialize project manager."""
self.projects: dict[str, BasePlugin] = {}
self.logger = logging.getLogger("ProjectManager")
def discover_projects(self) -> None:
"""
Discover projects from environment variables.
Scans environment for project configurations and creates
plugin instances.
"""
self.logger.info("Starting project discovery...")
# Get all registered plugin types
plugin_types = registry.get_registered_types()
for plugin_type in plugin_types:
self._discover_plugin_type(plugin_type)
self.logger.info(f"Discovery complete. Found {len(self.projects)} projects.")
def _discover_plugin_type(self, plugin_type: str) -> None:
"""
Discover all projects of a specific plugin type.
Args:
plugin_type: Type of plugin (e.g., 'wordpress')
"""
prefix = plugin_type.upper() + "_"
# Build list of longer prefixes from other plugin types to avoid collisions.
# e.g. WORDPRESS_ must not match WORDPRESS_ADVANCED_ env vars.
plugin_types = registry.get_registered_types()
longer_prefixes = [
pt.upper() + "_"
for pt in plugin_types
if pt != plugin_type and pt.upper().startswith(plugin_type.upper() + "_")
]
# Find all project IDs for this plugin type
project_ids = set()
env_pattern = re.compile(f"^{prefix}([A-Z0-9_]+?)_(.+)$")
for env_key in os.environ.keys():
# Skip env vars that belong to a more specific plugin type
if any(env_key.startswith(lp) for lp in longer_prefixes):
continue
match = env_pattern.match(env_key)
if match:
project_id = match.group(1).lower()
project_ids.add(project_id)
# Create plugin instance for each project
for project_id in project_ids:
try:
config = self._load_project_config(plugin_type, project_id)
if config:
self._create_project_instance(plugin_type, project_id, config)
except Exception as e:
self.logger.debug(f"Legacy ProjectManager: skipped {plugin_type}/{project_id}: {e}")
def _load_project_config(self, plugin_type: str, project_id: str) -> dict[str, Any] | None:
"""
Load configuration for a project from environment.
Args:
plugin_type: Plugin type
project_id: Project ID
Returns:
Dict with configuration or None if incomplete
"""
prefix = f"{plugin_type.upper()}_{project_id.upper()}_"
config = {}
# Collect all config keys for this project
for env_key, env_value in os.environ.items():
if env_key.startswith(prefix):
# Extract config key (everything after prefix)
config_key = env_key[len(prefix) :].lower()
config[config_key] = env_value
if not config:
return None
self.logger.debug(f"Loaded config for {plugin_type}/{project_id}: {list(config.keys())}")
return config
def _create_project_instance(
self, plugin_type: str, project_id: str, config: dict[str, Any]
) -> None:
"""
Create a plugin instance for a project.
Args:
plugin_type: Plugin type
project_id: Project ID
config: Project configuration
"""
try:
# Create plugin instance
plugin = registry.create_instance(plugin_type, project_id, config)
# Store with full identifier
full_id = f"{plugin_type}_{project_id}"
self.projects[full_id] = plugin
self.logger.info(f"Created project: {full_id}")
except Exception as e:
raise Exception(f"Failed to instantiate {plugin_type}/{project_id}: {e}")
def get_project(self, full_id: str) -> BasePlugin | None:
"""
Get a project plugin instance.
Args:
full_id: Full project identifier (plugin_type_project_id)
Returns:
Plugin instance or None
"""
return self.projects.get(full_id)
def get_all_projects(self) -> dict[str, BasePlugin]:
"""Get all project instances."""
return self.projects.copy()
def get_projects_by_type(self, plugin_type: str) -> dict[str, BasePlugin]:
"""
Get all projects of a specific type.
Args:
plugin_type: Plugin type to filter by
Returns:
Dict of project_id -> plugin
"""
prefix = plugin_type + "_"
return {
full_id: plugin
for full_id, plugin in self.projects.items()
if full_id.startswith(prefix)
}
def get_all_tools(self) -> list[dict[str, Any]]:
"""
Get all MCP tools from all projects.
Returns:
List of tool definitions
"""
all_tools = []
for full_id, plugin in self.projects.items():
try:
tools = plugin.get_tools()
all_tools.extend(tools)
self.logger.debug(f"Loaded {len(tools)} tools from {full_id}")
except Exception as e:
self.logger.error(f"Error loading tools from {full_id}: {e}", exc_info=True)
self.logger.debug(f"Total tools loaded: {len(all_tools)}")
return all_tools
async def check_all_health(self) -> dict[str, dict[str, Any]]:
"""
Check health of all projects.
Returns:
Dict mapping project ID to health status
"""
health_results = {}
for full_id, plugin in self.projects.items():
try:
health = await plugin.health_check()
health_results[full_id] = health
except Exception as e:
health_results[full_id] = {
"healthy": False,
"message": f"Health check failed: {str(e)}",
}
return health_results
def get_project_info(self, full_id: str) -> dict[str, Any] | None:
"""
Get information about a specific project.
Args:
full_id: Full project identifier
Returns:
Project info dict or None
"""
plugin = self.get_project(full_id)
if plugin:
return plugin.get_project_info()
return None
def list_projects(self) -> list[dict[str, Any]]:
"""
List all projects with basic information.
Returns:
List of project info dicts
"""
return [
{"id": full_id, "type": plugin.get_plugin_name(), "project_id": plugin.project_id}
for full_id, plugin in self.projects.items()
]
# Global project manager instance
_project_manager: ProjectManager | None = None
def get_project_manager() -> ProjectManager:
"""Get the global project manager instance."""
global _project_manager
if _project_manager is None:
_project_manager = ProjectManager()
_project_manager.discover_projects()
return _project_manager
