Class PluginManager

src/google/adk/plugins/plugin_manager.py:60–371

Manages the registration and execution of plugins. The PluginManager is an internal class that orchestrates the invocation of plugin callbacks at key points in the SDK's execution lifecycle. It maintains a list of registered plugins and ensures they are called in the order they were registe

class PluginManager:
  """Manages the registration and execution of plugins.

  The PluginManager is an internal class that orchestrates the invocation of
  plugin callbacks at key points in the SDK's execution lifecycle. It maintains
  a list of registered plugins and ensures they are called in the order they
  were registered.

  The core execution logic implements an "early exit" strategy: if any plugin
  callback returns a non-`None` value, the execution of subsequent plugins for
  that specific event is halted, and the returned value is propagated up the
  call stack. This allows plugins to short-circuit operations like agent runs,
  tool calls, or model requests.
  """

  def __init__(
      self,
      plugins: Optional[List[BasePlugin]] = None,
      close_timeout: float = 5.0,
  ):
    """Initializes the plugin service.

    Args:
      plugins: An optional list of plugins to register upon initialization.
      close_timeout: The timeout in seconds for each plugin's close method.
    """
    self.plugins: List[BasePlugin] = []
    self._close_timeout = close_timeout
    self._skip_closing_plugins = False
    if plugins:
      for plugin in plugins:
        self.register_plugin(plugin)

  def set_skip_closing_plugins(self, value: bool) -> None:
    """Controls whether `close()` will tear down the registered plugins.

    Set to True when the plugins are owned by another component (e.g. a parent
    `Runner` whose plugin list this manager is sharing). When set, subsequent
    calls to `close()` become a no-op so the shared plugins are not torn down
    while still in use.

    Args:
      value: True to skip closing the plugins; False (default) to close them
        normally.
    """
    self._skip_closing_plugins = value

  def register_plugin(self, plugin: BasePlugin) -> None:
    """Registers a new plugin.

    Args:
      plugin: The plugin instance to register.

    Raises:
      ValueError: If a plugin with the same name is already registered.
    """
    if any(p.name == plugin.name for p in self.plugins):
      raise ValueError(f"Plugin with name '{plugin.name}' already registered.")