Obtaining a Connection

Every device interaction starts by obtaining a ConnectionProxy instance. The SDK offers two entry points: a context manager (connect()) and a factory method (get_connection()).

`connect()` -- context manager (recommended)

connect() wraps the full lifecycle -- plugin resolution, connection setup, and teardown -- in a contextmanager. The connection is always closed on exit, even when an exception propagates.

    @classmethod
    @contextmanager
    def connect(
        cls,
        device: DeviceTypeDto,
        connection_type: ConnectionType | None = None,
        connection_library: ConnectionLibrary | None = None,
        *,
        connect: bool = True,
        fallback_to_default: bool = False,
        validate_capabilities: bool = True,
    ) -> Iterator[Self]:
        """
        Context manager for connection lifecycle.

        This is the recommended way to use the proxy. The connection is
        automatically closed when exiting the context, even if an exception occurs.

        Args:
            device: The device to connect to.
            connection_type: Type of connection (e.g., "ssh", "api").
            connection_library: Library to use (e.g., "netmiko", "scrapli").
            fallback_to_default: If True, fall back to platform defaults.
            validate_capabilities: If True, warn about missing capabilities.
            connect: If True, establish the connection immediately. If False, only
                initialize the connection wrapper.

        Yields:
            A configured proxy instance.

        Example:
            with MyProxy.connect(device, "ssh", "scrapli") as proxy:
                version = proxy.get_version()
                print(f"Device version: {version}")
            # Connection automatically closed here
        """
        proxy = cls.get_connection(
            device=device,
            connection_type=connection_type,
            connection_library=connection_library,
            connect=connect,
            fallback_to_default=fallback_to_default,
            validate_capabilities=validate_capabilities,
        )
        try:
            yield proxy
        finally:

Usage inside a function block:

with MyProxy.connect(device, "ssh", "scrapli") as proxy:
    version = proxy.get_version()

Synchronous context manager

connect() is a synchronous context manager (with, not async with). The underlying libraries (scrapli, netmiko, ncclient, ...) are blocking. This is fine inside function blocks because the neops runtime executes each function block invocation in a thread pool.

`get_connection()` -- manual lifecycle

When you need more control -- for example, to inspect the resolved plugin before connecting, or to keep a connection across several steps -- use get_connection() with an explicit close().

    @classmethod
    def get_connection(
        cls,
        device: DeviceTypeDto,
        connection_type: ConnectionType | None = None,
        connection_library: ConnectionLibrary | None = None,
        *,
        connect: bool = True,
        fallback_to_default: bool = False,
        validate_capabilities: bool = True,
    ) -> Self:
        """
        Get a connection proxy for the specified device.

        This is the primary factory method for creating proxy instances.
        It resolves the appropriate plugin using capability-aware resolution,
        creates the connection, and optionally validates capabilities.

        Args:
            device: The device to connect to.
            connection_type: Type of connection (e.g., "ssh", "api").
                Required unless fallback_to_default=True.
            connection_library: Library to use (e.g., "netmiko", "scrapli").
                Required unless fallback_to_default=True.
            connect: If True, establish the connection immediately. If False, only
                initialize the connection wrapper. Defaults to True.
            fallback_to_default: If True, fall back to platform/connection_type defaults
                when connection_library is not specified.
            validate_capabilities: If True, warn about missing plugin capabilities.

        Returns:
            A configured proxy instance of the called class type.

        Raises:
            PluginNotFoundError: If no matching plugin is found.
            AmbiguousPluginError: If multiple plugins match and no default resolves it.
            ConnectionValidationError: If device is missing required attributes.
            ConnectionCreationError: If connection cannot be established.

        Example:
            Exact plugin selection::

                proxy = MyProxy.get_connection(device, "ssh", "netmiko")

            Default SSH plugin for platform::

                proxy = MyProxy.get_connection(device, "ssh", fallback_to_default=True)

            Platform default plugin::

                proxy = MyProxy.get_connection(device, fallback_to_default=True)
        """
        # Extract required capabilities from this proxy class
        required_capabilities = cls._capability_interfaces

        # Resolve the plugin with capability-aware resolution
        plugin = resolve_plugin(
            device=device,
            connection_type=connection_type,
            connection_library=connection_library,
            required_capabilities=required_capabilities,
            fallback_to_default=fallback_to_default,
            proxy_class_name=cls.__name__,
        )

        # Validate plugin capabilities and warn about missing implementations
        if validate_capabilities:
            cls._validate_plugin_capabilities(plugin, device)

        # Initialize and connect
        plugin.initialize_connection(connect=connect)

        logger.info(
            f"Created {cls.__name__} proxy for device {device.id} "
            f"(platform={device.platform}) using plugin {plugin.__class__.__name__}"
        )

Always pair it with close():

proxy = MyProxy.get_connection(device, "ssh", "scrapli")
try:
    version = proxy.get_version()
finally:
    proxy.close()

Warning

