Netbox Sync BGP Peerings Task

task api name: sync_bgp_peerings

Synchronises BGP sessions between live network devices and the NetBox BGP plugin. Supports idempotent create, update, and optional deletion of stale sessions.

How It Works

Data collection — NetBox worker fetches every existing BGP session directly from the NetBox BGP plugin so sessions with duplicate names remain distinct. In parallel, Nornir service parses BGP neighbor data from devices using parse_ttp with get="bgp_neighbors".

Normalisation and diff — Both datasets are normalised to flat comparable dictionaries keyed by the five-part identity (device, local address, local ASN, remote address, remote ASN). Session names remain updateable attributes. DeepDiff classifies each session as create, update, delete, or in_sync.

NetBox writes — Sessions are created, updated, or (when process_deletions=True) deleted in NetBox. IP addresses and ASNs are resolved from IPAM or created on-demand. On create, all related objects (peer groups, routing policies, prefix lists) are resolved or created by name.

Dry run — When dry_run=True no writes occur; the raw diff report is returned.

1. Client submits sync_bgp_peerings request to NetBox worker
2. NetBox worker fetches existing BGP sessions from NetBox BGP plugin
3. NetBox worker requests Nornir service to parse BGP neighbor data from devices
4. Nornir fetches and parses BGP state from the network
5. Nornir returns parsed BGP session data to NetBox worker
6. NetBox worker computes diff, then writes creates/updates/deletes to NetBox

With Approval - Pass with_approval=True to use interactive NFCLI workflow. Sync task displays its preview, and waits for approval before applying changes. Declining at that point will return dry-run result.

Note

When both dry-run and with_approval are True, dry-run logic ignored.

Prerequisites

• NetBox BGP plugin (netbox-bgp) must be installed and enabled on the NetBox instance.
• Nornir service must have a TTP getter that handles get="bgp_neighbors".

Session Naming Convention

By default, NetBox session names follow:

{device}_{name}

For example, device ceos-leaf-1 with a parsed session to-spine-1 becomes ceos-leaf-1_to-spine-1.

Use the name_template parameter to customise the naming scheme. The template is a Python format string with the following variables:

Variable	Description
device	Device name (e.g. ceos-leaf-1)
name	Parsed session name
description	Session description
local_address	Local IP address string
local_as	Local AS number string
remote_address	Remote IP address string
remote_as	Remote AS number string
vrf	VRF name or None
state	Device-reported state (e.g. established)
peer_group	Peer group name or None
import_policies	List of import policy names or None
export_policies	List of export policy names or None
prefix_list_in	Inbound prefix list name or None
prefix_list_out	Outbound prefix list name or None

Example:

name_template="{device}_BGP_{name}"
# ceos-leaf-1_BGP_to-spine-1

Execution Modes

Dry-run mode (dry_run=True) returns the diff without any NetBox writes:

{
    "<device>": {
        "create":  ["<session_name>", ...],
        "delete":  ["<session_name>", ...],
        "update":  {
            "<session_name>": {
                "<field>": {"old_value": ..., "new_value": ...},
            },
        },
        "in_sync": ["<session_name>", ...],
    }
}

With Approval — Pass with_approval=True to use interactive NFCLI workflow. Sync task displays its preview, and waits for approval before applying changes. Declining at that point will return dry-run result.

Note

When both dry-run and with_approval are True, dry-run logic ignored.

Live-run mode (dry_run=False, default) applies changes and returns the same structure.

Output

Existing sessions are matched by device, local address and AS, and remote address and AS. Updates use their NetBox IDs, including when a session has no name. An unnamed matched session receives the rendered name_template name; dry-run and live update results show that rendered name.

Dry-run mode returns what would happen without writing to NetBox.

Live-run mode applies changes and returns the same structure showing what was done.

Description Preservation

preserve_description controls descriptions on sessions that already exist in NetBox. Its default value, None, keeps the NetBox description when the live description is empty and otherwise uses the live value. Set it to True to always keep the NetBox description. Set it to False to always use the live description, including an empty string. Newly created sessions always use the live description.

Filtering

Three optional filters narrow which live sessions are considered during sync. Existing NetBox sessions remain in the five-tuple-keyed target dataset so a matching live session is updated instead of incorrectly treated as a new session when filtered attributes differ.

Parameter	Match logic	Applied to
filter_by_remote_as	Exact match — live session's remote AS must equal one of the provided integer values	Live only
filter_by_peer_group	Exact match — live session's peer group name must equal one of the provided values	Live only
filter_by_description	Glob match — live session's description must match the provided glob pattern (e.g. *uplink*)	Live only

