Making Requests¶
netbox_sdk supports two async request styles:
- the low-level
NetBoxApiClientfor direct HTTP calls - the higher-level facade returned by
api()for PyNetBox-style workflows - the versioned typed client returned by
typed_api()for validated Pydantic I/O
Both layers are async and can be used in the same project.
Facade usage¶
The facade is the shortest path for common NetBox SDK operations:
Retrieve a single record¶
Filter and iterate¶
devices = nb.dcim.devices.filter(site="lon", role="leaf-switch")
async for device in devices:
print(device.name)
Create, modify, and save¶
device = await nb.dcim.devices.create(
name="sw-lon-01",
site={"id": 1},
role={"id": 2},
device_type={"id": 3},
)
device.name = "sw-lon-01-renamed"
await device.save()
Selection custom fields (NetBox 4.7)¶
NetBox 4.7 returns selection and multiple-selection custom field values as
objects ({"value": "datacenter", "label": "Data Center"}). Writes still accept
the raw value. Copying a 4.7 read payload into a write body would submit the
object form. Use custom_fields_for_write(..., selection=...) so only named
selection fields are unwrapped. JSON custom fields, even ones whose payload
is exactly {value, label}, stay intact unless you name them:
from netbox_sdk import custom_fields_for_write
device = await nb.dcim.devices.get(42)
await nb.dcim.devices.patch(
42,
custom_fields=custom_fields_for_write(
device.custom_fields,
selection=("site_role", "regions"),
),
)
Use detail endpoints¶
prefix = await nb.ipam.prefixes.get(123)
available = await prefix.available_ips.list()
new_ip = await prefix.available_ips.create({})
new_prefix = await prefix.available_prefixes.create({"prefix_length": 28})
Rack elevation JSON vs SVG¶
rack = await nb.dcim.racks.get(5)
units = await rack.elevation.list()
svg = await rack.elevation.list(render="svg")
Plugins¶
plugins = await nb.plugins.installed_plugins()
async for olt in nb.plugins.gpon.olts.filter(status="active"):
print(olt.name)
Typed client usage¶
Use the typed client when you want validated request payloads, validated response payloads, and version-specific endpoint models.
from netbox_sdk import typed_api
nb = typed_api(
"https://netbox.example.com",
token="mytoken",
netbox_version="4.5",
)
Retrieve a typed record¶
Validate a request before HTTP¶
from netbox_sdk import TypedRequestValidationError
try:
await nb.ipam.prefixes.available_ips.create(7, body=[{"prefix_length": "invalid"}])
except TypedRequestValidationError as exc:
print(exc)
Version selection¶
typed_api("https://netbox.example.com", token="tok", netbox_version="4.5")
typed_api("https://netbox.example.com", token="tok", netbox_version="4.5.5")
The typed client supports the stable NetBox release lines 4.7, 4.6, 4.5, 4.4, and 4.3. Generated
models for those release lines are committed in the repository and shipped
with the package. The CI live-NetBox suite exercises v4.7.0, v4.6.6,
v4.6.3, v4.6.2, and v4.5.10.
Low-level client usage¶
Standard direct HTTP calls go through NetBoxApiClient.request(). Security-
sensitive protocols that must not use redirects or cached/stale responses can
use request_bounded().
Basic usage¶
from netbox_sdk import Config, NetBoxApiClient
cfg = Config(base_url="https://netbox.example.com", token_version="v1", token_secret="mytoken")
client = NetBoxApiClient(cfg)
HTTPS and TLS verification¶
Config accepts optional ssl_verify:
- Omitted or
None: verify HTTPS certificates (default), with CLI/TUI prompts on first verification failure if you have not stored a preference yet. True/False: always verify or disable verification for HTTPS (no interactive prompt on failure).
cfg = Config(
base_url="https://netbox.example.com",
token_version="v1",
token_secret="mytoken",
ssl_verify=False, # only for trusted lab instances; insecure on untrusted networks
)
client = NetBoxApiClient(cfg)
For environment variables, saved profiles, and interactive recovery, see Configuration — HTTPS and TLS verification.
Fresh bounded responses¶
response = await client.request_bounded(
"GET",
"/api/plugins/example/mcp/",
max_response_bytes=256 * 1024,
)
request_bounded() preserves the configured credential and normal timeout but
bypasses the filesystem cache and stale-if-error, disables redirects, rejects a
declared Content-Length above the bound, and aborts while streaming if the
decompressed body crosses the bound. It returns ApiResponse for every HTTP
status; callers must still reject unexpected statuses such as 3xx.
response.body_size_bytes preserves the streamed, decompressed byte count from
before character decoding so aggregate protocol budgets cannot be reduced by a
server-selected charset.
GET — list resources¶
response = await client.request("GET", "/api/dcim/devices/")
data = response.json()
print(data["count"]) # total number of results
print(data["results"]) # list of device objects
GET — filter results¶
response = await client.request(
"GET",
"/api/dcim/devices/",
query={"site": "lon", "role": "leaf-switch", "limit": "50"},
)
GET — retrieve single object¶
response = await client.request("GET", "/api/dcim/devices/42/")
device = response.json()
print(device["name"])
POST — create¶
response = await client.request(
"POST",
"/api/dcim/devices/",
payload={
"name": "sw-lon-01",
"device_type": {"id": 3},
"site": {"id": 1},
"role": {"id": 2},
},
)
if response.status == 201:
new_device = response.json()
print(f"Created device ID {new_device['id']}")
PUT — full update¶
response = await client.request(
"PUT",
"/api/dcim/devices/42/",
payload={
"name": "sw-lon-01-renamed",
"device_type": {"id": 3},
"site": {"id": 1},
"role": {"id": 2},
},
)
PATCH — partial update¶
response = await client.request(
"PATCH",
"/api/dcim/devices/42/",
payload={"name": "sw-lon-01-renamed"},
)
DELETE¶
response = await client.request("DELETE", "/api/dcim/devices/42/")
if response.status == 204:
print("Deleted")
Response object¶
request() always returns ApiResponse:
class ApiResponse(BaseModel):
status: int # HTTP status code
text: str # raw response body
headers: dict[str, str] # response headers
def json(self) -> Any: ... # parse text as JSON
Check the status code before parsing:
response = await client.request("GET", "/api/dcim/devices/99/")
if response.status == 404:
print("Device not found")
elif response.status == 200:
device = response.json()
Pagination¶
NetBox list endpoints return paginated results. Iterate pages manually:
async def list_all_devices(client: NetBoxApiClient) -> list[dict]:
results = []
offset = 0
limit = 100
while True:
response = await client.request(
"GET",
"/api/dcim/devices/",
query={"limit": str(limit), "offset": str(offset)},
)
data = response.json()
results.extend(data["results"])
if not data.get("next"):
break
offset += limit
return results
GraphQL¶
query = """
{
device_list(site: "lon") {
id
name
device_type { model }
primary_ip4 { address }
}
}
"""
response = await client.graphql(query)
data = response.json()["data"]["device_list"]
With variables:
query = "query($id: Int!) { device(id: $id) { name status } }"
response = await client.graphql(query, variables={"id": 42})
Connection health check¶
Before making API calls, verify connectivity:
from netbox_sdk import ConnectionProbe
probe = await client.probe_connection()
if probe.ok:
print(f"Connected — NetBox API {probe.version}")
else:
print(f"Failed (HTTP {probe.status}): {probe.error}")
probe_connection() returns ok=True for any reachable status below 400, or 403 (which means the URL is valid but the token is wrong — still reachable).
HTTP caching¶
GET requests to /api/... paths are automatically cached to disk under the
NetBox SDK config root, typically ~/.config/netbox-sdk/http-cache/. The
X-NBX-Cache response header shows the cache outcome:
| Value | Meaning |
|---|---|
MISS |
First fetch, stored to cache |
HIT |
Served from fresh cache |
REVALIDATED |
Server returned 304, cache TTL extended |
STALE |
Network error; stale data served |
Tune the cache policy:
from netbox_sdk.http_cache import CachePolicy, HttpCacheStore
from netbox_sdk.config import cache_dir
# Default: 60s fresh, 300s stale-if-error for list endpoints
store = HttpCacheStore(cache_dir())
POST/PUT/PATCH/DELETE requests are never cached. A successful (2xx)
mutating request also invalidates cached reads for the affected path: the
exact path, its containing collection/list path, and — for bulk operations,
which always target the list path — each written object's own detail path
(derived from the id fields in the bulk payload). This means a
read-before-write, write, read-after-write sequence never observes a stale
cached body for the object just written.
The cache key incorporates every request header that can change the response
representation for an otherwise-identical method/path/query, not just the
bearer token. In particular, a NetBox Branching X-NetBox-Branch header is
part of the cache identity, so the same token reading two different branch
schemas never collides on the same cached entry.
Schema-driven requests¶
Use netbox_sdk.services to resolve action names to HTTP calls:
from netbox_sdk import build_schema_index
from netbox_sdk.services import resolve_dynamic_request, run_dynamic_command
idx = build_schema_index()
# Resolve to a ResolvedRequest
req = resolve_dynamic_request(
idx, "dcim", "devices", "list",
object_id=None, query={"site": "lon"}, payload=None,
)
# req.method == "GET", req.path == "/api/dcim/devices/", req.query == {"site": "lon"}
# Or run it directly
response = await run_dynamic_command(
client, idx, "dcim", "devices", "list",
object_id=None,
query_pairs=["site=lon"],
body_json=None,
body_file=None,
)
Supported actions: list, get, create, update, patch, delete.
Multipart uploads¶
NetBoxApiClient.request() automatically switches to multipart form data when a
payload contains file-like values:
with open("rack-photo.png", "rb") as handle:
response = await client.request(
"POST",
"/api/extras/image-attachments/",
payload={
"object_type": "dcim.device",
"object_id": 42,
"name": "Front view",
"image": ("rack-photo.png", handle, "image/png"),
},
)
Supported file inputs include plain file objects and tuples in the form
(filename, file_obj) or (filename, file_obj, content_type).
Status and OpenAPI helpers¶
The low-level client now exposes convenience helpers for common API metadata:
Token provisioning is also available: