NetBox Get BGP Peerings Task

task api name: get_bgp_peerings

This task integrates with the NetBox BGP Plugin and fetches BGP peerings for devices.

Inputs

Parameter	Required	Description
devices	Yes	Device names to retrieve BGP peerings for
instance	No	NetBox instance name to target
branch	No	NetBox Branching plugin branch name to read from
dry_run	No	Return query content without running it
cache	No	Cache usage mode: True, False, refresh, or force

How It Works

• Requires NetBox BGP plugin: The worker verifies the plugin is installed on the target instance before proceeding.
• Resolves device IDs: It calls get_devices() to map provided device names to NetBox device IDs for accurate API queries.
• Fetches sessions via REST: Uses pynetbox plugins.bgp.session.filter(device_id=...) to retrieve sessions per device.
• Returns structured data: The result is a dictionary keyed by device name; each device contains a dictionary keyed by BGP session name, with the full session dict as value.
• Smart caching: Per-device cache key get_bgp_peerings::<device> is used. Modes:
• True: Uses cache when up-to-date; performs smart update by comparing last_updated and fetching only changed/new sessions.
• False: Bypasses cache entirely and does not write to cache.
• refresh: Forces re-fetch from NetBox and overwrites cache.
• force: Returns cached data if present without freshness checks.

Output

Returns BGP session data keyed by device name and session name:

{
    "fceos4": {
        "fceos4-fceos5-eth105": {
            "name": "fceos4-fceos5-eth105",
            "description": "BGP peering between fceos4 and fceos5 on eth105",
            "status": {"label": "Active", "value": "active"},
            "local_address": {"address": "10.0.2.1/30"},
            "remote_address": {"address": "10.0.2.2/30"},
            "local_as": {"asn": 65100},
            "remote_as": {"asn": 65101},
            "peer_group": {"name": "TEST_BGP_PEER_GROUP_1"},
            "...": "...",
        },
    },
}

Notes / Gotchas

• Supported and tested NetBox version is 4.4 and later.
• NetBox BGP plugin required: If missing, the task fails early with an error. Confirm plugin availability and version compatibility.
• Device name must exist: Unknown devices are skipped with warnings; verify names beforehand or use get_devices to inspect inventory.
• Session key uniqueness: Sessions in the result are keyed by name. If session names are not unique per device, later entries overwrite earlier ones.
• Partial-field queries: Smart update relies on fields="id,last_updated,name". Older NetBox versions may not support fields, which can affect cache comparison.
• Large datasets: Fetching many devices or sessions may be slow; prefer cache or limit devices for interactive runs.

Examples

CLIPython

Get BGP peerings for devices:

nf#netbox get bgp-peerings devices fceos4 fceos5

Refresh cached BGP peering data:

nf#netbox get bgp-peerings devices fceos4 cache refresh

Preview query content:

nf#netbox get bgp-peerings devices fceos4 dry-run

Context manager:

from norfab.core.nfapi import NorFab

with NorFab(inventory="./inventory.yaml") as nf:
    client = nf.make_client()

    result = client.run_job(
        "netbox",
        "get_bgp_peerings",
        workers="any",
        kwargs={
            "devices": ["fceos4", "fceos5"],
        },
    )

Direct lifecycle:

from norfab.core.nfapi import NorFab

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

    result = client.run_job(
        "netbox",
        "get_bgp_peerings",
        workers="any",
        kwargs={
            "devices": ["fceos4"],
            "cache": "refresh",
        },
    )
finally:
    nf.destroy()

NORFAB Netbox Get BGP Peerings Command Shell Reference

NorFab shell supports these command options for Netbox get_bgp_peerings task:

nf# man tree netbox.get.bgp-peerings
root
└── netbox:    Netbox service
    └── get:    Query data from Netbox
        └── bgp-peerings:    Query Netbox BGP Peerings data
            ├── timeout:    Job timeout
            ├── workers:    Filter worker to target, default 'any'
            ├── verbose-result:    Control output details, default 'False'
            ├── progress:    Display progress events, default 'True'
            ├── instance:    Netbox instance name to target
            ├── dry-run:    Only return query content, do not run it
            ├── branch:    Branching plugin branch name to use
            ├── *devices:    Device names to query data for
            └── cache:    How to use cache, default 'True'
nf#

Python API Reference

Retrieve device BGP peerings from NetBox using REST API.

Parameters:

Name	Type	Description	Default
job	Job	NorFab Job object containing relevant metadata	required
instance	str	NetBox instance name.	None
devices	list	List of devices to retrieve BGP peerings for.	None
cache	Union[bool, str]	Cache usage options: True: Use data stored in cache if it is up to date, refresh it otherwise. False: Do not use cache and do not update cache. refresh: Ignore data in cache and replace it with data fetched from NetBox. force: Use data in cache without checking if it is up to date.	None

