Netbox Get Topology Task

task api name: get_topology

Overview

The get_topology task retrieves network topology data from NetBox and returns it as a structured list of nodes (devices) and links (physical cable connections). The output is designed to feed network visualisation applications such as the NorFab web UI topology viewer, D3.js graphs, or any tool that consumes node-link graph data.

Nodes carry rich device metadata — platform, primary management IP, role, site, manufacturer, and tags. Links carry interface-level details — interface type, speed, MTU, cable type/status, and tags. Every physical cable appears exactly once in the links list regardless of which side of the connection was queried first.

Inputs

Parameter	Required	Description
devices	Conditional	Device names to include in the topology
device_contains	Conditional	Case-insensitive substring to filter device names
device_regex	Conditional	Regex pattern to filter device names
role	Conditional	Device role slugs to filter by
platform	Conditional	Platform slugs to filter by
manufacturers	Conditional	Manufacturer slugs to filter by
status	Conditional	Device status values to filter by
sites	Conditional	Site slugs to filter by
Nornir filters	Conditional	Host filters such as FL, FB, FG, FC, or FN
instance	No	NetBox instance name to target
branch	No	NetBox Branching plugin branch name to read from
dry_run	No	Return query parameters without executing NetBox calls
timeout	No	Timeout in seconds for Nornir host resolution

At least one NetBox device filter or Nornir host filter is required.

Output

Returns topology data with nodes and links.

{
  "nodes": [
    {
      "id": "spine-1",
      "name": "spine-1",
      "type": "arista_eos",
      "ip": "10.0.0.1/32",
      "status": "active",
      "role": "spine",
      "site": "dc-nyc",
      "tags": ["core", "production"],
      "manufacturer": "Arista",
      "device_type": "DCS-7050CX3-32S"
    }
  ],
  "links": [
    {
      "source": "spine-1",
      "target": "leaf-1",
      "src_iface": "Ethernet1",
      "dst_iface": "Ethernet49",
      "type": "1000base-t",
      "speed": 1000000,
      "mtu": 9214,
      "tags": [],
      "cable_type": "smf",
      "cable_status": "connected",
      "cable_label": ""
    }
  ]
}

Examples

CLIPython

Retrieve topology for specific devices:

nf#netbox get topology devices spine-1 leaf-1 leaf-2

Filter by name substring:

nf#netbox get topology device-contains spine

Filter by role and status:

nf#netbox get topology role spine leaf status active

Dry run to inspect query parameters:

nf#netbox get topology devices spine-1 leaf-1 dry-run

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_topology",
        workers="any",
        kwargs={"devices": ["spine-1", "leaf-1", "leaf-2"]},
    )

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_topology",
        workers="any",
        kwargs={
            "role": ["spine", "leaf"],
            "status": ["active"],
        },
    )

    for worker, res in result.items():
        nodes = res["result"]["nodes"]
        links = res["result"]["links"]
        print(f"Nodes: {len(nodes)}, Links: {len(links)}")

finally:
    nf.destroy()

Warning

Fetching broad topologies can be slow on large NetBox installations. Use filters to narrow the scope when possible.

Filtering Devices

The task provides several filtering parameters, all resolved via the NetBox REST API before interface connections are queried.

Parameter	Description
devices	Exact device name list
device_contains	Case-insensitive substring match on device name (NetBox name__ic filter)
device_regex	Regular-expression match on device name (NetBox name__re filter)
role	List of device role slugs (e.g. ["spine", "leaf"])
platform	List of platform slugs (e.g. ["arista-eos", "cisco-ios"])
manufacturers	List of manufacturer slugs (e.g. ["arista", "cisco"])
status	List of status values (e.g. ["active", "planned"])

Filters can be combined freely. For example, to get all active spine devices made by Arista:

result = client.run_job(
    "netbox",
    "get_topology",
    workers="any",
    kwargs={
        "role": ["spine"],
        "manufacturers": ["arista"],
        "status": ["active"],
    },
)

Or filter by a name pattern:

