NetBox Create BGP Peering Task¤

task api name: create_bgp_peering

Creates one or many BGP sessions in NetBox. Supports single-session mode (individual keyword arguments) and bulk mode (bulk_create list of dicts). IP addresses and ASNs are resolved from IPAM or created on-demand. local_ip_query and remote_ip_query can find existing addresses with NetBox filters such as literal device and interface names. Explicit addresses take precedence over queries. When local_interface is provided the local address is resolved from IPAM; for P2P subnets (/30, /31, /127) the remote address is derived automatically. Optionally creates a mirror (reverse) session on the remote device.

How It Works¤

When deriving a missing remote address, the task uses the first interface IP in the resolved local address's family. Without a supplied or queried local address, it uses the first interface IP. Use family: 4 or family: 6 in a scoped local_ip_query to select a family. The query must match exactly one address.

Supply tenant as an existing tenant name for a new session. Bulk entries specify their own tenant, and reverse sessions inherit it. Existing sessions remain unchanged.

Client submits create_bgp_peering request to NetBox worker
NetBox worker validates that the BGP plugin is installed
Worker resolves the RIR ID once (when rir is provided) for on-demand ASN creation
For each spec, supplied IP queries find addresses first; when local_interface is supplied, the interface supplies a missing local address and can derive a missing P2P peer address
Worker pre-fetches existing sessions for targeted devices and indexes their device names, IP addresses, and ASN numbers
For each session spec:
- Idempotency check: existing sessions and repeated bulk entries are reported in exists by object identity, regardless of name
- In dry-run mode, new sessions are reported in create without writes
- Local and remote IPs and ASNs are resolved from IPAM after the duplicate check; normal runs can create missing references
- Optional fields (peer group, routing policies, prefix lists) are resolved or created by name
- When create_reverse=True a mirror session is built by swapping local/remote IPs and ASNs
Remaining payloads are sent to NetBox in bulk-create calls

Object and multiobject custom fields accept related object names or IDs when creating a session. Use update_bgp_peering to change an existing session. When custom_fields is supplied, the task reuses reference-field definitions in lookup_cache or fetches and caches them for the next call. Calls without supplied custom fields do not fetch those definitions.

Prerequisites¤

NetBox BGP plugin (netbox-bgp) must be installed and enabled on the NetBox instance.

Branching Support¤

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

Output¤

Normal mode returns the created, existing, and skipped BGP session details. In dry-run mode, the task returns the session names that would be created and the sessions that already exist.

Dry Run Mode¤

dry_run=True returns session names without any NetBox writes. It resolves ASN queries and compares session values with existing sessions:

{
    "create": ["<session_name>", ...],
    "exists": ["<session_name>", ...],
}

Sessions that already exist in NetBox appear in exists regardless of dry-run mode.

Session Naming¤

By default, sessions are named using:

{device}_{local_address}_{remote_address}

Use the name_template parameter for a custom naming scheme. The template is a Python format string with the variables device, local_address, and remote_address.

Example:

name_template="{device}_BGP_{local_address}"
# ceos-spine-1_BGP_10.0.0.1

Reverse (Mirror) Session¤

When create_reverse=True (the default), the task also creates a mirror session on the remote device by swapping local and remote IPs and ASNs. The remote device is identified from IPAM by looking up the interface that holds the remote address.

Set create_reverse=False to suppress mirror session creation.

Note

sync_bgp_peerings passes create_reverse=False when delegating to create_bgp_peering because it manages both sides of a session independently via the diff.

VRF Custom Field¤

The VRF reference is always stored in a BGP session custom field that must be configured as type Object in NetBox pointing to the VRF content-type. This means NetBox stores a reference to a single VRF object, not a plain string.

By default vrf_custom_field="vrf" means the VRF object reference is written into custom_fields["vrf"]. Pass a different name to target a different custom field:

result = client.run_job(
    "netbox",
    "create_bgp_peering",
    workers="any",
    kwargs={
        "name": "ceos-spine-1_10.0.0.1_10.0.0.2",
        "device": "ceos-spine-1",
        "local_address": "10.0.0.1",
        "remote_address": "10.0.0.2",
        "local_as": 65001,
        "remote_as": 65002,
        "rir": "lab",
        "vrf": "PROD_VRF",
        "vrf_custom_field": "tenant_vrf",  # Object-type custom field -> VRF
    },
)

The vrf parameter accepts a VRF name which is resolved to a NetBox VRF object ID before being stored as an object reference in the custom field.

Examples¤

CLIPython

Create a single BGP session:

nf#netbox create bgp-peering name ceos-spine-1_10.0.0.1_10.0.0.2 device ceos-spine-1 local-address 10.0.0.1 remote-address 10.0.0.2 local-as 65001 remote-as 65002 rir lab

Dry run - preview what would be created:

nf#netbox create bgp-peering name ceos-spine-1_10.0.0.1_10.0.0.2 device ceos-spine-1 local-address 10.0.0.1 remote-address 10.0.0.2 local-as 65001 remote-as 65002 dry-run

Create session from interface (P2P peer derived automatically):

nf#netbox create bgp-peering device ceos-spine-1 local-interface Ethernet1 local-as 65001 rir lab

Create session into a NetBox branch:

nf#netbox create bgp-peering name ceos-spine-1_10.0.0.1_10.0.0.2 device ceos-spine-1 local-address 10.0.0.1 remote-address 10.0.0.2 local-as 65001 remote-as 65002 rir lab branch my-bgp-branch

Context manager - create a single session:

from norfab.core.nfapi import NorFab

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

    result = client.run_job(
        "netbox",
        "create_bgp_peering",
        workers="any",
        kwargs={
            "name": "ceos-spine-1_10.0.0.1_10.0.0.2",
            "device": "ceos-spine-1",
            "local_address": "10.0.0.1",
            "remote_address": "10.0.0.2",
            "local_as": 65001,
            "remote_as": 65002,
            "rir": "lab",
        },
    )

Context manager - bulk create several sessions:

from norfab.core.nfapi import NorFab

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

    result = client.run_job(
        "netbox",
        "create_bgp_peering",
        workers="any",
        kwargs={
            "bulk_create": [
                {
                    "name": "ceos-spine-1_10.0.0.1_10.0.0.2",
                    "device": "ceos-spine-1",
                    "local_address": "10.0.0.1",
                    "remote_address": "10.0.0.2",
                    "local_as": 65001,
                    "remote_as": 65002,
                    "description": "spine-1 to leaf-1",
                },
                {
                    "name": "ceos-spine-2_10.0.0.5_10.0.0.6",
                    "device": "ceos-spine-2",
                    "local_address": "10.0.0.5",
                    "remote_address": "10.0.0.6",
                    "local_as": 65001,
                    "remote_as": 65003,
                    "description": "spine-2 to leaf-1",
                },
            ],
            "peer_group": "EBGP_LEAFS",
            "import_policies": ["IMPORT_FROM_LEAFS"],
            "export_policies": ["EXPORT_TO_LEAFS"],
            "prefix_list_in": "PL-IN-LEAFS",
            "prefix_list_out": "PL-OUT-LEAFS",
            "rir": "lab",
            "message": "Create leaf BGP peerings",
        },
    )

Context manager - create from local interfaces:

from norfab.core.nfapi import NorFab

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

    result = client.run_job(
        "netbox",
        "create_bgp_peering",
        workers="any",
        kwargs={
            "bulk_create": [
                {
                    "name": "ceos-spine-1_Ethernet1",
                    "device": "ceos-spine-1",
                    "local_interface": "Ethernet1",
                    "local_as": 65001,
                },
                {
                    "name": "ceos-spine-1_Ethernet2",
                    "device": "ceos-spine-1",
                    "local_interface": "Ethernet2",
                    "local_as": 65001,
                },
            ],
            "rir": "lab",
            "create_reverse": True,
        },
    )

Direct lifecycle - dry run and then create from interface data:

from norfab.core.nfapi import NorFab

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

    preview = client.run_job(
        "netbox",
        "create_bgp_peering",
        workers="any",
        kwargs={
            "name": "ceos-spine-1_10.0.0.1_10.0.0.2",
            "device": "ceos-spine-1",
            "local_address": "10.0.0.1",
            "remote_address": "10.0.0.2",
            "local_as": 65001,
            "remote_as": 65002,
            "dry_run": True,
        },
    )

    result = client.run_job(
        "netbox",
        "create_bgp_peering",
        workers="any",
        kwargs={
            "device": "ceos-spine-1",
            "local_interface": "Ethernet1",
            "local_as": 65001,
            "remote_as_query": {"asn": 65002},
            "rir": "lab",
            "branch": "my-bgp-branch",
        },
    )
finally:
    nf.destroy()

Direct lifecycle - use a custom name template and VRF custom field:

from norfab.core.nfapi import NorFab

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

    result = client.run_job(
        "netbox",
        "create_bgp_peering",
        workers="any",
        kwargs={
            "device": "ceos-spine-1",
            "local_interface": "Ethernet[1-2]",
            "local_as": 65001,
            "remote_as_query": {"asn": 65002},
            "name_template": "{{ device }}_{{ vrf }}_{{ remote_address }}",
            "vrf": "PROD",
            "vrf_custom_field": "tenant_vrf",
            "peer_group": "EBGP_LEAFS",
            "create_reverse": False,
            "rir": "lab",
            "message": "Create PROD BGP peerings from interface range",
        },
    )
finally:
    nf.destroy()

Direct lifecycle - resolve ASNs from an IPAM ASN query:

from norfab.core.nfapi import NorFab

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

    result = client.run_job(
        "netbox",
        "create_bgp_peering",
        workers="any",
        kwargs={
            "device": "ceos-spine-1",
            "local_ip_query": {"device": "ceos-spine-1", "interface": "Ethernet1"},
            "remote_ip_query": {"device": "ceos-leaf-1", "interface": "Ethernet1"},
            "local_as_query": {"asn": 65001},
            "remote_as_query": {"asn": 65002},
            "name_template": "{{ device }}_{{ vrf }}_{{ remote_address }}",
            "rir": "lab",
        },
    )
finally:
    nf.destroy()

NORFAB NetBox Create BGP Peering Command Shell Reference¤

NorFab shell supports these command options for the NetBox create_bgp_peering task:

