Skip to content

Writing Plugins

A connection plugin ties a platform, connection type, and connection library together and implements capability methods for that combination. This guide walks through creating a plugin from scratch.

Plugin Anatomy

Every plugin has three responsibilities:

  1. Declare identity – platform, connection_type, connection_library class attributes.
  2. Create the connection – _create_connection() returns a BaseConnection wrapper.
  3. Implement capabilities – concrete methods for each capability interface the plugin claims.

Step 1: Configure the Connection

When the SDK ships a base for your library (scrapli, netmiko, napalm), reuse its connection wrapper. Library arguments a platform needs go into connect_defaults on a subclass; explicit connect(**kwargs) still win over them. For a platform the SDK does not map to a library device type, platform names it there:

class FRRNetmikoConnection(NetmikoConnection):
    """Netmiko connection tuned for FRRouting's Linux shell."""

    # "frr-example" is not in the SDK's netmiko platform mapping, so name the device_type here
    connect_defaults: ClassVar[dict[str, Any]] = {"platform": "linux", "fast_cli": False}

For a library without a base, subclass BaseConnection directly:

  • Call self._require_credentials("MyLib") first: it returns the typed ip, username and password or raises ConnectionValidationError naming every missing field. A connection that brings its own login (e.g. key auth) calls self._require_ip("MyLib") instead.
  • Lazy-import the third-party dependency inside connect() so missing packages only fail when the plugin is actually used.
  • Store the raw client in self._connection only after a successful connect, and raise ConnectionCreationError for connection failures.
  • Implement _close_raw(connection) to close the client. BaseConnection.disconnect() calls it, swallows and logs its errors, and forgets the connection, so disconnect() is idempotent.

Upgrading plugins written against earlier SDK versions

  • BaseConnection subclasses must implement _close_raw(connection); disconnect() is now concrete. A subclass that only overrides disconnect() fails to instantiate with an abstract-method TypeError.
  • A device whose neops platform is missing from the library’s platform mapping raises PlatformNotSupportedError; it no longer falls back to netmiko/scrapli generic or napalm mock. Add the mapping, or pass platform= explicitly.
  • ConnectionStillAliveError is gone: disconnect() always forgets the connection, so clearing it can no longer be refused.
  • The scrapli raw connection is a scrapli 2 scrapli.Cli, not scrapli.Scrapli: send_command() is send_input(input_=...), textfsm_parse_output() is textfsm_parse() (raises ParsingException when no template exists), and raw_result is result_raw (bytes).

Step 2: Create the Plugin Class

Subclass a library-specific base like NetmikoConnectionPluginBase (or ConnectionPlugin directly), inherit capability interfaces, and set platform. A library base already sets connection_type and connection_library; point connection_class at your connection subclass from Step 1.

Without a base, set all three class attributes and implement _create_connection() to return your BaseConnection wrapper. Then implement every method declared by the inherited capability interfaces: registering a plugin that leaves any of them abstract raises PluginRegistrationError naming the missing methods, so the worker fails at startup instead of on every job that resolves the plugin.

# Its own platform: the SDK already ships a plugin for "frr", and loading this
# example must neither replace it nor compete with it during resolution.
@register_connection_plugin(default_for_platform=True, default_for_connection_type=True)
class FRRNetmikoPlugin(NetmikoConnectionPluginBase, NetworkDeviceInformationCapability):
    """Plugin for FRRouting devices over SSH via Netmiko."""

    platform = "frr-example"
    connection_class = FRRNetmikoConnection

    def get_information(self) -> InformationResult:
        output = self._run("/usr/bin/vtysh -c 'show version'")
        self.logger.info("Retrieved FRR version")

        return InformationResult(
            hostname=self.device.hostname or "",
            vendor="FRRouting",
            model="",
            serial="",
            software_release=self._parse_version(output),
        )

    def get_config(self) -> ConfigSnapshotResult:
        return ConfigSnapshotResult(
            running=self._run("/usr/bin/vtysh -c 'show running-config'"),
            startup=self._run("cat /etc/frr/frr.conf"),
        )

    def get_interfaces(self) -> InterfacesResult:
        # `ip -br link show` lines read `<name>[@<parent>] <operstate> ...`
        interfaces = InterfacesResult()
        for fields in self._run_lines("ip -br link show", min_fields=2):
            status = InterfaceStatus.DOWN if fields[1] == "DOWN" else InterfaceStatus.UP
            interfaces.append(InterfaceResult(name=fields[0].split("@")[0], status=status, description=""))
        return interfaces

    def get_neighbors(self) -> NeighborsResult:
        # `ip neighbor show` lines read `<ip> dev <interface> [lladdr <mac>] <STATE>`
        neighbors = NeighborsResult()
        for fields in self._run_lines("ip neighbor show", min_fields=4):
            if fields[-1] in ("FAILED", "INCOMPLETE"):
                continue
            neighbors.append(
                NeighborResult(
                    hostname="", ip=fields[0], capabilities=[], local_interface=fields[2], neighbor_interface=""
                )
            )
        return neighbors

    def _run(self, command: str) -> str:
        return str(self.require_raw_connection().send_command(command))

    def _run_lines(self, command: str, min_fields: int) -> list[list[str]]:
        rows = (line.split() for line in self._run(command).splitlines())
        return [fields for fields in rows if len(fields) >= min_fields]

    @staticmethod
    def _parse_version(output: str) -> str:
        for line in output.splitlines():
            if "FRRouting" in line:
                return line.strip()
        return ""

