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 existing BGP sessions from NetBox BGP plugin. 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 dicts keyed by session name convention {device_name}_{session_name}. 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

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".

Branching Support

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

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

Filtering Sessions

Three optional filters narrow which sessions are considered during sync. Filters apply to both the NetBox dataset and the live device dataset before the diff is computed. Sessions that do not match are silently excluded from creates, updates, and deletes.

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

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

When process_deletions=True, only sessions that pass the filters are candidates for deletion. Sessions excluded by a filter are never deleted.

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>", ...],
    }
}

Deletion Behaviour

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.

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

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
            ├── 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.

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
process_deletions	bool	If True, delete NetBox sessions not found on device.	False
timeout	int	Timeout in seconds for Nornir parse_ttp job.	60
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'
**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,
    process_deletions: bool = False,
    timeout: int = 60,
    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",
    **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.

    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.
        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']``.
        **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={},
        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)
    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
    job.event(f"fetching BGP session data from NetBox for {len(devices)} device(s)")
    nb_sessions_result = self.get_bgp_peerings(
        job=job, instance=instance, devices=devices, cache="refresh"
    )
    if nb_sessions_result.errors:
        job.event("failed to fetch BGP session data from NetBox", severity="ERROR")
        ret.errors.extend(nb_sessions_result.errors)
        ret.failed = True
        return ret

    # 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 sname, nb_session in nb_sessions_result.result.get(
            device_name, {}
        ).items():
            try:
                normalised = normalise_nb_bgp_session(
                    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
            if not bgp_session_matches_filters(
                normalised,
                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
            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) after applying filters"
    )

    # 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")
    for wname, wdata in parse_data.items():
        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")
                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"
    )

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

    # 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
    else:
        ret.diff = full_diff

    # Per-device result tracking
    device_results = {
        device_name: {
            "created": [],
            "updated": [],
            "deleted": [],
            "in_sync": actions["in_sync"],
        }
        for device_name, actions in full_diff.items()
    }

    # 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 = {"name": normalised_nb[device_name][identity]["name"]}
            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,
        )
        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)
    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,
        )
        ret.errors.extend(update_result.errors)
        updated_names = set(update_result.result.get("updated", []))
        for device_name, actions in full_diff.items():
            for sname in actions["update"]:
                if sname in updated_names:
                    device_results[device_name]["updated"].append(sname)
    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)
        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")

    ret.result = device_results
    job.event("BGP peerings sync complete")

    return ret