Skip to content

Plugins

Plugin system for extending AgentProbe with custom evaluators, adapters, reporters, and storage backends.

Base Classes

agentprobe.plugins.base

Base plugin classes with lifecycle hooks and typed subclasses.

Plugins extend AgentProbe's functionality by implementing lifecycle hooks and providing factory methods for evaluators, adapters, reporters, or storage backends.

PluginBase

Bases: ABC

Abstract base for all AgentProbe plugins.

Plugins implement lifecycle hooks that are called at various points during test execution. Subclasses should override only the hooks they need — default implementations are no-ops.

Attributes:

Name Type Description
name str

Unique plugin identifier.
plugin_type PluginType

The type of extension this plugin provides.
version str

Plugin version string.
Source code in src/agentprobe/plugins/base.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
class PluginBase(ABC):
    """Abstract base for all AgentProbe plugins.

    Plugins implement lifecycle hooks that are called at various points
    during test execution. Subclasses should override only the hooks
    they need — default implementations are no-ops.

    Attributes:
        name: Unique plugin identifier.
        plugin_type: The type of extension this plugin provides.
        version: Plugin version string.
    """

    @property
    @abstractmethod
    def name(self) -> str:
        """Return the unique plugin name."""
        ...

    @property
    @abstractmethod
    def plugin_type(self) -> PluginType:
        """Return the plugin type."""
        ...

    @property
    def version(self) -> str:
        """Return the plugin version."""
        return "0.1.0"

    def on_load(self) -> None:  # noqa: B027
        """Called when the plugin is loaded into the registry."""

    def on_unload(self) -> None:  # noqa: B027
        """Called when the plugin is unloaded from the registry."""

    def on_test_start(self, test_name: str, **kwargs: Any) -> None:  # noqa: B027
        """Called before a test case begins execution.

        Args:
            test_name: Name of the test about to run.
            **kwargs: Additional context.
        """

    def on_test_end(self, test_name: str, **kwargs: Any) -> None:  # noqa: B027
        """Called after a test case completes execution.

        Args:
            test_name: Name of the test that completed.
            **kwargs: Additional context.
        """

    def on_suite_start(self, **kwargs: Any) -> None:  # noqa: B027
        """Called before a test suite begins execution.

        Args:
            **kwargs: Additional context.
        """

    def on_suite_end(self, **kwargs: Any) -> None:  # noqa: B027
        """Called after a test suite completes execution.

        Args:
            **kwargs: Additional context.
        """

name abstractmethod property

Return the unique plugin name.

plugin_type abstractmethod property

Return the plugin type.

version property

Return the plugin version.

on_load()

Called when the plugin is loaded into the registry.

Source code in src/agentprobe/plugins/base.py 
57
58
def on_load(self) -> None:  # noqa: B027
    """Called when the plugin is loaded into the registry."""

on_unload()

Called when the plugin is unloaded from the registry.

Source code in src/agentprobe/plugins/base.py 
60
61
def on_unload(self) -> None:  # noqa: B027
    """Called when the plugin is unloaded from the registry."""

on_test_start(test_name, **kwargs)

Called before a test case begins execution.

Parameters:

Name Type Description Default
test_name str

Name of the test about to run.

 required
**kwargs Any

Additional context.

 {}
Source code in src/agentprobe/plugins/base.py 
63
64
65
66
67
68
69
def on_test_start(self, test_name: str, **kwargs: Any) -> None:  # noqa: B027
    """Called before a test case begins execution.

    Args:
        test_name: Name of the test about to run.
        **kwargs: Additional context.
    """

on_test_end(test_name, **kwargs)

Called after a test case completes execution.

Parameters:

Name Type Description Default
test_name str

Name of the test that completed.

 required
**kwargs Any

Additional context.

 {}
Source code in src/agentprobe/plugins/base.py 
71
72
73
74
75
76
77
def on_test_end(self, test_name: str, **kwargs: Any) -> None:  # noqa: B027
    """Called after a test case completes execution.

    Args:
        test_name: Name of the test that completed.
        **kwargs: Additional context.
    """

on_suite_start(**kwargs)

Called before a test suite begins execution.

Parameters:

Name Type Description Default
**kwargs Any

Additional context.

 {}
Source code in src/agentprobe/plugins/base.py 
79
80
81
82
83
84
def on_suite_start(self, **kwargs: Any) -> None:  # noqa: B027
    """Called before a test suite begins execution.

    Args:
        **kwargs: Additional context.
    """

on_suite_end(**kwargs)

Called after a test suite completes execution.

Parameters:

Name Type Description Default
**kwargs Any

Additional context.

 {}
Source code in src/agentprobe/plugins/base.py 
86
87
88
89
90
91
def on_suite_end(self, **kwargs: Any) -> None:  # noqa: B027
    """Called after a test suite completes execution.

    Args:
        **kwargs: Additional context.
    """

EvaluatorPlugin

Bases: PluginBase

Plugin that provides a custom evaluator.

Subclasses must implement create_evaluator() to return an object satisfying :class:~agentprobe.core.protocols.EvaluatorProtocol.

