Plugins

You can stop here

You do not need to understand plugins to use the SDK. Built-in plugins already support common platforms. Read this page when you need to add support for a new platform or connection library.

Plugins are the extension point of the connection system. They bridge the gap between a platform-agnostic proxy and the vendor-specific commands and libraries needed to talk to real devices.

ConnectionPlugin base class

Every plugin inherits from ConnectionPlugin, which is generic over two type parameters:

class ConnectionPlugin(Generic[AnyConnection, RawConnection], ABC):
    """
    Abstract base class for connection plugins.

    Plugins are the primary extension point for the connection system.
    They handle:

    1. **Connection creation** - Create BaseConnection instances for a specific
       platform and library combination.

    2. **Capability implementation** - Implement capability interfaces to provide
       platform-specific functionality (e.g., get_version(), get_inventory()).

    The plugin lifecycle:
    1. Plugin is instantiated with a Device
    2. initialize_connection() creates and connects the BaseConnection
    3. Plugin methods can use get_raw_connection() to access the library
    4. teardown_connection() cleans up the connection

    Example:
        Create a plugin for Cisco IOS using Scrapli::

            @register_connection_plugin(default_for_platform=True)
            class IOSScrapliPlugin(ConnectionPlugin[ScrapliConnection, Scrapli], DeviceInfoCapability):
                platform = "ios"
                connection_type = "ssh"
                connection_library = "scrapli"

                def _create_connection(self) -> ScrapliConnection:
                    return ScrapliConnection(self.device)

                def get_version(self) -> dict[str, str | None]:
                    conn = self.get_raw_connection()
                    output = conn.send_command("show version")
                    # Parse and return version info
                    return {"version": "..."}

    Attributes:
        platform: The platform this plugin supports (e.g., "ios").
        connection_type: Type of connection (e.g., "ssh", "api").
        connection_library: Library used (e.g., "netmiko", "scrapli").
        device: The Device instance this plugin is operating on.

    Note:
        The plugin uses the device passed to __init__ for all operations.
        Capability methods should use self.device and self.get_raw_connection()
        to interact with the device.
    """

    platform: ConnectionPlatform
    connection_type: ConnectionType
    connection_library: ConnectionLibrary

Type parameter	Meaning
`AnyConnection`	The `BaseConnection` subclass this plugin creates (e.g. `ScrapliConnection`).
`RawConnection`	The raw library object exposed by `get_raw_connection()` (e.g. `scrapli.Scrapli`).

Three class attributes are required on every concrete plugin:

Attribute	Example	Purpose
`platform`	`"ios"`	Device platform short name.
`connection_type`	`"ssh"`	Protocol category.
`connection_library`	`"scrapli"`	Underlying library.

Base plugins

The SDK ships base plugins that handle connection creation for each supported library. Platform plugins inherit from these rather than implementing _create_connection() from scratch.

Base plugin	Library	Connection type	Raw connection type
`ScrapliConnectionPluginBase`	scrapli	`ssh`	`scrapli.Scrapli`
`NetmikoConnectionPluginBase`	netmiko	`ssh`	`netmiko.BaseConnection`
`NapalmConnectionPluginBase`	napalm	`ssh`	`napalm.base.NetworkDriver`
`NETCONFConnectionPluginBase`	ncclient	`netconf`	`ncclient.manager.Manager`
`RESTCONFConnectionPluginBase`	httpx	`restconf`	`httpx.Client`
`APIConnectionPluginBase`	httpx	`api`	`httpx.Client`

Each base plugin pairs a BaseConnection wrapper (handling connect/disconnect/ is_alive) with a ConnectionPlugin subclass that implements _create_connection():

class ScrapliConnectionPluginBase(ConnectionPlugin[ScrapliConnection, "ScrapliType"]):
    """Base class for Scrapli connection plugins."""

    def _create_connection(self) -> ScrapliConnection:
        return ScrapliConnection(self.device)

    connection_type = "ssh"
    connection_library = "scrapli"

Lazy imports

Third-party libraries are imported inside BaseConnection.connect(), not at module level. If a library is not installed, the error surfaces as ConnectionCreationError only when the plugin is actually used.

Platform plugins

A platform plugin extends a base plugin and implements one or more capability interfaces. Here is the IOS Scrapli plugin:

"""
iOS-specific Scrapli connection plugin.
"""

from __future__ import annotations

