Netbox Create IP Task¤

task api name: create_ip

Task to create next available IP from prefix or get existing IP address.

Netbox service create_ip task integrated with Nornir service and can be called using netbox.create_ip Jinja2 filter, allowing to allocate IP addresses in Netbox on the fly while rendering configuration templates.

Branching Support¤

Create IP task is branch aware and can create IP addresses within the branch. Netbox Branching Plugin need to be installed on Netbox instance.

NORFAB Netbox Create IP Command Shell Reference¤

NorFab shell supports these command options for Netbox create_ip task:

nf#man tree netbox.create.ip
root
└── netbox:    Netbox service
    └── create:    Create objects in Netbox
            ├── 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:    Do not commit to database
            ├── *prefix:    Prefix to allocate IP address from, can also provide prefix name or filters
            ├── device:    Device name to associate IP address with
            ├── interface:    Device interface name to associate IP address with
            ├── description:    IP address description
            ├── vrf:    VRF to associate with IP address
            ├── tags:    Tags to add to IP address
            ├── dns_name:    IP address DNS name
            ├── tenant:    Tenant name to associate with IP address
            ├── comments:    IP address comments field
            ├── role:    IP address functional role
            └── branch:    Branching plugin branch name to use
nf#

Python API Reference¤

Allocate the next available IP address from a given subnet.

.. warning::

**Dry-run limitation**: when ``dry_run=True``, ``mask_len`` is ignored and
the candidate IP address is allocated directly from the parent prefix using
the parent prefix mask length — no child subnet is created.

This task finds or creates an IP address in NetBox, updates its metadata, optionally links it to a device/interface, and supports a dry run mode for previewing changes.

Parameters:

Name	Type	Description	Default
`prefix`	`str`	The prefix from which to allocate the IP address, could be: IPv4 prefix string e.g. 10.0.0.0/24 IPv6 prefix string e.g. 2001::/64 Prefix description string to filter by Dictionary with prefix filters to feed `pynetbox` get method e.g. `{"prefix": "10.0.0.0/24", "site": "foo", "role": "bar"}`, site and role referred to by slugs not by their names.	required
`description`	`str`	A description for the allocated IP address.	`None`
`device`	`str`	The device associated with the IP address.	`None`
`interface`	`str`	The interface associated with the IP address.	`None`
`vrf`	`str`	The VRF (Virtual Routing and Forwarding) instance.	`None`
`tags`	`list`	A list of tags to associate with the IP address.	`None`
`dns_name`	`str`	The DNS name for the IP address.	`None`
`tenant`	`str`	The tenant associated with the IP address.	`None`
`comments`	`str`	Additional comments for the IP address.	`None`
`instance`	`str`	The NetBox instance to use.	`None`
`dry_run`	`bool`	If True, do not actually allocate the IP address.	`False`
`branch`	`str`	Branch name to use, need to have branching plugin installed, automatically creates branch if it does not exist in Netbox.	`None`
`mask_len`	`int`	mask length to use for IP address on creation or to update existing IP address. On new IP address creation will create child subnet of `mask_len` within parent `prefix`, new subnet not created for existing IP addresses. Ignored when `dry_run=True`.	`None`
`create_peer_ip`	`bool`	If True creates IP address for link peer - remote device interface connected to requested device and interface	`True`

Returns:

Name	Type	Description
`dict`	`Result`	A dictionary containing the result of the IP allocation.

Tasks execution follow these steps:

Tries to find an existing IP in NetBox matching the device/interface/description. If found, uses it; otherwise, proceeds to create a new IP.
If prefix is a string, determines if it's an IP network or a description. Builds a filter dictionary for NetBox queries, optionally including VRF.
Queries NetBox for the prefix using the constructed filter.
If dry_run is True, fetches the next available IP but doesn't create it.
If not a dry run, creates the next available IP in the prefix.
Updates IP attributes (description, VRF, tenant, DNS name, comments, role, tags) if provided and different from current values. Handles interface assignment and can set the IP as primary for the device.
If changes were made and not a dry run, saves the IP and device updates to NetBox.

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

