Coverage for agentos/plugins/discovery.py: 0%

1""" 

2Plugin Discovery — entry_points based plugin auto-discovery for AgentOS. 

3 

4Scans installed packages for entry_points registered under the 

5'agentos.plugins' group and loads them without manual registration. 

6""" 

7 

8from __future__ import annotations 

9 

10import importlib.metadata 

11import logging 

12from collections.abc import Callable 

13from dataclasses import dataclass, field 

14from typing import Any, Protocol 

15 

16logger = logging.getLogger(__name__) 

17 

18 

19class PluginProtocol(Protocol): 

20 """Minimal protocol that discovered plugins must satisfy.""" 

21 

22 name: str 

23 version: str 

24 

25 def initialize(self) -> None: ... 

26 def shutdown(self) -> None: ... 

27 

28 

29@dataclass 

30class DiscoveredPlugin: 

31 """Represents a plugin discovered via entry_points.""" 

32 

33 name: str 

34 version: str 

35 entry_point_group: str 

36 entry_point_name: str 

37 package_name: str 

38 module_path: str 

39 metadata: dict[str, Any] = field(default_factory=dict) 

40 instance: Any | None = field(default=None, repr=False) 

41 

42 @property 

43 def is_loaded(self) -> bool: 

44 return self.instance is not None 

45 

46 

47@dataclass 

48class DiscoveryResult: 

49 """Result of a plugin discovery scan.""" 

50 

51 plugins: list[DiscoveredPlugin] 

52 total_found: int 

53 total_loaded: int 

54 errors: list[str] = field(default_factory=list) 

55 scan_duration_ms: float = 0.0 

56 

57 

58class PluginDiscovery: 

59 """Scans installed packages for AgentOS plugins via entry_points.""" 

60 

61 DEFAULT_GROUPS = [ 

62 "agentos.plugins", 

63 "agentos.tools", 

64 "agentos.models", 

65 "agentos.middleware", 

66 ] 

67 

68 def __init__(self, groups: list[str] | None = None): 

69 self._groups = groups or self.DEFAULT_GROUPS 

70 self._discovered: dict[str, DiscoveredPlugin] = {} 

71 self._loaders: dict[str, Callable] = {} 

72 

73 @property 

74 def discovered(self) -> dict[str, DiscoveredPlugin]: 

75 return dict(self._discovered) 

76 

77 @property 

78 def groups(self) -> list[str]: 

79 return list(self._groups) 

80 

81 def register_loader(self, group: str, loader: Callable) -> None: 

82 """Register a custom loader for a specific entry_point group.""" 

83 self._loaders[group] = loader 

84 

85 def scan(self, groups: list[str] | None = None) -> DiscoveryResult: 

86 """Scan for plugins across specified (or all registered) groups.""" 

87 import time 

88 

89 start = time.perf_counter() 

90 target_groups = groups or self._groups 

91 plugins: list[DiscoveredPlugin] = [] 

92 errors: list[str] = [] 

93 

94 for group in target_groups: 

95 try: 

96 entry_points = importlib.metadata.entry_points(group=group) 

97 except TypeError: 

98 # Python 3.11 fallback 

99 all_eps = importlib.metadata.entry_points() 

100 entry_points = [] 

101 for ep in all_eps: 

102 if ep.group == group: 

103 entry_points.append(ep) 

104 

105 for ep in entry_points: 

106 try: 

107 pkg = ep.dist.name if ep.dist else "unknown" 

108 plugin = DiscoveredPlugin( 

109 name=ep.name, 

110 version=ep.dist.version if ep.dist else "0.0.0", 

111 entry_point_group=group, 

112 entry_point_name=ep.name, 

113 package_name=pkg, 

114 module_path=ep.value, 

115 metadata={"group": group}, 

116 ) 

117 plugins.append(plugin) 

118 self._discovered[f"{group}:{ep.name}"] = plugin 

119 except Exception as e: 

120 errors.append(f"Failed to parse {ep.name} in {group}: {e}") 

121 

122 elapsed = (time.perf_counter() - start) * 1000 

123 return DiscoveryResult( 

124 plugins=plugins, 

125 total_found=len(plugins), 

126 total_loaded=0, 

127 errors=errors, 

128 scan_duration_ms=elapsed, 

129 ) 

130 

131 def load_plugin(self, name: str, group: str = "agentos.plugins") -> DiscoveredPlugin | None: 

132 """Load a specific discovered plugin by name and group.""" 

133 key = f"{group}:{name}" 

134 plugin = self._discovered.get(key) 

135 if plugin is None: 

136 logger.warning(f"Plugin '{key}' not found in discovered set.") 

137 return None 

138 

139 try: 

140 loader = self._loaders.get(group, _default_plugin_loader) 

141 instance = loader(plugin.module_path) 

142 plugin.instance = instance 

143 if hasattr(instance, "initialize"): 

144 instance.initialize() 

145 return plugin 

146 except Exception as e: 

147 logger.error(f"Failed to load plugin '{key}': {e}") 

148 return None 

149 

150 def load_all(self, group: str | None = None) -> dict[str, DiscoveredPlugin]: 

151 """Load all discovered plugins, optionally scoped to one group.""" 

152 loaded: dict[str, DiscoveredPlugin] = {} 

153 for key, plugin in self._discovered.items(): 

154 if group and not key.startswith(f"{group}:"): 

155 continue 

156 result = self.load_plugin(plugin.name, plugin.entry_point_group) 

157 if result and result.is_loaded: 

158 loaded[key] = result 

159 return loaded 

160 

161 def get_by_package(self, package_name: str) -> list[DiscoveredPlugin]: 

162 """Get all discovered plugins from a specific package.""" 

163 return [p for p in self._discovered.values() if p.package_name == package_name] 

164 

165 def get_by_group(self, group: str) -> list[DiscoveredPlugin]: 

166 """Get all discovered plugins in a specific group.""" 

167 return [p for p in self._discovered.values() if p.entry_point_group == group] 

168 

169 def summary(self) -> dict[str, Any]: 

170 """Return a summary of all discovered plugins.""" 

171 groups_summary: dict[str, int] = {} 

172 for p in self._discovered.values(): 

173 groups_summary[p.entry_point_group] = groups_summary.get(p.entry_point_group, 0) + 1 

174 return { 

175 "total_plugins": len(self._discovered), 

176 "by_group": groups_summary, 

177 "loaded": sum(1 for p in self._discovered.values() if p.is_loaded), 

178 } 

179 

180 def clear(self) -> None: 

181 """Clear all discovered plugins.""" 

182 self._discovered.clear() 

183 

184 

185def _default_plugin_loader(module_path: str) -> Any: 

186 """Default loader: import the module and look for a Plugin class.""" 

187 import importlib 

188 

189 module = importlib.import_module(module_path) 

190 # Try common class names 

191 for attr_name in ("Plugin", "AgentOSPlugin", "plugin", "__plugin__"): 

192 if hasattr(module, attr_name): 

193 return getattr(module, attr_name) 

194 return module