feat(plugins): add MultiAgentPlugin for Swarm and Graph orchestrators

AGENTS.md

@@ -130,8 +130,11 @@ strands-agents/
 │   │
 │   ├── plugins/                          # Plugin system
 │   │   ├── plugin.py                     # Plugin base class
+│   │   ├── multiagent_plugin.py          # MultiAgentPlugin base class
 │   │   ├── decorator.py                  # @hook decorator
-│   │   └── registry.py                   # PluginRegistry for tracking plugins
+│   │   ├── registry.py                   # PluginRegistry for tracking agent plugins
+│   │   ├── multiagent_registry.py        # Registry for tracking orchestrator plugins
+│   │   └── _discovery.py                 # Shared hook/tool discovery utilities
 │   │
 │   ├── handlers/                         # Event handlers
 │   │   └── callback_handler.py           # Callback handling

src/strands/__init__.py

@@ -4,7 +4,7 @@
 from .agent.agent import Agent
 from .agent.base import AgentBase
 from .event_loop._retry import ModelRetryStrategy
-from .plugins import Plugin
+from .plugins import MultiAgentPlugin, Plugin
 from .tools.decorator import tool
 from .types._snapshot import Snapshot
 from .types.tools import ToolContext
@@ -17,6 +17,7 @@
     "agent",
     "models",
     "ModelRetryStrategy",
+    "MultiAgentPlugin",
     "Plugin",
     "Skill",
     "Snapshot",

src/strands/multiagent/graph.py

@@ -37,6 +37,8 @@
 )
 from ..hooks.registry import HookProvider, HookRegistry
 from ..interrupt import Interrupt, _InterruptState
+from ..plugins.multiagent_plugin import MultiAgentPlugin
+from ..plugins.multiagent_registry import _MultiAgentPluginRegistry
 from ..session import SessionManager
 from ..telemetry import get_tracer
 from ..types._events import (
@@ -253,6 +255,7 @@
         self._id: str = _DEFAULT_GRAPH_ID
         self._session_manager: SessionManager | None = None
         self._hooks: list[HookProvider] | None = None
+        self._plugins: list[MultiAgentPlugin] | None = None
 
     def add_node(self, executor: AgentBase | MultiAgentBase, node_id: str | None = None) -> GraphNode:
         """Add an AgentBase or MultiAgentBase instance as a node to the graph."""
@@ -370,6 +373,15 @@
         self._hooks = hooks
         return self
 
+    def set_plugins(self, plugins: list[MultiAgentPlugin]) -> "GraphBuilder":
+        """Set plugins for the graph.
+
+        Args:
+            plugins: List of multi-agent plugins for extending graph behavior
+        """
+        self._plugins = plugins
+        return self
+
     def build(self) -> "Graph":
         """Build and validate the graph with configured settings."""
         if not self.nodes:
@@ -397,6 +409,7 @@
             reset_on_revisit=self._reset_on_revisit,
             session_manager=self._session_manager,
             hooks=self._hooks,
+            plugins=self._plugins,
             id=self._id,
         )
 
@@ -416,7 +429,7 @@
 class Graph(MultiAgentBase):
     """Directed Graph multi-agent orchestration with configurable revisit behavior."""
 
     def __init__(
         self,
         nodes: dict[str, GraphNode],
         edges: set[GraphEdge],
@@ -427,6 +440,7 @@
         reset_on_revisit: bool = False,
         session_manager: SessionManager | None = None,
         hooks: list[HookProvider] | None = None,
+        plugins: list[MultiAgentPlugin] | None = None,
         id: str = _DEFAULT_GRAPH_ID,
         trace_attributes: Mapping[str, AttributeValue] | None = None,
     ) -> None:
@@ -442,6 +456,7 @@
             reset_on_revisit: Whether to reset node state when revisited (default: False)
             session_manager: Session manager for persisting graph state and execution history (default: None)
             hooks: List of hook providers for monitoring and extending graph execution behavior (default: None)
+            plugins: List of multi-agent plugins for extending graph behavior (default: None)
             id: Unique graph id (default: None)
             trace_attributes: Custom trace attributes to apply to the agent's trace span (default: None)
         """
@@ -469,6 +484,11 @@
             for hook in hooks:
                 self.hooks.add_hook(hook)
 
+        self._plugin_registry = _MultiAgentPluginRegistry(self)
+        if plugins:
+            for plugin in plugins:
+                self._plugin_registry.add_and_init(plugin)
+
         self._resume_next_nodes: list[GraphNode] = []
         self._resume_from_session = False
         self.id = id

src/strands/multiagent/swarm.py

@@ -36,6 +36,8 @@
 )
 from ..hooks.registry import HookProvider, HookRegistry
 from ..interrupt import Interrupt, _InterruptState
+from ..plugins.multiagent_plugin import MultiAgentPlugin
+from ..plugins.multiagent_registry import _MultiAgentPluginRegistry
 from ..session import SessionManager
 from ..telemetry import get_tracer
 from ..tools.decorator import tool
@@ -247,6 +249,7 @@ def __init__(
         repetitive_handoff_min_unique_agents: int = 0,
         session_manager: SessionManager | None = None,
         hooks: list[HookProvider] | None = None,
+        plugins: list[MultiAgentPlugin] | None = None,
         id: str = _DEFAULT_SWARM_ID,
         trace_attributes: Mapping[str, AttributeValue] | None = None,
     ) -> None:
@@ -266,6 +269,7 @@ def __init__(
                 Disabled by default (default: 0)
             session_manager: Session manager for persisting graph state and execution history (default: None)
             hooks: List of hook providers for monitoring and extending graph execution behavior (default: None)
+            plugins: List of multi-agent plugins for extending swarm behavior (default: None)
             trace_attributes: Custom trace attributes to apply to the agent's trace span (default: None)
         """
         super().__init__()
