MCPcopy Create free account
hub / github.com/MemTensor/MemOS / discover

Method discover

src/memos/plugins/manager.py:100–145  ·  view source on GitHub ↗

Discover and load all installed plugins via entry_points.

(self)

Source from the content-addressed store, hash-verified

98 return winners
99
100 def discover(self) -> None:
101 """Discover and load all installed plugins via entry_points."""
102 if self._discovered:
103 return
104
105 try:
106 eps = importlib.metadata.entry_points()
107 if hasattr(eps, "select"):
108 plugin_eps = eps.select(group=ENTRY_POINT_GROUP)
109 else:
110 plugin_eps = eps.get(ENTRY_POINT_GROUP, [])
111 except Exception:
112 logger.exception("Failed to query entry_points")
113 return
114
115 candidates: list[tuple[str, MemOSPlugin]] = []
116 for ep in plugin_eps:
117 try:
118 plugin_cls = ep.load()
119 plugin = plugin_cls()
120 if not isinstance(plugin, MemOSPlugin):
121 logger.warning("Plugin %s does not extend MemOSPlugin, skipped", ep.name)
122 continue
123 candidates.append((ep.name, plugin))
124 except Exception:
125 logger.exception("Failed to load plugin: %s", ep.name)
126
127 winners = self._select_plugin_winners(candidates)
128 for plugin_name, plugin in winners.items():
129 if not self._is_plugin_enabled(plugin):
130 logger.info(
131 "Plugin discovered but disabled: %s v%s",
132 plugin.name,
133 plugin.version,
134 )
135 continue
136 plugin.on_load()
137 self._plugins[plugin_name] = plugin
138 logger.info(
139 "Plugin discovered: %s v%s (priority=%s)",
140 plugin.name,
141 plugin.version,
142 plugin.priority,
143 )
144
145 self._discovered = True
146
147 def init_components(self, context: dict) -> None:
148 """Initialize runtime components contributed by loaded plugins."""

Calls 8

_is_plugin_enabledMethod · 0.95
selectMethod · 0.80
getMethod · 0.65
infoMethod · 0.65
loadMethod · 0.45
itemsMethod · 0.45
on_loadMethod · 0.45