Netbox REST Task

task api name: rest

Sends a direct HTTP request to the configured NetBox REST API. Use this task when a dedicated NetBox worker task does not exist for the endpoint you need.

Inputs

Parameter	Required	Description
api	Yes	NetBox API path under /api, for example dcim/devices
method	No	HTTP method to use, default get
instance	No	NetBox instance name to target
**kwargs	No	Additional request arguments passed to requests, such as params, json, or data

Output

Returns the decoded JSON response from NetBox when possible. If the response is not JSON, the task returns response text or the HTTP status code.

Notes / Gotchas

• api is joined under the configured NetBox URL as /api/<api>/.
• Request headers, token, SSL verification, and default timeout are set by the worker.
• On HTTP errors, the task returns the response status code in result.
• The current NFCLI command model does not expose rest, so use the Python API or other task surfaces.

Examples

Python

Context manager:

from norfab.core.nfapi import NorFab

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

    result = client.run_job(
        "netbox",
        "rest",
        workers="any",
        kwargs={
            "api": "dcim/devices",
            "method": "get",
            "params": {"q": "ceos-leaf"},
        },
    )

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",
        "rest",
        workers="any",
        kwargs={
            "api": "extras/tags",
            "method": "post",
            "json": {
                "name": "automation-managed",
                "slug": "automation-managed",
            },
        },
    )
finally:
    nf.destroy()

Python API Reference

Sends a request to the Netbox REST API.

Parameters:

Name	Type	Description	Default
instance	str	The Netbox instance name to get parameters for.	None
method	str	The HTTP method to use for the request (e.g., 'get', 'post'). Defaults to "get".	'get'
api	str	The API endpoint to send the request to. Defaults to "".	''
**kwargs	Any	Additional arguments to pass to the request (e.g., params, data, json).	{}

Returns:

Type	Description
Result	Union[dict, list]: The JSON response from the API, parsed into a dictionary or list.

Raises:

Type	Description
HTTPError	If the HTTP request returned an unsuccessful status code.

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

@Task(
    input=RestInput,
    output=RestResult,
    fastapi={"methods": ["POST"], "schema": NetboxFastApiArgs.model_json_schema()},
    mcp={
        "annotations": {
            "title": "Call NetBox REST API",
            "readOnlyHint": False,
            "destructiveHint": True,
            "idempotentHint": False,
            "openWorldHint": True,
        }
    },
)
def rest(
    self,
    job: Job,
    instance: Union[None, str] = None,
    method: str = "get",
    api: str = "",
    **kwargs: Any,
) -> Result:
    """
    Sends a request to the Netbox REST API.

    Args:
        instance (str, optional): The Netbox instance name to get parameters for.
        method (str, optional): The HTTP method to use for the request (e.g., 'get', 'post'). Defaults to "get".
        api (str, optional): The API endpoint to send the request to. Defaults to "".
        **kwargs: Additional arguments to pass to the request (e.g., params, data, json).

    Returns:
        Union[dict, list]: The JSON response from the API, parsed into a dictionary or list.

    Raises:
        requests.exceptions.HTTPError: If the HTTP request returned an unsuccessful status code.
    """
    ret = Result(task=f"{self.name}:rest", result={})
    nb_params = self._get_instance_params(instance)
    api = api.strip("/")

    log.info(f"{self.name} - REST {method.upper()} '{nb_params['url']}/api/{api}/'")
    kwargs.setdefault(
        "timeout", (self.netbox_connect_timeout, self.netbox_read_timeout)
    )

    # send request to Netbox REST API
    response = getattr(self.netbox_http_session, method)(
        url=f"{nb_params['url']}/api/{api}/",
        headers={
            "Content-Type": "application/json",
            "Accept": "application/json",
            "Authorization": f"Token {nb_params['token']}",
        },
        verify=nb_params.get("ssl_verify", True),
        **kwargs,
    )

    try:
        response.raise_for_status()
    except Exception as e:
        log.error(
            f"{self.name} - REST {method.upper()} '{nb_params['url']}/api/{api}/' failed, status {response.status_code}, error: {e}"
        )
        ret.result = response.status_code
        return ret

    try:
        ret.result = response.json()
    except Exception as e:
        log.debug(f"Failed to decode json, error: {e}")
        ret.result = response.text if response.text else response.status_code

    return ret