Source code in src/agentprobe/plugins/base.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
class EvaluatorPlugin(PluginBase):
    """Plugin that provides a custom evaluator.

    Subclasses must implement ``create_evaluator()`` to return an object
    satisfying :class:`~agentprobe.core.protocols.EvaluatorProtocol`.
    """

    @property
    def plugin_type(self) -> PluginType:
        """Return EVALUATOR type."""
        return PluginType.EVALUATOR

    @abstractmethod
    def create_evaluator(self) -> EvaluatorProtocol:
        """Create and return an evaluator instance.

        Returns:
            An evaluator conforming to EvaluatorProtocol.
        """
        ...

plugin_type property

Return EVALUATOR type.

create_evaluator() abstractmethod

Create and return an evaluator instance.

Returns:

Type Description
EvaluatorProtocol

An evaluator conforming to EvaluatorProtocol.
Source code in src/agentprobe/plugins/base.py 
106
107
108
109
110
111
112
113
@abstractmethod
def create_evaluator(self) -> EvaluatorProtocol:
    """Create and return an evaluator instance.

    Returns:
        An evaluator conforming to EvaluatorProtocol.
    """
    ...

AdapterPlugin

Bases: PluginBase

Plugin that provides a custom adapter.

Subclasses must implement create_adapter() to return an object satisfying :class:~agentprobe.core.protocols.AdapterProtocol.

Source code in src/agentprobe/plugins/base.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
class AdapterPlugin(PluginBase):
    """Plugin that provides a custom adapter.

    Subclasses must implement ``create_adapter()`` to return an object
    satisfying :class:`~agentprobe.core.protocols.AdapterProtocol`.
    """

    @property
    def plugin_type(self) -> PluginType:
        """Return ADAPTER type."""
        return PluginType.ADAPTER

    @abstractmethod
    def create_adapter(self) -> AdapterProtocol:
        """Create and return an adapter instance.

        Returns:
            An adapter conforming to AdapterProtocol.
        """
        ...

plugin_type property

Return ADAPTER type.

create_adapter() abstractmethod

Create and return an adapter instance.

Returns:

Type Description
AdapterProtocol

An adapter conforming to AdapterProtocol.
Source code in src/agentprobe/plugins/base.py 
128
129
130
131
132
133
134
135
@abstractmethod
def create_adapter(self) -> AdapterProtocol:
    """Create and return an adapter instance.

    Returns:
        An adapter conforming to AdapterProtocol.
    """
    ...

ReporterPlugin

Bases: PluginBase

Plugin that provides a custom reporter.

Subclasses must implement create_reporter() to return an object satisfying :class:~agentprobe.core.protocols.ReporterProtocol.

Source code in src/agentprobe/plugins/base.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
class ReporterPlugin(PluginBase):
    """Plugin that provides a custom reporter.

    Subclasses must implement ``create_reporter()`` to return an object
    satisfying :class:`~agentprobe.core.protocols.ReporterProtocol`.
    """

    @property
    def plugin_type(self) -> PluginType:
        """Return REPORTER type."""
        return PluginType.REPORTER

    @abstractmethod
    def create_reporter(self) -> ReporterProtocol:
        """Create and return a reporter instance.

        Returns:
            A reporter conforming to ReporterProtocol.
        """
        ...

plugin_type property

Return REPORTER type.

create_reporter() abstractmethod

Create and return a reporter instance.

Returns:

Type Description
ReporterProtocol

A reporter conforming to ReporterProtocol.
Source code in src/agentprobe/plugins/base.py 
150
151
152
153
154
155
156
157
@abstractmethod
def create_reporter(self) -> ReporterProtocol:
    """Create and return a reporter instance.

    Returns:
        A reporter conforming to ReporterProtocol.
    """
    ...

StoragePlugin

Bases: PluginBase

Plugin that provides a custom storage backend.

Subclasses must implement create_storage() to return an object satisfying :class:~agentprobe.core.protocols.StorageProtocol.

Source code in src/agentprobe/plugins/base.py 
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
class StoragePlugin(PluginBase):
    """Plugin that provides a custom storage backend.

    Subclasses must implement ``create_storage()`` to return an object
    satisfying :class:`~agentprobe.core.protocols.StorageProtocol`.
    """

    @property
    def plugin_type(self) -> PluginType:
        """Return STORAGE type."""
        return PluginType.STORAGE

    @abstractmethod
    def create_storage(self) -> StorageProtocol:
        """Create and return a storage instance.

        Returns:
            A storage backend conforming to StorageProtocol.
        """
        ...

plugin_type property

Return STORAGE type.

create_storage() abstractmethod

Create and return a storage instance.

Returns:

Type Description
StorageProtocol

A storage backend conforming to StorageProtocol.
Source code in src/agentprobe/plugins/base.py 
172
173
174
175
176
177
178
179
@abstractmethod
def create_storage(self) -> StorageProtocol:
    """Create and return a storage instance.

    Returns:
        A storage backend conforming to StorageProtocol.
    """
    ...

Registry

agentprobe.plugins.registry