Returns:

Name	Type	Description
dict	Result	Dictionary keyed by device name with BGP peerings details.

Example return data for single bgp peering - same data as returned by NetBox built-in REST API explorer:

"fceos4-fceos5-eth105": {
    "comments": "",
    "created": "2026-04-08T10:29:13.404863Z",
    "custom_fields": {},
    "description": "BGP peering between fceos4 and fceos5 on eth105",
    "device": {
        "description": "",
        "display": "fceos4 (UUID-123451)",
        "id": 119,
        "name": "fceos4",
        "url": "http://netbox.lab:8000/api/dcim/devices/119/"
    },
    "display": "fceos4 (UUID-123451):fceos4-fceos5-eth105",
    "export_policies": [
        {
            "description": "",
            "display": "RPL1",
            "id": 1,
            "name": "RPL1",
            "url": "http://netbox.lab:8000/api/plugins/bgp/routing-policy/1/"
        }
    ],
    "id": 7,
    "import_policies": [
        {
            "description": "",
            "display": "RPL1",
            "id": 1,
            "name": "RPL1",
            "url": "http://netbox.lab:8000/api/plugins/bgp/routing-policy/1/"
        }
    ],
    "last_updated": "2026-04-18T21:39:38.947199Z",
    "local_address": {
        "address": "10.0.2.1/30",
        "description": "",
        "display": "10.0.2.1/30",
        "family": {
            "label": "IPv4",
            "value": 4
        },
        "id": 489,
        "url": "http://netbox.lab:8000/api/ipam/ip-addresses/489/"
    },
    "local_as": {
        "asn": 65100,
        "description": "BGP ASN for fceos4",
        "display": "AS65100",
        "id": 3,
        "url": "http://netbox.lab:8000/api/ipam/asns/3/"
    },
    "name": "fceos4-fceos5-eth105",
    "peer_group": {
        "description": "Test BGP peer group 1 for standard peerings",
        "display": "TEST_BGP_PEER_GROUP_1",
        "id": 3,
        "name": "TEST_BGP_PEER_GROUP_1",
        "url": "http://netbox.lab:8000/api/plugins/bgp/bgppeergroup/3/"
    },
    "prefix_list_in": {
        "description": "",
        "display": "PFL1",
        "id": 1,
        "name": "PFL1",
        "url": "http://netbox.lab:8000/api/plugins/bgp/prefix-list/1/"
    },
    "prefix_list_out": {
        "description": "",
        "display": "PFL1",
        "id": 1,
        "name": "PFL1",
        "url": "http://netbox.lab:8000/api/plugins/bgp/prefix-list/1/"
    },
    "remote_address": {
        "address": "10.0.2.2/30",
        "description": "",
        "display": "10.0.2.2/30",
        "family": {
            "label": "IPv4",
            "value": 4
        },
        "id": 490,
        "url": "http://netbox.lab:8000/api/ipam/ip-addresses/490/"
    },
    "remote_as": {
        "asn": 65101,
        "description": "BGP ASN for fceos5",
        "display": "AS65101",
        "id": 4,
        "url": "http://netbox.lab:8000/api/ipam/asns/4/"
    },
    "site": {
        "description": "",
        "display": "SALTNORNIR-LAB",
        "id": 13,
        "name": "SALTNORNIR-LAB",
        "slug": "saltnornir-lab",
        "url": "http://netbox.lab:8000/api/dcim/sites/13/"
    },
    "status": {
        "label": "Active",
        "value": "active"
    },
    "tags": [],
    "tenant": {
        "description": "",
        "display": "SALTNORNIR",
        "id": 10,
        "name": "SALTNORNIR",
        "slug": "saltnornir",
        "url": "http://netbox.lab:8000/api/tenancy/tenants/10/"
    },
    "url": "http://netbox.lab:8000/api/plugins/bgp/bgpsession/7/",
    "virtualmachine": null
}

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

