Netbox Update BGP Peering Task

task api name: update_bgp_peering

Updates one or many existing BGP sessions in NetBox. Supports single-session mode (name plus field kwargs) and bulk mode (bulk_update list of dicts). Only non-None fields are updated. Idempotency is enforced via DeepDiff: sessions with no effective changes are reported in in_sync and no write is performed.

How it Works

1. Client submits update_bgp_peering request to NetBox worker
2. NetBox worker validates that the BGP plugin is installed
3. Worker resolves the RIR ID once (when rir is provided) for on-demand ASN creation
4. For each session spec, the current session is fetched from NetBox by name
5. The proposed field values are compared against current NetBox values using DeepDiff
6. If there are no differences, the session is added to in_sync — no write is performed
7. In dry-run mode — the diff is returned without writing
8. Otherwise, the changed fields are resolved (IPs, ASNs, policies) and the session is updated

Prerequisites

• NetBox BGP plugin (netbox-bgp) must be installed and enabled on the NetBox instance.
• The session must already exist in NetBox (use create_bgp_peering to create new sessions).

Branching Support

update_bgp_peering is branch-aware. Pass branch=<name> to write all changes into a NetBox Branching Plugin branch instead of main.

Dry Run Mode

dry_run=True returns the diff without any NetBox writes:

{
    "update": [
        {"name": "<session_name>", "diff": { ... }},
        ...
    ],
    "in_sync": ["<session_name>", ...],
}

Sessions with no effective changes appear in in_sync regardless of dry-run mode.

VRF Custom Field

The VRF reference is always stored in a BGP session custom field that must be configured as type Object in NetBox pointing to the VRF content-type. By default vrf_custom_field="vrf" means the VRF object reference is read from and written to custom_fields["vrf"]. This must match the value used when the session was created.

result = client.run_job(
    "netbox",
    "update_bgp_peering",
    workers="any",
    kwargs={
        "name": "ceos-spine-1_10.0.0.1_10.0.0.2",
        "vrf": "NEW_VRF",
        "vrf_custom_field": "tenant_vrf",  # Object-type custom field -> VRF
    },
)

Examples

CLIPython

Update a single BGP session description:

nf#netbox update bgp-peering name ceos-spine-1_10.0.0.1_10.0.0.2 description "updated description"

Update status:

nf#netbox update bgp-peering name ceos-spine-1_10.0.0.1_10.0.0.2 status planned

Dry run — preview diff without writing:

nf#netbox update bgp-peering name ceos-spine-1_10.0.0.1_10.0.0.2 description "new desc" dry-run

Update session in a NetBox branch:

nf#netbox update bgp-peering name ceos-spine-1_10.0.0.1_10.0.0.2 description "branch update" branch my-bgp-branch

from norfab.core.nfapi import NorFab

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

# update a single BGP session
result = client.run_job(
    "netbox",
    "update_bgp_peering",
    workers="any",
    kwargs={
        "name": "ceos-spine-1_10.0.0.1_10.0.0.2",
        "description": "updated description",
        "status": "active",
    },
)

# dry run — preview diff without writing
result = client.run_job(
    "netbox",
    "update_bgp_peering",
    workers="any",
    kwargs={
        "name": "ceos-spine-1_10.0.0.1_10.0.0.2",
        "description": "new description",
        "dry_run": True,
    },
)

# bulk update
result = client.run_job(
    "netbox",
    "update_bgp_peering",
    workers="any",
    kwargs={
        "bulk_update": [
            {
                "name": "ceos-spine-1_10.0.0.1_10.0.0.2",
                "description": "spine-1 session",
                "status": "active",
            },
            {
                "name": "ceos-spine-2_10.0.0.3_10.0.0.4",
                "description": "spine-2 session",
                "import_policies": ["IMPORT_FROM_LEAF"],
            },
        ],
    },
)

# update into a NetBox branch
result = client.run_job(
    "netbox",
    "update_bgp_peering",
    workers="any",
    kwargs={
        "name": "ceos-spine-1_10.0.0.1_10.0.0.2",
        "description": "branch update",
        "branch": "my-bgp-branch",
    },
)

nf.destroy()

NORFAB Netbox Update BGP Peering Command Shell Reference

