Skip to content

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_information(), send_command()).

    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:
        Plugins usually subclass a library base (which fixes connection_type,
        connection_library and _create_connection) and add capabilities::

            @register_connection_plugin(default_for_platform=True)
            class IOSScrapliPlugin(ScrapliConnectionPluginBase, NetworkDeviceInformationCapability):
                platform = "ios"

                def get_information(self) -> InformationResult:
                    output = self.require_raw_connection().send_input(input_="show version")
                    return parse_information(output)

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

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.Cli
ScrapliNetconfConnectionPluginBase scrapli netconf scrapli.Netconf
NetmikoConnectionPluginBase netmiko ssh netmiko.BaseConnection
NapalmConnectionPluginBase napalm ssh napalm.base.NetworkDriver

Each base plugin pairs a BaseConnection wrapper (handling connect and is_alive; BaseConnection.disconnect() closes it via _close_raw()) with a ConnectionPlugin subclass that implements _create_connection():

class ScrapliConnection(ScrapliSessionConnection["ScrapliType"]):
    """Wrapper for a scrapli ``Cli`` (SSH/telnet) session.

    ``connect()`` picks the scrapli definition from the device's neops platform (see
    ``get_scrapli_platform``); ``platform=`` overrides that lookup. Login, host-key and
    ``connect_defaults`` handling: see ``ScrapliSessionConnection``.
    """

    # TODO: native async via scrapli 2's *_async methods once the connection stack supports it.
    def connect(self, **kwargs: Any) -> None:
        self._import_scrapli_or_raise()
        from scrapli import Cli

        options = self.connect_defaults | kwargs
        definition = options.pop("platform", None) or get_scrapli_platform(self.device)
        self._open_session(Cli, {"definition_file_or_name": definition} | options)

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

    Subclasses that need platform-specific ``Cli`` arguments point
    ``connection_class`` at a ScrapliConnection subclass with ``connect_defaults``.
    """

    connection_type = "ssh"
    connection_library = "scrapli"
    connection_class: ClassVar[type[ScrapliConnection]] = ScrapliConnection

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

A connection reads ip, username and password from the device. The CLI connections (scrapli, netmiko, napalm) also map the device’s neops platform to the library’s platform name; an unmapped platform raises PlatformNotSupportedError before any connection attempt, and platform= bypasses the mapping. NETCONF is platform-neutral and needs no mapping. To tune library arguments per platform, subclass the connection with connect_defaults and point the plugin’s connection_class at it – see Writing Plugins. For scrapli that covers port, transport_options, and auth_options for key auth or an enable secret; explicit connect(**kwargs) win over connect_defaults.

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:

# TODO: The different ConnectionPlugin will get overcrowded.
@register_connection_plugin(default_for_platform=True, default_for_connection_type=True)
class CiscoIOSScrapliPlugin(
    ScrapliConnectionPluginBase,
    NetworkDeviceConfigurationCapability,
    NetworkDeviceExecutionCapability,
    NetworkDeviceInformationCapability,
    NetworkDeviceLifecycleCapability,
):
    """Plugin for Scrapli SSH connections to Cisco IOS devices."""

    platform = "ios"
    detection_signature = IOS_SIGNATURE

    def _candidate_config(self, timeout: int | None) -> IOSCandidateConfig:
        return IOSCandidateConfig(self.require_raw_connection(), self.send_interactive, _config_timeout(timeout))

    def _diff_running_around(self, change: Callable[[], None]) -> ConfigApplyResult:
        pre_config = self.get_config().running
        change()
        post_config = self.get_config().running
        return ConfigApplyResult(diff=get_config_unified_diff(self.platform, pre_config, post_config))

    # =============================================================================
    # Network Device Configuration Capability
    # =============================================================================

    def send_configs(
        self, configs: list[str], timeout: int | None = None, expect_prompt: bool = True
    ) -> ConfigApplyResult:
        raw_connection = self.require_raw_connection()

        timeout_ns = _timeout_ns(_config_timeout(timeout))

        def send() -> None:
            self.logger.debug(
                "Sending device config via Scrapli for device {}: {}", self.device.id, describe_config(configs)
            )
            if not configs:
                return
            if expect_prompt:
                raw_connection.send_inputs(
                    configs,
                    requested_mode="configuration",
                    stop_on_indicated_failure=False,
                    operation_timeout_ns=timeout_ns,
                )
            else:
                # A line such as `banner motd #` leaves no prompt to wait for until its text ends.
                raw_connection.enter_mode("configuration", operation_timeout_ns=timeout_ns)
                for line in configs:
                    raw_connection.write_and_return(input_=line)
            raw_connection.enter_mode("privileged_exec", operation_timeout_ns=timeout_ns)

        return self._diff_running_around(send)

    def merge_config(self, config: str, timeout: int | None = None) -> ConfigApplyResult:
        candidate = self._candidate_config(timeout)

        def merge() -> None:
            self.logger.debug("Adding config file to device via Scrapli for device {}", self.device.id)
            candidate.write(_ending_with_end(config))
            self.logger.debug("Merging config via Scrapli for device {}: {}", self.device.id, describe_config(config))
            candidate.merge_into_running()
            self.logger.debug("Cleaning up config file via Scrapli for device {}", self.device.id)
            candidate.delete()

        return self._diff_running_around(merge)

    def replace_running_config(self, config: str, timeout: int | None = None) -> ConfigApplyResult:
        candidate = self._candidate_config(timeout)

        def replace() -> None:
            self.logger.debug("Adding config file to device via Scrapli for device {}", self.device.id)
            # TODO: Replace running config does currently not work with banners in config
            # https://github.com/scrapli/scrapli_cfg/issues/55
            # https://github.com/zebbra/neops-worker-sdk-py/issues/151
            candidate.write(_ending_with_end(config))
            self.logger.debug(
                "Replacing running config via Scrapli for device {}: {}", self.device.id, describe_config(config)
            )
            candidate.replace_running()
            candidate.delete()

        return self._diff_running_around(replace)

    def replace_startup_config(self, config: str, timeout: int | None = None) -> ConfigApplyResult:
        running_config = self.get_config().running
        candidate = self._candidate_config(timeout)

        self.logger.debug("Adding config file to device via Scrapli for device {}", self.device.id)
        candidate.write(_ending_with_end(config))
        self.logger.debug(
            "Replacing startup config via Scrapli for device {}: {}", self.device.id, describe_config(config)
        )
        candidate.copy_to_startup()
        self.logger.debug("Cleaning up config file via Scrapli for device {}", self.device.id)
        candidate.delete()

        startup_config = self.get_config().startup
        return ConfigApplyResult(diff=get_config_unified_diff(self.platform, running_config, startup_config))

    def save_config(self) -> None:
        self.logger.debug("Saving device config via Scrapli for device {}", self.device.id)
        self.send_command("write memory", timeout=CONFIG_OPERATION_TIMEOUT)

    # =============================================================================
    # Network Device Execution Capability
    # =============================================================================

    def send_command(self, command: str, timeout: int | None = None) -> CommandResult:
        raw_connection = self.require_raw_connection()

        self.logger.debug("Sending command via Scrapli for device {}: {}", self.device.id, command)
        response = raw_connection.send_input(input_=command, operation_timeout_ns=_timeout_ns(timeout))
        return CommandResult(output=response.result, parsed_output=_parsed_rows(response))

    def send_command_expect(self, command: str, expect: str, timeout: int | None = None) -> CommandResult:
        self.logger.debug(
            "Sending command with expect via Scrapli for device {}: {}, expecting: {}", self.device.id, command, expect
        )
        return self.send_interactive([(command, expect)], timeout=timeout)

    def send_interactive(self, interactions: list[tuple[str, str]], timeout: int | None = None) -> CommandResult:
        raw_connection = self.require_raw_connection()

        self.logger.debug(
            "Sending interactive commands via Scrapli for device {}: {}",
            self.device.id,
            describe_interactions(interactions),
        )
        outputs: list[str] = []
        for input_, expect in interactions:
            step = self._interaction_step(raw_connection, input_, expect, _timeout_ns(timeout))
            outputs.append(step.output)
            if step.back_at_prompt:
                break
        return CommandResult(output="\n".join(outputs), parsed_output=None)

    @staticmethod
    def _interaction_step(raw_connection: Cli, input_: str, expect: str, timeout_ns: int | None) -> InteractionStep:
        """Send one answer; an empty ``expect`` waits for the device prompt and the sequence goes on, as in scrapli 1."""
        if expect:
            return interaction_step(raw_connection, input_, expect, _BACK_AT_PROMPT, timeout_ns)
        output = raw_connection.send_input(
            input_=input_, input_handling=InputHandling.IGNORE, operation_timeout_ns=timeout_ns
        ).result
        return InteractionStep(output, back_at_prompt=False)

    # =============================================================================
    # Network Device Information Capability
    # =============================================================================

    def get_information(self) -> InformationResult:
        self.logger.debug("Getting device information via Scrapli for device {}", self.device.id)
        result = self.send_command("show version")
        if not result.parsed_output:
            raise ValueError(
                f"Failed to parse device information from 'show version' command for device {self.device.id}"
            )
        device_info = result.parsed_output[0]

        # NOTE: For stacked devices, the hardware model and serial number are used from the first member.
        return InformationResult(
            hostname=device_info["hostname"],
            vendor="Cisco",
            model=device_info["hardware"][0] if device_info["hardware"] else "",
            serial=device_info["serial"][0] if device_info["serial"] else "",
            software_release=device_info["version"],
        )

    def get_config(self) -> ConfigSnapshotResult:
        raw_connection = self.require_raw_connection()

        self.logger.debug("Getting device config via Scrapli for device {}", self.device.id)
        timeout_ns = _timeout_ns(CONFIG_OPERATION_TIMEOUT)
        running_config = raw_connection.send_input(input_="show running-config", operation_timeout_ns=timeout_ns).result
        startup_config = raw_connection.send_input(input_="show startup-config", operation_timeout_ns=timeout_ns).result
        return ConfigSnapshotResult(running=clean_config(running_config), startup=clean_config(startup_config))

    def get_interfaces(self) -> InterfacesResult:
        interfaces_result = InterfacesResult()

        self.logger.debug("Getting interfaces via Scrapli for device {}", self.device.id)
        result = self.send_command("show interfaces")
        if not result.parsed_output:
            raise ValueError(f"Failed to parse interfaces from 'show interfaces' command for device {self.device.id}")
        interfaces = result.parsed_output

        for intf in interfaces:
            interfaces_result.append(
                InterfaceResult(
                    name=intf["interface"],
                    status=_interface_status(intf["link_status"], intf["protocol_status"]),
                    description=intf["description"],
                )
            )

        return interfaces_result

    def get_neighbors(self) -> NeighborsResult:
        neighbors_result = NeighborsResult()

        # Get CDP neighbors (Priority 1)
        self.logger.debug("Getting CDP neighbors via Scrapli for device {}", self.device.id)
        result = self.send_command("show cdp neighbors detail")
        if result.parsed_output:
            neighbors = result.parsed_output

            for nbr in neighbors:
                capabilities = [cap.strip() for cap in nbr["capabilities"].split(",")]
                neighbors_result.append(
                    NeighborResult(
                        hostname=nbr["neighbor_name"].split(".")[0],  # Device ID
                        ip=nbr["mgmt_address"],  # IP address
                        capabilities=capabilities,  # Capabilities
                        local_interface=nbr["local_interface"],  # Interface
                        neighbor_interface=nbr["neighbor_interface"],  # Port ID (outgoing port)
                        model=reported(nbr, "platform"),
                        system_description=reported(nbr, "neighbor_description"),  # Version
                    )
                )

        # Get LLDP neighbors (Priority 2)
        self.logger.debug("Getting LLDP neighbors via Scrapli for device {}", self.device.id)
        result = self.send_command("show lldp neighbors detail")
        if result.parsed_output:
            neighbors = result.parsed_output

            for nbr in neighbors:
                # Only append neighbor to result if they don't already exist based on hostname or management IP and local interface
                # TODO: Fix abbreviated interface names in LLDP for local interfaces
                if not any(
                    (
                        existing_nbr.hostname == nbr["neighbor_name"].split(".")[0]
                        or existing_nbr.ip == nbr["mgmt_address"]
                    )
                    and existing_nbr.local_interface == nbr["local_interface"]
                    for existing_nbr in neighbors_result
                ):
                    neighbors_result.append(
                        NeighborResult(
                            hostname=nbr["neighbor_name"].split(".")[0]  # System Name or Chassis id
                            if nbr["neighbor_name"]
                            else nbr["chassis_id"],
                            ip=nbr["mgmt_address"],  # IP
                            capabilities=nbr["capabilities"].split(),  # Enabled Capabilities
                            local_interface=nbr["local_interface"],  # Local Intf
                            neighbor_interface=nbr["neighbor_interface"]  # Port Description or Port id
                            if nbr["neighbor_interface"]
                            else nbr["neighbor_port_id"],
                            vendor=reported(nbr, "manufacturer"),  # LLDP-MED inventory
                            model=reported(nbr, "platform"),
                            software_release=reported(nbr, "sw_revision"),
                            system_description=reported(nbr, "neighbor_description"),  # System Description
                        )
                    )

        return neighbors_result

    # =============================================================================
    # Network Device Lifecycle Capability
    # =============================================================================

    def reload(self, save_config: bool = True) -> None:
        if save_config:
            self.logger.debug("Saving config before reload for device {}", self.device.id)
            self.save_config()

        self.logger.debug("Reloading device via Scrapli for device {}", self.device.id)
        self._answer_reload_prompts("reload", save_config)
        # Confirm without reading on: the device drops the session as it goes down.
        self.require_raw_connection().write_return()

    def schedule_reload(self, delay: int, save_config: bool = True) -> None:
        if save_config:
            self.logger.debug("Saving config before scheduling reload for device {}", self.device.id)
            self.save_config()

        self.logger.debug(
            "Scheduling device reload via Scrapli for device {} with delay {} minutes", self.device.id, delay
        )
        self._answer_reload_prompts(f"reload in {delay}", save_config)
        self.send_command_expect("", expect="#")

    def _answer_reload_prompts(self, command: str, save_config: bool) -> None:
        """Run a reload command up to its final ``[confirm]``, answering the unsaved-config question."""
        result = self.send_command_expect(command, expect="?")
        if "?" not in result.output:
            raise ConnectionStateError(
                f"The device refused '{command}': {result.output.strip()}", device_id=device_id_of(self.device)
            )
        if "Save?" in result.output:
            self.send_command_expect("yes" if save_config else "no", expect="[confirm]")

    def cancel_reload(self) -> None:
        self.logger.debug("Canceling scheduled device reload via Scrapli for device {}", self.device.id)
        self.send_command("reload cancel")

Key points:

  • Inherits ScrapliConnectionPluginBase (connection creation is handled).
  • Inherits NetworkDeviceConfigurationCapability, NetworkDeviceExecutionCapability, NetworkDeviceInformationCapability and NetworkDeviceLifecycleCapability, and implements their methods (get_information(), send_command(), merge_config(), …).
  • Sets platform = "ios" to bind to Cisco IOS devices.
  • Uses self.require_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

    return decorator
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.

Both flags in use, from the FRR example plugin. It registers under its own platform, frr-example, because the SDK already ships a plugin for frr (neops_worker_sdk/connection/plugins/frr/netmiko_plugin.py). A second plugin for an SDK platform either replaces the SDK’s (same platform, type and library; a user default also silently overrides the SDK default) or competes with it, so a base function block’s connect(device, "ssh") raises AmbiguousPluginError:

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

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_information(), etc.
    Capability_Calls --> Connected
    Connected --> Torn_Down: teardown_connection()
    Torn_Down --> [*]
  1. Instantiation – Plugin(device) stores the device reference. No connection is created yet.
  2. initialize_connection(connect=True) – calls _create_connection() to build the BaseConnection wrapper, then calls wrapper.connect() to open the session.
  3. Capability methods – the plugin uses self.get_raw_connection() to obtain the library’s native connection object and execute commands.
  4. 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.
        """
        self.logger.debug(f"Initializing connection for device {self.device.id}")
        self._active_connection = self._create_connection()
        # Hand the plugin's context (and any caller context) down to the wrapper.
        self._active_connection.logger = self.logger.child(self._active_connection)

        if connect:
            self.logger.debug(f"Connecting to device {self.device.id}")
            self._active_connection.connect()
            self.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:
            self.logger.debug(f"Tearing down connection for device {self.device.id}")
            self._active_connection.disconnect()
            self._active_connection = None
            self.logger.info(f"Connection closed for device {self.device.id}")

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_information(self) -> InformationResult:
                output = self.require_raw_connection().send_input(input_="show version")
                return parse_version(output)
        """
        if not self._active_connection:
            return None
        return self._active_connection._connection

    def require_raw_connection(self) -> RawConnection:
        """The raw connection object, raising when none is active.

        Capability methods that cannot work without a session use this instead
        of re-implementing the None check around get_raw_connection().
        """
        connection = self.get_raw_connection()
        if connection is None:
            raise ConnectionCreationError(
                "No active connection; call initialize_connection() first.",
                device_id=device_id_of(self.device),
            )
        return connection

Inside a capability method this looks like:

def get_information(self) -> InformationResult:
    output = self.require_raw_connection().send_input(input_="show version")
    ...

Warning

get_raw_connection() does not check liveness. It returns None before initialize_connection(), and raises ConnectionNotInitializedError when the wrapper exists but never connected (connect=False, or a failed connect()). require_raw_connection() raises ConnectionCreationError instead of returning None; prefer it unless your method has a meaningful answer without a session.