nf# man tree netbox.create.bgp-peering
root
└── netbox:    Netbox service
    └── create:    Create Netbox objects
        └── bgp-peering:    Create BGP peering session(s)
            ├── timeout:    Job timeout
            ├── workers:    Filter worker to target, default 'any'
            ├── verbose-result:    Control output details, default 'False'
            ├── progress:    Display progress events, default 'True'
            ├── instance:    Netbox instance name to target
            ├── branch:    Branching plugin branch name to use
            ├── dry-run:    Return names without writing to NetBox
            ├── name:    Session name (single mode)
            ├── device:    Local device name (single mode)
            ├── local-address:    Local IP address string
            ├── remote-address:    Remote IP address string
            ├── local-as:    Local AS number string
            ├── remote-as:    Remote AS number string
            ├── status:    Session status, default 'active'
            ├── description:    Session description
            ├── vrf:    VRF name
            ├── peer-group:    Peer group name
            ├── import-policies:    Import routing policy names
            ├── export-policies:    Export routing policy names
            ├── prefix-list-in:    Inbound prefix list name
            ├── prefix-list-out:    Outbound prefix list name
            ├── local-interface:    Local interface name or bracket-range pattern
            ├── asn-source:    Dot-path or IPAM query dict for ASN resolution
            ├── name-template:    Format string for session names
            ├── create-reverse:    Also create mirror session on remote device, default 'True'
            ├── bulk-create:    JSON list of session dicts for bulk creation
            ├── rir:    RIR name for ASN creation
            ├── vrf-custom-field:    BGP session field for VRF reference, default 'vrf'
            └── message:    Changelog message for NetBox write operations
nf#

Python API Reference¤

Create one or many BGP sessions in NetBox.

Supports single-session mode (individual keyword arguments) and bulk mode (bulk_create list of dicts). IP addresses and ASNs are resolved from IPAM or created on demand. When local_interface is provided the local address is resolved from IPAM. If remote_address is omitted, the remote address is derived from the first IP assigned to that interface matching the resolved local address's family, or the first IP overall when no local address is supplied. Derivation supports IPv4 /30 or /31 and IPv6 /127 subnets. This uses subnet arithmetic and does not require a cable or remote interface recorded in NetBox. For example, 10.0.0.1/30 yields 10.0.0.2. Supplying only local_address without local_interface does not trigger derivation. Unless dry_run is enabled, a missing peer IP is created in IPAM using the most specific containing NetBox prefix's mask, or /32 for IPv4 and /128 for IPv6 if no containing prefix exists. The new IP is not assigned to a remote device or interface automatically. Existing sessions and repeated bulk entries are matched by device name, local IP and ASN, and remote IP and ASN values; names do not identify sessions.

Parameters:

Name	Type	Description	Default
`job`	`Job`	NorFab Job object.	required
`instance`	`str`	NetBox instance name.	`None`
`name`	`str`	Session name. Derived from `name_template` when omitted.	`None`
`device`	`str`	Local device name. Required in single-session mode.	`None`
`local_address`	`str`	Local IP address string. Derived from `local_interface` when omitted.	`None`
`remote_address`	`str`	Remote IP address string. When omitted, derived from the first IP assigned to `local_interface` matching the resolved local address's family (or the first IP overall when no local address is supplied) for IPv4 /30 or /31 and IPv6 /127 subnets. An explicit value takes precedence over derivation.	`None`
`local_as`	`int`	Local AS number. Takes precedence over `local_as_query`.	`None`
`remote_as`	`int`	Remote AS number. Takes precedence over `remote_as_query`.	`None`
`status`	`str`	Session status. Default `'active'`.	`'active'`
`description`	`str`	Session description.	`None`
`tenant`	`str`	Existing tenant name for new sessions, including reverse sessions. Bulk entries specify their own tenant. NetBox rejects unknown tenant names. Existing sessions remain unchanged.	`None`
`vrf`	`str`	NetBox VRF name for a new session. The task resolves or creates the VRF and stores its object reference in `custom_fields[vrf_custom_field]`. `global` and `default` assign no VRF. If the custom field is missing, VRF assignment is skipped. Existing sessions are left unchanged.	`None`
`peer_group`	`str`	Peer group name (resolved or created).	`None`
`import_policies`	`list`	List of import routing-policy names.	`None`
`export_policies`	`list`	List of export routing-policy names.	`None`
`custom_fields`	`dict`	Custom-field values. Object and multiobject fields accept related object names or IDs when creating a session. Existing sessions are left unchanged.	`None`
`prefix_list_in`	`str`	Inbound prefix-list name.	`None`
`prefix_list_out`	`str`	Outbound prefix-list name.	`None`
`local_interface`	`str`	Local interface name or bracket-range pattern.	`None`
`local_as_query`	`dict`	Filters passed to `nb.ipam.asns.get` to find the local ASN when `local_as` is omitted. The `role` value is treated as a name and converted to a slug for filtering.	`None`
`remote_as_query`	`dict`	Filters passed to `nb.ipam.asns.get` to find the remote ASN when `remote_as` is omitted. The `role` value is treated as a name and converted to a slug for filtering.	`None`
`local_ip_query`	`dict`	Filters passed to `nb.ipam.ip_addresses.get` when `local_address` is omitted.	`None`
`remote_ip_query`	`dict`	Filters passed to `nb.ipam.ip_addresses.get` when `remote_address` is omitted. Filters may include literal device and interface names.	`None`
`name_template`	`str`	Jinja2 template string for session names. Default `'{{device}}_{{vrf}}_{{remote_address}}'`. When the session dictionary omits `vrf`, the template uses `default`; an explicit `vrf=None` renders as `None`.	`'{{device}}_{{vrf}}_{{remote_address}}'`
`create_reverse`	`bool`	When `True` also create a mirror session on the remote device with local and remote IPs/ASNs swapped. Default `True`.	`True`
`bulk_create`	`list`	List of session dicts for bulk creation. Each entry can supply its own ASN and IP address queries.	`None`
`rir`	`str`	RIR name used when auto-creating ASNs.	`None`
`message`	`str`	Changelog message recorded on every NetBox write.	`None`
`branch`	`str`	NetBox branching plugin branch name.	`None`
`dry_run`	`bool`	When `True` return session names without writing.	`False`
`vrf_custom_field`	`str`	Name of the BGP session custom field that stores the VRF object reference. The custom field must be of type Object in NetBox pointing to the VRF content-type. The value is always a single VRF object reference written into `custom_fields[vrf_custom_field]`. Default `'vrf'` means `custom_fields['vrf']`.	`'vrf'`
`lookup_cache`	`dict`	Reusable NetBox lookup data. Object and multiobject custom-field definitions are cached under `custom_fields['netbox_bgp.bgpsession']` when supplied custom fields need resolution.	`None`