from neops_worker_sdk.connection.capabilities import DeviceInfoCapability
from neops_worker_sdk.connection.exceptions import ConnectionCreationError
from neops_worker_sdk.connection.plugins.base.scrapli_plugin import (
    ScrapliConnectionPluginBase,
)
from neops_worker_sdk.connection.registry import register_connection_plugin
from neops_worker_sdk.logger.logger import Logger

logger = Logger()


@register_connection_plugin()
class IOSCiscoScrapliPlugin(ScrapliConnectionPluginBase, DeviceInfoCapability):
    """Plugin for Scrapli SSH connections to iOS devices."""

    platform = "ios"
    # FIX: import path updated to connection base plugin

    def get_version(self) -> dict[str, str | None]:
        raw_connection = self.get_raw_connection()
        if not raw_connection:
            # FIX: use ConnectionCreationError for missing connection (not ValueError)
            raise ConnectionCreationError(
                "No connection available; initialize the connection before calling get_version().",
                device_id=str(self.device.id) if self.device.id else None,
            )

        logger.debug("Fetching IOS version via Scrapli for device %s", self.device.id)
        output = raw_connection.send_command("show version")
        structured_result = output.textfsm_parse_output()

        if structured_result and isinstance(structured_result, list):
            result = structured_result[0]
            return {
                "vendor": "Cisco",
                "model": result.get("hardware", [None])[0] if result.get("hardware") else None,
                "serial": result.get("serial", [None])[0] if result.get("serial") else None,
                "software_release": result.get("version", None),
                "raw_output": str(output.raw_result),
            }

        # Fallback if textfsm parsing fails
        logger.warning(
            "TextFSM parsing failed for IOS show version on device %s; returning raw output with None fields.",
            self.device.id,
        )
        return {
            "vendor": "Cisco",
            "model": None,
            "serial": None,
            "software_release": None,
            "raw_output": str(output.raw_result),
        }

Key points:

Inherits ScrapliConnectionPluginBase (connection creation is handled).
Inherits DeviceInfoCapability (declares get_version() as a capability).
Sets platform = "ios" to bind to Cisco IOS devices.
Uses self.get_raw_connection() to access the Scrapli session.

A platform plugin can implement as many capabilities as needed. The resolver matches proxies to plugins based on which capabilities the plugin provides.

Registration

Plugins register themselves at import time via the @register_connection_plugin() decorator:

def register_connection_plugin(
    default_for_platform: bool = False,
    default_for_connection_type: bool = False,
) -> Callable[[type[ConnectionPlugin]], type[ConnectionPlugin]]:
    """
    Decorator to register a connection plugin.

    Args:
        default_for_platform: If True, this plugin becomes the default for its platform.
            Used when only device is provided to ``get_connection()``.
        default_for_connection_type: If True, this plugin becomes the default for its
            platform + connection_type combination. Used when library is not specified.

    Returns:
        Decorator function that registers the plugin class.

    Raises:
        PluginRegistrationError: If the plugin is invalid or a duplicate default is registered.

    Example:
        Standard registration::

            @register_connection_plugin()
            class MyPlugin(ConnectionPlugin):
                platform = "ios"
                connection_type = "ssh"
                connection_library = "netmiko"

        Register as platform default::

            @register_connection_plugin(default_for_platform=True)
            class IOSDefaultPlugin(ConnectionPlugin):
                platform = "ios"
                connection_type = "ssh"
                connection_library = "scrapli"

        Register as connection type default::

            @register_connection_plugin(default_for_connection_type=True)
            class IOSSSHDefaultPlugin(ConnectionPlugin):
                platform = "ios"
                connection_type = "ssh"
                connection_library = "netmiko"
    """

    def decorator(cls: type[ConnectionPlugin]) -> type[ConnectionPlugin]:
        # Validate plugin class
        _validate_plugin_class(cls)

        platform = cls.platform
        connection_type = cls.connection_type
        connection_library = cls.connection_library

        # Register in main registry
        _plugin_registry.setdefault(platform, {}).setdefault(connection_type, {})[connection_library] = cls

        logger.debug(
            f"Registered plugin {cls.__name__} for "
            f"platform={platform}, type={connection_type}, library={connection_library}"
        )

        # Handle platform default registration
        if default_for_platform:
            _register_platform_default(cls, platform)

        # Handle connection type default registration
        if default_for_connection_type:
            _register_connection_type_default(cls, platform, connection_type)

        return cls