result = client.run_job(
    "netbox",
    "get_topology",
    workers="any",
    kwargs={"device_contains": "spine"},
)

Adjacent Nodes

When a device in the filtered set has a physical cable connected to a device outside the filtered set, the remote device is automatically fetched from NetBox and added to the nodes list. This ensures the topology graph is always consistent — every link endpoint has a corresponding node entry — and lets you visualise partial topologies without orphaned links.

For example, if you query only spine-1 but it has cables to leaf-1 and leaf-2, the result will contain three nodes (spine-1, leaf-1, leaf-2) and all links between them.

Adjacent nodes are fetched in a single additional REST call after the interface connection query, so the overhead is minimal regardless of how many extra devices are discovered.

Pass dry_run=True to inspect the REST filter parameters and GraphQL query that would be sent to NetBox without executing any network calls:

result = client.run_job(
    "netbox",
    "get_topology",
    workers="any",
    kwargs={"devices": ["spine-1", "leaf-1"], "dry_run": True},
)

NORFAB Netbox Get Topology Command Shell Reference

NorFab shell supports these command options for Netbox get_topology task:

nf# man tree netbox.get.topology
root
└── netbox:    Netbox service
    └── get:    Query data from Netbox
        └── topology:    Query Netbox topology 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
            ├── branch:    Branching plugin branch name to use
            ├── dry-run:    Do not commit to database
            ├── devices:    List of device names to build topology for
            ├── device-contains:    Case-insensitive substring filter on device name
            ├── device-regex:    Regex filter on device name
            ├── role:    List of device role slugs to filter by
            ├── platform:    List of platform slugs to filter by
            ├── manufacturers:    List of manufacturer slugs to filter by
            ├── status:    List of device status values to filter by
            └── sites:    List of site slugs to filter by
nf#

Output Structure Reference

Node Fields

Field	Type	Description
id	string	Device name used as a unique graph node identifier
name	string	Device name
type	string or null	Device platform slug (e.g. arista_eos, cisco_iosxr)
ip	string or null	Primary management IP with prefix length (e.g. 10.0.0.1/32)
status	string or null	NetBox device status value (e.g. active, planned)
role	string or null	Device role name (e.g. spine, leaf, access-switch)
site	string or null	Site name the device belongs to
tags	list of strings	Tag names assigned to the device
manufacturer	string or null	Manufacturer name (e.g. Arista, Cisco)
device_type	string or null	Device type model name (e.g. DCS-7050CX3-32S)

Link Fields

Field	Type	Description
source	string	Source device name
target	string	Target device name
src_iface	string	Source interface name
dst_iface	string	Destination interface name
type	string or null	Interface type value (e.g. 1000base-t, 10gbase-x-sfpp)
speed	integer or null	Interface speed in Kbps
mtu	integer or null	Interface MTU in bytes
tags	list of strings	Tag names assigned to the interface
cable_type	string or null	Cable type value (e.g. smf, mmf, cat6)
cable_status	string or null	Cable status value (e.g. connected, planned)
cable_label	string or null	Cable label text

Python API Reference

Retrieve network topology from NetBox as a list of nodes and links.

Fetches device data and physical interface connections to build a topology graph suitable for network visualisation tools. Nodes represent devices; links represent physical cabled connections between them. Links are deduplicated so each cable appears only once.

Parameters:

Name	Type	Description	Default
job	Job	NorFab Job object containing relevant metadata.	required
instance	str	NetBox instance name; uses the default instance when omitted.	None
devices	list	List of device names to include in the topology. When omitted all devices in NetBox are fetched. After building links, any remote device connected to a filtered device that is not already in the list is automatically fetched and added as an adjacent node so every link endpoint always has a corresponding node entry.	None
device_contains	str	Case-insensitive substring to filter device names by (e.g. "spine" matches spine-1, dc1-spine).	None
device_regex	str	Regex pattern to filter device names by (e.g. "spine-[0-9]+").	None
role	list	List of device role slugs to filter by (e.g. ["spine", "leaf"]).	None
platform	list	List of platform slugs to filter by (e.g. ["arista-eos", "cisco-ios"]).	None
manufacturers	list	List of manufacturer slugs to filter by (e.g. ["arista", "cisco"]).	None
status	list	List of device status values to filter by (e.g. ["active", "planned"]).	None
sites	list	List of site slugs to filter by (e.g. ["dc-1", "dc-2"]).	None
dry_run	bool	When True returns GraphQL query parameters for both the device and interface queries without executing them. Defaults to False.	False
branch	str	NetBox branching plugin branch name to use.	None
timeout	int	Timeout in seconds for Nornir host resolution when Fx filter arguments are used. Defaults to 60.	600
**kwargs	dict	Nornir host filter arguments (e.g. FC, FL, FB, FG, FO, FP, FH, FR, FT) passed to get_nornir_hosts to resolve additional devices from the Nornir inventory. Resolved hosts are merged with the devices list.	{}

Returns:

Type	Description
Result	GetTopologyResult with result.nodes and result.links:
Result	nodes — one entry per device:: { "id": "spine-1", "name": "spine-1", "type": "arista_eos", "ip": "10.0.0.1/32", "status": "active", "role": "spine", "site": "dc-1", "tags": ["core"], "manufacturer": "Arista", "device_type": "DCS-7050CX3-32S", "color": "aa1409" }
Result	links — one entry per physical cable:: { "source": "spine-1", "target": "leaf-1", "src_iface": "Ethernet1", "dst_iface": "Ethernet49", "type": "1000base-t", "speed": 1000000, "mtu": 9214, "tags": [], "cable_type": "smf", "cable_status": "connected", "cable_label": "" }

Raises:

Type	Description
UnsupportedNetboxVersion	If the NetBox instance version is below 4.4.0.

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