@Task(
    input=GetBgpPeeringsInput,
    output=GetBgpPeeringsResult,
    fastapi={"methods": ["GET"], "schema": NetboxFastApiArgs.model_json_schema()},
    mcp={
        "annotations": {
            "title": "Get BGP Peerings",
            "readOnlyHint": True,
            "destructiveHint": False,
            "idempotentHint": True,
            "openWorldHint": True,
        }
    },
)
def get_bgp_peerings(
    self,
    job: Job,
    instance: Union[None, str] = None,
    devices: Union[None, list] = None,
    cache: Union[None, bool, str] = None,
) -> Result:
    """
    Retrieve device BGP peerings from NetBox using REST API.

    Args:
        job: NorFab Job object containing relevant metadata
        instance (str, optional): NetBox instance name.
        devices (list, optional): List of devices to retrieve BGP peerings for.
        cache (Union[bool, str], optional): Cache usage options:

            - True: Use data stored in cache if it is up to date, refresh it otherwise.
            - False: Do not use cache and do not update cache.
            - refresh: Ignore data in cache and replace it with data fetched from NetBox.
            - force: Use data in cache without checking if it is up to date.

    Returns:
        dict: Dictionary keyed by device name with BGP peerings details.

    Example return data for single bgp peering - same data as returned by NetBox built-in REST
    API explorer:

    ```
    "fceos4-fceos5-eth105": {
        "comments": "",
        "created": "2026-04-08T10:29:13.404863Z",
        "custom_fields": {},
        "description": "BGP peering between fceos4 and fceos5 on eth105",
        "device": {
            "description": "",
            "display": "fceos4 (UUID-123451)",
            "id": 119,
            "name": "fceos4",
            "url": "http://netbox.lab:8000/api/dcim/devices/119/"
        },
        "display": "fceos4 (UUID-123451):fceos4-fceos5-eth105",
        "export_policies": [
            {
                "description": "",
                "display": "RPL1",
                "id": 1,
                "name": "RPL1",
                "url": "http://netbox.lab:8000/api/plugins/bgp/routing-policy/1/"
            }
        ],
        "id": 7,
        "import_policies": [
            {
                "description": "",
                "display": "RPL1",
                "id": 1,
                "name": "RPL1",
                "url": "http://netbox.lab:8000/api/plugins/bgp/routing-policy/1/"
            }
        ],
        "last_updated": "2026-04-18T21:39:38.947199Z",
        "local_address": {
            "address": "10.0.2.1/30",
            "description": "",
            "display": "10.0.2.1/30",
            "family": {
                "label": "IPv4",
                "value": 4
            },
            "id": 489,
            "url": "http://netbox.lab:8000/api/ipam/ip-addresses/489/"
        },
        "local_as": {
            "asn": 65100,
            "description": "BGP ASN for fceos4",
            "display": "AS65100",
            "id": 3,
            "url": "http://netbox.lab:8000/api/ipam/asns/3/"
        },
        "name": "fceos4-fceos5-eth105",
        "peer_group": {
            "description": "Test BGP peer group 1 for standard peerings",
            "display": "TEST_BGP_PEER_GROUP_1",
            "id": 3,
            "name": "TEST_BGP_PEER_GROUP_1",
            "url": "http://netbox.lab:8000/api/plugins/bgp/bgppeergroup/3/"
        },
        "prefix_list_in": {
            "description": "",
            "display": "PFL1",
            "id": 1,
            "name": "PFL1",
            "url": "http://netbox.lab:8000/api/plugins/bgp/prefix-list/1/"
        },
        "prefix_list_out": {
            "description": "",
            "display": "PFL1",
            "id": 1,
            "name": "PFL1",
            "url": "http://netbox.lab:8000/api/plugins/bgp/prefix-list/1/"
        },
        "remote_address": {
            "address": "10.0.2.2/30",
            "description": "",
            "display": "10.0.2.2/30",
            "family": {
                "label": "IPv4",
                "value": 4
            },
            "id": 490,
            "url": "http://netbox.lab:8000/api/ipam/ip-addresses/490/"
        },
        "remote_as": {
            "asn": 65101,
            "description": "BGP ASN for fceos5",
            "display": "AS65101",
            "id": 4,
            "url": "http://netbox.lab:8000/api/ipam/asns/4/"
        },
        "site": {
            "description": "",
            "display": "SALTNORNIR-LAB",
            "id": 13,
            "name": "SALTNORNIR-LAB",
            "slug": "saltnornir-lab",
            "url": "http://netbox.lab:8000/api/dcim/sites/13/"
        },
        "status": {
            "label": "Active",
            "value": "active"
        },
        "tags": [],
        "tenant": {
            "description": "",
            "display": "SALTNORNIR",
            "id": 10,
            "name": "SALTNORNIR",
            "slug": "saltnornir",
            "url": "http://netbox.lab:8000/api/tenancy/tenants/10/"
        },
        "url": "http://netbox.lab:8000/api/plugins/bgp/bgpsession/7/",
        "virtualmachine": null
    }
    ```
    """
    instance = instance or self.default_instance
    devices = devices or []
    log.info(
        f"{self.name} - Get BGP peerings: Fetching BGP peerings for {len(devices)} device(s) from '{instance}' NetBox"
    )
    cache = self.cache_use if cache is None else cache
    ret = Result(
        task=f"{self.name}:get_bgp_peerings",
        result={d: {} for d in devices},
        resources=[instance],
    )

    # Check if BGP plugin is installed
    if not self.has_plugin("netbox_bgp", instance, strict=True):
        ret.errors.append(f"{instance} NetBox instance has no BGP Plugin installed")
        ret.failed = True
        return ret

    self.cache.expire()

    # Get device details to collect device IDs
    devices_result = self.get_devices(
        job=job, devices=devices, instance=instance, cache=False
    )
    if devices_result.errors:
        ret.errors.append(
            f"Failed to retrieve device details: {devices_result.errors}"
        )
        return ret

    nb = self._get_pynetbox(instance)

    for device_name in devices:
        # Skip devices not found in NetBox
        if device_name not in devices_result.result:
            msg = f"Device '{device_name}' not found in NetBox"
            job.event(msg, resource=instance, severity="WARNING")
            log.warning(msg)
            ret.errors.append(msg)
            continue

        device_id = devices_result.result[device_name]["id"]
        cache_key = f"get_bgp_peerings::{device_name}"
        cached_data = self.cache.get(cache_key)

        # Mode: force with cached data - use cache directly
        if cache == "force" and cached_data is not None:
            ret.result[device_name] = cached_data
            job.event(
                f"using cached BGP peerings for '{device_name}' (forced)",
                resource=instance,
            )
            continue

        # Mode: cache disabled - fetch without caching
        if cache is False:
            bgp_sessions = self.bulk_filter(
                nb.plugins.bgp.session, device_id=device_id
            )
            ret.result[device_name] = {s.name: dict(s) for s in bgp_sessions}
            job.event(
                f"retrieved {len(ret.result[device_name])} BGP session(s) for '{device_name}'",
                resource=instance,
            )
            continue

        # Mode: refresh or no cached data - fetch and cache
        if cache == "refresh" or cached_data is None:
            if cache == "refresh" and cached_data is not None:
                self.cache.delete(cache_key, retry=True)
            bgp_sessions = self.bulk_filter(
                nb.plugins.bgp.session, device_id=device_id
            )
            ret.result[device_name] = {s.name: dict(s) for s in bgp_sessions}
            self.cache.set(
                cache_key, ret.result[device_name], expire=self.cache_ttl
            )
            job.event(
                f"fetched and cached {len(ret.result[device_name])} BGP session(s) for '{device_name}'",
                resource=instance,
            )
            continue

        # Mode: cache=True with cached data - smart update (only fetch changed sessions)
        ret.result[device_name] = dict(cached_data)
        job.event(
            f"retrieved {len(cached_data)} BGP session(s) from cache for '{device_name}'",
            resource=instance,
        )

        # Fetch brief session info to compare timestamps
        brief_sessions = self.bulk_filter(
            nb.plugins.bgp.session,
            device_id=device_id,
            fields="id,last_updated,name",
        )
        netbox_sessions = {
            s.id: {"name": s.name, "last_updated": s.last_updated}
            for s in brief_sessions
        }

        # Build lookup maps
        cached_by_id = {s["id"]: name for name, s in cached_data.items()}
        session_ids_to_fetch = []
        sessions_to_remove = []

        # Find stale sessions (exist in both but timestamps differ) and deleted sessions
        for session_name, cached_session in cached_data.items():
            cached_id = cached_session["id"]
            if cached_id in netbox_sessions:
                if (
                    cached_session["last_updated"]
                    != netbox_sessions[cached_id]["last_updated"]
                ):
                    session_ids_to_fetch.append(cached_id)
            else:
                sessions_to_remove.append(session_name)

        # Find new sessions in NetBox not in cache
        for nb_id in netbox_sessions:
            if nb_id not in cached_by_id:
                session_ids_to_fetch.append(nb_id)

        # Remove deleted sessions
        for session_name in sessions_to_remove:
            ret.result[device_name].pop(session_name, None)
            log.info(
                f"removed deleted session '{session_name}' from cache for '{device_name}'"
            )

        # Fetch updated/new sessions
        if session_ids_to_fetch:
            job.event(
                f"fetching {len(session_ids_to_fetch)} updated BGP session(s) for '{device_name}'",
                resource=instance,
            )
            for session in self.bulk_filter(
                nb.plugins.bgp.session, id=session_ids_to_fetch
            ):
                ret.result[device_name][session.name] = dict(session)

        # Update cache if any changes occurred
        if session_ids_to_fetch or sessions_to_remove:
            self.cache.set(
                cache_key, ret.result[device_name], expire=self.cache_ttl
            )
            job.event(f"updated cache for '{device_name}'", resource=instance)
        else:
            job.event(
                f"using cache, it is up to date for '{device_name}'",
                resource=instance,
            )

    return ret