Netbox Get Connections Task

task api name: get_connections

Retrieves connection details for interfaces and ports on one or more devices from NetBox. The task returns physical links, virtual and LAG-derived links, console connections, provider network connections, cable details, and remote MAC addresses where NetBox has that data.

How It Works

1. Client submits a get_connections request to the NetBox worker
2. Worker resolves the target NetBox instance
3. Worker queries NetBox GraphQL for interface and port objects on the selected devices
4. Optional interfaces and interface_regex filters narrow the ports included in the GraphQL result
5. Worker normalises physical, virtual, LAG, console, and provider network connections into a device/interface keyed dictionary
6. In dry-run mode, the raw GraphQL query result is returned instead of the normalised connection map

Prerequisites

• Target devices must exist in NetBox.
• NetBox version must be 4.4.0 or newer.

Inputs

Parameter	Required	Description
devices	Yes	List of device names to retrieve connections for
interfaces	No	Exact interface or port names to include
interface_regex	No	Case-insensitive regex pattern to match interface and port names
instance	No	NetBox instance name to target
dry_run	No	Return raw GraphQL query content instead of processed connection data
cache	No	Cache usage mode accepted by the command model

Output

Normal mode returns connection data keyed by device name and interface or port name:

{
    "<device>": {
        "<interface>": {
            "breakout": False,
            "remote_device": "<peer-device>",
            "remote_device_status": "active",
            "remote_interface": "<peer-interface>",
            "remote_interface_label": "<peer-label>",
            "remote_termination_type": "interface",
            "termination_type": "interface",
            "remote_mac_addresses": ["00:11:22:33:44:55"],
            "cable": {
                "type": "smf",
                "status": "connected",
                "label": "",
                "tags": [],
                "custom_fields": {},
                "peer_termination_type": "interface",
                "peer_device": "<peer-device>",
                "peer_interface": "<peer-interface>",
            },
        },
    },
}

Provider network connections include provider instead of remote device/interface details.

Filtering

Use interfaces when you know exact port names. Use interface_regex when you want to match a group of ports. If both are provided, NetBox returns only ports matching the explicit names and the regex.

Dry Run Mode

dry_run=True returns the raw GraphQL result used by the task. This is useful when checking the NetBox query output or troubleshooting why a connection was not normalised.

Examples

CLIPython

Query all connections for two devices:

nf#netbox get connections devices fceos4 fceos5

Query selected interface names only:

nf#netbox get connections devices fceos4 fceos5 interfaces eth101 eth103

Query connections matching an interface regex:

nf#netbox get connections devices fceos4 interface-regex "eth10[1-4]"

Return the raw GraphQL query result:

nf#netbox get connections devices fceos4 dry-run

Query a specific NetBox instance:

nf#netbox get connections devices fceos4 instance lab

Context manager:

from norfab.core.nfapi import NorFab

with NorFab(inventory="./inventory.yaml") as nf:
    client = nf.make_client()

    result = client.run_job(
        "netbox",
        "get_connections",
        workers="any",
        kwargs={
            "devices": ["fceos4", "fceos5"],
        },
    )

Direct lifecycle:

from norfab.core.nfapi import NorFab

nf = NorFab(inventory="./inventory.yaml")
try:
    nf.start()
    client = nf.make_client()

    result = client.run_job(
        "netbox",
        "get_connections",
        workers="any",
        kwargs={
            "devices": ["fceos4", "fceos5"],
            "interfaces": ["eth101", "eth103"],
        },
    )

    preview = client.run_job(
        "netbox",
        "get_connections",
        workers="any",
        kwargs={
            "devices": ["fceos4"],
            "interface_regex": "eth10[1-4]",
            "dry_run": True,
        },
    )

    instance_result = client.run_job(
        "netbox",
        "get_connections",
        workers="any",
        kwargs={
            "devices": ["fceos4"],
            "instance": "lab",
        },
    )
finally:
    nf.destroy()

NORFAB Netbox Get Connections Command Shell Reference

NorFab shell supports these command options for Netbox get_connections task:

Note

The command model accepts cache, but the current task implementation does not use it when running the GraphQL query.

nf# man tree netbox.get.connections
root
└── netbox:    Netbox service
    └── get:    Query data from Netbox
        └── connections:    Query Netbox connections data for devices
            ├── timeout:    Job timeout
            ├── workers:    Filter worker to target, default 'any'
            ├── verbose-result:    Control output details, default 'False'
            ├── progress:    Display progress events, default 'True'
            ├── instance:    Netbox instance name to target
            ├── dry-run:    Only return query content, do not run it
            ├── devices:    Device names to query data for
            ├── interfaces:    Interface and port names to query data for
            ├── interface-regex:    Regex pattern to match interfaces and ports
            └── cache:    How to use cache, default 'True'