Plugin registry: stores and retrieves plugins by name and type.

Provides a simple dict-based registry with type filtering support.

PluginRegistry

Registry for managing loaded plugins.

Stores plugins by name and provides lookup by name or type. Prevents duplicate registrations.

Source code in src/agentprobe/plugins/registry.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
class PluginRegistry:
    """Registry for managing loaded plugins.

    Stores plugins by name and provides lookup by name or type.
    Prevents duplicate registrations.
    """

    def __init__(self) -> None:
        self._plugins: dict[str, PluginBase] = {}

    def register(self, plugin: PluginBase) -> None:
        """Register a plugin.

        Args:
            plugin: The plugin to register.

        Raises:
            PluginError: If a plugin with the same name is already registered.
        """
        if plugin.name in self._plugins:
            raise PluginError(f"Plugin '{plugin.name}' is already registered")
        self._plugins[plugin.name] = plugin
        logger.info("Registered plugin '%s' (type=%s)", plugin.name, plugin.plugin_type)

    def unregister(self, name: str) -> None:
        """Unregister a plugin by name.

        Args:
            name: The plugin name to remove.

        Raises:
            PluginError: If no plugin with the given name is registered.
        """
        if name not in self._plugins:
            raise PluginError(f"Plugin '{name}' is not registered")
        del self._plugins[name]
        logger.info("Unregistered plugin '%s'", name)

    def get(self, name: str) -> PluginBase | None:
        """Get a plugin by name.

        Args:
            name: The plugin name to look up.

        Returns:
            The plugin if found, otherwise None.
        """
        return self._plugins.get(name)

    def list_plugins(self) -> list[PluginBase]:
        """Return all registered plugins.

        Returns:
            A list of all plugins in registration order.
        """
        return list(self._plugins.values())

    def list_by_type(self, plugin_type: PluginType) -> list[PluginBase]:
        """Return all plugins of a given type.

        Args:
            plugin_type: The type to filter by.

        Returns:
            A list of matching plugins.
        """
        return [p for p in self._plugins.values() if p.plugin_type == plugin_type]

    def clear(self) -> None:
        """Remove all registered plugins."""
        self._plugins.clear()
        logger.info("Plugin registry cleared")

    def __len__(self) -> int:
        return len(self._plugins)

    def __contains__(self, name: str) -> bool:
        return name in self._plugins

register(plugin)

Register a plugin.

Parameters:

Name Type Description Default
plugin PluginBase

The plugin to register.

 required

Raises:

Type Description
PluginError

If a plugin with the same name is already registered.
Source code in src/agentprobe/plugins/registry.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
def register(self, plugin: PluginBase) -> None:
    """Register a plugin.

    Args:
        plugin: The plugin to register.

    Raises:
        PluginError: If a plugin with the same name is already registered.
    """
    if plugin.name in self._plugins:
        raise PluginError(f"Plugin '{plugin.name}' is already registered")
    self._plugins[plugin.name] = plugin
    logger.info("Registered plugin '%s' (type=%s)", plugin.name, plugin.plugin_type)

unregister(name)

Unregister a plugin by name.

Parameters:

Name Type Description Default
name str

The plugin name to remove.

 required

Raises:

Type Description
PluginError

If no plugin with the given name is registered.
Source code in src/agentprobe/plugins/registry.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
def unregister(self, name: str) -> None:
    """Unregister a plugin by name.

    Args:
        name: The plugin name to remove.

    Raises:
        PluginError: If no plugin with the given name is registered.
    """
    if name not in self._plugins:
        raise PluginError(f"Plugin '{name}' is not registered")
    del self._plugins[name]
    logger.info("Unregistered plugin '%s'", name)

get(name)

Get a plugin by name.

Parameters:

Name Type Description Default
name str

The plugin name to look up.

 required

Returns:

Type Description
PluginBase | None

The plugin if found, otherwise None.
Source code in src/agentprobe/plugins/registry.py 
55
56
57
58
59
60
61
62
63
64
def get(self, name: str) -> PluginBase | None:
    """Get a plugin by name.

    Args:
        name: The plugin name to look up.

    Returns:
        The plugin if found, otherwise None.
    """
    return self._plugins.get(name)

list_plugins()

Return all registered plugins.

Returns:

Type Description
list[PluginBase]

A list of all plugins in registration order.
Source code in src/agentprobe/plugins/registry.py 
66
67
68
69
70
71
72
def list_plugins(self) -> list[PluginBase]:
    """Return all registered plugins.

    Returns:
        A list of all plugins in registration order.
    """
    return list(self._plugins.values())

list_by_type(plugin_type)

Return all plugins of a given type.

Parameters:

Name Type Description Default
plugin_type PluginType

The type to filter by.

 required

Returns:

Type Description
list[PluginBase]

A list of matching plugins.
Source code in src/agentprobe/plugins/registry.py 
74
75
76
77
78
79
80
81
82
83
def list_by_type(self, plugin_type: PluginType) -> list[PluginBase]:
    """Return all plugins of a given type.

    Args:
        plugin_type: The type to filter by.

    Returns:
        A list of matching plugins.
    """
    return [p for p in self._plugins.values() if p.plugin_type == plugin_type]