Do not rely on garbage collection to close connections. Always call close() when you use get_connection() directly.

`close()` -- explicit teardown

close() delegates to the plugin's teardown_connection() and marks the proxy as closed. It is safe to call multiple times.

    def close(self) -> None:
        """
        Explicitly close the connection and cleanup resources.

        This method is safe to call multiple times. After calling close(),
        any capability method calls will fail.

        Example:
            proxy = MyProxy.get_connection(device, "ssh", "netmiko")
            try:
                result = proxy.get_version()
            finally:
                proxy.close()
        """
        if self._closed:
            logger.debug(f"Connection already closed for device {self._device.id}")
            return

        logger.debug(f"Closing connection for device {self._device.id}")
        self._plugin.teardown_connection()
        self._closed = True

After close(), capability method calls will fail because the plugin's raw connection is gone. Treat a closed proxy as unusable.

Parameters

Both connect() and get_connection() accept the same parameters:

Parameter	Type	Default	Description
`device`	`Device`	(required)	The device to connect to. Must have `platform`, `ip`, `username`, `password`.
`connection_type`	`str \\| None`	`None`	Protocol: `"ssh"`, `"api"`, `"netconf"`, `"restconf"`, etc. Required unless `fallback_to_default=True`.
`connection_library`	`str \\| None`	`None`	Library: `"scrapli"`, `"netmiko"`, `"httpx"`, `"ncclient"`, etc. Required unless `fallback_to_default=True`.
`connect`	`bool`	`True`	Establish the connection immediately. Set to `False` to create the wrapper without connecting (advanced/testing).
`fallback_to_default`	`bool`	`False`	Allow the resolver to select a registered default plugin when the library is omitted or resolution is ambiguous.
`validate_capabilities`	`bool`	`True`	Warn at creation time if the resolved plugin is missing methods declared by the proxy's capabilities.

Deferred connect

Pass connect=False to create the plugin and connection wrapper without opening a live session. You can inspect the resolved plugin before establishing the connection:

proxy = MyProxy.get_connection(device, "ssh", "scrapli", connect=False)
# inspect proxy.plugin before connecting
proxy.plugin._active_connection.connect()

Warning

initialize_connection() always creates a new connection wrapper. Do not call it a second time — connect the existing wrapper instead.

Error handling

Exception hierarchy

All connection exceptions inherit from NeopsConnectionError (not Python's built-in ConnectionError):

classDiagram
    class NeopsConnectionError
    NeopsConnectionError <|-- ConnectionCreationError
    NeopsConnectionError <|-- ConnectionValidationError
    NeopsConnectionError <|-- ConnectionStateError
    ConnectionStateError <|-- ConnectionNotInitializedError
    ConnectionStateError <|-- ConnectionAlreadyInitializedError
    ConnectionStateError <|-- ConnectionStillAliveError
    NeopsConnectionError <|-- PluginNotFoundError
    NeopsConnectionError <|-- PluginRegistrationError
    NeopsConnectionError <|-- AmbiguousPluginError
    NeopsConnectionError <|-- NotImplementedForThisPlatform

Exceptions by phase

Phase	Exception	Typical cause
Validation	`ConnectionValidationError`	Device missing `ip`, `username`, `password`, or `platform`.
Resolution	`PluginNotFoundError`	No plugin registered for the platform/type/library combination.
Resolution	`AmbiguousPluginError`	Multiple plugins match and no default resolves the tie.
Connection	`ConnectionCreationError`	Auth failure, unreachable host, or missing third-party library.
Capability call	`NotImplementedForThisPlatform`	Plugin does not implement the called capability method. Includes rich context (proxy, method, interface, platform, plugin, device id).

Example: structured error handling

from neops_worker_sdk.connection.exceptions import (
    AmbiguousPluginError,
    ConnectionCreationError,
    ConnectionValidationError,
    NotImplementedForThisPlatform,
    PluginNotFoundError,
)

try:
    with MyProxy.connect(device, "ssh", "scrapli") as proxy:
        version = proxy.get_version()
except ConnectionValidationError:
    # Device data incomplete -- fix device fields, retrying won't help
    raise
except PluginNotFoundError:
    # Plugin modules not imported, or no plugin for this combination
    raise
except AmbiguousPluginError:
    # Specify library explicitly or opt into defaults
    raise
except ConnectionCreationError:
    # Network / auth / dependency problem -- consider retry with backoff
    raise
except NotImplementedForThisPlatform:
    # Capability method missing on the resolved plugin
    raise

Catch-all

If you only need a single handler for all connection errors:

from neops_worker_sdk.connection.exceptions import NeopsConnectionError

try:
    with MyProxy.connect(device, "ssh", "scrapli") as proxy:
        ...
except NeopsConnectionError as exc:
    logger.error("Connection failed: %s", exc)

Obtaining a Connection

connect() -- context manager (recommended)

get_connection() -- manual lifecycle

close() -- explicit teardown