@Task(
    fastapi={"methods": ["POST"], "schema": NetboxFastApiArgs.model_json_schema()},
    input=CreateIpInput,
    output=CreateIpResult,
    mcp={
        "annotations": {
            "title": "Create IP Address",
            "readOnlyHint": False,
            "destructiveHint": False,
            "idempotentHint": False,
            "openWorldHint": True,
        }
    },
)
def create_ip(
    self,
    job: Job,
    prefix: Union[str, dict],
    device: Union[None, str] = None,
    interface: Union[None, str] = None,
    description: Union[None, str] = None,
    vrf: Union[None, str] = None,
    tags: Union[None, list] = None,
    dns_name: Union[None, str] = None,
    tenant: Union[None, str] = None,
    comments: Union[None, str] = None,
    role: Union[None, str] = None,
    status: Union[None, str] = None,
    is_primary: Union[None, bool] = None,
    instance: Union[None, str] = None,
    dry_run: Union[None, bool] = False,
    branch: Union[None, str] = None,
    mask_len: Union[None, int] = None,
    create_peer_ip: Union[None, bool] = True,
) -> Result:
    """
    Allocate the next available IP address from a given subnet.

    .. warning::

        **Dry-run limitation**: when ``dry_run=True``, ``mask_len`` is ignored and
        the candidate IP address is allocated directly from the parent prefix using
        the parent prefix mask length — no child subnet is created.

    This task finds or creates an IP address in NetBox, updates its metadata,
    optionally links it to a device/interface, and supports a dry run mode for
    previewing changes.

    Args:
        prefix (str): The prefix from which to allocate the IP address, could be:

            - IPv4 prefix string e.g. 10.0.0.0/24
            - IPv6 prefix string e.g. 2001::/64
            - Prefix description string to filter by
            - Dictionary with prefix filters to feed `pynetbox` get method
                e.g. `{"prefix": "10.0.0.0/24", "site": "foo", "role": "bar"}`,
                site and role referred to by slugs not by their names.

        description (str, optional): A description for the allocated IP address.
        device (str, optional): The device associated with the IP address.
        interface (str, optional): The interface associated with the IP address.
        vrf (str, optional): The VRF (Virtual Routing and Forwarding) instance.
        tags (list, optional): A list of tags to associate with the IP address.
        dns_name (str, optional): The DNS name for the IP address.
        tenant (str, optional): The tenant associated with the IP address.
        comments (str, optional): Additional comments for the IP address.
        instance (str, optional): The NetBox instance to use.
        dry_run (bool, optional): If True, do not actually allocate the IP address.
        branch (str, optional): Branch name to use, need to have branching plugin
            installed, automatically creates branch if it does not exist in Netbox.
        mask_len (int, optional): mask length to use for IP address on creation or to
            update existing IP address. On new IP address creation will create child
            subnet of `mask_len` within parent `prefix`, new subnet not created for
            existing IP addresses. Ignored when ``dry_run=True``.
        create_peer_ip (bool, optional): If True creates IP address for link peer -
            remote device interface connected to requested device and interface

    Returns:
        dict: A dictionary containing the result of the IP allocation.

    Tasks execution follow these steps:

    1. Tries to find an existing IP in NetBox matching the device/interface/description.
        If found, uses it; otherwise, proceeds to create a new IP.

    2. If prefix is a string, determines if it's an IP network or a description.
        Builds a filter dictionary for NetBox queries, optionally including VRF.

    3. Queries NetBox for the prefix using the constructed filter.

    4. If dry_run is True, fetches the next available IP but doesn't create it.

    5. If not a dry run, creates the next available IP in the prefix.

    6. Updates IP attributes (description, VRF, tenant, DNS name, comments, role, tags)
        if provided and different from current values. Handles interface assignment and
        can set the IP as primary for the device.

    7. If changes were made and not a dry run, saves the IP and device updates to NetBox.
    """
    instance = instance or self.default_instance
    log.info(
        f"{self.name} - Create IP: Allocating IP from '{prefix}' for '{device}:{interface}' in '{instance}' Netbox"
    )
    ret = Result(task=f"{self.name}:create_ip", result={}, resources=[instance])
    tags = tags or []
    has_changes = False
    nb_ip = None
    nb_device = None
    create_peer_ip_data = {}
    nb = self._get_pynetbox(instance, branch=branch)
    job.event(
        f"creating IP address in '{instance}' for "
        f"'{device}:{interface}' from prefix '{prefix}', dry_run={dry_run}"
    )

    # source parent prefix from Netbox
    if isinstance(prefix, str):
        # try converting prefix to network, if fails prefix is not an IP network
        try:
            _ = ipaddress.ip_network(prefix)
            is_network = True
        except Exception:
            is_network = False
        if is_network is True and vrf:
            prefix = {"prefix": prefix, "vrf__name": vrf}
        elif is_network is True:
            prefix = {"prefix": prefix}
        elif is_network is False and vrf:
            prefix = {"description": prefix, "vrf__name": vrf}
        elif is_network is False:
            prefix = {"description": prefix}
    nb_prefixes = list(nb.ipam.prefixes.filter(**prefix))
    if not nb_prefixes:
        raise NetboxAllocationError(
            f"Unable to source parent prefix from Netbox - {prefix}"
        )

    # try to source existing IP from netbox
    for nb_prefix in nb_prefixes:
        if device and interface and description:
            nb_ip = nb.ipam.ip_addresses.get(
                device=device,
                interface=interface,
                description=description,
                parent=str(nb_prefix),
            )
        elif device and interface:
            nb_ip = nb.ipam.ip_addresses.get(
                device=device, interface=interface, parent=str(nb_prefix)
            )
        elif description:
            nb_ip = nb.ipam.ip_addresses.get(
                description=description, parent=str(nb_prefix)
            )
        if nb_ip:
            break
    # if no existing IP found, find parent prefix that has available subnets to allocate
    else:
        for nb_prefix in nb_prefixes:
            parent_prefix_len = int(str(nb_prefix).split("/")[1])
            # check if can create child subnet in prefix
            if mask_len and mask_len != parent_prefix_len:
                try:
                    candidate = self.create_prefix(
                        job=job,
                        parent=str(nb_prefix),
                        prefixlen=mask_len,
                        instance=instance,
                        branch=branch,
                        dry_run=True,
                    )
                    if candidate.failed is False and candidate.result.get("prefix"):
                        break
                except Exceptions as e:
                    # error might be expected
                    log.debug(f"Error while trying to allocate child subnet: {e}")
                    continue
            elif nb_prefix.available_ips.list():
                break
        else:
            raise NetboxAllocationError(
                f"No subnets of {mask_len} lenght or IPs available in parent prefix - {prefix}"
            )

    # create new IP address
    if not nb_ip:
        # check if interface has link peer that has IP within parent prefix
        if device and interface:
            connection = self.get_connections(
                job=job,
                devices=[device],
                interface_regex=interface,
                instance=instance,
            )
            if interface in connection.result[device]:
                peer = connection.result[device][interface]
                # do not process breakout cables
                if isinstance(peer["remote_interface"], list):
                    peer["remote_interface"] = None
                # try to source peer ip subnet
                nb_peer_ip = None
                if peer["remote_device"] and peer["remote_interface"]:
                    nb_peer_ip = nb.ipam.ip_addresses.get(
                        device=peer["remote_device"],
                        interface=peer["remote_interface"],
                        parent=str(nb_prefix),
                    )
                # try to source peer ip subnet
                nb_peer_prefix = None
                if nb_peer_ip:
                    peer_ip = ipaddress.ip_interface(nb_peer_ip.address)
                    nb_peer_prefix = nb.ipam.prefixes.get(
                        prefix=str(peer_ip.network),
                        vrf__name=vrf,
                    )
                elif create_peer_ip and peer["remote_interface"]:
                    create_peer_ip_data = {
                        "device": peer["remote_device"],
                        "interface": peer["remote_interface"],
                        "vrf": vrf,
                        "branch": branch,
                        "tenant": tenant,
                        "dry_run": dry_run,
                        "tags": tags,
                        "status": status,
                        "create_peer_ip": False,
                        "instance": instance,
                        "mask_len": mask_len,
                    }
                # use peer subnet to create IP address
                if nb_peer_prefix:
                    nb_prefix = nb_peer_prefix
                    mask_len = None  # cancel subnet creation
                    job.event(
                        f"using link peer '{peer['remote_device']}:{peer['remote_interface']}' "
                        f"prefix '{nb_peer_prefix}' to create IP address"
                    )
        # if mask_len provided create new subnet
        if mask_len and not dry_run and mask_len != parent_prefix_len:
            if mask_len < parent_prefix_len:
                raise ValueError(
                    f"Mask length '{mask_len}' must be longer then '{parent_prefix_len}' prefix length"
                )
            prefix_status = status
            if prefix_status not in ["active", "reserved", "deprecated"]:
                prefix_status = None
            # find next available subnet with available IPs
            while not job.is_timed_out():
                child_subnet = self.create_prefix(
                    job=job,
                    parent=str(nb_prefix),
                    prefixlen=mask_len,
                    vrf=vrf,
                    tags=tags,
                    tenant=tenant,
                    status=prefix_status,
                    instance=instance,
                    branch=branch,
                )
                if child_subnet.failed is True:
                    raise NetboxAllocationError(
                        f"Unable to create child subnet of mask length "
                        f"'{mask_len}' inside '{prefix}' parent prefix"
                    )

                child_prefix = {"prefix": child_subnet.result["prefix"]}
                if vrf:
                    child_prefix["vrf__name"] = vrf
                nb_child_prefix = nb.ipam.prefixes.get(**child_prefix)

                if not nb_child_prefix:
                    raise NetboxAllocationError(
                        f"Unable to source child prefix of mask length "
                        f"'{mask_len}' from '{child_prefix}' parent prefix"
                    )

                # check how many IPs available in child prefix
                available_ip_count = len(nb_child_prefix.available_ips.list())
                if not available_ip_count:
                    msg = f"{nb_child_prefix} - created child prefix but it has no available IPs, continue"
                    log.info(msg)
                    job.event(msg)
                elif create_peer_ip and available_ip_count < 2:
                    msg = f"{nb_child_prefix} - created child prefix but it has less than 2 available IPs, create peer ip is True, need two IPs, continue"
                    log.info(msg)
                    job.event(msg)
                else:
                    nb_prefix = nb_child_prefix
                    break
            else:
                raise NetboxAllocationError(
                    f"Unable to find a child subnet of mask length '{mask_len}' "
                    f"with sufficient available IPs inside '{nb_prefix}'"
                )
        # execute dry run on new IP
        if dry_run is True:
            nb_ip = nb_prefix.available_ips.list()[0]
            ret.status = "unchanged"
            ret.dry_run = True
            ret.result = {
                "address": str(nb_ip),
                "description": description,
                "vrf": vrf,
                "device": device,
                "interface": interface,
            }
            # add branch to results
            if branch is not None:
                ret.result["branch"] = branch
            job.event(f"dry-run: would create IP address '{nb_ip}'")
            return ret
        # create new IP
        else:
            nb_ip = nb_prefix.available_ips.create()
            job.event(
                f"created '{nb_ip}' IP address for '{device}:{interface}' within '{nb_prefix}' prefix"
            )
        ret.status = "created"
    else:
        job.event(f"using existing IP address {nb_ip}")
        ret.status = "updated"

    # update IP address parameters
    if description and description != nb_ip.description:
        nb_ip.description = description
        has_changes = True
    if vrf and vrf != nb_ip.vrf:
        nb_ip.vrf = {"name": vrf}
        has_changes = True
    if tenant and tenant != nb_ip.tenant:
        nb_ip.tenant = {"name": tenant}
        has_changes = True
    if dns_name and dns_name != nb_ip.dns_name:
        nb_ip.dns_name = dns_name
        has_changes = True
    if comments and comments != nb_ip.comments:
        nb_ip.comments = comments
        has_changes = True
    if role and role != nb_ip.role:
        nb_ip.role = role
        has_changes = True
    if tags and not any(t in nb_ip.tags for t in tags):
        for t in tags:
            if t not in nb_ip.tags:
                nb_ip.tags.append({"name": t})
                has_changes = True
    if device and interface:
        nb_interface = nb.dcim.interfaces.get(device=device, name=interface)
        if not nb_interface:
            raise NetboxAllocationError(
                f"Unable to source '{device}:{interface}' interface from Netbox"
            )
        if not nb_ip.assigned_object or (
            nb_ip.assigned_object.id != nb_interface.id
        ):
            nb_ip.assigned_object_id = nb_interface.id
            nb_ip.assigned_object_type = "dcim.interface"
            if is_primary is not None:
                nb_device = nb.dcim.devices.get(name=device)
                nb_device.primary_ip4 = nb_ip.id
            has_changes = True
            log.info(
                f"{device}:{nb_interface} - association {nb_ip} IP address with interface"
            )
    if mask_len and not str(nb_ip).endswith(f"/{mask_len}"):
        address = str(nb_ip).split("/")[0]
        nb_ip.address = f"{address}/{mask_len}"
        has_changes = True

    # save IP address into Netbox
    if dry_run is True:
        ret.status = "unchanged"
        ret.dry_run = True
    elif has_changes:
        nb_ip.save()
        job.event(f"updated '{str(nb_ip)}' IP address parameters")
        # make IP primary for device
        if is_primary is True and nb_device:
            nb_device.save()
    else:
        ret.status = "unchanged"

    # form and return results
    ret.result = {
        "address": str(nb_ip),
        "description": str(nb_ip.description),
        "vrf": str(nb_ip.vrf) if not vrf else nb_ip.vrf["name"],
        "device": device,
        "interface": interface,
    }
    # add branch to results
    if branch is not None:
        ret.result["branch"] = branch

    # create IP address for peer
    if create_peer_ip and create_peer_ip_data:
        job.event(
            f"creating IP address for link peer '{create_peer_ip_data['device']}:{create_peer_ip_data['interface']}'"
        )
        peer_ip = self.create_ip(
            **create_peer_ip_data, prefix=str(nb_prefix), job=job
        )
        if peer_ip.failed == False:
            ret.result["peer"] = peer_ip.result

    job.event(f"IP address task complete for '{str(nb_ip)}'")
    return ret