clear()

Remove all registered plugins.

Source code in src/agentprobe/plugins/registry.py 
85
86
87
88
def clear(self) -> None:
    """Remove all registered plugins."""
    self._plugins.clear()
    logger.info("Plugin registry cleared")

Loader

agentprobe.plugins.loader

Plugin loader: discovers and loads plugins from entry points and paths.

Supports two discovery mechanisms: 1. Entry points (importlib.metadata) for installed packages. 2. File paths (importlib.util) for local plugin files.

PluginLoader

Loads plugin classes from various sources.

Discovers and instantiates plugins from Python entry points, file paths, or direct class references.

Source code in src/agentprobe/plugins/loader.py 
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
class PluginLoader:
    """Loads plugin classes from various sources.

    Discovers and instantiates plugins from Python entry points,
    file paths, or direct class references.
    """

    def __init__(self, entry_point_group: str = "agentprobe.plugins") -> None:
        """Initialize the plugin loader.

        Args:
            entry_point_group: Entry point group name for discovery.
        """
        self._entry_point_group = entry_point_group

    def load_from_entry_points(self) -> list[PluginBase]:
        """Discover and load plugins from installed package entry points.

        Returns:
            A list of loaded plugin instances.
        """
        plugins: list[PluginBase] = []
        try:
            eps = importlib.metadata.entry_points()
        except Exception:
            logger.exception("Failed to read entry points")
            return plugins

        group_eps: list[Any] = []
        if hasattr(eps, "select"):
            group_eps = list(eps.select(group=self._entry_point_group))
        elif isinstance(eps, dict):
            group_eps = eps.get(self._entry_point_group, [])  # pragma: no cover

        for ep in group_eps:
            try:
                plugin_cls = ep.load()
                plugin = self._instantiate(plugin_cls, source=f"entry_point:{ep.name}")
                plugins.append(plugin)
            except Exception:
                logger.exception("Failed to load plugin from entry point '%s'", ep.name)

        return plugins

    def load_from_path(self, path: str | Path) -> PluginBase:
        """Load a plugin from a Python file path.

        The file must contain exactly one class that subclasses PluginBase.

        Args:
            path: Path to the Python file containing the plugin.

        Returns:
            The loaded plugin instance.

        Raises:
            PluginError: If the file cannot be loaded or contains no plugin class.
        """
        file_path = Path(path)
        if not file_path.exists():
            raise PluginError(f"Plugin file not found: {file_path}")
        if not file_path.suffix == ".py":
            raise PluginError(f"Plugin file must be a .py file: {file_path}")

        module_name = f"agentprobe_plugin_{file_path.stem}"

        try:
            spec = importlib.util.spec_from_file_location(module_name, file_path)
            if spec is None or spec.loader is None:
                raise PluginError(f"Cannot create module spec for: {file_path}")
            module = importlib.util.module_from_spec(spec)
            spec.loader.exec_module(module)
        except PluginError:
            raise
        except Exception as exc:
            raise PluginError(f"Failed to load module from {file_path}: {exc}") from exc

        plugin_classes = [
            obj
            for name in dir(module)
            if not name.startswith("_")
            and isinstance(obj := getattr(module, name), type)
            and issubclass(obj, PluginBase)
            and obj is not PluginBase
        ]

        if not plugin_classes:
            raise PluginError(f"No PluginBase subclass found in {file_path}")

        return self._instantiate(plugin_classes[0], source=str(file_path))

    def load_from_class(self, cls: type[PluginBase]) -> PluginBase:
        """Load a plugin from a direct class reference.

        Args:
            cls: The plugin class to instantiate.

        Returns:
            The loaded plugin instance.

        Raises:
            PluginError: If the class is not a PluginBase subclass.
        """
        if not (isinstance(cls, type) and issubclass(cls, PluginBase)):
            raise PluginError(f"{cls} is not a PluginBase subclass")
        return self._instantiate(cls, source=f"class:{cls.__name__}")

    def load_all(
        self,
        directories: list[str] | None = None,
    ) -> list[PluginBase]:
        """Load plugins from entry points and optional directories.

        Args:
            directories: Additional directories to scan for .py plugin files.

        Returns:
            A list of all loaded plugin instances.
        """
        plugins = self.load_from_entry_points()

        for dir_path in directories or []:
            path = Path(dir_path)
            if not path.is_dir():
                logger.warning("Plugin directory not found: %s", dir_path)
                continue
            for py_file in sorted(path.glob("*.py")):
                if py_file.name.startswith("_"):
                    continue
                try:
                    plugin = self.load_from_path(py_file)
                    plugins.append(plugin)
                except PluginError:
                    logger.exception("Failed to load plugin from %s", py_file)

        return plugins

    def _instantiate(self, cls: type, source: str) -> PluginBase:
        """Instantiate a plugin class with error handling.

        Args:
            cls: The class to instantiate.
            source: Description of where the class came from (for logging).

        Returns:
            The plugin instance.

        Raises:
            PluginError: If instantiation fails.
        """
        try:
            instance = cls()
        except Exception as exc:
            raise PluginError(f"Failed to instantiate plugin from {source}: {exc}") from exc

        if not isinstance(instance, PluginBase):
            raise PluginError(f"Plugin from {source} is not a PluginBase instance")

        logger.info("Loaded plugin '%s' from %s", instance.name, source)
        return instance