@@ -299,6 +303,11 @@ def __init__(
         if self.session_manager:
             self.hooks.add_hook(self.session_manager)
 
+        self._plugin_registry = _MultiAgentPluginRegistry(self)
+        if plugins:
+            for plugin in plugins:
+                self._plugin_registry.add_and_init(plugin)
+
         self._resume_from_session = False
 
         self._setup_swarm(nodes)

src/strands/plugins/__init__.py

@@ -1,13 +1,16 @@
-"""Plugin system for extending agent functionality.
+"""Plugin system for extending agent and orchestrator functionality.
 
 This module provides a composable mechanism for building objects that can
-extend agent behavior through automatic hook and tool registration.
+extend agent and multi-agent orchestrator behavior through automatic hook
+and tool registration.
 """
 
 from .decorator import hook
+from .multiagent_plugin import MultiAgentPlugin
 from .plugin import Plugin
 
 __all__ = [
+    "MultiAgentPlugin",
     "Plugin",
     "hook",
 ]

src/strands/plugins/_discovery.py

@@ -0,0 +1,119 @@
+"""Shared utility for discovering decorated methods on plugin instances.
+
+This module provides helper functions used by both Plugin and MultiAgentPlugin
+to scan for @hook (and optionally @tool) decorated methods, and shared registry
+utilities for plugin initialization and hook registration.
+"""
+
+import inspect
+import logging
+from collections.abc import Awaitable, Callable
+from typing import Any, cast
+
+from .._async import run_async
+from ..hooks.registry import HookCallback, HookRegistry
+from ..tools.decorator import DecoratedFunctionTool
+
+logger = logging.getLogger(__name__)
+
+
+def discover_hooks(instance: object, plugin_name: str) -> list[HookCallback]:
+    """Scan an instance's class hierarchy for @hook decorated methods.
+
+    Walks the MRO in reverse so parent class hooks come first, but child
+    overrides win (only the child's version is included).
+
+    Args:
+        instance: The plugin instance to scan.
+        plugin_name: The plugin name (used for debug logging).
+
+    Returns:
+        List of bound hook callback methods in declaration order.
+    """
+    hooks: list[HookCallback] = []
+    seen: set[str] = set()
+
+    for cls in reversed(type(instance).__mro__):
+        for attr_name in cls.__dict__:
+            if attr_name in seen:
+                continue
+            seen.add(attr_name)
+
+            try:
+                bound = getattr(instance, attr_name)
+            except Exception:
+                continue
+
+            if hasattr(bound, "_hook_event_types") and callable(bound):
+                hooks.append(bound)
+                logger.debug("plugin=<%s>, hook=<%s> | discovered hook method", plugin_name, attr_name)
+
+    return hooks
+
+
+def discover_tools(instance: object, plugin_name: str) -> list[DecoratedFunctionTool]:
+    """Scan an instance's class hierarchy for @tool decorated methods.
+
+    Walks the MRO in reverse so parent class tools come first, but child
+    overrides win (only the child's version is included).
+
+    Args:
+        instance: The plugin instance to scan.
+        plugin_name: The plugin name (used for debug logging).
+
+    Returns:
+        List of DecoratedFunctionTool instances in declaration order.
+    """
+    tools: list[DecoratedFunctionTool] = []
+    seen: set[str] = set()
+
+    for cls in reversed(type(instance).__mro__):
+        for attr_name in cls.__dict__:
+            if attr_name in seen:
+                continue
+            seen.add(attr_name)
+
+            try:
+                bound = getattr(instance, attr_name)
+            except Exception:
+                continue
+
+            if isinstance(bound, DecoratedFunctionTool):
+                tools.append(bound)
+                logger.debug("plugin=<%s>, tool=<%s> | discovered tool method", plugin_name, attr_name)
+
+    return tools
+
+
+def call_init_method(init_method: Callable[..., Any], target: Any) -> None:
+    """Call a plugin's init method, handling both sync and async implementations.
+
+    Args:
+        init_method: The init_agent or init_multi_agent method to call.
+        target: The agent or orchestrator instance to pass to the init method.
+    """
+    if inspect.iscoroutinefunction(init_method):
+        async_init = cast(Callable[..., Awaitable[None]], init_method)
+        run_async(lambda: async_init(target))
+    else:
+        init_method(target)
+
+
+def register_hooks(plugin_name: str, hooks: list[HookCallback], registry: HookRegistry) -> None:
+    """Register discovered hook callbacks with a hook registry.
+
+    Args:
+        plugin_name: The plugin name (used for debug logging).
+        hooks: List of hook callbacks to register.
+        registry: The HookRegistry to register callbacks with.
+    """
+    for hook_callback in hooks:
+        event_types = getattr(hook_callback, "_hook_event_types", [])
+        for event_type in event_types:
+            registry.add_callback(event_type, hook_callback)
+            logger.debug(
+                "plugin=<%s>, hook=<%s>, event_type=<%s> | registered hook",
+                plugin_name,
+                getattr(hook_callback, "__name__", repr(hook_callback)),
+                event_type.__name__,
+            )

src/strands/plugins/multiagent_plugin.py

@@ -0,0 +1,119 @@
+"""MultiAgentPlugin base class for extending multi-agent orchestrator functionality.
+
+This module defines the MultiAgentPlugin base class, which provides a composable way to
+add behavior changes to multi-agent orchestrators (Swarm, Graph) through automatic hook
+registration and custom initialization.
+
+MultiAgentPlugin is the orchestrator-level counterpart to Plugin (which targets individual agents).
+A class can implement both Plugin and MultiAgentPlugin to provide functionality at both levels.
+"""
+
+from abc import ABC, abstractmethod
+from collections.abc import Awaitable
+from typing import TYPE_CHECKING
+
+from ..hooks.registry import HookCallback
+from ._discovery import discover_hooks
+
+if TYPE_CHECKING:
+    from ..multiagent.base import MultiAgentBase
+
+
+class MultiAgentPlugin(ABC):
+    """Base class for objects that extend multi-agent orchestrator functionality.
+
+    MultiAgentPlugins provide a composable way to add behavior changes to orchestrators
+    (Swarm, Graph). They support automatic discovery and registration of methods decorated
+    with @hook.
+
+    Unlike agent-level Plugin, MultiAgentPlugin does not support @tool decorated methods
+    since orchestrators do not have tool registries.
+
+    Attributes:
+        name: A stable string identifier for the plugin (must be provided by subclass)
+        hooks: Hooks attached to the orchestrator, auto-discovered from @hook decorated methods
+
+    Example using decorators (recommended):
+        ```python
+        from strands.plugins import MultiAgentPlugin, hook
+        from strands.hooks import BeforeNodeCallEvent, AfterNodeCallEvent
+
+        class MonitoringPlugin(MultiAgentPlugin):
+            name = "monitoring"
+
+            @hook
+            def on_before_node(self, event: BeforeNodeCallEvent):
+                print(f"Node {event.node_id} starting")
+
+            @hook
+            def on_after_node(self, event: AfterNodeCallEvent):
+                print(f"Node {event.node_id} completed")
+        ```
+
+    Example with custom initialization:
+        ```python
+        class MyPlugin(MultiAgentPlugin):
+            name = "my-plugin"
+
+            def init_multi_agent(self, orchestrator: MultiAgentBase) -> None:
+                # Custom initialization logic
+                pass
+        ```
+
+    Dual-use example (both agent and orchestrator):
+        ```python
+        from strands.plugins import Plugin, MultiAgentPlugin, hook
+        from strands.hooks import BeforeInvocationEvent, BeforeNodeCallEvent
+
+        class ObservabilityPlugin(Plugin, MultiAgentPlugin):
+            name = "observability"
+
+            @hook
+            def on_agent_invocation(self, event: BeforeInvocationEvent):
+                print("Agent invocation started")
+
+            @hook
+            def on_node_call(self, event: BeforeNodeCallEvent):
+                print(f"Node {event.node_id} starting")
+
+            def init_agent(self, agent):
+                pass  # Agent-level setup
+
+            def init_multi_agent(self, orchestrator):
+                pass  # Orchestrator-level setup
+        ```
+    """
+
+    @property
+    @abstractmethod
+    def name(self) -> str:
+        """A stable string identifier for the plugin."""
+        ...
+
+    def __init__(self) -> None:
+        """Initialize the plugin and discover decorated hook methods.
+
+        Scans the class for methods decorated with @hook and stores references
+        for later registration when the plugin is attached to an orchestrator.
+
+        Uses a guard to prevent double-discovery when used with multiple inheritance
+        (e.g., a class that inherits from both Plugin and MultiAgentPlugin).
+        """
+        if not hasattr(self, "_hooks"):
+            self._hooks: list[HookCallback] = discover_hooks(self, self.name)
+
+    @property
+    def hooks(self) -> list[HookCallback]:
+        """List of hooks the plugin provides, auto-discovered from @hook decorated methods."""
+        return self._hooks
+
+    def init_multi_agent(self, orchestrator: "MultiAgentBase") -> None | Awaitable[None]:
+        """Initialize the plugin with the orchestrator instance.
+
+        Override this method to add custom initialization logic. Decorated
+        hooks are automatically registered by the plugin registry.
+
+        Args:
+            orchestrator: The multi-agent orchestrator instance to initialize with.
+        """
+        return None