Automate NetBox with the API

A scoped automation user, pynetbox or curl to read and write, a CSV import that survives errors, webhooks on device changes, and a custom script run from the API.

intermediate~45 min hands-on
#netbox#api#pynetbox#python#webhooks#automation

Not validated end to end yet — be the first.Report a problem

Draft — not yet run end to end. This page was written but its author has not yet run it on a real machine. Commands may be wrong: read before you run, and tell us what breaks.

The gistFrom your script to your receiver
NetBox server
HTTPS · tokenwriteseventPOST
Your scriptpynetbox · curl
REST APIhttps:///api
NetBoxpermissions · validation
PostgreSQLobjects · change log
RQ workernetbox-rq
Webhook receiver

Your script talks to the REST API with the automation user's token. NetBox writes the objects, queues an event, and the RQ worker on the server delivers the webhook to your receiver.

Everything you clicked on the previous pages is an HTTP call, and NetBox is only worth it once other systems read and write it. This page runs on your machine, not the server: a user made for scripts, pynetbox or curl to read and write, a CSV import that survives a bad row, webhooks when a device changes, and a script that runs inside NetBox on demand.

Before you start

$python3 -m venv ~/.venvs/netbox
$source ~/.venvs/netbox/bin/activate
$pip install pynetbox
Check
$pip show pynetbox | head -1
Expected output
Name: pynetbox

An automation user and its token

$api=https:///api
$auth="Authorization: Token "
$curl -sk -H "$auth" -H 'Content-Type: application/json' "$api/users/users/" \
$  -d "{\"username\": \"\", \"password\": \"$(openssl rand -base64 24)\"}"

Note the id in the answer; it goes into the users list of the permissions. Three of them:

$curl -sk -H "$auth" -H 'Content-Type: application/json' "$api/users/permissions/" -d '{
$  "name": "automation: view", "actions": ["view"], "users": [ID],
$  "object_types": ["dcim.site", "dcim.manufacturer", "dcim.devicerole", "dcim.devicetype",
$                   "dcim.device", "dcim.interface", "ipam.ipaddress"]
$}'

Now the token, with an expiry 90 days out:

$key=$(openssl rand -hex 20)
$exp=$(python3 -c 'import datetime as d; print((d.datetime.now(d.timezone.utc) + d.timedelta(days=90)).strftime("%Y-%m-%dT%H:%M:%SZ"))')
$curl -sk -H "$auth" -H 'Content-Type: application/json' "$api/users/tokens/" \
$  -d "{\"user\": ID, \"key\": \"$key\", \"expires\": \"$exp\", \"write_enabled\": true, \"description\": \"automation\"}"
$echo "$key"

Put the printed key in your values as AUTOMATION_TOKEN. From here on, data calls use it.

Check
$curl -sk -o /dev/null -w '%{http_code}' -H 'Authorization: Token ' https:///api/dcim/sites/
Expected output
200
Check
$curl -sk -o /dev/null -w '%{http_code}' -H 'Authorization: Token ' https:///api/users/users/
Expected output
403

Read and write

sites.py
import re
import pynetbox
nb = pynetbox.api("https://", token="")
for site in nb.dcim.sites.all():
    print(site.id, site.name, site.status)
slug = re.sub(r"[^a-z0-9]+", "-", "".lower()).strip("-")
site = nb.dcim.sites.get(slug=slug)
if site is None:
    site = nb.dcim.sites.create(name="", slug=slug, status="active")
    print("created", site.name)
elif site.update({"description": "managed by "}):
    print("updated", site.name)
else:
    print("unchanged", site.name)
$python sites.py && python sites.py
Check
$python -c 'import pynetbox; nb = pynetbox.api("https://", token=""); print(nb.dcim.sites.get(name="").status.value)'
Expected output
active

A device end to end