nf#

Python API Reference

Retrieve interface connection details for specified devices from Netbox.

This task retrieves these connections:

• Physical interfaces connections
• Child/virtual interfaces connections using parent interface connections details
• Lag interfaces connections using member ports connections details
• Lag child interfaces connections using member ports connections details
• Console port and console server ports connections
• Connections to provider networks for physical, child/virtual and lag interfaces

This task also fetches MAC addresses for remote interfaces, sub-interfaces, LAG and LAG sub-interfaces

Parameters:

Name	Type	Description	Default
job	Job	NorFab Job object containing relevant metadata	required
devices	list	List of device names to retrieve connections for.	required
interfaces	list	List of interface, console port, console server port, or power outlet names to retrieve connections for.	None
instance	str	Netbox instance name for the GraphQL query.	None
dry_run	bool	If True, perform a dry run without making actual changes.	False
interface_regex	str	Regex pattern to match interfaces, console ports and console server ports by name, case insensitive.	None

Returns:

Name	Type	Description
dict	Result	A dictionary containing per-interface connection details for each device: { "netbox-worker-1.2": { "r1": { "eth101": { "breakout": False, "cable": { "custom_fields": {}, "label": "", "peer_device": None, "peer_interface": None, "peer_termination_type": "circuittermination", "status": "connected", "tags": [], "tenant": None, "type": "smf" }, "remote_device": "fceos4", "remote_device_status": "active", "remote_interface": "eth101", "remote_interface_label": "port101", "remote_termination_type": "interface", "termination_type": "interface", "remote_mac_addresses": [ "00:11:22:33:44:01" ] } } } }

Raises:

Type	Description
Exception	If there is an error in the GraphQL query or data retrieval process.

Source code in norfab\workers\netbox_worker\connections_tasks.py