Decorator flag	Effect
(no flags)	Plugin registered for its `platform`/`connection_type`/`connection_library` triple.
`default_for_platform=True`	Becomes the fallback plugin when only a device is provided (with `fallback_to_default=True`).
`default_for_connection_type=True`	Becomes the fallback when platform + connection type are provided but library is omitted.

Example from the IOS Netmiko plugin:

"""
iOS-specific Netmiko connection plugin.
"""

from __future__ import annotations

from neops_worker_sdk.connection.plugins.base.netmiko_plugin import (
    NetmikoConnectionPluginBase,
)
from neops_worker_sdk.connection.registry import register_connection_plugin


@register_connection_plugin(default_for_platform=True, default_for_connection_type=True)
class IOSNetmikoPlugin(NetmikoConnectionPluginBase):
    """Plugin for Netmiko SSH connections to iOS devices."""

    platform = "ios"

Import-time registration

Plugins self-register when their module is imported. Import the built-in plugin package to ensure the SDK plugins are available:

import neops_worker_sdk.connection.plugins  # noqa: F401

Plugin lifecycle

stateDiagram-v2
    [*] --> Instantiated: Plugin(device)
    Instantiated --> Connected: initialize_connection(connect=True)
    Instantiated --> Wrapper_Only: initialize_connection(connect=False)
    Wrapper_Only --> Connected: _active_connection.connect()
    Connected --> Capability_Calls: get_version(), etc.
    Capability_Calls --> Connected
    Connected --> Torn_Down: teardown_connection()
    Torn_Down --> [*]

Instantiation -- Plugin(device) stores the device reference. No connection is created yet.
initialize_connection(connect=True) -- calls _create_connection() to build the BaseConnection wrapper, then calls wrapper.connect() to open the session.
Capability methods -- the plugin uses self.get_raw_connection() to obtain the library's native connection object and execute commands.
teardown_connection() -- calls wrapper.disconnect() and clears the internal reference.

    def initialize_connection(self, connect: bool = True) -> None:
        """
        Initialize and optionally establish the connection.

        This method:
        1. Creates a new BaseConnection via _create_connection()
        2. Optionally calls connect() on it

        Args:
            connect: If True (default), establish the connection immediately.
                    If False, just create the connection wrapper.

        Raises:
            ConnectionCreationError: If connection creation or establishment fails.
        """
        logger.debug(f"Initializing connection for device {self.device.id}")
        self._active_connection = self._create_connection()

        if connect:
            logger.debug(f"Connecting to device {self.device.id}")
            self._active_connection.connect()
            logger.info(f"Connected to device {self.device.id}")

    def teardown_connection(self) -> None:
        """
        Teardown the active connection.

        This method:
        1. Disconnects the active connection if it exists
        2. Sets _active_connection to None

        This method is safe to call even if no connection exists.
        """
        if self._active_connection:
            logger.debug(f"Tearing down connection for device {self.device.id}")
            self._active_connection.disconnect()
            self._active_connection = None
            logger.info(f"Connection closed for device {self.device.id}")

    def is_connection_alive(self) -> bool:
        """
        Check if the plugin's connection is alive.

        Returns:
            True if a connection exists and is alive, False otherwise.
        """

`get_raw_connection()` -- escape hatch

When the capability system does not cover your use case, you can drop down to the library's native connection object:

    def get_raw_connection(self) -> RawConnection | None:
        """
        Get the raw connection object from the underlying library.

        This provides direct access to the library's connection object
        for capability methods that need to execute commands.

        Returns:
            The raw connection object, or None if not connected or dead.

        Example:
            def get_version(self) -> dict:
                conn = self.get_raw_connection()
                if not conn:
                    raise ConnectionCreationError("Not connected")
                output = conn.send_command("show version")
                return parse_version(output)
        """
        if not self._active_connection or not self._active_connection.is_alive:
            return None
        return self._active_connection._connection

Inside a capability method this looks like:

def get_version(self) -> dict[str, str | None]:
    raw = self.get_raw_connection()
    if not raw:
        raise ConnectionCreationError(
            "No connection available",
            device_id=str(self.device.id),
        )
    output = raw.send_command("show version")
    ...

Warning

get_raw_connection() returns None when the connection is not alive. Always check the return value before using it.