Multiple filters may be combined; all must pass (AND logic). filter_by_description uses Python fnmatch glob syntax — * matches any sequence of characters, ? matches a single character.

Note

Because filters apply only to live data, existing NetBox sessions without a matching filtered live five-tuple appear as deletion candidates. They are deleted only when process_deletions=True.

Deletion Behavior

Default process_deletions=False — sessions present in NetBox but absent on the device are left untouched.

Set process_deletions=True to delete stale sessions. Only sessions for the explicitly targeted devices are considered.

Warning

Anything the TTP getter does not return (parser gap, unreachable device) will be deleted when process_deletions=True. Use dry-run to check before break.

Branching Support

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

Bulk Request Batching

List-based NetBox writes are sent as sequential requests containing at most batch_size objects. The default is 1000; set any integer greater than zero to tune the request size. Each batch emits matching progress event and log messages. If a request fails, the task stops and returns the results recorded for earlier successful batches.

Examples

CLIPython

Sync BGP sessions for a list of devices:

nf#netbox sync bgp-peerings devices ceos-leaf-1 ceos-leaf-2 rir lab

Preview changes without writing to NetBox (dry run):

nf#netbox sync bgp-peerings devices ceos-leaf-1 dry-run

Sync and delete stale sessions no longer present on devices:

nf#netbox sync bgp-peerings devices ceos-leaf-1 ceos-leaf-2 rir lab process-deletions

Use a custom session naming template:

nf#netbox sync bgp-peerings devices ceos-leaf-1 rir lab name-template "{device}_BGP_{name}"

Sync sessions into a NetBox branch:

nf#netbox sync bgp-peerings devices ceos-leaf-1 rir lab branch my-bgp-branch

Sync using Nornir host filters instead of explicit device names:

nf#netbox sync bgp-peerings rir lab FG spine-group

Sync only sessions with a specific remote AS:

nf#netbox sync bgp-peerings devices ceos-leaf-1 rir lab filter-by-remote-as 65001

Sync only sessions belonging to a peer group:

nf#netbox sync bgp-peerings devices ceos-leaf-1 rir lab filter-by-peer-group EBGP_PEERS

Sync only sessions whose description matches a glob pattern:

nf#netbox sync bgp-peerings devices ceos-leaf-1 rir lab filter-by-description "*spine uplink*"

from norfab.core.nfapi import NorFab

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

# sync BGP sessions for specific devices
result = client.run_job(
    "netbox",
    "sync_bgp_peerings",
    workers="any",
    kwargs={
        "devices": ["ceos-leaf-1", "ceos-leaf-2"],
        "rir": "lab",
    },
)

# dry run — preview creates/updates/deletes without writing
result = client.run_job(
    "netbox",
    "sync_bgp_peerings",
    workers="any",
    kwargs={
        "devices": ["ceos-leaf-1"],
        "rir": "lab",
        "dry_run": True,
    },
)

# sync and remove stale sessions
result = client.run_job(
    "netbox",
    "sync_bgp_peerings",
    workers="any",
    kwargs={
        "devices": ["ceos-leaf-1", "ceos-leaf-2"],
        "rir": "lab",
        "process_deletions": True,
    },
)

# custom session name template
result = client.run_job(
    "netbox",
    "sync_bgp_peerings",
    workers="any",
    kwargs={
        "devices": ["ceos-leaf-1"],
        "rir": "lab",
        "name_template": "{device}_BGP_{name}",
    },
)

# sync into a NetBox branch
result = client.run_job(
    "netbox",
    "sync_bgp_peerings",
    workers="any",
    kwargs={
        "devices": ["ceos-leaf-1"],
        "rir": "lab",
        "branch": "my-bgp-branch",
    },
)

# use Nornir host filters instead of explicit device names
result = client.run_job(
    "netbox",
    "sync_bgp_peerings",
    workers="any",
    kwargs={
        "rir": "lab",
        "FG": "spine-group",
    },
)

# sync only sessions with a specific remote AS
result = client.run_job(
    "netbox",
    "sync_bgp_peerings",
    workers="any",
    kwargs={
        "devices": ["ceos-leaf-1"],
        "rir": "lab",
        "filter_by_remote_as": [65001, 65002],
    },
)

# sync only sessions belonging to a peer group
result = client.run_job(
    "netbox",
    "sync_bgp_peerings",
    workers="any",
    kwargs={
        "devices": ["ceos-leaf-1"],
        "rir": "lab",
        "filter_by_peer_group": ["EBGP_PEERS"],
    },
)

