Proxies and Capabilities
A ConnectionProxy is the user-facing object you interact with in a function block. It declares which capabilities your code requires and delegates every capability method call to the resolved plugin at runtime.
What is a ConnectionProxy?
A proxy is a thin class you define by inheriting from ConnectionProxy and one
or more capability interfaces:
from neops_worker_sdk.connection.proxy import ConnectionProxy
from neops_worker_sdk.connection.capabilities import NetworkDeviceInformationCapability
class DeviceInformationProxy(ConnectionProxy, NetworkDeviceInformationCapability):
pass
The proxy itself contains no implementation. It declares what it needs; the plugin system provides how it works for each platform.
What is a CapabilityInterface?
A capability interface is an abstract contract – a set of @abstractmethod
declarations that plugins can implement.
"""
Base capability interface for the connection proxy system.
"""
from abc import ABC
class CapabilityInterface(ABC): # noqa: B024 - Intentionally has no abstract methods; serves as base marker class
"""
Base class for all capability interfaces.
Subclass this to define a capability that plugins can implement.
The ProxyMeta metaclass will auto-generate fallback implementations
for any abstract methods not implemented by the resolved plugin.
The ProxyMeta metaclass identifies capability interfaces via inheritance
(i.e., subclasses of this base class).
How Fallback Methods Work:
When a proxy inherits from a capability interface, the metaclass:
1. Collects all abstract methods from the interface
2. For each method, checks if the resolved plugin implements it
3. If the plugin has the method -> delegates to plugin
4. If not -> calls the auto-generated fallback that raises NotImplementedForThisPlatform
Note:
This class inherits from ABC but has no abstract methods itself.
It serves as a base marker class. Subclasses should define their
own abstract methods using the @abstractmethod decorator.
Example:
Create a custom capability interface::
class ConfigCapability(CapabilityInterface):
@abstractmethod
def get_running_config(self) -> str:
'''Get the running configuration.'''
...
@abstractmethod
def get_startup_config(self) -> str:
'''Get the startup configuration.'''
...
"""
A concrete capability looks like this:
"""
Network Device information capability interface.
"""
from abc import abstractmethod
from dataclasses import dataclass
from enum import Enum
from neops_worker_sdk.connection.capabilities.base import CapabilityInterface
class InterfaceStatus(Enum):
UP = "up"
DOWN = "down"
DISABLED = "disabled"
ERROR_DISABLED = "error_disabled"
@dataclass
class InformationResult:
hostname: str
vendor: str
model: str
serial: str
software_release: str
@dataclass
class ConfigSnapshotResult:
running: str
startup: str
@dataclass
class InterfaceResult:
name: str
status: InterfaceStatus
description: str
class InterfacesResult(list[InterfaceResult]):
pass
@dataclass
class NeighborResult:
"""
One neighbor of the device. ``system_description`` is the first line of the neighbor's
LLDP System Description, or the first line after CDP ``Version :``, unparsed.
"""
hostname: str
ip: str
capabilities: list[str]
local_interface: str
neighbor_interface: str
vendor: str | None = None
model: str | None = None
software_release: str | None = None
system_description: str | None = None
class NeighborsResult(list[NeighborResult]):
pass
class NetworkDeviceInformationCapability(CapabilityInterface):
"""
Capability interface for basic network device information.
This is a common capability that most plugins should implement.
It provides basic device identification and version information.
"""
@abstractmethod
def get_information(self) -> InformationResult:
"""
Get network device version and identification information.
Returns:
A InformationResult containing network device information. Standard keys:
- hostname: Device hostname
- vendor: Device vendor (e.g., "Cisco", "Juniper")
- model: Device model (e.g., "ISR4451", "MX480")
- serial: Device serial number
- software_release: Software/firmware version
Example:
>>> plugin.get_information()
InformationResult(
hostname="switch01",
vendor="Cisco",
model="C1000-8P-E-2G-L",
serial="FTX1234A567",
software_release="15.2(4)E",
)
"""
...
@abstractmethod
def get_config(self) -> ConfigSnapshotResult:
"""
Get the network device configuration.
Returns:
A ConfigSnapshotResult containing the network device configuration. Standard keys:
- running: The current running configuration
- startup: The startup configuration (if applicable)
Example:
>>> plugin.get_config()
ConfigSnapshotResult(
running="version 15.2\nservice timestamps debug datetime msec\nservice timestamps log datetime msec\nno service password-encryption\n...",
startup="version 15.2\nservice timestamps debug datetime msec\nservice timestamps log datetime msec\nno service password-encryption\n..."
)
"""
...
@abstractmethod
def get_interfaces(self) -> InterfacesResult:
"""
Get the network device interfaces.
Returns:
An InterfacesResult containing a list of interfaces. Each interface has:
- name: Interface name (e.g., "GigabitEthernet0/1")
- status: Interface status (e.g., "up", "down", "disabled")
- description: Interface description
Example:
>>> plugin.get_interfaces()
InterfacesResult([
InterfaceResult(name="GigabitEthernet0/1", status=InterfaceStatus.UP, description="Uplink to Core"),
InterfaceResult(name="GigabitEthernet0/2", status=InterfaceStatus.DOWN, description="Access Port"),
...
])
"""
...
@abstractmethod
def get_neighbors(self) -> NeighborsResult:
"""
Get the network device neighbors.
Returns:
A NeighborsResult containing a list of neighbors. Each neighbor has:
- hostname: Neighbor hostname (if available)
- ip: Neighbor IP address (if available)
- capabilities: List of neighbor capabilities (e.g., ["Switch", "Router"])
- local_interface: Local interface connected to the neighbor
- neighbor_interface: Neighbor interface connected to the local device
- vendor, model, software_release: Structured fields the neighbour reported about
itself, or None. vendor and software_release come only from LLDP-MED inventory
(manufacturer, software revision), which endpoints such as phones and access points
send far more often than switches do. model comes from LLDP-MED (e.g.
"WS-C3850-24P") or from the CDP platform (e.g. "cisco WS-C2960X-48FPD-L"), so its
format differs by source.
- system_description: The first line of the neighbour's free-text self-description,
unparsed: the first line of the LLDP System Description, or the first line after
CDP "Version :". None when not reported.
Example:
>>> plugin.get_neighbors()
NeighborsResult([
NeighborResult(
hostname="switch02",
ip="10.0.0.2",
capabilities=["Router", "Switch", "IGMP"],
local_interface="GigabitEthernet1/0/1",
neighbor_interface="GigabitEthernet1/0/1",
model="cisco WS-C2960X-48FPD-L",
system_description="Cisco IOS Software, C2960X Software, Version 15.2(7)E8",
)
])
"""
...
Each @abstractmethod in the interface becomes a method you can call on the
proxy. The proxy’s metaclass auto-generates fallback implementations so that
the proxy class itself is always instantiable – even when the resolved plugin
only implements a subset of the declared methods.
Composing multiple capabilities
Proxies support multiple inheritance. List every capability your code needs:
class FullProxy(ConnectionProxy, NetworkDeviceInformationCapability, NetworkDeviceConfigurationCapability):
pass
The plugin resolver will prefer plugins that implement all listed
capabilities. If no single plugin covers everything, the resolver picks the
best-match plugin (the one implementing the most) and logs a warning. Methods
the plugin does not implement will raise NotImplementedForThisPlatform at
call time.
Fallback methods
When a resolved plugin does not implement a capability method, the proxy does
not crash on creation. Instead, the metaclass injects a fallback that raises
NotImplementedForThisPlatform with rich context:
from neops_worker_sdk.connection.exceptions import NotImplementedForThisPlatform
try:
result = proxy.get_inventory()
except NotImplementedForThisPlatform as exc:
# exc.context contains:
# proxy_class, method_name, interface_name,
# platform, plugin_class, device_id
logger.warning("Capability not available: %s", exc)
You can also override a fallback on the proxy class itself to provide a safe default instead of raising:
class SafeProxy(ConnectionProxy, NetworkDeviceInformationCapability, InventoryCapability):
def get_inventory(self) -> list:
"""Return empty list when the plugin has no inventory support."""
return []
Because __getattribute__ checks the plugin first, the override only takes
effect when the plugin lacks the method.
Listing capabilities
Every proxy exposes list_capabilities() for introspection:
@classmethod
def list_capabilities(cls) -> dict[str, list[str]]:
"""
List all capabilities and their methods for this proxy class.
This is useful for introspection and debugging.
Returns:
Dict mapping interface names to lists of method names.
Example:
>>> MyProxy.list_capabilities()
{
"NetworkDeviceInformationCapability": ["get_information", "get_config", "get_interfaces", "get_neighbors"],
"NetworkDeviceLifecycleCapability": ["reload", "schedule_reload", "cancel_reload"],
}
"""
result: dict[str, list[str]] = {}
for interface in cls._capability_interfaces:
methods = [name for name, iface in cls._capability_methods.items() if iface is interface]
if methods: # Only include interfaces with methods
result[interface.__name__] = methods
return result
Example output:
>>> MyProxy.list_capabilities()
{
"NetworkDeviceInformationCapability": ["get_config", "get_information", "get_interfaces", "get_neighbors"],
"InventoryCapability": ["get_inventory", "get_serial_numbers"],
}
Method delegation
When you call proxy.get_information(), the proxy routes the call to the plugin
that handles your device:
flowchart LR
call["proxy.get_information()"] --> check{"Is it a\ncapability method?"}
check -- yes --> plugin{"Plugin implements\nget_information()?"}
plugin -- yes --> delegate["plugin.get_information()"]
plugin -- no --> fallback["Return fallback\n(raises when called)"]
check -- no --> self["Normal attribute lookup"]In short: capability methods go to the plugin; everything else stays on the
proxy. If the plugin does not implement the method, the auto-generated fallback
raises NotImplementedForThisPlatform when you call it.
You can stop here
If you only need built-in capabilities like get_information(), this is all you
need. The pages that follow explain how to add new capabilities and
plugins — read them when you outgrow the built-in options.
Under the hood: ProxyMeta
The ConnectionProxy uses a custom metaclass called ProxyMeta (inheriting
from ABCMeta). When Python creates a new proxy subclass, ProxyMeta:
- Collects all
CapabilityInterfacesubclasses from the MRO. - Tracks which abstract methods belong to which interface
(
_capability_methodsmapping). - Generates fallback methods for every abstract method not already
defined in the class namespace. Each fallback raises
NotImplementedForThisPlatformwith aCapabilityContextthat includes the proxy class, method name, interface name, platform, plugin, and device id. - Clears the
__abstractmethods__set so the proxy can be instantiated without implementing the abstract methods itself.
This means you never need to write boilerplate raise NotImplementedError
stubs in your proxy classes.
For a full walkthrough of the metaclass internals, see Architecture Deep Dive.
