Nornir Create Host from NetBox Task

task api name: create_host_from_netbox

Creates or replaces runtime Nornir hosts from explicit NetBox device names. The Nornir service owns the inventory write, while the NetBox service supplies Nornir-compatible host inventory through get_nornir_inventory.

Hosts created by this task are in-memory runtime hosts. They are not written to YAML inventory files and disappear after worker restart unless startup inventory or NetBox-backed refresh also includes them.

Inputs

Parameter	Required	Description
devices	Yes	NetBox device names to fetch and add as Nornir hosts.
instance	No	NetBox instance name to target.
netbox_workers	No	NetBox worker or workers to query. Defaults to any. CLI alias: netbox-workers.
timeout	No	Timeout for the NetBox inventory request. Defaults to 600 seconds.
interfaces	No	Include interface data, or provide interface task kwargs. When omitted, the worker netbox inventory option is used if configured.
connections	No	Include connection data, or provide connection task kwargs. When omitted, the worker netbox inventory option is used if configured.
circuits	No	Include circuit data, or provide circuit task kwargs. When omitted, the worker netbox inventory option is used if configured.
bgp_peerings	No	Include BGP peering data, or provide BGP task kwargs. CLI alias: bgp-peerings.
nbdata	No	Include NetBox device data in host data. When omitted, the worker netbox inventory option is used if configured.
primary_ip	No	Primary IP family to use for hostname. CLI alias: primary-ip. When omitted, the worker netbox inventory option is used if configured.
cache	No	Cache usage mode passed to NetBox retrieval tasks. Defaults to True in the task.
groups	No	Additional Nornir group names to attach to created hosts.
dry_run	No	Preview created, updated, and missing hosts without changing Nornir. CLI alias: dry-run.
progress	No	Emit progress events. Defaults to True.

The task starts with the current Nornir worker netbox inventory section when present, then overlays explicit task arguments for NetBox inventory-build options such as interfaces, connections, circuits, bgp_peerings, nbdata, primary_ip, and cache. The devices and instance arguments are always taken from the task request.

Output

Returns created, updated, and missing host or device names:

{
    "created": ["leaf-1"],
    "updated": ["leaf-2"],
    "missing": ["leaf-3"],
}

created and updated use the returned Nornir host names. missing is calculated by comparing requested device names with returned host names.

Examples

CLIPython

nf# nornir inventory create-host-from-netbox devices leaf-1 leaf-2 workers nornir-worker-1

nf# nornir inventory create-host-from-netbox devices leaf-1 instance prod cache refresh workers nornir-worker-1

nf# nornir inventory create-host-from-netbox devices leaf-1 interfaces connections workers nornir-worker-1

nf# nornir inventory create-host-from-netbox devices leaf-1 dry-run workers nornir-worker-1

import pprint

from norfab.core.nfapi import NorFab

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

result = client.run_job(
    "nornir",
    "create_host_from_netbox",
    workers=["nornir-worker-1"],
    kwargs={
        "devices": ["leaf-1", "leaf-2"],
        "interfaces": True,
        "connections": True,
        "netbox_workers": "any",
    },
)

pprint.pprint(result)
nf.destroy()

Notes

• The returned Nornir host name can differ from the NetBox device name when config_context.nornir.name is set in NetBox. The current task reports created and updated using returned Nornir host names.
• If some requested devices are missing, the task emits a warning and still creates or updates hosts for the devices NetBox returned.
• dry_run=True sets the top-level result dry_run flag and does not mutate runtime inventory.
• Runtime inventory writes are delegated to runtime_inventory with a load action, so host create/replace semantics remain the existing runtime inventory semantics.

Python API Reference

Create or replace runtime Nornir hosts from explicit NetBox device names.

The task asks the NetBox service to build Nornir-compatible host data, predicts which returned host names are new or existing, and delegates the actual in-memory host replacement to runtime_inventory.

Source code in norfab\workers\nornir_worker\inventory_tasks.py