Returns:

Type	Description
`Result`	Normal run:: {"created": ["name1", ...], "exists": ["name2", ...]}
`Result`	Dry run:: {"create": ["name1", ...], "exists": ["name2", ...]}

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

@Task(
    fastapi={"methods": ["POST"], "schema": NetboxFastApiArgs.model_json_schema()},
    input=CreateBgpPeeringInput,
    output=CreateBgpPeeringResult,
    mcp={
        "annotations": {
            "title": "Create BGP Peering",
            "readOnlyHint": False,
            "destructiveHint": False,
            "idempotentHint": False,
            "openWorldHint": True,
        }
    },
)
def create_bgp_peering(
    self,
    job: Job,
    instance: Union[None, str] = None,
    batch_size: int = 1000,
    # single-session mode
    name: Union[None, str] = None,
    device: Union[None, str] = None,
    local_address: Union[None, str] = None,
    remote_address: Union[None, str] = None,
    local_as: Union[None, int] = None,
    remote_as: Union[None, int] = None,
    status: str = "active",
    description: Union[None, str] = None,
    vrf: Union[None, str] = None,
    peer_group: Union[None, str] = None,
    import_policies: Union[None, list] = None,
    export_policies: Union[None, list] = None,
    prefix_list_in: Union[None, str] = None,
    prefix_list_out: Union[None, str] = None,
    tags: Union[None, list] = None,
    custom_fields: Union[None, dict] = None,
    # interface-driven resolution
    local_interface: Union[None, str] = None,
    local_as_query: Union[None, dict] = None,
    remote_as_query: Union[None, dict] = None,
    local_ip_query: Union[None, dict] = None,
    remote_ip_query: Union[None, dict] = None,
    name_template: Union[None, str] = "{{device}}_{{vrf}}_{{remote_address}}",
    # mirror session
    create_reverse: bool = True,
    # bulk mode
    bulk_create: Union[None, list] = None,
    # shared
    rir: Union[None, str] = None,
    message: Union[None, str] = None,
    branch: Union[None, str] = None,
    dry_run: bool = False,
    vrf_custom_field: str = "vrf",
    lookup_cache: Union[None, dict] = None,
    tenant: Union[None, str] = None,
) -> Result:
    """
    Create one or many BGP sessions in NetBox.

    Supports single-session mode (individual keyword arguments) and bulk mode
    (``bulk_create`` list of dicts).  IP addresses and ASNs are resolved from
    IPAM or created on demand.  When ``local_interface`` is provided the local
    address is resolved from IPAM. If ``remote_address`` is omitted, the
    remote address is derived from the first IP assigned to that interface
    matching the resolved local address's family, or the first IP overall
    when no local address is supplied. Derivation supports IPv4 /30 or /31
    and IPv6 /127 subnets. This uses subnet arithmetic
    and does not require a cable or remote interface recorded in NetBox.
    For example, ``10.0.0.1/30`` yields ``10.0.0.2``. Supplying only
    ``local_address`` without ``local_interface`` does not trigger derivation.
    Unless ``dry_run`` is enabled, a missing peer IP is created in IPAM
    using the most specific containing NetBox prefix's mask, or /32 for
    IPv4 and /128 for IPv6 if no containing prefix exists. The new IP is
    not assigned to a remote device or interface automatically.
    Existing sessions and repeated bulk
    entries are matched by device name, local IP and ASN, and remote IP and
    ASN values; names do not identify sessions.

    Args:
        job: NorFab Job object.
        instance (str, optional): NetBox instance name.
        name (str, optional): Session name. Derived from ``name_template`` when omitted.
        device (str, optional): Local device name. Required in single-session mode.
        local_address (str, optional): Local IP address string. Derived from ``local_interface`` when omitted.
        remote_address (str, optional): Remote IP address string. When omitted,
            derived from the first IP assigned to ``local_interface`` matching
            the resolved local address's family (or the first IP overall
            when no local address is supplied) for
            IPv4 /30 or /31 and IPv6 /127 subnets. An explicit value takes
            precedence over derivation.
        local_as (int, optional): Local AS number. Takes precedence over ``local_as_query``.
        remote_as (int, optional): Remote AS number. Takes precedence over ``remote_as_query``.
        status (str): Session status. Default ``'active'``.
        description (str, optional): Session description.
        tenant (str, optional): Existing tenant name for new sessions, including
            reverse sessions. Bulk entries specify their own tenant. NetBox
            rejects unknown tenant names. Existing sessions remain unchanged.
        vrf (str, optional): NetBox VRF name for a new session. The task
            resolves or creates the VRF and stores its object reference in
            ``custom_fields[vrf_custom_field]``. ``global`` and ``default``
            assign no VRF. If the custom field is missing, VRF assignment is
            skipped. Existing sessions are left unchanged.
        peer_group (str, optional): Peer group name (resolved or created).
        import_policies (list, optional): List of import routing-policy names.
        export_policies (list, optional): List of export routing-policy names.
        custom_fields (dict, optional): Custom-field values. Object and
            multiobject fields accept related object names or IDs when
            creating a session. Existing sessions are left unchanged.
        prefix_list_in (str, optional): Inbound prefix-list name.
        prefix_list_out (str, optional): Outbound prefix-list name.
        local_interface (str, optional): Local interface name or bracket-range pattern.
        local_as_query (dict, optional): Filters passed to ``nb.ipam.asns.get``
            to find the local ASN when ``local_as`` is omitted. The ``role``
            value is treated as a name and converted to a slug for filtering.
        remote_as_query (dict, optional): Filters passed to ``nb.ipam.asns.get``
            to find the remote ASN when ``remote_as`` is omitted. The ``role``
            value is treated as a name and converted to a slug for filtering.
        local_ip_query (dict, optional): Filters passed to
            ``nb.ipam.ip_addresses.get`` when ``local_address`` is omitted.
        remote_ip_query (dict, optional): Filters passed to
            ``nb.ipam.ip_addresses.get`` when ``remote_address`` is omitted.
            Filters may include literal device and interface names.
        name_template (str, optional): Jinja2 template string for session names.
            Default ``'{{device}}_{{vrf}}_{{remote_address}}'``. When the
            session dictionary omits ``vrf``, the template uses ``default``;
            an explicit ``vrf=None`` renders as ``None``.
        create_reverse (bool): When ``True`` also create a mirror session on the
            remote device with local and remote IPs/ASNs swapped. Default ``True``.
        bulk_create (list, optional): List of session dicts for bulk creation.
            Each entry can supply its own ASN and IP address queries.
        rir (str, optional): RIR name used when auto-creating ASNs.
        message (str, optional): Changelog message recorded on every NetBox write.
        branch (str, optional): NetBox branching plugin branch name.
        dry_run (bool): When ``True`` return session names without writing.
        vrf_custom_field (str): Name of the BGP session custom field that stores
            the VRF object reference.  The custom field must be of type Object in
            NetBox pointing to the VRF content-type.  The value is always a single
            VRF object reference written into ``custom_fields[vrf_custom_field]``.
            Default ``'vrf'`` means ``custom_fields['vrf']``.
        lookup_cache (dict, optional): Reusable NetBox lookup data. Object
            and multiobject custom-field definitions are cached under
            ``custom_fields['netbox_bgp.bgpsession']`` when supplied custom
            fields need resolution.

    Returns:
        Normal run::

            {"created": ["name1", ...], "exists": ["name2", ...]}

        Dry run::

            {"create": ["name1", ...], "exists": ["name2", ...]}
    """
    instance = instance or self.default_instance
    ret = Result(
        task=f"{self.name}:create_bgp_peering",
        result={},
        resources=[instance],
    )

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

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

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

    lookup_cache = lookup_cache if lookup_cache is not None else {}
    name_template = name_template or "{{device}}_{{vrf}}_{{remote_address}}"

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

    # Build initial bgp_sessions list
    bgp_sessions = bulk_create or [
        {
            "name": name,
            "device": device,
            "local_address": local_address,
            "remote_address": remote_address,
            "local_as": local_as,
            "remote_as": remote_as,
            "local_as_query": local_as_query,
            "remote_as_query": remote_as_query,
            "local_ip_query": local_ip_query,
            "remote_ip_query": remote_ip_query,
            "status": status,
            "description": description,
            "vrf": vrf,
            "tenant": tenant,
            "peer_group": peer_group,
            "import_policies": import_policies,
            "export_policies": export_policies,
            "prefix_list_in": prefix_list_in,
            "prefix_list_out": prefix_list_out,
            "custom_fields": custom_fields,
            "local_interface": local_interface,
        }
    ]
    job.event(
        f"preparing {len(bgp_sessions)} BGP session create request(s), dry_run={dry_run}"
    )

    # Step 5a: Resolve interfaces, IPs, and discover remote devices for every
    # bgp_session in one pass.  Range expansion turns one bgp_session into many when a bracket
    # pattern is used in local_interface.
    job.event("resolving BGP session interface and peer details")
    base_bgp_sessions = []

    for bgp_session in bgp_sessions:
        bgp_session = dict(bgp_session)
        bgp_session_device = bgp_session["device"]
        bgp_session_local_interface = bgp_session.get("local_interface")

        # Query addresses before interface and name resolution. Explicit
        # addresses take precedence over queries.
        query_failed = False
        for side in ("local", "remote"):
            address_field = f"{side}_address"
            query_field = f"{side}_ip_query"
            query = bgp_session.get(query_field)
            if bgp_session.get(address_field) is not None or query is None:
                continue
            try:
                ip_obj = nb.ipam.ip_addresses.get(**query)
            except ValueError as exc:
                ip_obj = None
                msg = f"{query_field} failed for '{bgp_session_device}': {exc}"
            else:
                msg = (
                    f"{query_field} found no IP for '{bgp_session_device}': "
                    f"{query}"
                )
            if ip_obj is None:
                job.event(msg, severity="ERROR")
                log.error(f"{self.name} - {msg}")
                ret.errors.append(msg)
                query_failed = True
                break
            bgp_session[address_field] = ip_obj.address.split("/")[0]
        if query_failed:
            continue

        if bgp_session_local_interface:
            # Expand bracket-range pattern e.g. "Ethernet[1-4]/1.101"
            intf_names = expand_alphanumeric_range(bgp_session_local_interface)

            # Batch fetch interfaces and their IPs in two calls instead of 2N
            intf_by_name = {
                i.name: i
                for i in self.bulk_filter(
                    endpoint=nb.dcim.interfaces,
                    name=intf_names,
                    device=bgp_session_device,
                    fields="id,name",
                )
            }
            found_intf_ids = [i.id for i in intf_by_name.values()]
            ips_by_intf_id: Dict[int, list] = {}
            if found_intf_ids:
                for _ip in self.bulk_filter(
                    endpoint=nb.ipam.ip_addresses,
                    interface_id=found_intf_ids,
                    fields="id,address,assigned_object_id",
                ):
                    ips_by_intf_id.setdefault(_ip.assigned_object_id, []).append(
                        _ip
                    )

            for intf_name in intf_names:
                intf = intf_by_name.get(intf_name)
                if not intf:
                    msg = f"interface '{intf_name}' not found on device '{bgp_session_device}'"
                    job.event(msg, severity="WARNING")
                    log.warning(f"{self.name} - {msg}")
                    ret.errors.append(msg)
                    continue

                ip_list = ips_by_intf_id.get(intf.id, [])
                local_address = bgp_session.get("local_address")
                # Select IPs of the same family as the local address.
                if local_address and not bgp_session.get("remote_address"):
                    family = ipaddress.ip_interface(local_address).version
                    ip_list = [
                        ip
                        for ip in ip_list
                        if ipaddress.ip_interface(ip.address).version == family
                    ]
                if not ip_list:
                    msg = f"no IP address assigned to interface '{intf_name}' on '{bgp_session_device}'"
                    if local_address:
                        msg += f" matching the family of local address '{local_address}'"
                    job.event(msg, severity="WARNING")
                    log.warning(f"{self.name} - {msg}")
                    ret.errors.append(msg)
                    continue

                ip_cidr = ip_list[0].address  # e.g. "10.0.0.1/31"
                local_addr = (
                    bgp_session.get("local_address") or ip_cidr.split("/")[0]
                )

                bgp_session_remote_address = bgp_session.get("remote_address")

                if not bgp_session_remote_address:
                    peer_ip = get_p2p_peer_ip(ip_cidr)
                    if peer_ip:
                        bgp_session_remote_address = peer_ip
                if not bgp_session_remote_address:
                    msg = (
                        f"remote_address not provided and could not derive peer IP "
                        f"from interface '{intf_name}' address '{ip_cidr}' on "
                        f"device '{bgp_session_device}'"
                    )
                    job.event(msg, severity="ERROR")
                    log.error(f"{self.name} - {msg}")
                    ret.errors.append(msg)
                    continue

                new_bgp_session = dict(bgp_session)
                new_bgp_session["device"] = bgp_session_device
                new_bgp_session["local_address"] = local_addr
                new_bgp_session["remote_address"] = bgp_session_remote_address
                new_bgp_session.setdefault("remote_device", None)
                base_bgp_sessions.append(new_bgp_session)
        else:
            if not bgp_session.get("remote_address"):
                msg = (
                    f"remote_address not provided for BGP session on device "
                    f"'{bgp_session_device}'"
                )
                job.event(msg, severity="ERROR")
                log.error(f"{self.name} - {msg}")
                ret.errors.append(msg)
                continue
            bgp_session_copy = dict(bgp_session)
            bgp_session_copy.setdefault("remote_device", None)
            base_bgp_sessions.append(bgp_session_copy)

    preseed_bgp_lookup_cache(
        nb, base_bgp_sessions, lookup_cache, job, self.bulk_filter, self.name
    )

    reverse_sessions = []
    if create_reverse:
        for bgp_session in base_bgp_sessions:
            remote_device = bgp_session.get(
                "remote_device"
            ) or _remote_device_from_ip(
                _cached_bgp_ip_object(bgp_session["remote_address"], lookup_cache)
            )
            if not remote_device:
                session_label = (
                    bgp_session.get("name")
                    or bgp_session.get("remote_address")
                    or "unknown"
                )
                msg = (
                    "failed to resolve reverse session remote device name for "
                    f"'{session_label}'"
                )
                job.event(msg, severity="ERROR")
                log.error(f"{self.name} - {msg}")
                ret.errors.append(msg)
                continue
            mirror_session = dict(bgp_session)
            mirror_session["device"] = remote_device
            mirror_session["local_address"] = bgp_session["remote_address"]
            mirror_session["remote_address"] = bgp_session["local_address"]
            mirror_session["local_as"] = bgp_session.get("remote_as")
            mirror_session["remote_as"] = bgp_session.get("local_as")
            mirror_session["local_as_query"] = bgp_session.get("remote_as_query")
            mirror_session["remote_as_query"] = bgp_session.get("local_as_query")
            mirror_session["local_ip_query"] = bgp_session.get("remote_ip_query")
            mirror_session["remote_ip_query"] = bgp_session.get("local_ip_query")
            mirror_session["remote_device"] = bgp_session["device"]
            mirror_session["local_interface"] = None
            mirror_session["name"] = None
            reverse_sessions.append(mirror_session)
    bgp_sessions = base_bgp_sessions + reverse_sessions

    all_device_names = {
        name: _cached_bgp_device_info(name, lookup_cache)
        for bgp_session in bgp_sessions
        for name in (
            bgp_session["device"],
            bgp_session.get("remote_device"),
        )
        if name
    }
    job.event(f"resolved {len(bgp_sessions)} BGP session create candidate(s)")

    # Step 5c: Pre-fetch existing sessions for idempotency (single API call)
    existing_sessions = set()
    existing_device_names = [
        name for name, info in all_device_names.items() if info
    ]
    if existing_device_names:
        job.event(
            f"checking existing BGP sessions on {len(existing_device_names)} device(s)"
        )
        try:
            existing = self.bulk_filter(
                nb.plugins.bgp.session,
                device=existing_device_names,
                fields="device,local_address,local_as,remote_address,remote_as",
            )
            existing_sessions = {
                (
                    session.device.name,
                    _normalise_bgp_identity_ip(session.local_address.address),
                    session.local_as.asn,
                    _normalise_bgp_identity_ip(session.remote_address.address),
                    session.remote_as.asn,
                )
                for session in existing
            }
            job.event(f"found {len(existing_sessions)} existing BGP session(s)")
        except Exception as exc:
            msg = (
                f"could not pre-fetch BGP sessions for {existing_device_names}: "
                f"{exc}"
            )
            job.event(msg, severity="ERROR")
            log.error(f"{self.name} - {msg}")
            ret.errors.append(msg)
            ret.failed = True
            return ret

    # Step 6: Process each resolved bgp_session
    if dry_run is True:
        ret.dry_run = True
        result = {"create": [], "exists": []}
    else:
        result = {"created": [], "exists": []}

    payloads = []
    if any(session.get("custom_fields") is not None for session in bgp_sessions):
        cache_bgp_custom_field_definitions(nb, lookup_cache, self.bulk_filter)

    job.event("building BGP session create payloads")
    for bgp_session in bgp_sessions:
        bgp_session_device = bgp_session.get("device")
        bgp_session_local_address = bgp_session.get("local_address")
        bgp_session_remote_address = bgp_session.get("remote_address")
        bgp_session_local_as = bgp_session.get("local_as")
        bgp_session_remote_as = bgp_session.get("remote_as")
        remote_device = bgp_session.get("remote_device")

        # Determine session name
        sname = bgp_session.get("name")
        if not sname:
            try:
                sname = render_bgp_session_name(
                    name_template, bgp_session, lookup_cache
                )
            except Exception as exc:
                msg = (
                    f"failed to render name_template '{name_template}' for session "
                    f"'{bgp_session.get('name')}' on '{bgp_session_device}': {exc}"
                )
                job.event(msg, severity="ERROR")
                ret.errors.append(msg)
                log.error(f"{self.name} - {msg}")
                continue
            bgp_session["name"] = sname

        # Explicit ASN numbers win; otherwise query NetBox for each side.
        resolved_asns = {
            "local": bgp_session_local_as,
            "remote": bgp_session_remote_as,
        }
        query_failed = False
        for side in ("local", "remote"):
            query_field = f"{side}_as_query"
            query = bgp_session.get(query_field)
            if resolved_asns[side] is not None or query is None:
                continue
            try:
                if "role" in query:
                    query["role"] = slugify(query["role"])
                asn_obj = nb.ipam.asns.get(**query)
            except ValueError as exc:
                asn_obj = None
                msg = f"{query_field} failed for '{sname}': {exc}"
            else:
                msg = f"{query_field} found no ASN for '{sname}': {query}"
            # NetBox may reject ambiguous filters or return no matching ASN.
            if asn_obj is None:
                job.event(msg, severity="ERROR")
                log.error(f"{self.name} - {msg}")
                ret.errors.append(msg)
                query_failed = True
                break
            resolved_asns[side] = asn_obj.asn
        if query_failed:
            continue
        bgp_session_local_as = resolved_asns["local"]
        bgp_session_remote_as = resolved_asns["remote"]

        # Resolve device ID and site ID (step 6f)
        dev_info = all_device_names.get(bgp_session_device)
        if not dev_info:
            msg = (
                f"device '{bgp_session_device}' not found in NetBox, "
                f"skipping '{sname}'"
            )
            job.event(msg, severity="WARNING")
            log.warning(f"{self.name} - {msg}")
            ret.errors.append(msg)
            continue

        device_id = dev_info["id"]
        site_id = dev_info["site_id"]
        identity = bgp_session_identity(
            bgp_session_device,
            {
                "local_address": bgp_session_local_address,
                "local_as": bgp_session_local_as,
                "remote_address": bgp_session_remote_address,
                "remote_as": bgp_session_remote_as,
            },
        )
        if identity in existing_sessions:
            result["exists"].append(sname)
            continue
        if dry_run:
            existing_sessions.add(identity)
            result["create"].append(sname)
            continue

        local_ip_id = resolve_ip(
            bgp_session_local_address, nb, job, ret, self.name, lookup_cache
        )
        remote_ip_id = resolve_ip(
            bgp_session_remote_address, nb, job, ret, self.name, lookup_cache
        )
        local_as_id = resolve_asn(
            bgp_session_local_as, nb, rir_id, job, ret, self.name, lookup_cache
        )
        remote_as_id = resolve_asn(
            bgp_session_remote_as, nb, rir_id, job, ret, self.name, lookup_cache
        )

        resolution_errors = []
        if not local_ip_id:
            resolution_errors.append(f"local IP '{bgp_session_local_address}'")
        if not remote_ip_id:
            resolution_errors.append(f"remote IP '{bgp_session_remote_address}'")
        if not local_as_id:
            resolution_errors.append(f"local ASN '{bgp_session_local_as}'")
        if not remote_as_id:
            resolution_errors.append(f"remote ASN '{bgp_session_remote_as}'")
        if resolution_errors:
            msg = f"skipping '{sname}': could not resolve {', '.join(resolution_errors)}"
            job.event(msg, severity="WARNING")
            log.warning(f"{self.name} - {msg}")
            ret.errors.append(msg)
            continue

        existing_sessions.add(identity)
        addr_family = get_addr_family(bgp_session_local_address)

        payload = {
            "name": sname,
            "description": bgp_session.get("description") or "",
            "device": device_id,
            "local_address": local_ip_id,
            "local_as": local_as_id,
            "remote_address": remote_ip_id,
            "remote_as": remote_as_id,
            "status": bgp_session.get("status", "active"),
            "site": site_id,
        }
        if bgp_session.get("tenant") is not None:
            payload["tenant"] = {"name": bgp_session["tenant"]}
        if bgp_session.get("custom_fields") is not None:
            payload["custom_fields"] = dict(bgp_session["custom_fields"])
        if "tags" in bgp_session:
            payload["tags"] = merge_array_values(
                [], bgp_session["tags"], attribute="name"
            )

        # Optional fields (step 6h)
        # Tags are already converted to NetBox name references in payload.
        resolved_fields = resolve_bgp_session_payload_fields(
            {
                k: bgp_session[k]
                for k in (
                    "vrf",
                    "peer_group",
                    "import_policies",
                    "export_policies",
                    "prefix_list_in",
                    "prefix_list_out",
                    "custom_fields",
                )
                if bgp_session.get(k)
            },
            nb,
            rir_id,
            job,
            ret,
            self.name,
            addr_family,
            lookup_cache=lookup_cache,
            vrf_custom_field=vrf_custom_field,
        )
        if "custom_fields" in resolved_fields:
            payload.setdefault("custom_fields", {}).update(
                resolved_fields.pop("custom_fields")
            )
        payload.update(resolved_fields)
        if bgp_session.get("custom_fields") is not None:
            # Resolve names after VRF resolution adds its object reference.
            _, payload["custom_fields"] = merge_resolved_custom_fields(
                self,
                nb,
                None,
                payload["custom_fields"],
                lookup_cache["custom_fields"]["netbox_bgp.bgpsession"]["object"],
                lookup_cache["custom_fields"]["netbox_bgp.bgpsession"][
                    "multiobject"
                ],
            )

        payloads.append(payload)
    if dry_run is True:
        job.event(
            f"dry-run: {len(result['create'])} BGP session(s) would be created, "
            f"{len(result['exists'])} already exist"
        )
        ret.result = result
        job.event("BGP session create task complete")
        return ret
    job.event(
        f"prepared {len(payloads)} BGP session create payload(s), "
        f"{len(result['exists'])} already exist"
    )

    # Bulk create (step 7)
    if payloads:
        job.event(f"creating {len(payloads)} BGP session(s)")
        total_batches = (len(payloads) + batch_size - 1) // batch_size
        for batch_start in range(0, len(payloads), batch_size):
            batch = payloads[batch_start : batch_start + batch_size]
            batch_number = batch_start // batch_size + 1
            msg = f"creating BGP session batch {batch_number}/{total_batches} ({len(batch)} session(s))"
            job.event(msg)
            log.info(msg)
            try:
                nb.plugins.bgp.session.create(batch)
            except Exception as exc:
                msg = f"failed to create BGP session batch {batch_number}/{total_batches}: {exc}"
                job.event(msg, severity="ERROR")
                ret.errors.append(msg)
                ret.failed = True
                ret.result = result
                log.error(f"{self.name} - {msg}")
                return ret
            result["created"].extend(payload["name"] for payload in batch)
        msg = f"created {len(payloads)} BGP session(s)"
        job.event(msg)
        log.info(f"{self.name} - {msg}")
    else:
        job.event("no BGP sessions to create")

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