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: Create a BaseConnection Wrapper

The BaseConnection wraps the raw library client. Key rules:

  • Validate device fields early (ip, username, password).
  • 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.
  • Raise ConnectionValidationError for missing fields and ConnectionCreationError for connection failures.
  • Make disconnect() defensive -- log warnings, never raise.
class FRRNetmikoConnection(BaseConnection[NetmikoType]):
    """Netmiko wrapper for FRRouting devices."""

    def connect(self, **kwargs: Any) -> None:
        try:
            from netmiko import ConnectHandler
        except ImportError as exc:
            raise ConnectionCreationError(
                "netmiko is not installed. Add it to your dependencies.",
                device_id=str(self.device.id) if self.device.id else None,
                cause=exc,
            ) from exc

        if not self.device.ip:
            raise ConnectionValidationError(
                "Device has no IP address.",
                device_id=str(self.device.id) if self.device.id else None,
                missing_fields=["ip"],
            )

        try:
            self._connection = ConnectHandler(
                device_type="linux",
                host=self.device.ip,
                username=self.device.username,
                password=self.device.password,
                **kwargs,
            )
        except Exception as exc:
            raise ConnectionCreationError(
                "Failed to connect to FRR device.",
                device_id=str(self.device.id) if self.device.id else None,
                cause=exc,
            ) from exc

    def disconnect(self) -> None:
        if self._connection:
            self._connection.disconnect()
            self._connection = None

    @property
    def is_alive(self) -> bool:
        return bool(self._connection and self._connection.is_alive())

Step 2: Create the Plugin Class

Subclass ConnectionPlugin (or a library-specific base like ScrapliConnectionPluginBase), inherit capability interfaces, and set the three required class attributes.

Implement _create_connection() to return your BaseConnection wrapper, then implement every method declared by the inherited capability interfaces.

@register_connection_plugin(default_for_platform=True, default_for_connection_type=True)
class FRRNetmikoPlugin(
    ConnectionPlugin[FRRNetmikoConnection, NetmikoType],
    DeviceInfoCapability,
):
    """Plugin for FRRouting devices over SSH via Netmiko."""

    platform = "frr"
    connection_type = "ssh"
    connection_library = "netmiko"

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

    def get_version(self) -> dict[str, str | None]:
        conn = self.get_raw_connection()
        if not conn:
            raise ConnectionCreationError(
                "No active connection.",
                device_id=str(self.device.id) if self.device.id else None,
            )

        output = str(conn.send_command("show version"))
        logger.info("Retrieved FRR version for device %s", self.device.id)

        return {
            "vendor": "FRRouting",
            "model": None,
            "serial": None,
            "software_release": self._parse_version(output),
            "raw_output": output,
        }

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

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.plugins.base.scrapli_plugin import (
    ScrapliConnectionPluginBase,
)

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

    def get_version(self) -> dict[str, str | None]:
        ...

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 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

Use the SDK Logger() with 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.

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().

Non-defensive disconnect()

Cleanup must not raise. Log warnings on errors and set self._connection = None.

is_alive that throws

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

Next Steps