@Task(
    fastapi={"methods": ["POST"]},
    input=CreateHostFromNetboxInput,
    output=CreateHostFromNetboxResult,
    mcp={
        "annotations": {
            "title": "Create Nornir Hosts from NetBox Devices",
            "readOnlyHint": False,
            "destructiveHint": False,
            "idempotentHint": True,
            "openWorldHint": True,
        }
    },
)
def create_host_from_netbox(
    self,
    job: Job,
    devices: list[str],
    instance: str | None = None,
    netbox_workers: str | list[str] | None = "any",
    timeout: int | None = 600,
    interfaces: bool | dict | None = None,
    connections: bool | dict | None = None,
    circuits: bool | dict | None = None,
    bgp_peerings: bool | dict | None = None,
    nbdata: bool | None = None,
    primary_ip: str | None = None,
    cache: bool | str | None = True,
    groups: list[str] | None = None,
    dry_run: bool | None = False,
    progress: bool | None = True,
) -> Result:
    """
    Create or replace runtime Nornir hosts from explicit NetBox device names.

    The task asks the NetBox service to build Nornir-compatible host data,
    predicts which returned host names are new or existing, and delegates
    the actual in-memory host replacement to ``runtime_inventory``.
    """
    # Normalize optional list inputs used later in simple list operations.
    devices = devices or []
    groups = groups or []
    netbox_hosts = None
    ret = Result(
        task=f"{self.name}:create_host_from_netbox",
        result={"created": [], "updated": [], "missing": []},
    )

    # Start with the worker's configured NetBox inventory options, when any.
    if isinstance(self.nornir_worker_inventory.get("netbox"), dict):
        nornir_netbox_options = self.nornir_worker_inventory["netbox"].copy()
    else:
        nornir_netbox_options = {}

    # Task arguments that are present override worker inventory options.
    if interfaces is not None:
        nornir_netbox_options["interfaces"] = interfaces
    if connections is not None:
        nornir_netbox_options["connections"] = connections
    if circuits is not None:
        nornir_netbox_options["circuits"] = circuits
    if bgp_peerings is not None:
        nornir_netbox_options["bgp_peerings"] = bgp_peerings
    if nbdata is not None:
        nornir_netbox_options["nbdata"] = nbdata
    if primary_ip is not None:
        nornir_netbox_options["primary_ip"] = primary_ip
    if cache is not None:
        nornir_netbox_options["cache"] = cache
    nornir_netbox_options["devices"] = devices
    nornir_netbox_options["instance"] = instance

    # NetBox owns host data construction; Nornir owns the runtime inventory write.
    job.event(
        f"fetching Nornir inventory for {len(devices)} NetBox device(s) "
        f"from '{netbox_workers or 'any'}' worker(s)"
    )
    nb_inventory_data = self.client.run_job(
        service="netbox",
        task="get_nornir_inventory",
        workers=netbox_workers,
        kwargs=nornir_netbox_options,
        timeout=timeout,
    )

    if nb_inventory_data is None:
        msg = f"{self.name} - NetBox get_nornir_inventory returned no data"
        log.error(msg)
        ret.failed = True
        ret.status = "failed"
        ret.errors = [msg]
        return ret

    # Use the first NetBox worker that returns host inventory.
    for wname, wdata in nb_inventory_data.items():
        if wdata.get("failed") is False and wdata.get("result", {}).get("hosts"):
            netbox_hosts = wdata["result"]["hosts"]
            break

    if not netbox_hosts:
        msg = (
            f"{self.name} - NetBox worker(s) "
            f"'{', '.join(list(nb_inventory_data.keys()))}' returned no hosts data"
        )
        log.error(msg)
        ret.failed = True
        ret.status = "failed"
        ret.errors = [msg]
        return ret

    # calculate devices that are missing from netbox
    missing = sorted(set(devices) - set(netbox_hosts))
    if missing:
        msg = (
            f"{self.name} - NetBox returned no host data for: {', '.join(missing)}"
        )
        log.warning(msg)
        job.event(msg, severity="WARNING")

    # add host groups
    for host_data in netbox_hosts.values():
        host_data["groups"] = host_data.get("groups", []) + groups

    existing_hosts = set(self.nr.inventory.hosts)
    created = sorted(set(netbox_hosts) - existing_hosts)
    updated = sorted(set(netbox_hosts) & existing_hosts)
    ret.result = {"created": created, "updated": updated, "missing": missing}

    # Dry run stops after prediction and does not mutate Nornir runtime inventory.
    if dry_run is True:
        job.event("dry-run requested, Nornir runtime inventory not changed")
        ret.dry_run = True
        return ret

    # Runtime inventory load calls create_host once for each returned NetBox host.
    inventory_actions = [
        {"call": "create_host", "name": host_name, **host_data}
        for host_name, host_data in netbox_hosts.items()
    ]
    inventory_result = self.runtime_inventory(
        job=job,
        action="load",
        data=inventory_actions,
        progress=progress if progress is not None else False,
    )
    if inventory_result.failed:
        # Keep this task's created/updated/missing prediction and attach runtime errors.
        ret.failed = inventory_result.failed
        ret.status = inventory_result.status
        ret.errors.extend(inventory_result.errors)
        ret.messages.extend(inventory_result.messages)
        if inventory_result.result:
            ret.messages.append(str(inventory_result.result))
        return ret

    job.event(
        f"created {len(ret.result['created'])}, updated {len(ret.result['updated'])}, "
        f"missing {len(ret.result['missing'])} Nornir host(s) from NetBox"
    )
    return ret