Step 3: Register the Plugin

The @register_connection_plugin() decorator registers the plugin in the global registry at import time.

Parameter Effect
default_for_platform=True This plugin is selected when only the device is provided
default_for_connection_type=True This plugin is selected when the library is omitted

Both are optional. Only set defaults when the plugin is the obvious choice for most users of that platform. See Resolution and Defaults for the full algorithm.

Using a Library-Specific Base

The SDK provides base plugins that handle _create_connection() for common libraries. When one exists for your library, inherit from it instead of ConnectionPlugin directly:

from neops_worker_sdk.connection.capabilities.network_device_information import (
    InformationResult,
    NetworkDeviceInformationCapability,
)
from neops_worker_sdk.connection.plugins.base.scrapli_plugin import (
    ScrapliConnectionPluginBase,
)

@register_connection_plugin()
class MyPlatformScrapliPlugin(ScrapliConnectionPluginBase, NetworkDeviceInformationCapability):
    platform = "my_platform"

    def get_information(self) -> InformationResult:
        ...

Available bases: ScrapliConnectionPluginBase, NetmikoConnectionPluginBase, and others under neops_worker_sdk.connection.plugins.base.

Using a base saves you from implementing _create_connection() and the BaseConnection wrapper entirely. You only need to set platform and implement capability methods.

Default Registration

Only set defaults when the plugin is the clear choice for a given platform or platform+type combination. Defaults are opt-in at the call site (fallback_to_default=True), so they do not affect callers who specify the library explicitly.

If your package registers a plugin as a default and the SDK already has one for that scope, your plugin silently takes precedence. See Resolution and Defaults for override rules.

Checklist

  • platform, connection_type, and connection_library are set
  • _create_connection() returns a valid BaseConnection
  • Every inherited capability method has a concrete implementation
  • Exceptions use ConnectionCreationError / ConnectionValidationError (not ValueError)
  • Logging uses the plugin’s self.logger with appropriate levels
  • Plugin is decorated with @register_connection_plugin()
  • Parsing failures return a safe structure with raw_output when helpful

Logging and Exception Guidelines

Log through the plugin’s self.logger. It already carries the plugin class, the device (entity_type, entity_id, device_ip, platform) and the running job. Follow these conventions:

Level When
logger.info() User-visible lifecycle events (connected, disconnected, executed a command)
logger.warning() Recoverable issues (parsing fallbacks, cleanup errors, health check failures)
logger.error() Unexpected failures you cannot recover from cleanly
logger.debug() Internal state transitions and verbose details

Always include device_id in exception context. Use cause when wrapping third-party exceptions to preserve the original traceback. Use missing_fields on ConnectionValidationError so callers can identify what data is missing.

Raise ConnectionValidationError, NotImplementedForThisPlatform or any other PreContactError only before the call has sent anything to the device: function blocks report such a failure as having changed nothing, which takes the step out of the rollback.

Common Pitfalls

Top-level third-party imports

Importing a library at module level makes every worker fail if that dependency is missing. Use lazy imports inside connect().

Overriding disconnect()

Implement _close_raw() instead. BaseConnection.disconnect() already makes cleanup idempotent and never raises.

is_alive that throws

Treat is_alive like a health check: catch all exceptions, return False, and log a warning.

Next Steps