__init__(entry_point_group='agentprobe.plugins')

Initialize the plugin loader.

Parameters:

Name Type Description Default
entry_point_group str

Entry point group name for discovery.

 'agentprobe.plugins'
Source code in src/agentprobe/plugins/loader.py 
30
31
32
33
34
35
36
def __init__(self, entry_point_group: str = "agentprobe.plugins") -> None:
    """Initialize the plugin loader.

    Args:
        entry_point_group: Entry point group name for discovery.
    """
    self._entry_point_group = entry_point_group

load_from_entry_points()

Discover and load plugins from installed package entry points.

Returns:

Type Description
list[PluginBase]

A list of loaded plugin instances.
Source code in src/agentprobe/plugins/loader.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
def load_from_entry_points(self) -> list[PluginBase]:
    """Discover and load plugins from installed package entry points.

    Returns:
        A list of loaded plugin instances.
    """
    plugins: list[PluginBase] = []
    try:
        eps = importlib.metadata.entry_points()
    except Exception:
        logger.exception("Failed to read entry points")
        return plugins

    group_eps: list[Any] = []
    if hasattr(eps, "select"):
        group_eps = list(eps.select(group=self._entry_point_group))
    elif isinstance(eps, dict):
        group_eps = eps.get(self._entry_point_group, [])  # pragma: no cover

    for ep in group_eps:
        try:
            plugin_cls = ep.load()
            plugin = self._instantiate(plugin_cls, source=f"entry_point:{ep.name}")
            plugins.append(plugin)
        except Exception:
            logger.exception("Failed to load plugin from entry point '%s'", ep.name)

    return plugins

load_from_path(path)

Load a plugin from a Python file path.

The file must contain exactly one class that subclasses PluginBase.

Parameters:

Name Type Description Default
path str | Path

Path to the Python file containing the plugin.

 required

Returns:

Type Description
PluginBase

The loaded plugin instance.

Raises:

Type Description
PluginError

If the file cannot be loaded or contains no plugin class.
Source code in src/agentprobe/plugins/loader.py 
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
def load_from_path(self, path: str | Path) -> PluginBase:
    """Load a plugin from a Python file path.

    The file must contain exactly one class that subclasses PluginBase.

    Args:
        path: Path to the Python file containing the plugin.

    Returns:
        The loaded plugin instance.

    Raises:
        PluginError: If the file cannot be loaded or contains no plugin class.
    """
    file_path = Path(path)
    if not file_path.exists():
        raise PluginError(f"Plugin file not found: {file_path}")
    if not file_path.suffix == ".py":
        raise PluginError(f"Plugin file must be a .py file: {file_path}")

    module_name = f"agentprobe_plugin_{file_path.stem}"

    try:
        spec = importlib.util.spec_from_file_location(module_name, file_path)
        if spec is None or spec.loader is None:
            raise PluginError(f"Cannot create module spec for: {file_path}")
        module = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(module)
    except PluginError:
        raise
    except Exception as exc:
        raise PluginError(f"Failed to load module from {file_path}: {exc}") from exc

    plugin_classes = [
        obj
        for name in dir(module)
        if not name.startswith("_")
        and isinstance(obj := getattr(module, name), type)
        and issubclass(obj, PluginBase)
        and obj is not PluginBase
    ]

    if not plugin_classes:
        raise PluginError(f"No PluginBase subclass found in {file_path}")

    return self._instantiate(plugin_classes[0], source=str(file_path))

load_from_class(cls)

Load a plugin from a direct class reference.

Parameters:

Name Type Description Default
cls type[PluginBase]

The plugin class to instantiate.

 required

Returns:

Type Description
PluginBase

The loaded plugin instance.

Raises:

Type Description
PluginError

If the class is not a PluginBase subclass.
Source code in src/agentprobe/plugins/loader.py 
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
def load_from_class(self, cls: type[PluginBase]) -> PluginBase:
    """Load a plugin from a direct class reference.

    Args:
        cls: The plugin class to instantiate.

    Returns:
        The loaded plugin instance.

    Raises:
        PluginError: If the class is not a PluginBase subclass.
    """
    if not (isinstance(cls, type) and issubclass(cls, PluginBase)):
        raise PluginError(f"{cls} is not a PluginBase subclass")
    return self._instantiate(cls, source=f"class:{cls.__name__}")

load_all(directories=None)

Load plugins from entry points and optional directories.

Parameters:

Name Type Description Default
directories list[str] | None

Additional directories to scan for .py plugin files.

 None

Returns:

Type Description
list[PluginBase]

