feat(hooks): add Plugin Protocol for agent extensibility (#1733)

Co-authored-by: Strands Agent <217235299+strands-agent@users.noreply.github.com>

Author: Unshure

AGENTS.md

@@ -126,6 +126,10 @@ strands-agents/
 │   │   ├── events.py                     # Hook event definitions
 │   │   └── registry.py                   # Hook registration
 │   │
+│   ├── plugins/                          # Plugin system
+│   │   ├── plugin.py                     # Plugin Protocol definition
+│   │   └── registry.py                   # PluginRegistry for tracking plugins
+│   │
 │   ├── handlers/                         # Event handlers
 │   │   └── callback_handler.py           # Callback handling
 │   │
@@ -171,6 +175,7 @@ strands-agents/
 │       ├── session/
 │       ├── telemetry/
 │       ├── hooks/
+│       ├── plugins/
 │       ├── handlers/
 │       ├── experimental/
 │       └── utils/

src/strands/__init__.py

@@ -4,6 +4,7 @@
 from .agent.agent import Agent
 from .agent.base import AgentBase
 from .event_loop._retry import ModelRetryStrategy
+from .plugins import Plugin
 from .tools.decorator import tool
 from .types.tools import ToolContext
 
@@ -13,6 +14,7 @@
     "agent",
     "models",
     "ModelRetryStrategy",
+    "Plugin",
     "tool",
     "ToolContext",
     "types",

src/strands/plugins/__init__.py

@@ -0,0 +1,25 @@
+"""Plugin system for extending agent functionality.
+
+This module provides a composable mechanism for building objects that can
+extend agent behavior through a standardized initialization pattern.
+
+Example Usage:
+    ```python
+    from strands.plugins import Plugin
+
+    class LoggingPlugin:
+        name = "logging"
+
+        def init_plugin(self, agent: Agent) -> None:
+            agent.add_hook(self.on_model_call, BeforeModelCallEvent)
+
+        def on_model_call(self, event: BeforeModelCallEvent) -> None:
+            print(f"Model called for {event.agent.name}")
+    ```
+"""
+
+from .plugin import Plugin
+
+__all__ = [
+    "Plugin",
+]

src/strands/plugins/plugin.py

@@ -0,0 +1,43 @@
+"""Plugin protocol for extending agent functionality.
+
+This module defines the Plugin Protocol, which provides a composable way to
+add behavior changes to agents through a standardized initialization pattern.
+"""
+
+from collections.abc import Awaitable
+from typing import TYPE_CHECKING, Protocol, runtime_checkable
+
+if TYPE_CHECKING:
+    from ..agent import Agent
+
+
+@runtime_checkable
+class Plugin(Protocol):
+    """Protocol for objects that extend agent functionality.
+
+    Plugins provide a composable way to add behavior changes to agents.
+    They are initialized with an agent instance and can register hooks,
+    modify agent attributes, or perform other setup tasks.
+
+    Attributes:
+        name: A stable string identifier for the plugin
+
+    Example:
+        ```python
+        class MyPlugin:
+            name = "my-plugin"
+
+            def init_plugin(self, agent: Agent) -> None:
+                agent.add_hook(self.on_model_call, BeforeModelCallEvent)
+        ```
+    """
+
+    name: str
+
+    def init_plugin(self, agent: "Agent") -> None | Awaitable[None]:
+        """Initialize the plugin with an agent instance.
+
+        Args:
+            agent: The agent instance to extend.
+        """
+        ...

src/strands/plugins/registry.py

@@ -0,0 +1,73 @@
+"""Plugin registry for managing plugins attached to an agent.
+
+This module provides the _PluginRegistry class for tracking and managing
+plugins that have been initialized with an agent instance.
+"""
+
+import inspect
+import logging
+from collections.abc import Awaitable, Callable
+from typing import TYPE_CHECKING, cast
+
+from .._async import run_async
+from .plugin import Plugin
+
+if TYPE_CHECKING:
+    from ..agent import Agent
+
+logger = logging.getLogger(__name__)
+
+
+class _PluginRegistry:
+    """Registry for managing plugins attached to an agent.
+
+    The _PluginRegistry tracks plugins that have been initialized with an agent,
+    providing methods to add plugins and invoke their initialization.
+
+    Example:
+        ```python
+        registry = _PluginRegistry(agent)
+
+        class MyPlugin:
+            name = "my-plugin"
+
+            def init_plugin(self, agent: Agent) -> None:
+                pass
+
+        plugin = MyPlugin()
+        registry.add_and_init(plugin)
+        ```
+    """
+
+    def __init__(self, agent: "Agent") -> None:
+        """Initialize a plugin registry with an agent reference.
+
+        Args:
+            agent: The agent instance that plugins will be initialized with.
+        """
+        self._agent = agent
+        self._plugins: dict[str, Plugin] = {}
+
+    def add_and_init(self, plugin: Plugin) -> None:
+        """Add and initialize a plugin with the agent.
+
+        This method registers the plugin and calls its init_plugin method.
+        Handles both sync and async init_plugin implementations automatically.
+
+        Args:
+            plugin: The plugin to add and initialize.
+
+        Raises:
+            ValueError: If a plugin with the same name is already registered.
+        """
+        if plugin.name in self._plugins:
+            raise ValueError(f"plugin_name=<{plugin.name}> | plugin already registered")
+
+        logger.debug("plugin_name=<%s> | registering and initializing plugin", plugin.name)
+        self._plugins[plugin.name] = plugin
+
+        if inspect.iscoroutinefunction(plugin.init_plugin):
+            async_plugin_init = cast(Callable[..., Awaitable[None]], plugin.init_plugin)
+            run_async(lambda: async_plugin_init(self._agent))
+        else:
+            plugin.init_plugin(self._agent)

tests/strands/plugins/__init__.py

@@ -0,0 +1 @@
+"""Tests for the plugins module."""

tests/strands/plugins/test_plugins.py

@@ -0,0 +1,192 @@
+"""Tests for the plugin system."""
+
+import unittest.mock
+
+import pytest
+
+from strands.plugins import Plugin
+from strands.plugins.registry import _PluginRegistry
+
+# Plugin Protocol Tests
+
+
+def test_plugin_protocol_is_runtime_checkable():
+    """Test that Plugin Protocol is runtime checkable with isinstance."""
+
+    class MyPlugin:
+        name = "my-plugin"
+
+        def init_plugin(self, agent):
+            pass
+
+    plugin = MyPlugin()
+    assert isinstance(plugin, Plugin)
+
+
+def test_plugin_protocol_sync_implementation():
+    """Test Plugin Protocol works with synchronous init_plugin."""
+
+    class SyncPlugin:
+        name = "sync-plugin"
+
+        def init_plugin(self, agent):
+            agent.custom_attribute = "initialized by plugin"
+
+    plugin = SyncPlugin()
+    mock_agent = unittest.mock.Mock()
+
+    # Verify the plugin matches the protocol
+    assert isinstance(plugin, Plugin)
+    assert plugin.name == "sync-plugin"
+
+    # Execute init_plugin synchronously
+    plugin.init_plugin(mock_agent)
+    assert mock_agent.custom_attribute == "initialized by plugin"
+
+
+@pytest.mark.asyncio
+async def test_plugin_protocol_async_implementation():
+    """Test Plugin Protocol works with asynchronous init_plugin."""
+
+    class AsyncPlugin:
+        name = "async-plugin"
+
+        async def init_plugin(self, agent):
+            agent.custom_attribute = "initialized by async plugin"
+
+    plugin = AsyncPlugin()
+    mock_agent = unittest.mock.Mock()
+
+    # Verify the plugin matches the protocol
+    assert isinstance(plugin, Plugin)
+    assert plugin.name == "async-plugin"
+
+    # Execute init_plugin asynchronously
+    await plugin.init_plugin(mock_agent)
+    assert mock_agent.custom_attribute == "initialized by async plugin"
+
+
+def test_plugin_protocol_requires_name():
+    """Test that Plugin Protocol requires a name property."""
+
+    class PluginWithoutName:
+        def init_plugin(self, agent):
+            pass
+
+    plugin = PluginWithoutName()
+    # A class without 'name' should not pass isinstance check
+    assert not isinstance(plugin, Plugin)
+
+
+def test_plugin_protocol_requires_init_plugin_method():
+    """Test that Plugin Protocol requires an init_plugin method."""
+
+    class PluginWithoutInitPlugin:
+        name = "incomplete-plugin"
+
+    plugin = PluginWithoutInitPlugin()
+    # A class without 'init_plugin' should not pass isinstance check
+    assert not isinstance(plugin, Plugin)
+
+
+def test_plugin_protocol_with_class_attribute_name():
+    """Test Plugin Protocol works when name is a class attribute."""
+
+    class PluginWithClassAttribute:
+        name: str = "class-attr-plugin"
+
+        def init_plugin(self, agent):
+            pass
+
+    plugin = PluginWithClassAttribute()
+    assert isinstance(plugin, Plugin)
+    assert plugin.name == "class-attr-plugin"
+
+
+def test_plugin_protocol_with_property_name():
+    """Test Plugin Protocol works when name is a property."""
+
+    class PluginWithProperty:
+        @property
+        def name(self):
+            return "property-plugin"
+
+        def init_plugin(self, agent):
+            pass
+
+    plugin = PluginWithProperty()
+    assert isinstance(plugin, Plugin)
+    assert plugin.name == "property-plugin"
+
+
+# _PluginRegistry Tests
+
+
+@pytest.fixture
+def mock_agent():
+    """Create a mock agent for testing."""
+    return unittest.mock.Mock()
+
+
+@pytest.fixture
+def registry(mock_agent):
+    """Create a fresh _PluginRegistry for each test."""
+    return _PluginRegistry(mock_agent)
+
+
+def test_plugin_registry_add_and_init_calls_init_plugin(registry, mock_agent):
+    """Test adding a plugin calls its init_plugin method."""
+
+    class TestPlugin:
+        name = "test-plugin"
+
+        def __init__(self):
+            self.initialized = False
+
+        def init_plugin(self, agent):
+            self.initialized = True
+            agent.plugin_initialized = True
+
+    plugin = TestPlugin()
+    registry.add_and_init(plugin)
+
+    assert plugin.initialized
+    assert mock_agent.plugin_initialized
+
+
+def test_plugin_registry_add_duplicate_raises_error(registry, mock_agent):
+    """Test that adding a duplicate plugin raises an error."""
+
+    class TestPlugin:
+        name = "test-plugin"
+
+        def init_plugin(self, agent):
+            pass
+
+    plugin1 = TestPlugin()
+    plugin2 = TestPlugin()
+
+    registry.add_and_init(plugin1)
+
+    with pytest.raises(ValueError, match="plugin_name=<test-plugin> | plugin already registered"):
+        registry.add_and_init(plugin2)
+
+
+def test_plugin_registry_add_and_init_with_async_plugin(registry, mock_agent):
+    """Test that add_and_init handles async plugins using run_async."""
+
+    class AsyncPlugin:
+        name = "async-plugin"
+
+        def __init__(self):
+            self.initialized = False
+
+        async def init_plugin(self, agent):
+            self.initialized = True
+            agent.async_plugin_initialized = True
+
+    plugin = AsyncPlugin()
+    registry.add_and_init(plugin)
+
+    assert plugin.initialized
+    assert mock_agent.async_plugin_initialized