Skip to main content

PluginRegistry

The PluginRegistry manages plugin discovery, loading, and lifecycle in openstackai.

Import​

from openstackai.plugins import PluginRegistry

Basic Usage​

from openstackai.plugins import PluginRegistry, Plugin

# Create registry
registry = PluginRegistry()

# Register plugins
registry.register(MyPlugin())
registry.register(AnotherPlugin())

# Use with agent
agent = Agent(plugins=registry.get_all())

Class Reference​

Constructor​

PluginRegistry(
    auto_discover: bool = False,  # Auto-discover plugins in PYTHONPATH
    plugin_dirs: list[str] = None # Additional directories to scan
)

Registration Methods​

register()​

Register a plugin instance.

registry.register(plugin: Plugin) -> None

Example:

plugin = WeatherPlugin(api_key="xxx")
registry.register(plugin)

register_class()​

Register a plugin class for lazy instantiation.

registry.register_class(
    plugin_class: type[Plugin],
    config: dict = None
) -> None

Example:

registry.register_class(
    WeatherPlugin,
    config={"api_key": "xxx"}
)

unregister()​

Remove a plugin from the registry.

registry.unregister(name: str) -> bool

Example:

if registry.unregister("weather"):
    print("Plugin removed")

Discovery Methods​

discover()​

Auto-discover plugins in specified paths.

registry.discover(
    paths: list[str] = None,  # Directories to scan
    pattern: str = "*_plugin.py"  # File pattern
) -> int  # Number of plugins found

Example:

count = registry.discover(paths=["./plugins"])
print(f"Found {count} plugins")

discover_entry_points()​

Load plugins from package entry points.

registry.discover_entry_points(group: str = "openstackai.plugins") -> int

Example:

# In plugin package's pyproject.toml:
# [project.entry-points."openstackai.plugins"]
# my_plugin = "my_package:MyPlugin"

registry.discover_entry_points()

Query Methods​

get()​

Get a plugin by name.

plugin = registry.get(name: str) -> Plugin | None

get_all()​

Get all registered plugins.

plugins = registry.get_all() -> list[Plugin]

has()​

Check if a plugin is registered.

if registry.has("weather"):
    print("Weather plugin available")

list_names()​

Get all plugin names.

names = registry.list_names() -> list[str]

list_functions()​

Get all functions across all plugins.

functions = registry.list_functions() -> dict[str, callable]
# Returns {"plugin.function": callable, ...}

Configuration​

configure()​

Configure a plugin with settings.

registry.configure(
    name: str,
    settings: dict
) -> None

Example:

registry.configure("weather", {
    "api_key": "new_key",
    "timeout": 60
})

configure_all()​

Configure multiple plugins at once.

registry.configure_all(config: dict) -> None

Example:

registry.configure_all({
    "weather": {"api_key": "xxx"},
    "search": {"endpoint": "production"}
})

Lifecycle Management​

load_all()​

Initialize all registered plugins.

registry.load_all() -> None

unload_all()​

Clean up all plugins.

registry.unload_all() -> None

Context Manager​

with PluginRegistry(auto_discover=True) as registry:
    agent = Agent(plugins=registry.get_all())
    # ... use agent
# Plugins automatically unloaded

Events​

Callbacks​

registry.on_plugin_loaded = lambda plugin: print(f"Loaded: {plugin.name}")
registry.on_plugin_error = lambda plugin, error: print(f"Error: {error}")

Global Registry​

For convenience, a global registry is available:

from openstackai.plugins import global_registry

# Register globally
global_registry.register(MyPlugin())

# Access from anywhere
plugin = global_registry.get("my_plugin")

Example: Plugin Directory Structure​

my_project/
├── plugins/
│   ├── __init__.py
│   ├── weather_plugin.py
│   ├── search_plugin.py
│   └── custom_plugin.py
└── main.py
# main.py
from openstackai.plugins import PluginRegistry
from openstackai import Agent

registry = PluginRegistry()
registry.discover(paths=["./plugins"])

agent = Agent(
    name="Assistant",
    plugins=registry.get_all()
)

See Also​