Skip to content

from collections.abc import Iterator

Writing Capabilities & Proxies

This page shows how to define capability interfaces, how to compose them into typed proxies, and how to provide custom fallbacks.

Define a capability interface

A capability is an abstract interface (ABC) that declares one or more methods.

from abc import abstractmethod

from neops_worker_sdk.connection.capabilities import CapabilityInterface


class DeviceInfoCapability(CapabilityInterface):
    @abstractmethod
    def get_version(self) -> dict[str, str | None]:
        """Return vendor/model/serial/software_release (and optionally raw output)."""
        ...

Guidelines:

  • Keep method names unique across interfaces (name collisions are hard to debug).
  • Add docstrings: they become the first place users look when they hit NotImplementedForThisPlatform.

Create a proxy by inheriting capabilities

The proxy is the API your function block code uses.

from neops_worker_sdk.connection.proxy import ConnectionProxy


class DeviceInfoProxy(ConnectionProxy, DeviceInfoCapability):
    pass

Warning: Always put ConnectionProxy first in the inheritance list.

class MyProxy(ConnectionProxy, CapA, CapB): ...

This keeps method resolution and fallback injection predictable. Concretely, it ensures ConnectionProxy.__getattribute__ intercepts capability method access and that metaclass-injected fallback methods behave as intended.

Note (IDEs and type checkers): Some static analyzers (PyCharm, Pyright, mypy, etc.) may still think your proxy class is “abstract” because capability methods are declared as @abstractmethod on the interfaces.

At runtime this is still valid Python: the ProxyMeta metaclass injects fallback implementations and clears __abstractmethods__, so the proxy is instantiable.

If your IDE complains, you may need to suppress the warning locally (e.g. PyCharm # noinspection PyAbstractClass, or a type-checker ignore on the class line) while keeping the runtime behavior unchanged.

Compose multiple capabilities

from abc import abstractmethod

from neops_worker_sdk.connection.proxy import ConnectionProxy
from neops_worker_sdk.connection.capabilities import CapabilityInterface


class InventoryCapability(CapabilityInterface):
    @abstractmethod
    def get_inventory(self) -> list[str]:
        ...


class StatusCapability(CapabilityInterface):
    @abstractmethod
    def get_status(self) -> str:
        ...


class InventoryAndStatusProxy(ConnectionProxy, InventoryCapability, StatusCapability):
    pass

The proxy now has a typed surface area that includes all methods from all inherited interfaces.

Extend connection type hints

The SDK type aliases include common values. If you want IDE hints for your own strings, define custom aliases with extra Literal values and override get_connection() / connect() to expose them on your proxy:

from collections.abc import Iterator
from contextlib import contextmanager
from typing import Literal, Self

from neops_worker_sdk.connection.proxy import ConnectionProxy
from neops_worker_sdk.connection.types import (
    ConnectionLibrary,
    ConnectionPlatform,
    ConnectionType,
)
from neops_worker_sdk.connection.capabilities import InventoryCapability, StatusCapability
from neops_worker_sdk.workflow import Device
from collections.abc import Iterator
from contextlib import contextmanager

type MyConnectionType = ConnectionType | Literal["grpc"]
type MyConnectionLibrary = ConnectionLibrary | Literal["my_library"]
type MyConnectionPlatform = ConnectionPlatform | Literal["my_platform"]


class MyConnectionProxy(ConnectionProxy, InventoryCapability, StatusCapability):
    @classmethod
    def get_connection(
            cls,
            device: Device,
            connection_type: MyConnectionType | None = None,
            connection_library: MyConnectionLibrary | None = None,
            *,
            connect: bool = True,
            fallback_to_default: bool = False,
            validate_capabilities: bool = True,
    ) -> Self:
        return super().get_connection(
            device=device,
            connection_type=connection_type,
            connection_library=connection_library,
            connect=connect,
            fallback_to_default=fallback_to_default,
            validate_capabilities=validate_capabilities,
        )

    @classmethod
    @contextmanager
    def connect(
            cls,
            device: Device,
            connection_type: MyConnectionType | None = None,
            connection_library: MyConnectionLibrary | None = None,
            *,
            connect: bool = True,
            fallback_to_default: bool = False,
            validate_capabilities: bool = True,
    ) -> Iterator[Self]:
        with super().connect(
                device=device,
                connection_type=connection_type,
                connection_library=connection_library,
                connect=connect,
                fallback_to_default=fallback_to_default,
                validate_capabilities=validate_capabilities,
        ) as proxy:
            yield proxy

Use the same aliases for plugin attributes so your platform/type/library values stay consistent:

from neops_worker_sdk.connection.types import ConnectionPlugin


class MyPlugin(ConnectionPlugin, InventoryCapability, StatusCapability):
    platform: MyConnectionPlatform = "my_platform"
    connection_type: MyConnectionType = "grpc"
    connection_library: MyConnectionLibrary = "my_library"

These aliases are for static typing only; the runtime still accepts any string values.

Custom proxy fallbacks (override instead of raising)

If a plugin doesn’t implement a capability method, the proxy has an auto-generated fallback method that raises NotImplementedForThisPlatform.

You can override that behavior by implementing the method on the proxy itself:

from neops_worker_sdk.connection.capabilities import InventoryCapability


class SafeInventoryProxy(ConnectionProxy, InventoryCapability):
    def get_inventory(self) -> list[str]:
        # Custom fallback: return a safe default instead of raising
        return []

This is useful when:

  • a capability is nice to have but not required for your workflow
  • you want to degrade gracefully (e.g., empty inventory if unsupported)

Introspection: what does this proxy require?

You can inspect what capabilities/methods a proxy declares:

caps = DeviceInfoProxy.list_capabilities()

The result is a dict mapping interface name → list of method names.

Handling unsupported methods

If a proxy fallback raises, you’ll get NotImplementedForThisPlatform with rich context:

from neops_worker_sdk.connection.exceptions import NotImplementedForThisPlatform

try:
    proxy.get_inventory()
except NotImplementedForThisPlatform as exc:
    # exc.context is a CapabilityContext dataclass (when available) with:
    # - proxy_class: str
    # - method_name: str
    # - interface_name: str
    # - platform: str | None
    # - plugin_class: str | None
    # - device_id: str | None
    raise

Use this for:

  • clear error messages back to workflows/users
  • platform “feature flags” (capability support is explicit)

Complete minimal example (end-to-end)

For a tiny “hello world” you can run end-to-end, see:

  • neops_worker_sdk/connection/draft_usage.py

It demonstrates defining capabilities and proxies, registering plugins, and what happens when capabilities are missing.