A list of all loaded plugin instances.
Source code in src/agentprobe/plugins/loader.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
def load_all(
    self,
    directories: list[str] | None = None,
) -> list[PluginBase]:
    """Load plugins from entry points and optional directories.

    Args:
        directories: Additional directories to scan for .py plugin files.

    Returns:
        A list of all loaded plugin instances.
    """
    plugins = self.load_from_entry_points()

    for dir_path in directories or []:
        path = Path(dir_path)
        if not path.is_dir():
            logger.warning("Plugin directory not found: %s", dir_path)
            continue
        for py_file in sorted(path.glob("*.py")):
            if py_file.name.startswith("_"):
                continue
            try:
                plugin = self.load_from_path(py_file)
                plugins.append(plugin)
            except PluginError:
                logger.exception("Failed to load plugin from %s", py_file)

    return plugins

Manager

agentprobe.plugins.manager

Plugin manager: orchestrates plugin lifecycle and event dispatch.

Coordinates loading, registration, lifecycle hooks, and factory collection across all loaded plugins.

PluginManager

Orchestrates plugin lifecycle and event dispatch.

Manages the full lifecycle of plugins: loading from sources, registering, dispatching lifecycle events, and collecting factory-created objects from typed plugins.

Attributes:

Name Type Description
registry

The plugin registry.
Source code in src/agentprobe/plugins/manager.py 
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
class PluginManager:
    """Orchestrates plugin lifecycle and event dispatch.

    Manages the full lifecycle of plugins: loading from sources,
    registering, dispatching lifecycle events, and collecting
    factory-created objects from typed plugins.

    Attributes:
        registry: The plugin registry.
    """

    def __init__(
        self,
        entry_point_group: str = "agentprobe.plugins",
    ) -> None:
        """Initialize the plugin manager.

        Args:
            entry_point_group: Entry point group for plugin discovery.
        """
        self.registry = PluginRegistry()
        self._loader = PluginLoader(entry_point_group=entry_point_group)

    def load_plugins(
        self,
        directories: list[str] | None = None,
        classes: list[type[PluginBase]] | None = None,
    ) -> list[PluginBase]:
        """Load and register plugins from all sources.

        Args:
            directories: Additional directories to scan.
            classes: Direct class references to load.

        Returns:
            A list of all loaded and registered plugins.
        """
        loaded: list[PluginBase] = []

        discovered = self._loader.load_all(directories=directories)
        for plugin in discovered:
            self._safe_register(plugin)
            loaded.append(plugin)

        for cls in classes or []:
            try:
                plugin = self._loader.load_from_class(cls)
                self._safe_register(plugin)
                loaded.append(plugin)
            except Exception:
                logger.exception("Failed to load plugin class %s", cls.__name__)

        return loaded

    def unload_all(self) -> None:
        """Unload all plugins, calling on_unload for each."""
        for plugin in self.registry.list_plugins():
            try:
                plugin.on_unload()
            except Exception:
                logger.exception("Error during on_unload for plugin '%s'", plugin.name)
        self.registry.clear()

    def dispatch_test_start(self, test_name: str, **kwargs: Any) -> None:
        """Dispatch on_test_start to all registered plugins.

        Errors from individual plugins are logged but do not propagate.

        Args:
            test_name: Name of the test starting.
            **kwargs: Additional context.
        """
        for plugin in self.registry.list_plugins():
            try:
                plugin.on_test_start(test_name, **kwargs)
            except Exception:
                logger.exception("Plugin '%s' failed in on_test_start", plugin.name)

    def dispatch_test_end(self, test_name: str, **kwargs: Any) -> None:
        """Dispatch on_test_end to all registered plugins.

        Args:
            test_name: Name of the test ending.
            **kwargs: Additional context.
        """
        for plugin in self.registry.list_plugins():
            try:
                plugin.on_test_end(test_name, **kwargs)
            except Exception:
                logger.exception("Plugin '%s' failed in on_test_end", plugin.name)

    def dispatch_suite_start(self, **kwargs: Any) -> None:
        """Dispatch on_suite_start to all registered plugins.

        Args:
            **kwargs: Additional context.
        """
        for plugin in self.registry.list_plugins():
            try:
                plugin.on_suite_start(**kwargs)
            except Exception:
                logger.exception("Plugin '%s' failed in on_suite_start", plugin.name)

    def dispatch_suite_end(self, **kwargs: Any) -> None:
        """Dispatch on_suite_end to all registered plugins.

        Args:
            **kwargs: Additional context.
        """
        for plugin in self.registry.list_plugins():
            try:
                plugin.on_suite_end(**kwargs)
            except Exception:
                logger.exception("Plugin '%s' failed in on_suite_end", plugin.name)

    def get_evaluators(self) -> list[EvaluatorProtocol]:
        """Collect evaluators from all EvaluatorPlugin instances.

        Returns:
            A list of evaluator instances.
        """
        evaluators: list[EvaluatorProtocol] = []
        for plugin in self.registry.list_by_type(PluginType.EVALUATOR):
            if isinstance(plugin, EvaluatorPlugin):
                try:
                    evaluators.append(plugin.create_evaluator())
                except Exception:
                    logger.exception("Plugin '%s' failed to create evaluator", plugin.name)
        return evaluators

    def get_adapters(self) -> list[AdapterProtocol]:
        """Collect adapters from all AdapterPlugin instances.

        Returns:
            A list of adapter instances.
        """
        adapters: list[AdapterProtocol] = []
        for plugin in self.registry.list_by_type(PluginType.ADAPTER):
            if isinstance(plugin, AdapterPlugin):
                try:
                    adapters.append(plugin.create_adapter())
                except Exception:
                    logger.exception("Plugin '%s' failed to create adapter", plugin.name)
        return adapters

    def get_reporters(self) -> list[ReporterProtocol]:
        """Collect reporters from all ReporterPlugin instances.

        Returns:
            A list of reporter instances.
        """
        reporters: list[ReporterProtocol] = []
        for plugin in self.registry.list_by_type(PluginType.REPORTER):
            if isinstance(plugin, ReporterPlugin):
                try:
                    reporters.append(plugin.create_reporter())
                except Exception:
                    logger.exception("Plugin '%s' failed to create reporter", plugin.name)
        return reporters

    def get_storage_backends(self) -> list[StorageProtocol]:
        """Collect storage backends from all StoragePlugin instances.

        Returns:
            A list of storage backend instances.
        """
        backends: list[StorageProtocol] = []
        for plugin in self.registry.list_by_type(PluginType.STORAGE):
            if isinstance(plugin, StoragePlugin):
                try:
                    backends.append(plugin.create_storage())
                except Exception:
                    logger.exception("Plugin '%s' failed to create storage", plugin.name)
        return backends

    def _safe_register(self, plugin: PluginBase) -> None:
        """Register a plugin, calling on_load after registration.

        Args:
            plugin: The plugin to register.
        """
        self.registry.register(plugin)
        try:
            plugin.on_load()
        except Exception:
            logger.exception("Plugin '%s' failed in on_load", plugin.name)