device.py
import re
import pynetbox
nb = pynetbox.api("https://", token="")
slug = lambda s: re.sub(r"[^a-z0-9]+", "-", s.lower()).strip("-")
def ensure(endpoint, lookup, **fields):
    """Get by unique fields, else create. Safe to run twice."""
    obj = endpoint.get(**lookup)
    return obj if obj is not None else endpoint.create(**fields)
site = ensure(nb.dcim.sites, {"slug": slug("")},
              name="", slug=slug(""), status="active")
mfr = ensure(nb.dcim.manufacturers, {"slug": "generic"}, name="Generic", slug="generic")
dtype = ensure(nb.dcim.device_types, {"slug": "generic-48p"},
               manufacturer=mfr.id, model="Generic 48-port switch", slug="generic-48p")
role = ensure(nb.dcim.device_roles, {"slug": "access-switch"},
              name="Access switch", slug="access-switch", color="2196f3")
dev = ensure(nb.dcim.devices, {"name": "sw-01", "site_id": site.id},
             name="sw-01", site=site.id, role=role.id, device_type=dtype.id, status="active")
iface = ensure(nb.dcim.interfaces, {"device_id": dev.id, "name": "Ethernet1"},
               device=dev.id, name="Ethernet1", type="1000base-t")
ip = ensure(nb.ipam.ip_addresses, {"address": "192.0.2.11/24"},
            address="192.0.2.11/24", status="active",
            assigned_object_type="dcim.interface", assigned_object_id=iface.id)
if dev.primary_ip4 is None or dev.primary_ip4.id != ip.id:
    dev.update({"primary_ip4": ip.id})
print("ok:", dev.name, "at", site.name, "primary", ip.address)
$python device.py
Check
$python -c 'import pynetbox; nb = pynetbox.api("https://", token=""); print(nb.dcim.devices.get(name="sw-01").primary_ip4.address)'
Expected output
192.0.2.11/24
If a create answers 400 or 403

A 400 comes with a JSON body naming the field: read it, it is precise ("slug": ["This field is required."], "status": ["\"Active\" is not a valid choice."]). A 403 on a create you expected to work is the constraint of the permission: the device was saved, checked against site__name, and rolled back. Check the site name in the permission against , spelling and case.

Bulk import from CSV

devices.csv
name,site,role,device_type,status
sw-02,paris-dc1,access-switch,generic-48p,planned
sw-03,paris-dc1,access-switch,generic-48p,planned
sw-04,paris-dc1,no-such-role,generic-48p,planned
import_devices.py
import csv
import pynetbox
nb = pynetbox.api("https://", token="")
created = skipped = failed = 0
with open("devices.csv", newline="") as f:
    for row in csv.DictReader(f):
        if nb.dcim.devices.get(name=row["name"], site=row["site"]) is not None:
            skipped += 1
            continue
        try:
            nb.dcim.devices.create(
                name=row["name"], status=row["status"],
                site={"slug": row["site"]}, role={"slug": row["role"]},
                device_type={"slug": row["device_type"]},
            )
            created += 1
        except pynetbox.RequestError as e:
            failed += 1
            print(f"FAILED {row['name']}: {e.error}")
print(f"created {created}, skipped {skipped}, failed {failed}")
$python import_devices.py
Check
$python import_devices.py | tail -1
Expected output
created 0, skipped 2, failed 1

Event rules and webhooks

$api=https:///api
$auth="Authorization: Token "
$curl -sk -H "$auth" -H 'Content-Type: application/json' "$api/extras/webhooks/" -d '{
$  "name": "device-changes", "payload_url": "",
$  "http_method": "POST", "http_content_type": "application/json"
$}'

With the webhook's id as action_object_id:

$curl -sk -H "$auth" -H 'Content-Type: application/json' "$api/extras/event-rules/" -d '{
$  "name": "device created or updated",
$  "object_types": ["dcim.device"], "event_types": ["object_created", "object_updated"],
$  "action_type": "webhook", "action_object_type": "extras.webhook", "action_object_id": ID,
$  "enabled": true
$}'

