Coverage for agentos/plugins/discovery.py: 0%
111 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-02 09:59 +0800
1"""
2Plugin Discovery — entry_points based plugin auto-discovery for AgentOS.
4Scans installed packages for entry_points registered under the
5'agentos.plugins' group and loads them without manual registration.
6"""
8from __future__ import annotations
10import importlib.metadata
11import logging
12from dataclasses import dataclass, field
13from typing import Any, Callable, Dict, List, Optional, Protocol
15logger = logging.getLogger(__name__)
18class PluginProtocol(Protocol):
19 """Minimal protocol that discovered plugins must satisfy."""
21 name: str
22 version: str
24 def initialize(self) -> None: ...
25 def shutdown(self) -> None: ...
28@dataclass
29class DiscoveredPlugin:
30 """Represents a plugin discovered via entry_points."""
32 name: str
33 version: str
34 entry_point_group: str
35 entry_point_name: str
36 package_name: str
37 module_path: str
38 metadata: Dict[str, Any] = field(default_factory=dict)
39 instance: Optional[Any] = field(default=None, repr=False)
41 @property
42 def is_loaded(self) -> bool:
43 return self.instance is not None
46@dataclass
47class DiscoveryResult:
48 """Result of a plugin discovery scan."""
50 plugins: List[DiscoveredPlugin]
51 total_found: int
52 total_loaded: int
53 errors: List[str] = field(default_factory=list)
54 scan_duration_ms: float = 0.0
57class PluginDiscovery:
58 """Scans installed packages for AgentOS plugins via entry_points."""
60 DEFAULT_GROUPS = [
61 "agentos.plugins",
62 "agentos.tools",
63 "agentos.models",
64 "agentos.middleware",
65 ]
67 def __init__(self, groups: Optional[List[str]] = None):
68 self._groups = groups or self.DEFAULT_GROUPS
69 self._discovered: Dict[str, DiscoveredPlugin] = {}
70 self._loaders: Dict[str, Callable] = {}
72 @property
73 def discovered(self) -> Dict[str, DiscoveredPlugin]:
74 return dict(self._discovered)
76 @property
77 def groups(self) -> List[str]:
78 return list(self._groups)
80 def register_loader(self, group: str, loader: Callable) -> None:
81 """Register a custom loader for a specific entry_point group."""
82 self._loaders[group] = loader
84 def scan(self, groups: Optional[List[str]] = None) -> DiscoveryResult:
85 """Scan for plugins across specified (or all registered) groups."""
86 import time
87 start = time.perf_counter()
88 target_groups = groups or self._groups
89 plugins: List[DiscoveredPlugin] = []
90 errors: List[str] = []
92 for group in target_groups:
93 try:
94 entry_points = importlib.metadata.entry_points(group=group)
95 except TypeError:
96 # Python 3.11 fallback
97 all_eps = importlib.metadata.entry_points()
98 entry_points = []
99 for ep in all_eps:
100 if ep.group == group:
101 entry_points.append(ep)
103 for ep in entry_points:
104 try:
105 pkg = ep.dist.name if ep.dist else "unknown"
106 plugin = DiscoveredPlugin(
107 name=ep.name,
108 version=ep.dist.version if ep.dist else "0.0.0",
109 entry_point_group=group,
110 entry_point_name=ep.name,
111 package_name=pkg,
112 module_path=ep.value,
113 metadata={"group": group},
114 )
115 plugins.append(plugin)
116 self._discovered[f"{group}:{ep.name}"] = plugin
117 except Exception as e:
118 errors.append(f"Failed to parse {ep.name} in {group}: {e}")
120 elapsed = (time.perf_counter() - start) * 1000
121 return DiscoveryResult(
122 plugins=plugins,
123 total_found=len(plugins),
124 total_loaded=0,
125 errors=errors,
126 scan_duration_ms=elapsed,
127 )
129 def load_plugin(
130 self, name: str, group: str = "agentos.plugins"
131 ) -> Optional[DiscoveredPlugin]:
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
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
150 def load_all(
151 self, group: Optional[str] = None
152 ) -> Dict[str, DiscoveredPlugin]:
153 """Load all discovered plugins, optionally scoped to one group."""
154 loaded: Dict[str, DiscoveredPlugin] = {}
155 for key, plugin in self._discovered.items():
156 if group and not key.startswith(f"{group}:"):
157 continue
158 result = self.load_plugin(plugin.name, plugin.entry_point_group)
159 if result and result.is_loaded:
160 loaded[key] = result
161 return loaded
163 def get_by_package(self, package_name: str) -> List[DiscoveredPlugin]:
164 """Get all discovered plugins from a specific package."""
165 return [
166 p for p in self._discovered.values()
167 if p.package_name == package_name
168 ]
170 def get_by_group(self, group: str) -> List[DiscoveredPlugin]:
171 """Get all discovered plugins in a specific group."""
172 return [
173 p for p in self._discovered.values()
174 if p.entry_point_group == group
175 ]
177 def summary(self) -> Dict[str, Any]:
178 """Return a summary of all discovered plugins."""
179 groups_summary: Dict[str, int] = {}
180 for p in self._discovered.values():
181 groups_summary[p.entry_point_group] = (
182 groups_summary.get(p.entry_point_group, 0) + 1
183 )
184 return {
185 "total_plugins": len(self._discovered),
186 "by_group": groups_summary,
187 "loaded": sum(1 for p in self._discovered.values() if p.is_loaded),
188 }
190 def clear(self) -> None:
191 """Clear all discovered plugins."""
192 self._discovered.clear()
195def _default_plugin_loader(module_path: str) -> Any:
196 """Default loader: import the module and look for a Plugin class."""
197 import importlib
199 module = importlib.import_module(module_path)
200 # Try common class names
201 for attr_name in ("Plugin", "AgentOSPlugin", "plugin", "__plugin__"):
202 if hasattr(module, attr_name):
203 return getattr(module, attr_name)
204 return module