NorFab shell supports these command options for the Netbox update_bgp_peering task:

nf# man tree netbox.update.bgp-peering
root
└── netbox:    Netbox service
    └── update:    Update Netbox objects
        └── bgp-peering:    Update BGP peering session(s)
            ├── 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:    Return diff without writing to NetBox
            ├── name:    Existing session name (single mode)
            ├── description:    New description value
            ├── status:    New status value
            ├── local-address:    New local IP address string
            ├── remote-address:    New remote IP address string
            ├── local-as:    New local AS number string
            ├── remote-as:    New remote AS number string
            ├── vrf:    New VRF name
            ├── peer-group:    New peer group name
            ├── import-policies:    New import routing policy names
            ├── export-policies:    New export routing policy names
            ├── prefix-list-in:    New inbound prefix list name
            ├── prefix-list-out:    New outbound prefix list name
            ├── bulk-update:    JSON list of session update dicts for bulk mode
            ├── rir:    RIR name for ASN creation
            ├── vrf-custom-field:    BGP session field for VRF reference, default 'vrf'
            └── message:    Changelog message for NetBox write operations
nf#

Python API Reference

Update one or many existing BGP sessions in NetBox.

Supports single-session mode (name plus field kwargs) and bulk mode (bulk_update list of dicts). Only non-None fields are updated. Idempotency is enforced: sessions with no effective changes are reported in in_sync and no write is performed.

Parameters:

Name	Type	Description	Default
job	Job	NorFab Job object.	required
instance	str	NetBox instance name.	None
name	str	Existing session name. Required in single-session mode.	None
description	str	New description value.	None
status	str	New status value.	None
local_address	str	New local IP address string.	None
remote_address	str	New remote IP address string.	None
local_as	str	New local AS number string.	None
remote_as	str	New remote AS number string.	None
vrf	str	New VRF name.	None
peer_group	str	New peer group name.	None
import_policies	list	New list of import routing-policy names.	None
export_policies	list	New list of export routing-policy names.	None
prefix_list_in	str	Inbound prefix-list name.	None
prefix_list_out	str	Outbound prefix-list name.	None
bulk_update	list	List of session update dicts for bulk mode.	None
rir	str	RIR name used when auto-creating ASNs.	None
message	str	Changelog message recorded on every NetBox write.	None
branch	str	NetBox branching plugin branch name.	None
dry_run	bool	When True return diff without writing.	False
vrf_custom_field	str	Name of the BGP session custom field that stores the VRF object reference. The custom field must be of type Object in NetBox pointing to the VRF content-type. The value is always a single VRF object reference read from and written into custom_fields[vrf_custom_field]. Default 'vrf' means custom_fields['vrf'].	'vrf'

Returns:

Type	Description
Result	Normal run:: {"updated": ["name1", ...], "in_sync": ["name2", ...]}
Result	Dry run:: { "update": [{"name": "name1", "diff": {...}}, ...], "in_sync": ["name2", ...], }

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