# sync only sessions whose description matches a glob pattern
result = client.run_job(
    "netbox",
    "sync_bgp_peerings",
    workers="any",
    kwargs={
        "devices": ["ceos-leaf-1"],
        "rir": "lab",
        "filter_by_description": "*spine uplink*",
        "preserve_description": None,
    },
)

nf.destroy()

VRF Custom Field

VRF reference could be stored in a BGP session custom field, for that custom field must be configured as type Object in NetBox pointing to the VRF content-type. By default vrf_custom_field="vrf" means custom_fields["vrf"] is used for all reads and writes. The parameter is forwarded to both create_bgp_peering and update_bgp_peering internally so all normalisation, diff, and write operations use the same field consistently.

result = client.run_job(
    "netbox",
    "sync_bgp_peerings",
    workers="any",
    kwargs={
        "devices": ["ceos-leaf-1"],
        "rir": "lab",
        "vrf_custom_field": "tenant_vrf",  # BGP Sessions custom field -> Object-type VRF
    },
)

NORFAB Netbox Sync BGP Peerings Command Shell Reference

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

nf# man tree netbox.sync.bgp-peerings
root
└── netbox:    Netbox service
    └── sync:    Sync Netbox data
        └── bgp-peerings:    Sync BGP peering sessions
            ├── 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
            ├── devices:    List of device names to process
            ├── status:    Status for created/updated sessions, default 'active'
            ├── process-deletions:    Delete NetBox sessions absent on device, default 'False'
            ├── rir:    RIR name for ASN creation (e.g. 'RFC 1918')
            ├── message:    Changelog message for all NetBox write operations
            ├── name-template:    Template for BGP session names in NetBox, default '{device}_{name}'
            ├── filter-by-remote-as:    Only sync sessions with matching remote AS number(s)
            ├── filter-by-peer-group:    Only sync sessions with matching peer group name(s)
            ├── filter-by-description:    Only sync sessions whose description matches this glob pattern
            ├── preserve-description:    Preserve NetBox descriptions always (true), when live text is empty (null), or never (false)
            ├── vrf-custom-field:    BGP session field for VRF reference, default 'vrf'
            ├── FO:    Filter hosts using Filter Object
            ├── FB:    Filter hosts by name using Glob Patterns
            ├── FH:    Filter hosts by hostname
            ├── FC:    Filter hosts containment of pattern in name
            ├── FR:    Filter hosts by name using Regular Expressions
            ├── FG:    Filter hosts by group
            ├── FP:    Filter hosts by hostname using IP Prefix
            ├── FL:    Filter hosts by names list
            ├── FM:    Filter hosts by platform
            └── FN:    Negate the match
nf#

Python API Reference

Synchronize BGP sessions between live devices and NetBox.

Collects BGP session data from devices via Nornir parse_ttp with get="bgp_neighbors", compares against existing NetBox BGP sessions and creates, updates, or (optionally) deletes sessions in NetBox accordingly. Matched sessions are updated by NetBox ID, including sessions with no name.

Parameters:

Name	Type	Description	Default
job	Job	NorFab Job object.	required
instance	str	NetBox instance name.	None
devices	list	List of device names to process.	None
status	str	Status to assign to created/updated sessions when not established on device.	'active'
dry_run	bool	If True, return diff report without writing to NetBox.	False
with_approval	bool	Preview changes, ask for review, then apply them.	False
process_deletions	bool	If True, delete NetBox sessions not found on device.	False
timeout	int	Timeout in seconds for Nornir parse_ttp job.	600
branch	str	NetBox branching plugin branch name.	None
rir	str	RIR name to use when creating new ASNs in NetBox (e.g. RFC 1918, ARIN).	None
message	str	Changelog message recorded in NetBox for all create, update, and delete operations.	None
name_template	str	Jinja2 template string for BGP session names written to NetBox. The template context includes the following values: device — NetBox device object, rendered as device name by default remote_device — NetBox remote device object resolved via remote_address or None name — parsed session name - {vrf}_{remote_address} by default description — session description local_address — local IP address string (e.g. 10.0.0.1) local_as — local AS number string (e.g. 65100) remote_address — remote IP address string remote_as — remote AS number string vrf — VRF name or None state — device-reported state (e.g. established) peer_group — peer group name or None Default: "{{device}}_{{name}}".	'{{device}}_{{name}}'
filter_by_remote_as	list of int	Only include sessions whose remote AS number matches one of the provided integer values. Applied to both NetBox and live device sessions.	None
filter_by_peer_group	list	Only include sessions whose peer group name matches one of the provided values. Applied to both NetBox and live device sessions.	None
filter_by_description	str	Only include sessions whose description matches this glob pattern (e.g. '*uplink*'). Applied to both NetBox and live device sessions.	None
ignore_peer_ranges	list	provide prefixes to ignore BGP peers	None
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'
preserve_description	bool	Description preservation policy. None preserves NetBox text when the live description is empty, True always preserves NetBox text, and False always uses live text.	None
**kwargs	object	Nornir host filters (e.g. FC, FL, FB).	{}