@Task(
    fastapi={"methods": ["GET"], "schema": NetboxFastApiArgs.model_json_schema()},
    input=GetTopologyInput,
    output=GetTopologyResult,
    mcp={
        "annotations": {
            "title": "Get Topology",
            "readOnlyHint": True,
            "destructiveHint": False,
            "idempotentHint": True,
            "openWorldHint": True,
        }
    },
)
def get_topology(
    self,
    job: Job,
    instance: Union[None, str] = None,
    devices: Union[None, list] = None,
    device_contains: Union[None, str] = None,
    device_regex: Union[None, str] = None,
    role: Union[None, list] = None,
    platform: Union[None, list] = None,
    manufacturers: Union[None, list] = None,
    status: Union[None, list] = None,
    sites: Union[None, list] = None,
    dry_run: bool = False,
    branch: Union[None, str] = None,
    timeout: int = 600,
    **kwargs: dict,
) -> Result:
    """
    Retrieve network topology from NetBox as a list of nodes and links.

    Fetches device data and physical interface connections to build a topology
    graph suitable for network visualisation tools. Nodes represent devices;
    links represent physical cabled connections between them. Links are
    deduplicated so each cable appears only once.

    Args:
        job: NorFab Job object containing relevant metadata.
        instance (str, optional): NetBox instance name; uses the default instance
            when omitted.
        devices (list, optional): List of device names to include in the topology.
            When omitted all devices in NetBox are fetched. After building links,
            any remote device connected to a filtered device that is not already
            in the list is automatically fetched and added as an adjacent node so
            every link endpoint always has a corresponding node entry.
        device_contains (str, optional): Case-insensitive substring to filter
            device names by (e.g. ``"spine"`` matches ``spine-1``, ``dc1-spine``).
        device_regex (str, optional): Regex pattern to filter device names by
            (e.g. ``"spine-[0-9]+"``).
        role (list, optional): List of device role slugs to filter by
            (e.g. ``["spine", "leaf"]``).
        platform (list, optional): List of platform slugs to filter by
            (e.g. ``["arista-eos", "cisco-ios"]``).
        manufacturers (list, optional): List of manufacturer slugs to filter by
            (e.g. ``["arista", "cisco"]``).
        status (list, optional): List of device status values to filter by
            (e.g. ``["active", "planned"]``).
        sites (list, optional): List of site slugs to filter by
            (e.g. ``["dc-1", "dc-2"]``).
        dry_run (bool, optional): When True returns GraphQL query parameters
            for both the device and interface queries without executing them.
            Defaults to False.
        branch (str, optional): NetBox branching plugin branch name to use.
        timeout (int, optional): Timeout in seconds for Nornir host resolution
            when Fx filter arguments are used. Defaults to 60.
        **kwargs: Nornir host filter arguments (e.g. ``FC``, ``FL``, ``FB``,
            ``FG``, ``FO``, ``FP``, ``FH``, ``FR``, ``FT``) passed to
            ``get_nornir_hosts`` to resolve additional devices from the Nornir
            inventory. Resolved hosts are merged with the ``devices`` list.

    Returns:
        GetTopologyResult with ``result.nodes`` and ``result.links``:

        **nodes** — one entry per device::

            {
                "id":           "spine-1",
                "name":         "spine-1",
                "type":         "arista_eos",
                "ip":           "10.0.0.1/32",
                "status":       "active",
                "role":         "spine",
                "site":         "dc-1",
                "tags":         ["core"],
                "manufacturer": "Arista",
                "device_type":  "DCS-7050CX3-32S",
                "color":        "aa1409"
            }

        **links** — one entry per physical cable::

            {
                "source":        "spine-1",
                "target":        "leaf-1",
                "src_iface":     "Ethernet1",
                "dst_iface":     "Ethernet49",
                "type":          "1000base-t",
                "speed":         1000000,
                "mtu":           9214,
                "tags":          [],
                "cable_type":    "smf",
                "cable_status":  "connected",
                "cable_label":   ""
            }

    Raises:
        UnsupportedNetboxVersion: If the NetBox instance version is below 4.4.0.
    """
    instance = instance or self.default_instance
    log.info(
        f"{self.name} - Get topology: Fetching topology for "
        + (f"{len(devices)} device(s)" if devices else "all devices")
        + f" from '{instance}'"
    )
    ret = GetTopologyResult(
        task=f"{self.name}:get_topology",
        result={"nodes": [], "links": []},
        resources=[instance],
    )

    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}"
        )

    devices = devices or []

    # resolve additional devices from Nornir Fx filter arguments
    if kwargs:
        nornir_hosts = self.get_nornir_hosts(kwargs, timeout)
        for host in nornir_hosts:
            if host not in devices:
                devices.append(host)

    # build pynetbox REST filter params
    device_filter_params: Dict[str, Any] = {}
    if devices:
        device_filter_params["name"] = devices
    if device_contains:
        device_filter_params["name__ic"] = device_contains
    if device_regex:
        device_filter_params["name__re"] = device_regex
    if role:
        device_filter_params["role"] = role
    if platform:
        device_filter_params["platform"] = platform
    if manufacturers:
        device_filter_params["manufacturer"] = manufacturers
    if status:
        device_filter_params["status"] = status
    if sites:
        device_filter_params["site"] = sites

    if dry_run is True:
        job.event("dry-run requested, returning topology query parameters")
        intf_dry = self.netbox_graphql(
            job=job,
            query=TOPOLOGY_INTERFACES_QUERY,
            variables={"devices": devices or ["*"], "offset": 0, "limit": 50},
            instance=instance,
            dry_run=True,
        )
        ret.dry_run = True
        ret.result = {
            "device_filter": device_filter_params,
            "graphql": intf_dry.result,
        }
        return ret

    nb = self._get_pynetbox(instance, branch=branch)

    # --- step 1: fetch device data for nodes via pynetbox REST ---
    job.event(
        "fetching device data from '{}'".format(instance)
        + (
            f" using {len(device_filter_params)} filter(s)"
            if device_filter_params
            else ""
        )
    )
    all_nb_devices = list(
        nb.dcim.devices.filter(
            **device_filter_params,
            fields="name,platform,primary_ip4,status,role,site,tags,device_type",
        )
    )
    device_names = set()
    role_slugs: set = set()
    for dev in all_nb_devices:
        device_names.add(dev.name)
        if dev.role:
            role_slugs.add(dev.role.slug)

    # fetch role colors
    role_colors: Dict[str, str] = {}
    if role_slugs:
        job.event("fetching role colors for {} role(s)".format(len(role_slugs)))
        for role in nb.dcim.device_roles.filter(slug=list(role_slugs)):
            role_colors[role.slug] = role.color

    for dev in all_nb_devices:
        color = role_colors.get(dev.role.slug) if dev.role else None
        ret.result.nodes.append(_build_node(dev, color=color))

    if not device_names:
        log.info(
            f"{self.name} - Get topology: No devices found, returning empty topology"
        )
        job.event("no devices found for topology")
        return ret

    # --- step 2: fetch interface connections for links ---
    job.event(f"fetching interface connections for {len(device_names)} device(s)")
    variables = {
        "devices": list(device_names),
        "offset": 0,
        "limit": 50,
    }
    query_result = self.netbox_graphql(
        job=job,
        query=TOPOLOGY_INTERFACES_QUERY,
        variables=variables,
        instance=instance,
        dry_run=False,
    )

    if query_result.failed:
        job.event(
            "failed to fetch topology interface connections", severity="ERROR"
        )
        ret.failed = True
        ret.errors.extend(query_result.errors)
        return ret

    all_interfaces = (
        query_result.result.get("interface", []) if query_result.result else []
    )
    job.event(f"processing {len(all_interfaces)} topology interface object(s)")

    # --- step 3: build deduplicated links ---
    seen_links: set = set()
    for intf in all_interfaces:
        for endpoint in intf.get("connected_endpoints") or []:
            if endpoint.get("__typename") != "InterfaceType":
                continue

            source = intf["device"]["name"]
            target = endpoint["device"]["name"]
            src_iface = intf["name"]
            dst_iface = endpoint["name"]

            # canonical key ensures (A:eth1->B:eth2) == (B:eth2->A:eth1)
            link_key = tuple(sorted([(source, src_iface), (target, dst_iface)]))
            if link_key in seen_links:
                continue
            seen_links.add(link_key)

            cable = intf.get("cable") or {}
            intf_type = intf.get("type") or {}
            link = {
                "source": source,
                "target": target,
                "src_iface": src_iface,
                "dst_iface": dst_iface,
                "type": intf_type,
                "speed": intf.get("speed"),
                "mtu": intf.get("mtu"),
                "tags": [t["name"] for t in (intf.get("tags") or [])],
                "cable_type": cable.get("type"),
                "cable_status": cable.get("status"),
                "cable_label": cable.get("label"),
            }
            ret.result.links.append(link)

    # --- step 4: add nodes for connected devices not yet in the nodes list ---
    linked_device_names = set()
    for link in ret.result.links:
        linked_device_names.add(link["source"])
        linked_device_names.add(link["target"])
    missing_device_names = linked_device_names - device_names
    if missing_device_names:
        job.event(
            f"fetching data for {len(missing_device_names)} additional connected device(s)"
        )
        extra_nb_devices = list(
            nb.dcim.devices.filter(
                name=list(missing_device_names),
                fields="name,platform,primary_ip4,status,role,site,tags,device_type",
            )
        )
        extra_role_slugs = {
            dev.role.slug
            for dev in extra_nb_devices
            if dev.role and dev.role.slug not in role_colors
        }
        if extra_role_slugs:
            for role in nb.dcim.device_roles.filter(slug=list(extra_role_slugs)):
                role_colors[role.slug] = role.color
        for dev in extra_nb_devices:
            device_names.add(dev.name)
            color = role_colors.get(dev.role.slug) if dev.role else None
            ret.result.nodes.append(_build_node(dev, color=color))

    log.info(
        f"{self.name} - Get topology: Built topology with "
        f"{len(ret.result.nodes)} nodes and {len(ret.result.links)} links"
    )
    job.event(
        f"topology build complete: {len(ret.result.nodes)} node(s), "
        f"{len(ret.result.links)} link(s)"
    )
    return ret