@Task(
    fastapi={"methods": ["PATCH"], "schema": NetboxFastApiArgs.model_json_schema()},
    input=UpdateBgpPeeringInput,
    output=UpdateBgpPeeringResult,
    mcp={
        "annotations": {
            "title": "Update BGP Peering",
            "readOnlyHint": False,
            "destructiveHint": True,
            "idempotentHint": True,
            "openWorldHint": True,
        }
    },
)
def update_bgp_peering(
    self,
    job: Job,
    # single-session mode
    name: Union[None, str] = None,
    description: Union[None, str] = None,
    status: Union[None, str] = None,
    local_address: Union[None, str] = None,
    remote_address: Union[None, str] = None,
    local_as: Union[None, int] = None,
    remote_as: Union[None, int] = None,
    vrf: Union[None, str] = None,
    peer_group: Union[None, str] = None,
    import_policies: Union[None, list] = None,
    export_policies: Union[None, list] = None,
    prefix_list_in: Union[None, str] = None,
    prefix_list_out: Union[None, str] = None,
    # bulk mode
    bulk_update: Union[None, list] = None,
    # shared
    rir: Union[None, str] = None,
    message: Union[None, str] = None,
    branch: Union[None, str] = None,
    dry_run: bool = False,
    instance: Union[None, str] = None,
    vrf_custom_field: str = "vrf",
    lookup_cache: Union[None, dict] = None,
) -> Result:
    """
    Update one or many existing BGP sessions in NetBox.

    Supports single-session mode (``name`` plus field kwargs) and bulk mode
    (``bulk_update`` list of dicts).  Only non-None fields are updated.
    Idempotency is enforced: sessions with no effective changes are reported in
    ``in_sync`` and no write is performed.

    Args:
        job: NorFab Job object.
        instance (str, optional): NetBox instance name.
        name (str, optional): Existing session name. Required in single-session mode.
        description (str, optional): New description value.
        status (str, optional): New status value.
        local_address (str, optional): New local IP address string.
        remote_address (str, optional): New remote IP address string.
        local_as (str, optional): New local AS number string.
        remote_as (str, optional): New remote AS number string.
        vrf (str, optional): New VRF name.
        peer_group (str, optional): New peer group name.
        import_policies (list, optional): New list of import routing-policy names.
        export_policies (list, optional): New list of export routing-policy names.
        prefix_list_in (str, optional): Inbound prefix-list name.
        prefix_list_out (str, optional): Outbound prefix-list name.
        bulk_update (list, optional): List of session update dicts for bulk mode.
        rir (str, optional): RIR name used when auto-creating ASNs.
        message (str, optional): Changelog message recorded on every NetBox write.
        branch (str, optional): NetBox branching plugin branch name.
        dry_run (bool): When ``True`` return diff without writing.
        vrf_custom_field (str): Name of the BGP session custom field that stores
            the VRF object reference.  The custom field must be of type Object in
            NetBox pointing to the VRF content-type.  The value is always a single
            VRF object reference read from and written into
            ``custom_fields[vrf_custom_field]``.  Default ``'vrf'`` means
            ``custom_fields['vrf']``.

    Returns:
        Normal run::

            {"updated": ["name1", ...], "in_sync": ["name2", ...]}

        Dry run::

            {
                "update": [{"name": "name1", "diff": {...}}, ...],
                "in_sync": ["name2", ...],
            }
    """
    instance = instance or self.default_instance
    ret = Result(
        task=f"{self.name}:update_bgp_peering",
        result={},
        resources=[instance],
    )

    # Validate BGP plugin
    job.event(f"validating NetBox BGP plugin for '{instance}'")
    if not self.has_plugin("netbox_bgp", instance, strict=True):
        msg = f"'{instance}' NetBox instance has no BGP plugin installed"
        job.event(msg, severity="ERROR")
        ret.errors.append(msg)
        ret.failed = True
        return ret

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

    if message:
        job.event("setting NetBox changelog message for BGP session update")
        nb.http_session.headers["X-Changelog-Message"] = message

    lookup_cache = lookup_cache if lookup_cache is not None else {}

    # Validate VRF custom field and RIR
    job.event("validating BGP session VRF custom field and RIR")
    vrf_custom_field = _resolve_vrf_custom_field(
        vrf_custom_field, nb, job, self.name
    )
    rir_id = _resolve_rir_id(rir, nb, job, self.name, lookup_cache)

    # Build list of sessions to update
    if bulk_update is not None:
        bgp_sessions = bulk_update
    else:
        _single_fields = (
            "description",
            "status",
            "local_address",
            "remote_address",
            "local_as",
            "remote_as",
            "vrf",
            "peer_group",
            "import_policies",
            "export_policies",
            "prefix_list_in",
            "prefix_list_out",
        )
        bgp_session = {
            k: v
            for k, v in locals().items()
            if k in _single_fields and v is not None
        }
        bgp_sessions = [{"name": name, **bgp_session}]
    job.event(
        f"preparing {len(bgp_sessions)} BGP session update request(s), dry_run={dry_run}"
    )

    result = {"updated": [], "in_sync": []}
    if dry_run is True:
        ret.dry_run = True
        result = {"update": [], "in_sync": []}

    # Fetch all existing sessions in a single batch call
    session_names = [s["name"] for s in bgp_sessions]
    job.event(f"fetching {len(session_names)} BGP session(s) from NetBox")
    nb_sessions_raw = self.bulk_filter(
        nb.plugins.bgp.session,
        "name",
        session_names,
        fields="id,name,description,status,local_address,remote_address,local_as,remote_as,custom_fields,peer_group,import_policies,export_policies,prefix_list_in,prefix_list_out",
    )
    normalised_nb = {
        s.name: normalise_nb_bgp_session(dict(s), vrf_custom_field=vrf_custom_field)
        for s in nb_sessions_raw
    }
    job.event(f"retrieved {len(normalised_nb)} BGP session(s) from NetBox")

    # Build updates dictionary by session name
    job.event("normalising BGP session update data")
    normalised_updates = {}
    for bgp_session in bgp_sessions:
        sname = bgp_session["name"]
        # Report sessions not found in NetBox
        if sname not in normalised_nb:
            msg = f"BGP session '{sname}' not found in NetBox, skipping update"
            job.event(msg, severity="ERROR")
            log.error(f"{self.name} - {msg}")
            ret.errors.append(msg)
        else:
            # Normalise policies to list
            if isinstance(bgp_session.get("import_policies"), str):
                bgp_session["import_policies"] = [bgp_session["import_policies"]]
            if isinstance(bgp_session.get("export_policies"), str):
                bgp_session["export_policies"] = [bgp_session["export_policies"]]
            bgp_session_data = {
                k: v for k, v in bgp_session.items() if k != "new_name"
            }
            bgp_session_data["name"] = bgp_session.get(
                "new_name", bgp_session_data["name"]
            )
            normalised_updates[sname] = dict(bgp_session_data)
            if "import_policies" in bgp_session_data:
                normalised_updates[sname]["import_policies"] = sorted(
                    bgp_session_data.get("import_policies") or []
                )
            if "export_policies" in bgp_session_data:
                normalised_updates[sname]["export_policies"] = sorted(
                    bgp_session_data.get("export_policies") or []
                )

    # Compare complete dictionaries using make_diff; classify in_sync vs changed
    job.event("calculating BGP session update diff")
    sessions_diff = self.make_diff(
        {"_": normalised_updates},
        {"_": normalised_nb},
    )["_"]

    changed_snames = set(sessions_diff["update"].keys())
    result["in_sync"].extend(sessions_diff["in_sync"])
    job.event(
        f"BGP session update diff complete: {len(changed_snames)} update, "
        f"{len(sessions_diff['in_sync'])} in sync"
    )

    if dry_run is True:
        job.event(
            "dry-run requested, returning BGP session update diff without changes"
        )
        result["update"] = [
            {"name": sname, "diff": changes}
            for sname, changes in sessions_diff["update"].items()
        ]
        ret.result = result
        ret.dry_run = True
        job.event("BGP session update task complete")
        return ret

    # Build update payloads — iterate over diff to get only changed fields per session
    job.event("building BGP session update payloads")
    update_payloads = []
    for sname, field_changes in sessions_diff["update"].items():
        nb_session = normalised_nb[sname]
        addr_family = get_addr_family(nb_session["local_address"] or "0.0.0.0")
        payload = {"id": nb_session["id"]}
        payload.update(
            resolve_bgp_session_payload_fields(
                {f: c["new_value"] for f, c in field_changes.items()},
                nb,
                rir_id,
                job,
                ret,
                self.name,
                addr_family,
                lookup_cache=lookup_cache,
                vrf_custom_field=vrf_custom_field,
            )
        )
        update_payloads.append(payload)
    job.event(f"prepared {len(update_payloads)} BGP session update payload(s)")

    # Bulk update in a single pynetbox call
    if update_payloads:
        job.event(f"updating {len(update_payloads)} BGP session(s)")
        try:
            nb.plugins.bgp.session.update(update_payloads)
            result["updated"].extend(changed_snames)
            msg = f"updated {len(update_payloads)} BGP session(s)"
            job.event(msg)
            log.info(f"{self.name} - {msg}")
        except Exception as e:
            msg = f"failed to bulk update BGP sessions: {e}"
            job.event(msg, severity="ERROR")
            ret.errors.append(msg)
            log.error(f"{self.name} - {msg}")
    else:
        job.event("no BGP sessions to update")

    ret.result = result
    job.event("BGP session update task complete")
    return ret