@Task(
    input=GetConnectionsInput,
    output=GetConnectionsResult,
    fastapi={"methods": ["GET"], "schema": NetboxFastApiArgs.model_json_schema()},
    mcp={
        "annotations": {
            "title": "Get Connections",
            "readOnlyHint": True,
            "destructiveHint": False,
            "idempotentHint": True,
            "openWorldHint": True,
        }
    },
)
def get_connections(
    self,
    job: Job,
    devices: list[str],
    interfaces: Union[None, list[str]] = None,
    interface_regex: Union[None, str] = None,
    instance: Union[None, str] = None,
    dry_run: bool = False,
    cache: Union[None, bool, str] = None,
) -> Result:
    """
    Retrieve interface connection details for specified devices from Netbox.

    This task retrieves these connections:

    - Physical interfaces connections
    - Child/virtual interfaces connections using parent interface connections details
    - Lag interfaces connections using member ports connections details
    - Lag child interfaces connections using member ports connections details
    - Console port and console server ports connections
    - Connections to provider networks for physical, child/virtual and lag interfaces

    This task also fetches MAC addresses for remote interfaces, sub-interfaces, LAG
    and LAG sub-interfaces

    Args:
        job: NorFab Job object containing relevant metadata
        devices (list): List of device names to retrieve connections for.
        interfaces (list, optional): List of interface, console port, console server
            port, or power outlet names to retrieve connections for.
        instance (str, optional): Netbox instance name for the GraphQL query.
        dry_run (bool, optional): If True, perform a dry run without making actual changes.
        interface_regex (str, optional): Regex pattern to match interfaces, console ports and
            console server ports by name, case insensitive.

    Returns:
        dict: A dictionary containing per-interface connection details for each device:

            ```
            {
                "netbox-worker-1.2": {
                    "r1": {
                        "eth101": {
                            "breakout": False,
                            "cable": {
                                "custom_fields": {},
                                "label": "",
                                "peer_device": None,
                                "peer_interface": None,
                                "peer_termination_type": "circuittermination",
                                "status": "connected",
                                "tags": [],
                                "tenant": None,
                                "type": "smf"
                            },
                            "remote_device": "fceos4",
                            "remote_device_status": "active",
                            "remote_interface": "eth101",
                            "remote_interface_label": "port101",
                            "remote_termination_type": "interface",
                            "termination_type": "interface",
                            "remote_mac_addresses": [
                                "00:11:22:33:44:01"
                            ]
                        }
                    }
                }
            }
            ```

    Raises:
        Exception: If there is an error in the GraphQL query or data retrieval process.
    """
    instance = instance or self.default_instance
    log.info(
        f"{self.name} - Get connections: Fetching connections for {len(devices)} device(s) from '{instance}'"
    )
    job.event(
        f"fetching connections for {len(devices)} device(s), dry_run={dry_run}"
    )
    ret = Result(
        task=f"{self.name}:get_connections",
        result={d: {} for d in devices},
        resources=[instance],
    )

    # query uses optional GraphQL variables for interface name and regex filters
    if not self.nb_version[instance] >= (4, 4, 0):
        raise UnsupportedNetboxVersion(
            f"{self.name} - Netbox version {self.nb_version[instance]} is not supported, "
            f"minimum required version is {self.compatible_ge_v4}"
        )

    variables = {
        "devices": devices,
        "interface_regex": interface_regex or ".*",
        "offset": 0,
        "limit": 50,
    }
    if interfaces:
        variables["interfaces"] = interfaces

    # retrieve full list of ports and process client-side to map connections
    query_result = self.netbox_graphql(
        job=job,
        query=CONNECTIONS_QUERY,
        variables=variables,
        instance=instance,
        dry_run=dry_run,
    )
    if query_result.errors:
        job.event(
            "failed to fetch connection data from NetBox GraphQL", severity="ERROR"
        )
        ret.errors.extend(query_result.errors)
        ret.failed = True
        return ret

    # return dry run result
    if dry_run is True:
        job.event("dry-run requested, returning connection GraphQL query result")
        ret.dry_run = True
        ret.result = query_result.result
        return ret

    all_ports = query_result.result
    if not all_ports:
        log.info(
            f"{self.name} - Get connections: No ports returned for {len(devices)} device(s)"
        )
        job.event("no connection ports returned from NetBox")
        return ret

    # extract physical interfaces connections
    port_count = sum(len(ports or []) for ports in all_ports.values())
    job.event(f"processing {port_count} connection port object(s)")
    for port_type, ports in all_ports.items():
        for port in ports:
            # skip ports that have no remote device connected
            endpoints = port["connected_endpoints"]
            if not endpoints or not all(i for i in endpoints):
                continue

            # extract required parameters
            cable = port.get("cable", {})
            device_name = port["device"]["name"]
            port_name = port["name"]
            link_peers = port["link_peers"]
            remote_termination_type = endpoints[0]["__typename"].lower()
            remote_termination_type = remote_termination_type.replace("type", "")

            # form initial connection dictionary
            connection = {
                "breakout": len(endpoints) > 1,
                "remote_termination_type": remote_termination_type,
                "termination_type": port_type,
                "remote_mac_addresses": sorted(
                    {
                        mac.get("mac_address")
                        for endpoint in endpoints
                        for mac in endpoint.get("mac_addresses") or []
                        if mac.get("mac_address")
                    }
                ),
            }

            # add remote connection details
            if remote_termination_type == "providernetwork":
                connection["remote_device"] = None
                connection["remote_device_status"] = None
                connection["remote_interface"] = None
                connection["remote_interface_label"] = None
                connection["provider"] = endpoints[0]["name"]
            else:
                remote_interface = endpoints[0]["name"]
                remote_label = endpoints[0]["label"]
                if len(endpoints) > 1:
                    remote_interface = [i["name"] for i in endpoints]
                    remote_label = [i["label"] for i in endpoints]
                connection["remote_interface"] = remote_interface
                connection["remote_interface_label"] = remote_label
                connection["remote_device"] = endpoints[0]["device"]["name"]
                connection["remote_device_status"] = endpoints[0]["device"][
                    "status"
                ]

            # add cable and its peer details
            if link_peers:
                peer_termination_type = link_peers[0]["__typename"].lower()
                peer_termination_type = peer_termination_type.replace("type", "")
                cable["peer_termination_type"] = peer_termination_type
                cable["peer_device"] = link_peers[0].get("device", {}).get("name")
                cable["peer_interface"] = link_peers[0].get("name")
                if len(link_peers) > 1:  # handle breakout cable
                    cable["peer_interface"] = [i["name"] for i in link_peers]
            connection["cable"] = cable

            # add physical connection to the results
            ret.result[device_name][port_name] = connection

    # extract virtual interfaces connections
    for port_type, ports in all_ports.items():
        for port in ports:
            # add child virtual interfaces connections
            if port["type"] != "virtual" or not port["parent"]:
                continue
            device_name = port["device"]["name"]
            interface_name = port["name"]
            parent = port["parent"]
            connection = {
                "remote_device": None,
                "remote_device_status": None,
                "remote_interface": None,
                "remote_termination_type": "virtual",
                "termination_type": "virtual",
                "remote_mac_addresses": [],
            }
            # find connection endpoint
            if parent["type"] == "lag":
                try:
                    endpoint = parent["member_interfaces"][0][
                        "connected_endpoints"
                    ][0]
                except Exception:
                    continue
            elif parent["connected_endpoints"]:
                try:
                    endpoint = parent["connected_endpoints"][0]
                except Exception:
                    continue
            else:
                log.error(
                    f"{device_name}:{interface_name} Parent has no connected endpoints"
                )
                continue
            remote_termination_type = endpoint["__typename"].lower()
            remote_termination_type = remote_termination_type.replace("type", "")
            # collect virtual interfaces facing provider
            if remote_termination_type == "providernetwork":
                connection["provider"] = endpoint["name"]
                connection["remote_termination_type"] = remote_termination_type
            # find matching remote virtual interface for LAG subif
            elif "." in interface_name and parent["type"] == "lag":
                connection["remote_device"] = endpoint["device"]["name"]
                connection["remote_device_status"] = endpoint["device"]["status"]
                if endpoint["lag"]:
                    subif_id = interface_name.split(".")[1]
                    for remote_child in endpoint["lag"]["child_interfaces"]:
                        if remote_child["name"].endswith(f".{subif_id}"):
                            connection["remote_interface"] = remote_child["name"]
                            connection["remote_mac_addresses"] = sorted(
                                {
                                    mac.get("mac_address")
                                    for mac in remote_child.get("mac_addresses")
                                    or []
                                    if mac.get("mac_address")
                                }
                            )
                            break
                    # no matching subinterface found, associate child interface with remote interface
                    else:
                        connection["remote_interface"] = endpoint["lag"]["name"]
                        connection["remote_termination_type"] = "lag"
                # no remote lag found, associate child interface with remote interface
                else:
                    connection["remote_interface"] = endpoint["name"]
                    connection["remote_interface_label"] = endpoint["label"]
            # find matching remote virtual interface for physical interface subif
            elif "." in interface_name:
                connection["remote_device"] = endpoint["device"]["name"]
                connection["remote_device_status"] = endpoint["device"]["status"]
                subif_id = interface_name.split(".")[1]
                for remote_child in endpoint["child_interfaces"]:
                    if remote_child["name"].endswith(f".{subif_id}"):
                        connection["remote_interface"] = remote_child["name"]
                        connection["remote_mac_addresses"] = sorted(
                            {
                                mac.get("mac_address")
                                for mac in remote_child.get("mac_addresses") or []
                                if mac.get("mac_address")
                            }
                        )
                        break
                # no matching subinterface found, associate child interface with remote interface
                else:
                    connection["remote_interface"] = endpoint["name"]
                    connection["remote_interface_label"] = endpoint["label"]
                    connection["remote_termination_type"] = remote_termination_type
            # add virtual interface connection to results
            ret.result[device_name][interface_name] = connection

    # extract LAG interfaces connections
    for port_type, ports in all_ports.items():
        for port in ports:
            if port["type"] != "lag":
                continue
            device_name = port["device"]["name"]
            interface_name = port["name"]
            connection = {
                "remote_device": None,
                "remote_device_status": None,
                "remote_interface": None,
                "remote_termination_type": "lag",
                "termination_type": "lag",
                "remote_mac_addresses": [],
            }
            try:
                endpoint = port["member_interfaces"][0]["connected_endpoints"][0]
            except Exception:
                continue
            remote_termination_type = endpoint["__typename"].lower()
            remote_termination_type = remote_termination_type.replace("type", "")
            # collect lag interfaces facing provider
            if remote_termination_type == "providernetwork":
                connection["provider"] = endpoint["name"]
            # find remote lag interface
            elif endpoint["lag"]:
                connection["remote_interface"] = endpoint["lag"]["name"]
                connection["remote_device"] = endpoint["device"]["name"]
                connection["remote_device_status"] = endpoint["device"]["status"]
                connection["remote_mac_addresses"] = sorted(
                    {
                        mac.get("mac_address")
                        for mac in endpoint["lag"].get("mac_addresses") or []
                        if mac.get("mac_address")
                    }
                )
            # if no remote lag, collect remote end interfaces
            else:
                connection["remote_interface"] = []
                connection["remote_interface_label"] = []
                connection["termination_type"] = "interface"
                connection["remote_device"] = endpoint["device"]["name"]
                connection["remote_device_status"] = endpoint["device"]["status"]
                for member in port["member_interfaces"]:
                    for endpoint in member["connected_endpoints"]:
                        connection["remote_interface"].append(endpoint["name"])
                        connection["remote_interface_label"].append(
                            endpoint["label"]
                        )
            # add lag interface connection to results
            ret.result[device_name][interface_name] = connection

    log.info(
        f"{self.name} - get_connections: completed with connection data for {len(ret.result)} device(s)"
    )
    job.event(f"retrieved connections for {len(ret.result)} device(s)")

    return ret