Returns:

Type	Description
Result	Normal run result keyed by device name:: { "": { "created": ["", ...], "updated": ["", ...], "deleted": ["", ...], "in_sync": ["", ...], } }
Result	Dry-run result keyed by device name:: { "": { "create": ["", ...], "delete": ["", ...], "update": {"": {"": {"old_value": ..., "new_value": ...}, ...}, ...}, "in_sync": ["", ...], } }

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

@Task(
    fastapi={"methods": ["POST"], "schema": NetboxFastApiArgs.model_json_schema()},
    input=SyncBgpPeeringsInput,
    output=SyncBgpPeeringsResult,
    mcp={
        "annotations": {
            "title": "Sync BGP Peerings",
            "readOnlyHint": False,
            "destructiveHint": True,
            "idempotentHint": True,
            "openWorldHint": True,
        }
    },
)
def sync_bgp_peerings(
    self,
    job: Job,
    instance: Union[None, str] = None,
    devices: Union[None, list] = None,
    status: str = "active",
    dry_run: bool = False,
    with_approval: bool = False,
    process_deletions: bool = False,
    timeout: int = 600,
    branch: str = None,
    rir: str = None,
    message: str = None,
    name_template: str = "{{device}}_{{name}}",
    filter_by_remote_as: Union[None, List[int]] = None,
    filter_by_peer_group: Union[None, list] = None,
    filter_by_description: Union[None, str] = None,
    ignore_peer_ranges: Union[None, list] = None,
    vrf_custom_field: str = "vrf",
    preserve_description: Union[None, bool] = None,
    batch_size: int = 1000,
    **kwargs: object,
) -> Result:
    """
    Synchronize BGP sessions between live devices and NetBox.

    Collects BGP session data from devices via Nornir ``parse_ttp`` with
    ``get="bgp_neighbors"``, compares against existing NetBox BGP sessions and
    creates, updates, or (optionally) deletes sessions in NetBox accordingly.
    Matched sessions are updated by NetBox ID, including sessions with no name.

    Args:
        job: NorFab Job object.
        instance (str, optional): NetBox instance name.
        devices (list, optional): List of device names to process.
        status (str): Status to assign to created/updated sessions when not ``established`` on device.
        dry_run (bool): If True, return diff report without writing to NetBox.
        with_approval (bool): Preview changes, ask for review, then apply them.
        process_deletions (bool): If True, delete NetBox sessions not found on device.
        timeout (int): Timeout in seconds for Nornir ``parse_ttp`` job.
        branch (str, optional): NetBox branching plugin branch name.
        rir (str, optional): RIR name to use when creating new ASNs in NetBox (e.g. ``RFC 1918``, ``ARIN``).
        message (str, optional): Changelog message recorded in NetBox for all create, update, and delete operations.
        name_template (str): Jinja2 template string for BGP session names written to NetBox.
            The template context includes the following values:

            - ``device`` — NetBox device object, rendered as device name by default
            - ``remote_device`` — NetBox remote device object resolved via remote_address or None
            - ``name`` — parsed session name - ``{vrf}_{remote_address}`` by default
            - ``description`` — session description
            - ``local_address`` — local IP address string (e.g. ``10.0.0.1``)
            - ``local_as`` — local AS number string (e.g. ``65100``)
            - ``remote_address`` — remote IP address string
            - ``remote_as`` — remote AS number string
            - ``vrf`` — VRF name or ``None``
            - ``state`` — device-reported state (e.g. ``established``)
            - ``peer_group`` — peer group name or ``None``

            Default: ``"{{device}}_{{name}}"``.
        filter_by_remote_as (list of int, optional): Only include sessions whose remote AS number
            matches one of the provided integer values. Applied to both NetBox and live device sessions.
        filter_by_peer_group (list, optional): Only include sessions whose peer group name matches
            one of the provided values. Applied to both NetBox and live device sessions.
        filter_by_description (str, optional): Only include sessions whose description matches
            this glob pattern (e.g. ``'*uplink*'``). Applied to both NetBox and live device sessions.
        ignore_peer_ranges (list, optional): provide prefixes to ignore BGP peers
        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']``.
        preserve_description (bool, optional): Description preservation policy.
            ``None`` preserves NetBox text when the live description is empty,
            ``True`` always preserves NetBox text, and ``False`` always uses
            live text.
        **kwargs: Nornir host filters (e.g. ``FC``, ``FL``, ``FB``).

    Returns:
        Normal run result keyed by device name::

            {
                "<device>": {
                    "created": ["<session_name>", ...],
                    "updated": ["<session_name>", ...],
                    "deleted": ["<session_name>", ...],
                    "in_sync": ["<session_name>", ...],
                }
            }

        Dry-run result keyed by device name::

            {
                "<device>": {
                    "create": ["<session_name>", ...],
                    "delete": ["<session_name>", ...],
                    "update": {"<session_name>": {"<field>": {"old_value": ..., "new_value": ...}, ...}, ...},
                    "in_sync": ["<session_name>", ...],
                }
            }
    """
    instance = instance or self.default_instance
    devices = devices or []
    ret = Result(
        task=f"{self.name}:sync_bgp_peerings",
        result=SyncActionSummaryMap().model_dump(),
        resources=[instance],
    )

    # Normalised session dicts keyed by device name -> BGP identity tuple -> field values
    normalised_nb = {}
    normalised_live = {}
    lookup_cache: dict = {}
    # handle ranges
    ignore_peer_ranges = ignore_peer_ranges or [
        "127.0.0.0/8",
        "224.0.0.0/24",
        "fe80::/10",
        "ff02::/16",
    ]
    ignore_peer_nets = [
        ipaddress.ip_network(str(pfx), strict=False) for pfx in ignore_peer_ranges
    ]

    # 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

    # Validate VRF custom field
    job.event("validating BGP session VRF custom field")
    nb = self._get_pynetbox(instance, branch=branch, job=job)
    vrf_custom_field = _resolve_vrf_custom_field(
        vrf_custom_field, nb, job, self.name
    )

    # Source additional devices from Nornir filters
    if kwargs:
        job.event("resolving devices from Nornir filters")
        nornir_hosts = self.get_nornir_hosts(kwargs, timeout)
        for host in nornir_hosts:
            if host not in devices:
                devices.append(host)
        job.event(
            f"resolved {len(nornir_hosts)} device(s) from Nornir filters, "
            f"{len(devices)} total device(s) selected"
        )

    if not devices:
        msg = "no devices specified"
        job.event(msg, severity="ERROR")
        ret.errors.append(msg)
        ret.failed = True
        return ret

    log.info(
        f"{self.name} - Sync BGP peerings: processing {len(devices)} device(s) in '{instance}'"
    )
    job.event(
        f"syncing BGP peerings for {len(devices)} device(s) in '{instance}', dry_run={dry_run}"
    )

    # Fetch existing NetBox BGP sessions directly. get_bgp_peerings returns
    # name-keyed data and cannot represent multiple sessions with the same name.
    job.event(f"fetching BGP session data from NetBox for {len(devices)} device(s)")
    devices_result = self.get_devices(
        job=job, devices=devices, instance=instance, cache=False, branch=branch
    )
    if devices_result.errors:
        job.event("failed to fetch device data from NetBox", severity="ERROR")
        ret.errors.extend(devices_result.errors)
        ret.failed = True
        return ret

    # Report devices missing from NetBox while continuing with valid devices.
    valid_devices = set(devices_result.result)
    missing_devices = sorted(set(devices) - valid_devices)
    for device_name in missing_devices:
        msg = f"device '{device_name}' not found in NetBox"
        job.event(msg, severity="ERROR")
        log.error(msg)
        ret.errors.append(msg)

    devices = [
        device_name for device_name in devices if device_name in valid_devices
    ]
    if not devices:
        return ret

    nb_sessions = {device_name: [] for device_name in devices}
    device_names_by_id = {
        device_data["id"]: device_name
        for device_name, device_data in devices_result.result.items()
        if device_name in devices
    }
    if device_names_by_id:
        fetched_sessions = self.bulk_filter(
            nb.plugins.bgp.session,
            device_id=list(device_names_by_id),
        )
        for session in fetched_sessions:
            device_name = device_names_by_id.get(session.device.id)
            if device_name:
                nb_sessions[device_name].append(session)

    # Fetch live BGP data from devices via Nornir parse_ttp
    job.event(
        f"fetching BGP session data from {len(devices)} device(s) via Nornir parse_ttp"
    )
    parse_data = self.client.run_job(
        "nornir",
        "parse_ttp",
        kwargs={"get": "bgp_neighbors", "FL": devices},
        workers="all",
        timeout=timeout,
    )

    # Build NetBox BGP sessions normalised dicts per device
    job.event("normalising NetBox BGP session data")
    for device_name in devices:
        # Normalise NetBox sessions for this device
        normalised_nb[device_name] = {}
        for nb_session in nb_sessions.get(device_name, []):
            sname = nb_session.name
            try:
                normalised = normalise_nb_bgp_session(
                    dict(nb_session), vrf_custom_field=vrf_custom_field
                )
            except Exception as e:
                log.warning(
                    f"{self.name} - failed to normalise NetBox session '{sname}' for '{device_name}': {e}"
                )
                continue
            identity = bgp_session_identity(device_name, normalised)
            normalised_nb[device_name][identity] = normalised
    nb_session_count = sum(len(sessions) for sessions in normalised_nb.values())
    job.event(f"normalised {nb_session_count} NetBox BGP session(s)")

    # pre-seed lookup cache from fetched NetBox sessions and all parsed live
    # sessions before rendering live session names.
    sessions_for_cache = []
    for device_name, sessions in normalised_nb.items():
        for session_data in sessions.values():
            sessions_for_cache.append({"device": device_name, **session_data})
    for wdata in parse_data.values():
        if wdata.get("failed"):
            continue
        for device_name, host_sessions in wdata.get("result", {}).items():
            for s in host_sessions:
                sessions_for_cache.append(
                    {
                        "device": device_name,
                        "local_address": s.get("local_address"),
                        "local_as": s["local_as"],
                        "remote_address": s["remote_address"],
                        "remote_as": s["remote_as"],
                        "vrf": s.get("vrf"),
                    }
                )
    if sessions_for_cache:
        job.event(
            f"pre-seeding BGP lookup cache from {len(sessions_for_cache)} NetBox/live session(s)"
        )
        preseed_bgp_lookup_cache(
            nb, sessions_for_cache, lookup_cache, job, self.bulk_filter, self.name
        )

    # Normalize live parse data per device
    job.event("normalising live BGP session data")
    failed_devices = set()
    for wname, wdata in parse_data.items():
        resources_failed = wdata.get("resources_failed") or []
        if resources_failed:
            failed_devices.update(resources_failed)
            ret.resources_failed = sorted(
                set(ret.resources_failed) | set(resources_failed)
            )
            msg = (
                f"{wname} failed to fetch BGP session data from devices "
                f"{', '.join(sorted(resources_failed))}"
            )
            job.event(msg, severity="ERROR")
            log.error(f"{self.name} - {msg}")
            ret.errors.append(msg)
        if wdata.get("failed"):
            msg = f"{wname} - failed to parse BGP session data from devices"
            log.warning(msg)
            job.event(msg, severity="WARNING")
            continue
        for device_name, host_sessions in (wdata.get("result") or {}).items():
            normalised_live.setdefault(device_name, {})
            for s in host_sessions:
                parsed_name = s.get("name")
                description = s.get("description") or ""
                local_address = s.get("local_address")
                if local_address:
                    try:
                        ipaddress.ip_address(local_address)
                    except ValueError:
                        msg = (
                            f"{parsed_name or 'unknown'} - skipping, parsed local ip "
                            f"'{local_address}' is not a valid IP address"
                        )
                        log.warning(msg)
                        job.event(msg, severity="WARNING")
                        continue
                local_as = _normalise_bgp_identity_asn(s["local_as"])
                remote_address = s["remote_address"]
                remote_as = _normalise_bgp_identity_asn(s["remote_as"])
                peer_group = s.get("peer_group")
                session_data = {
                    "device": device_name,
                    "name": parsed_name,
                    "description": description,
                    "local_address": local_address,
                    "local_as": local_as,
                    "remote_address": remote_address,
                    "remote_as": remote_as,
                    "vrf": s.get("vrf"),
                    "status": (
                        "active" if s.get("state") == "established" else status
                    ),
                    "peer_group": peer_group,
                    "import_policies": s.get("import_policies"),
                    "export_policies": s.get("export_policies"),
                    "prefix_list_in": s.get("prefix_list_in"),
                    "prefix_list_out": s.get("prefix_list_out"),
                }
                # attempt to resolve local ip if empty
                if not local_address:
                    local_address = resolve_local_ip_via_peer(
                        device_name, remote_address, nb
                    )
                    if not local_address:
                        msg = (
                            f"{parsed_name or remote_address or 'unknown'} "
                            "- skipping, no local ip in parsed data, failed to resolve using peer ip"
                        )
                        log.error(msg)
                        job.event(msg, severity="ERROR")
                        continue
                    session_data["local_address"] = local_address
                    msg = (
                        f"{parsed_name or remote_address} - resolved local ip "
                        f"'{local_address}' using peer ip"
                    )
                    log.info(msg)
                    job.event(msg)
                try:
                    session_name = render_bgp_session_name(
                        name_template, session_data, lookup_cache
                    )
                except Exception as exc:
                    msg = (
                        f"failed to render name_template '{name_template}' for session "
                        f"'{parsed_name}' on '{device_name}': {exc}"
                    )
                    ret.errors.append(msg)
                    job.event(msg, severity="ERROR")
                    log.error(f"{self.name} - {msg}")
                    continue
                if not bgp_session_matches_filters(
                    session_data,
                    filter_by_remote_as=filter_by_remote_as,
                    filter_by_peer_group=filter_by_peer_group,
                    filter_by_description=filter_by_description,
                    ignore_peer_nets=ignore_peer_nets,
                ):
                    continue
                session_data["name"] = session_name
                identity = bgp_session_identity(device_name, session_data)
                session_data.pop("device", None)
                normalised_live[device_name][identity] = session_data
    live_session_count = sum(len(sessions) for sessions in normalised_live.values())
    job.event(
        f"normalised {live_session_count} live BGP session(s) after applying filters"
    )

    for device_name in failed_devices:
        normalised_nb.pop(device_name, None)

    for device_name, sessions in normalised_live.items():
        for identity, session in sessions.items():
            if identity in normalised_nb.get(device_name, {}):
                session["description"] = apply_description_policy(
                    session["description"],
                    normalised_nb[device_name][identity]["description"],
                    preserve_description,
                )

    # Single diff on the full normalised datasets
    job.event("calculating BGP session sync diff")
    identity_diff = self.make_diff(normalised_live, normalised_nb)
    full_diff = bgp_session_name_from_identity_diff(
        identity_diff, normalised_live, normalised_nb
    )
    create_count = sum(len(actions["create"]) for actions in identity_diff.values())
    update_count = sum(len(actions["update"]) for actions in identity_diff.values())
    delete_count = sum(len(actions["delete"]) for actions in identity_diff.values())
    in_sync_count = sum(
        len(actions["in_sync"]) for actions in identity_diff.values()
    )
    job.event(
        "BGP session sync diff complete: "
        f"{create_count} create, {update_count} update, "
        f"{delete_count} delete, {in_sync_count} in sync"
    )

    ret.diff = full_diff
    ret.result = {
        device_name: SyncActionSummary(in_sync=actions["in_sync"]).model_dump()
        for device_name, actions in full_diff.items()
    }
    device_results = ret.result
    # Return dry-run results per device.
    if dry_run is True:
        job.event(
            "dry-run requested, returning BGP session sync diff without changes"
        )
        ret.result = full_diff
        ret.dry_run = True
        return ret
    if not sync_diff_has_changes(full_diff):
        job.event("no BGP peerings sync changes required")
        return ret
    if with_approval and not review_sync_task_result(
        job, "BGP peerings sync", full_diff
    ):
        ret.status = "skipped"
        ret.result = full_diff
        ret.dry_run = True
        ret.messages.append("review declined; changes were not applied")
        return ret

    # Build bulk_create list from full_diff — split pipe-separated policies to lists
    job.event("preparing BGP session create payloads")
    bulk_create = []
    for device_name, actions in identity_diff.items():
        for identity in actions["create"]:
            session_data = normalised_live[device_name][identity]
            bulk_create.append(
                {
                    "name": session_data.get("name"),
                    "device": device_name,
                    "description": session_data.get("description") or "",
                    "local_address": session_data.get("local_address"),
                    "local_as": session_data.get("local_as"),
                    "remote_address": session_data.get("remote_address"),
                    "remote_as": session_data.get("remote_as"),
                    "vrf": session_data.get("vrf"),
                    "status": session_data.get("status", "active"),
                    "peer_group": session_data.get("peer_group"),
                    "import_policies": session_data.get("import_policies"),
                    "export_policies": session_data.get("export_policies"),
                    "prefix_list_in": session_data.get("prefix_list_in"),
                    "prefix_list_out": session_data.get("prefix_list_out"),
                }
            )
    job.event(f"prepared {len(bulk_create)} BGP session create payload(s)")

    # Build bulk_update list from full_diff
    job.event("preparing BGP session update payloads")
    bulk_update = []
    for device_name, actions in identity_diff.items():
        for identity, field_changes in actions["update"].items():
            entry = {"id": normalised_nb[device_name][identity]["id"]}
            for field, change in field_changes.items():
                new_value = change["new_value"]
                if field == "name":
                    entry["new_name"] = new_value
                elif field in ("import_policies", "export_policies"):
                    entry[field] = new_value if new_value is not None else []
                elif field in ("prefix_list_in", "prefix_list_out"):
                    entry[field] = new_value if new_value else None
                else:
                    entry[field] = new_value
            bulk_update.append(entry)
    job.event(f"prepared {len(bulk_update)} BGP session update payload(s)")

    # Delegate writes to create_bgp_peering and update_bgp_peering
    if bulk_create:
        job.event(f"creating {len(bulk_create)} BGP session(s) from sync diff")
        create_result = self.create_bgp_peering(
            job=job,
            instance=instance,
            bulk_create=bulk_create,
            rir=rir,
            message=message,
            branch=branch,
            create_reverse=False,
            vrf_custom_field=vrf_custom_field,
            lookup_cache=lookup_cache,
            batch_size=batch_size,
        )
        ret.errors.extend(create_result.errors)
        created_names = create_result.result.get("created", [])
        for device_name, actions in full_diff.items():
            for sname in actions["create"]:
                if sname in created_names:
                    device_results[device_name]["created"].append(sname)
        if create_result.failed:
            ret.failed = True
            return ret
    else:
        job.event("no BGP sessions to create")

    if bulk_update:
        job.event(f"updating {len(bulk_update)} BGP session(s) from sync diff")
        update_result = self.update_bgp_peering(
            job=job,
            instance=instance,
            bulk_update=bulk_update,
            rir=rir,
            message=message,
            branch=branch,
            vrf_custom_field=vrf_custom_field,
            lookup_cache=lookup_cache,
            batch_size=batch_size,
        )
        ret.errors.extend(update_result.errors)
        updated_ids = set(update_result.result.get("updated", []))
        for device_name, actions in identity_diff.items():
            for identity in actions["update"]:
                old_session = normalised_nb[device_name][identity]
                old_name = old_session["name"]
                if old_session["id"] in updated_ids:
                    device_results[device_name]["updated"].append(
                        old_name or normalised_live[device_name][identity]["name"]
                    )
        if update_result.failed:
            ret.failed = True
            return ret
    else:
        job.event("no BGP sessions to update")

    # Deletion — batch-fetch all candidate sessions then delete individually
    if process_deletions:
        job.event("processing BGP session deletions")
        nb = self._get_pynetbox(instance, branch=branch, job=job)
        if message:
            nb.http_session.headers["X-Changelog-Message"] = message
        # Map session name → device name for all sessions to delete across devices
        all_deletions: Dict[str, str] = {}
        for device_name, actions in full_diff.items():
            for sname in actions["delete"]:
                all_deletions[sname] = device_name
        if all_deletions:
            job.event(f"fetching {len(all_deletions)} BGP session(s) to delete")
            sessions_to_delete = self.bulk_filter(
                nb.plugins.bgp.session,
                name=list(all_deletions),
                fields="id,name",
            )
            for session in sessions_to_delete:
                device_name = all_deletions[session.name]
                try:
                    session.delete()
                    device_results[device_name]["deleted"].append(session.name)
                    msg = (
                        f"deleted BGP session '{session.name}' for '{device_name}'"
                    )
                    job.event(msg)
                    log.info(f"{self.name} - {msg}")
                except Exception as e:
                    msg = f"failed to delete BGP session '{session.name}' for '{device_name}': {e}"
                    ret.errors.append(msg)
                    log.error(f"{self.name} - {msg}")
        else:
            job.event("no BGP sessions to delete")
    elif delete_count:
        job.event(
            f"skipping {delete_count} BGP session deletion(s), process_deletions=False"
        )
    else:
        job.event("no BGP sessions to delete")

    job.event("BGP peerings sync complete")

    return ret