ZerdaZerda

registry — PluginRegistry

On this page

Overview

PluginRegistry is the single source of truth for every registered plugin instance in a running process. It is a plain in-memory catalog — no persistence, no network calls — that additionally maintains a type index, a tag index, and keyword-search scoring so that both application code and an AIPluginSelector can find plugins without iterating the full instance dict by hand.

PluginRegistry

Python · 6 lines
class PluginRegistry:
    def __init__(self) -> None:
        self._plugins:    Dict[str, BasePlugin]
        self._classes:    Dict[str, Type[BasePlugin]]
        self._type_index: Dict[str, List[str]]   # type → [names]
        self._tag_index:  Dict[str, List[str]]   # tag → [names]

Thread safety

Per the class docstring: "designed for cooperative async; use a dedicated asyncio.Lock in the manager for concurrent registrations." The registry itself does not hold a lock — PluginManager does not currently wrap registry mutations in a lock either, based on the source of plugin_manager.py. In practice this means concurrent register()/unregister() calls from different coroutines on the same registry are not guaranteed atomic; serialize them at the call site if you register plugins concurrently.

register(plugin: BasePlugin) -> None

Adds a plugin instance under plugin.name, records its class, and indexes it by type and tags.

Raises: KeyError — if plugin.name is already registered.

Side effects: logger.info on success.

unregister(name: str) -> None

Removes a plugin, its class record, and de-indexes it from type/tag indices.

Raises: KeyError — if name is not registered.

get(name: str) -> Optional[BasePlugin]

Returns the plugin or None. Never raises.

get_or_raise(name: str) -> BasePlugin

Same lookup; Raises: KeyError if not found. This is what PluginManager.execute() calls internally.

exists(name: str) -> bool

all() -> List[BasePlugin]

names() -> List[str]

list_by_type(plugin_type: str) -> List[BasePlugin]

Returns plugins whose class declared that PLUGIN_TYPE string ("retrieval", "tool", "action", "processing", or any custom value a user-defined subclass sets).

list_by_tag(tag: str) -> List[BasePlugin]

Case-insensitive tag lookup (tag.lower()).

list_enabled() -> List[BasePlugin]

Filters self._plugins.values() by p.config.enabled — note this reflects the PluginConfig.enabled flag, not the runtime PluginStatus. A plugin with status == ERROR but config.enabled == True is still returned by this method.

search(query: str, top_k: int = 5) -> List[BasePlugin]

Keyword-based relevance search over name, description, tags, capabilities, and use_cases.

Algorithm:

  1. Tokenize query with re.findall(r"\w+", query.lower()) → a set of tokens.
  2. For each registered plugin, compute a weighted overlap score via the module-level _keyword_score helper (see below).
  3. Keep only plugins with score > 0.
  4. Sort descending by score, return the top top_k.

Complexity: O(P × F) where P is the number of plugins and F is the total text length across their searchable fields — every plugin's fields are re-tokenized on every call (no cached inverted index).

find_by_capability(capability: str) -> List[BasePlugin]

Case-insensitive substring match: returns plugins where capability.lower() is a substring of any entry in plugin.metadata.capabilities.

to_tool_descriptors(plugin_names: Optional[List[str]] = None) -> List[Dict[str, Any]]

Returns OpenAI-format function-calling descriptors for the given plugin names (or all registered plugins if None), filtered to only config.enabled plugins.

to_anthropic_tools(plugin_names: Optional[List[str]] = None) -> List[Dict[str, Any]]

Same, in Anthropic tool-use format.

snapshot() -> List[Dict[str, Any]]

A lightweight serializable summary per plugin: name, version, type (from PLUGIN_TYPE, defaults to "unknown"), status, enabled, tags, capabilities.

Dunder methods

  • __len__ — number of registered plugins.
  • __contains__ — name in registry.
  • __repr__ — f"PluginRegistry({len} plugins: {names})".

_keyword_score(tokens: set, meta: PluginMetadata) -> float (module-private)

The scoring function used by both PluginRegistry.search() and AIPluginSelector.select() (which imports it directly from this module).

Algorithm: for each of name (weight 3.0), description (weight 2.0), each tag (weight 2.5 each), each capability (weight 2.0 each), and each use_case (weight 1.5 each), tokenize the text and compute |query_tokens ∩ field_tokens| / max(|query_tokens|, 1), then multiply by that field's weight and sum. Fields with more query-token overlap, and fields weighted higher (name > tags > description > capabilities > use_cases), contribute more to the final score.

Complexity: O(n) per plugin in the number of words across its metadata fields.

Examples

1. Basic registration and lookup:

Python · 3 lines
registry = PluginRegistry()
registry.register(my_plugin)
found = registry.get("my_plugin")

2. Search:

Python · 3 lines
top = registry.search("find recent research papers", top_k=3)
for p in top:
    print(p.name)

3. Filter by type and tag:

Python · 2 lines
retrievers = registry.list_by_type("retrieval")
web_tools = registry.list_by_tag("web")

4. Error handling — double registration:

Python · 4 lines
try:
    registry.register(same_plugin_again)
except KeyError as e:
    print(e)  # "Plugin 'x' is already registered."

5. Exporting for an LLM call, restricted to a subset:

Python
tools = registry.to_tool_descriptors(plugin_names=["web_search", "calculator"])

Best practices

  • Call list_enabled() rather than all() when you specifically need only plugins available for execution — all() includes disabled and errored plugins.
  • search()/find_by_capability() are O(n) scans; for registries with hundreds of plugins and latency-sensitive selection, consider caching results or precomputing candidate sets per query type.

Common mistakes

  • Assuming unregister() is safe to call on a name that isn't registered — it raises KeyError, so guard with exists() first if the name might not be present.
  • Relying on registry-level thread safety for concurrent registration from multiple threads — the registry documents that it expects the caller to serialize concurrent registrations.

Related pages