__init__(entry_point_group='agentprobe.plugins')

Initialize the plugin manager.

Parameters:

Name Type Description Default
entry_point_group str

Entry point group for plugin discovery.

 'agentprobe.plugins'
Source code in src/agentprobe/plugins/manager.py 
45
46
47
48
49
50
51
52
53
54
55
def __init__(
    self,
    entry_point_group: str = "agentprobe.plugins",
) -> None:
    """Initialize the plugin manager.

    Args:
        entry_point_group: Entry point group for plugin discovery.
    """
    self.registry = PluginRegistry()
    self._loader = PluginLoader(entry_point_group=entry_point_group)

load_plugins(directories=None, classes=None)

Load and register plugins from all sources.

Parameters:

Name Type Description Default
directories list[str] | None

Additional directories to scan.

 None
classes list[type[PluginBase]] | None

Direct class references to load.

 None

Returns:

Type Description
list[PluginBase]

A list of all loaded and registered plugins.
Source code in src/agentprobe/plugins/manager.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def load_plugins(
    self,
    directories: list[str] | None = None,
    classes: list[type[PluginBase]] | None = None,
) -> list[PluginBase]:
    """Load and register plugins from all sources.

    Args:
        directories: Additional directories to scan.
        classes: Direct class references to load.

    Returns:
        A list of all loaded and registered plugins.
    """
    loaded: list[PluginBase] = []

    discovered = self._loader.load_all(directories=directories)
    for plugin in discovered:
        self._safe_register(plugin)
        loaded.append(plugin)

    for cls in classes or []:
        try:
            plugin = self._loader.load_from_class(cls)
            self._safe_register(plugin)
            loaded.append(plugin)
        except Exception:
            logger.exception("Failed to load plugin class %s", cls.__name__)

    return loaded

unload_all()

Unload all plugins, calling on_unload for each.

Source code in src/agentprobe/plugins/manager.py 
88
89
90
91
92
93
94
95
def unload_all(self) -> None:
    """Unload all plugins, calling on_unload for each."""
    for plugin in self.registry.list_plugins():
        try:
            plugin.on_unload()
        except Exception:
            logger.exception("Error during on_unload for plugin '%s'", plugin.name)
    self.registry.clear()

dispatch_test_start(test_name, **kwargs)

Dispatch on_test_start to all registered plugins.

Errors from individual plugins are logged but do not propagate.

Parameters:

Name Type Description Default
test_name str

Name of the test starting.

 required
**kwargs Any

Additional context.

 {}
Source code in src/agentprobe/plugins/manager.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def dispatch_test_start(self, test_name: str, **kwargs: Any) -> None:
    """Dispatch on_test_start to all registered plugins.

    Errors from individual plugins are logged but do not propagate.

    Args:
        test_name: Name of the test starting.
        **kwargs: Additional context.
    """
    for plugin in self.registry.list_plugins():
        try:
            plugin.on_test_start(test_name, **kwargs)
        except Exception:
            logger.exception("Plugin '%s' failed in on_test_start", plugin.name)

dispatch_test_end(test_name, **kwargs)

Dispatch on_test_end to all registered plugins.

Parameters:

Name Type Description Default
test_name str

Name of the test ending.

 required
**kwargs Any

Additional context.

 {}
Source code in src/agentprobe/plugins/manager.py 
112
113
114
115
116
117
118
119
120
121
122
123
def dispatch_test_end(self, test_name: str, **kwargs: Any) -> None:
    """Dispatch on_test_end to all registered plugins.

    Args:
        test_name: Name of the test ending.
        **kwargs: Additional context.
    """
    for plugin in self.registry.list_plugins():
        try:
            plugin.on_test_end(test_name, **kwargs)
        except Exception:
            logger.exception("Plugin '%s' failed in on_test_end", plugin.name)

dispatch_suite_start(**kwargs)

Dispatch on_suite_start to all registered plugins.

Parameters:

Name Type Description Default
**kwargs Any

Additional context.

 {}
Source code in src/agentprobe/plugins/manager.py 
125
126
127
128
129
130
131
132
133
134
135
def dispatch_suite_start(self, **kwargs: Any) -> None:
    """Dispatch on_suite_start to all registered plugins.

    Args:
        **kwargs: Additional context.
    """
    for plugin in self.registry.list_plugins():
        try:
            plugin.on_suite_start(**kwargs)
        except Exception:
            logger.exception("Plugin '%s' failed in on_suite_start", plugin.name)

dispatch_suite_end(**kwargs)

Dispatch on_suite_end to all registered plugins.

Parameters:

Name Type Description Default
**kwargs Any

Additional context.

 {}
Source code in src/agentprobe/plugins/manager.py 
137
138
139
140
141
142
143
144
145
146
147
def dispatch_suite_end(self, **kwargs: Any) -> None:
    """Dispatch on_suite_end to all registered plugins.

    Args:
        **kwargs: Additional context.
    """
    for plugin in self.registry.list_plugins():
        try:
            plugin.on_suite_end(**kwargs)
        except Exception:
            logger.exception("Plugin '%s' failed in on_suite_end", plugin.name)

get_evaluators()

Collect evaluators from all EvaluatorPlugin instances.

Returns:

Type Description
list[EvaluatorProtocol]

A list of evaluator instances.
Source code in src/agentprobe/plugins/manager.py 
149
150
151
152
153
154
155
156
157
158
159
160
161
162
def get_evaluators(self) -> list[EvaluatorProtocol]:
    """Collect evaluators from all EvaluatorPlugin instances.

    Returns:
        A list of evaluator instances.
    """
    evaluators: list[EvaluatorProtocol] = []
    for plugin in self.registry.list_by_type(PluginType.EVALUATOR):
        if isinstance(plugin, EvaluatorPlugin):
            try:
                evaluators.append(plugin.create_evaluator())
            except Exception:
                logger.exception("Plugin '%s' failed to create evaluator", plugin.name)
    return evaluators

get_adapters()

Collect adapters from all AdapterPlugin instances.

Returns:

Type Description
list[AdapterProtocol]

A list of adapter instances.
Source code in src/agentprobe/plugins/manager.py 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
def get_adapters(self) -> list[AdapterProtocol]:
    """Collect adapters from all AdapterPlugin instances.

    Returns:
        A list of adapter instances.
    """
    adapters: list[AdapterProtocol] = []
    for plugin in self.registry.list_by_type(PluginType.ADAPTER):
        if isinstance(plugin, AdapterPlugin):
            try:
                adapters.append(plugin.create_adapter())
            except Exception:
                logger.exception("Plugin '%s' failed to create adapter", plugin.name)
    return adapters

get_reporters()

Collect reporters from all ReporterPlugin instances.

Returns:

Type Description
list[ReporterProtocol]

A list of reporter instances.
Source code in src/agentprobe/plugins/manager.py 
179
180
181
182
183
184
185
186
187
188
189
190
191
192
def get_reporters(self) -> list[ReporterProtocol]:
    """Collect reporters from all ReporterPlugin instances.

    Returns:
        A list of reporter instances.
    """
    reporters: list[ReporterProtocol] = []
    for plugin in self.registry.list_by_type(PluginType.REPORTER):
        if isinstance(plugin, ReporterPlugin):
            try:
                reporters.append(plugin.create_reporter())
            except Exception:
                logger.exception("Plugin '%s' failed to create reporter", plugin.name)
    return reporters

get_storage_backends()

Collect storage backends from all StoragePlugin instances.

Returns:

Type Description
list[StorageProtocol]

A list of storage backend instances.
Source code in src/agentprobe/plugins/manager.py 
194
195
196
197
198
199
200
201
202
203
204
205
206
207
def get_storage_backends(self) -> list[StorageProtocol]:
    """Collect storage backends from all StoragePlugin instances.

    Returns:
        A list of storage backend instances.
    """
    backends: list[StorageProtocol] = []
    for plugin in self.registry.list_by_type(PluginType.STORAGE):
        if isinstance(plugin, StoragePlugin):
            try:
                backends.append(plugin.create_storage())
            except Exception:
                logger.exception("Plugin '%s' failed to create storage", plugin.name)
    return backends