A receiver to see the payloads. python -m http.server will not do, it never shows request bodies; ten lines of Flask do:

receiver.py
from urllib.parse import urlparse
from flask import Flask, request
target = urlparse("")
app = Flask(__name__)
@app.post(target.path or "/")
def hook():
    body = request.get_json(force=True)
    print(body.get("event"), body.get("model"), body.get("data", {}).get("name"), flush=True)
    return "", 204
app.run(host="0.0.0.0", port=target.port or 80)
$pip install flask && python receiver.py

From the NetBox server, prove it can reach the receiver, then change a device and watch the terminal:

Check
$curl -s -o /dev/null -w '%{http_code}' -X POST -H 'Content-Type: application/json' -d '{"event": "test", "model": "none", "data": {}}'
Expected output
204
$python -c 'import pynetbox; nb = pynetbox.api("https://", token=""); nb.dcim.devices.get(name="sw-01").update({"description": "hello webhook"})'
Check
$curl -sk -H 'Authorization: Token ' "https:///api/extras/event-rules/?name=device%20created%20or%20updated" | python3 -c 'import sys,json; print(json.load(sys.stdin)["count"])'
Expected output
1
If nothing arrives

In order: systemctl is-active netbox-rq on the server; the curl check above from the server, not from your machine (a firewall between the two is the usual cause); the event rule is enabled and lists dcim.device; Operations → Background Tasks shows the job and its error. With an https:// receiver and a private CA, set ssl_verification to false on the webhook for the test, then fix the CA.

Custom scripts

create_devices.py
from dcim.choices import DeviceStatusChoices
from dcim.models import Device, DeviceRole, DeviceType, Site
from extras.scripts import IntegerVar, ObjectVar, Script, StringVar
class CreateDevices(Script):
    class Meta:
        name = "Create N devices"
        description = "Numbered devices at one site, planned status"
        commit_default = False
    site = ObjectVar(model=Site)
    role = ObjectVar(model=DeviceRole)
    device_type = ObjectVar(model=DeviceType)
    prefix = StringVar(default="sw-", description="Name prefix")
    count = IntegerVar(default=2, min_value=1, max_value=50)
    def run(self, data, commit):
        for i in range(1, data["count"] + 1):
            name = f"{data['prefix']}{i:02d}"
            device, created = Device.objects.get_or_create(
                name=name, site=data["site"],
                defaults={"role": data["role"], "device_type": data["device_type"],
                          "status": DeviceStatusChoices.STATUS_PLANNED},
            )
            self.log_success(f"created {name}" if created else f"exists {name}")
        return f"{data['count']} devices checked at {data['site']}"

Upload it in the UI: Customization → Scripts → Add, pick the file. Run it once from the form with Commit unticked to see the log, then from the API:

Sensitive command — runs a remote script. Review before running.
$api=https:///api
$auth="Authorization: Token "
$curl -sk -H "$auth" "$api/extras/scripts/" | python3 -m json.tool | grep -E '"(id|name)"'
$curl -sk -H "$auth" -H 'Content-Type: application/json' "$api/extras/scripts/ID/" -d '{
$  "data": {"site": SITE_ID, "role": ROLE_ID, "device_type": TYPE_ID, "prefix": "sw-", "count": 3},
$  "commit": true
$}'
Check
$curl -sk -H 'Authorization: Token ' https:///api/extras/scripts/ | grep -q CreateDevices && echo found
Expected output
found

Done

NetBox now has a user made for scripts with rights bounded to , you can create a device with its interface and IP in one idempotent run, import a CSV without fearing a bad row, other systems hear about device changes, and a script inside NetBox does the repetitive work from a form or from the API. The series ends here; the next ones put this to use in a NetDevOps lab.

Did everything work?

If you followed this page to the end on a real machine, say so. Your validation is dated and records your stack, so the next reader on the same path knows it still works.

This copy is read-only. To report that it works, or that it does not, open an issue

Only your stack choices are recorded, never your values. The pseudonym stays on this browser.