Python for Network Engineers — Part 9: REST APIs — The Modern Control Plane

Netmiko connects over SSH and receives text. REST APIs connect over HTTPS and exchange JSON. As platforms have matured, REST APIs have become the preferred management interface — they’re faster, return structured data directly (no parsing required), support bulk operations, and can be called from any language or tool that speaks HTTP.

Every platform you work with either has a REST API today or is building towards one. Cisco DNA Centre, FortiManager, Arista CloudVision, Juniper Mist, Palo Alto Panorama, NSO — all REST. Even individual devices expose REST interfaces: Arista eAPI, Fortinet FortiOS REST API, Cisco IOS-XE RESTCONF. The engineer who can only talk CLI is increasingly limited.

This post uses the Arista cEOS lab from Parts 5–7 for live examples. The eAPI is already enabled in the startup configs we set up — no changes needed. We also walk through the FortiManager REST API in detail for readers managing Fortinet environments.

---

HTTP Fundamentals (Brief Refresher)

You understand networks. HTTP is just a protocol — you know what a request and response are. The parts that matter for API work:

Verbs define what you’re doing:

Verb	Meaning
GET	Retrieve data — no side effects
POST	Create a resource or trigger an action
PUT	Replace a resource entirely
PATCH	Update part of a resource
DELETE	Remove a resource

Not every API uses all verbs correctly. Some vendor APIs use POST for everything (FortiManager does). What matters is understanding what the API you’re working with expects.

Status codes tell you what happened:

Range	Meaning
2xx	Success
400	Bad request — your input is wrong
401	Unauthorised — authentication required or failed
403	Forbidden — authenticated but not allowed
404	Not found
429	Too many requests — you’re being rate limited
5xx	Server error — the API itself has a problem

Headers carry metadata: Content-Type: application/json tells the server your body is JSON; Authorization: Bearer <token> carries your auth token.

---

The requests Library

requests is the standard Python HTTP client. Clean API, sensible defaults, excellent documentation.

uv pip install requests

Making a GET Request

import requests

response = requests.get("https://httpbin.org/get")

print(response.status_code)    # 200
print(response.headers["Content-Type"])
data = response.json()          # parse JSON response body
print(data)

response.json() parses the response body as JSON and returns a Python dict or list. If the body isn’t valid JSON, it raises requests.exceptions.JSONDecodeError.

Query Parameters

Pass query parameters as a dict — requests encodes them into the URL:

response = requests.get(
    "https://api.example.com/devices",
    params={"limit": 100, "offset": 0, "status": "active"},
)
# Sends: GET /devices?limit=100&offset=0&status=active

POST with a JSON Body

payload = {
    "hostname": "new-device",
    "ip": "192.168.1.50",
    "type": "switch",
}

response = requests.post(
    "https://api.example.com/devices",
    json=payload,   # automatically sets Content-Type: application/json
)

Using json=payload is cleaner than data=json.dumps(payload) — requests sets the Content-Type header automatically.

Raise for Status

raise_for_status() raises an HTTPError exception if the response code is 4xx or 5xx. Call it immediately after every request — it’s the cleanest way to catch API errors without inspecting status_code manually:

response = requests.get("https://api.example.com/devices/999")
response.raise_for_status()   # raises HTTPError if 404 or 5xx
data = response.json()

---

Sessions

For multiple requests to the same host, use a Session. It persists headers, cookies, and connection settings across requests — reducing overhead and keeping your code clean:

session = requests.Session()
session.headers.update({
    "Content-Type": "application/json",
    "Accept": "application/json",
})
session.verify = False   # lab only — see SSL note below

response = session.get("https://192.168.100.10/rest/v1/interfaces")

You’ll use sessions for almost all API work. They’re the equivalent of keeping an SSH connection open rather than reconnecting for every command.

---

Authentication Patterns

Basic Auth

Username and password sent with every request, Base64-encoded:

from requests.auth import HTTPBasicAuth

response = requests.get(
    "https://api.example.com/devices",
    auth=HTTPBasicAuth("admin", "password"),
)

# Shorthand
response = requests.get(url, auth=("admin", "password"))

Arista eAPI uses basic auth. Fine for lab use, but credentials travel with every request — always use HTTPS.

Bearer Token (API Key)

The most common pattern for modern APIs. Authenticate once, receive a token, include it in a header for all subsequent requests:

session = requests.Session()
session.headers["Authorization"] = f"Bearer {api_token}"

response = session.get("https://api.example.com/devices")

Session-Based Auth (Login → Token → Use → Logout)

Some vendor APIs — FortiManager is a prominent example — require a login call that returns a session key, which you then include in every subsequent request:

# 1. Login
login_resp = session.post(api_url, json={
    "method": "exec",
    "params": [{"url": "/sys/login/user", "data": {"user": "admin", "passwd": "password"}}]
})
session_key = login_resp.json()["session"]

# 2. Use the session key
data_resp = session.post(api_url, json={
    "method": "get",
    "params": [{"url": "/dvmdb/device"}],
    "session": session_key,
})

# 3. Logout
session.post(api_url, json={
    "method": "exec",
    "params": [{"url": "/sys/logout"}],
    "session": session_key,
})

Use a context manager or try/finally to ensure logout happens even if the script fails mid-run — hanging sessions consume FortiManager licences.

---

Error Handling

import requests
from requests.exceptions import (
    ConnectionError,
    Timeout,
    HTTPError,
    JSONDecodeError,
)

try:
    response = session.get(url, timeout=30)
    response.raise_for_status()
    data = response.json()
except ConnectionError:
    print(f"Could not connect to {url} — host unreachable or port closed")
except Timeout:
    print(f"Request to {url} timed out after 30s")
except HTTPError as e:
    print(f"HTTP {e.response.status_code}: {e.response.text[:200]}")
except JSONDecodeError:
    print(f"Response was not valid JSON: {response.text[:200]}")

Always set timeout. Without it, a request to an unreachable host hangs indefinitely. Use a tuple (connect_timeout, read_timeout) for finer control:

response = session.get(url, timeout=(5, 30))
# 5 seconds to establish connection, 30 seconds to receive response

---

SSL Verification

By default, requests verifies the server’s TLS certificate against trusted CA roots. This is correct behaviour and you should not disable it in production.

In a lab with self-signed certificates (like our cEOS nodes), you’ll see an SSLError. The temptation is to set verify=False:

session.verify = False   # disables certificate verification

This makes the warning go away but exposes you to man-in-the-middle attacks. Use it only in isolated lab environments where you fully control the network path. In production, provide the CA certificate instead:

session.verify = "/path/to/ca-bundle.pem"

If you do use verify=False in a lab script, suppress the noisy InsecureRequestWarning:

import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

Add a comment explaining why — anyone reading the script later (including future you) should know this was a deliberate lab choice, not an oversight.

---

Pagination

Most APIs don’t return every result in one response. A management platform with 1,000 devices will return them in pages of 100 or 200. Your code needs to collect all pages:

def get_all(session, url, page_size=100, items_key="items"):
    """Fetch all pages from an offset/limit-paginated endpoint.

    Args:
        session: An authenticated requests.Session.
        url: The endpoint URL.
        page_size: Items per page.
        items_key: The JSON key containing the list of results.

    Yields:
        Individual items, one at a time.
    """
    offset = 0
    while True:
        response = session.get(url, params={"limit": page_size, "offset": offset})
        response.raise_for_status()
        data = response.json()

        items = data.get(items_key, [])
        yield from items

        if len(items) < page_size:
            break   # last page — fewer items than requested
        offset += page_size

Usage:

devices = list(get_all(session, "https://api.example.com/devices"))

The items_key parameter handles vendor differences — some return items, others return results, data, or devices. Check the API documentation or inspect the raw response to find the right key.

---

Arista eAPI

Arista EOS exposes a REST API called eAPI. It accepts show commands and returns structured JSON. The lab topology from Part 5 has eAPI enabled — the startup configs included:

management api http-commands
   no shutdown

eAPI lives at https://<device-ip>/command-api and accepts POST requests with a list of commands. Authentication is basic auth.

Enabling HTTPS on the cEOS Lab Nodes

SSH into each node and verify eAPI is accessible:

ssh [email protected]
# spine-01> show management api http-commands
# Enabled:            Yes
# HTTPS server:       running, set to use port 443

If HTTPS isn’t running, enable it:

spine-01# configure
spine-01(config)# management api http-commands
spine-01(config-mgmt-api-http-cmds)# protocol https
spine-01(config-mgmt-api-http-cmds)# no shutdown

Making eAPI Calls

import requests
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

EAPI_URL = "https://192.168.100.10/command-api"
AUTH = ("admin", "netauto")

def eapi_call(host, commands, auth=AUTH):
    """Send a list of commands to an Arista device via eAPI.
    
    Returns the list of response dicts, one per command.
    """
    url = f"https://{host}/command-api"
    payload = {
        "jsonrpc": "2.0",
        "method": "runCmds",
        "params": {
            "version": 1,
            "cmds": commands,
            "format": "json",
        },
        "id": 1,
    }
    response = requests.post(url, json=payload, auth=auth, verify=False, timeout=15)
    response.raise_for_status()
    result = response.json()
    
    if "error" in result:
        raise ValueError(f"eAPI error: {result['error']['message']}")
    
    return result["result"]

# Get version info
results = eapi_call("192.168.100.10", ["show version"])
version = results[0]
print(f"EOS version: {version['version']}")
print(f"Model:       {version['modelName']}")
print(f"Uptime:      {version['uptime']}")

# Get interface status — structured JSON, no TextFSM needed
results = eapi_call("192.168.100.10", ["show interfaces status"])
for intf_name, intf_data in results[0]["interfaceStatuses"].items():
    status = intf_data["linkStatus"]
    line_proto = intf_data["lineProtocolStatus"]
    print(f"{intf_name:<25} {status:<12} {line_proto}")

Compare this to the Netmiko equivalent from Part 5: send_command("show interfaces") returns a string that needs TextFSM or regex parsing. eapi_call("show interfaces status") returns a structured dict directly — no parsing needed.

Sending Multiple Commands in One Call

eAPI accepts a list of commands and returns a list of results — one result per command, in order. This reduces round trips:

results = eapi_call("192.168.100.10", [
    "show version",
    "show interfaces status",
    "show ip route summary",
])

version_data  = results[0]
intf_data     = results[1]
route_summary = results[2]

---

FortiManager REST API

FortiManager uses a JSON-RPC style API — all requests are POST to a single endpoint, with method and params in the body. It doesn’t follow strict REST conventions, but it’s consistent and well-documented.

Note: This section requires a FortiManager instance — physical appliance, VM (evaluation licences are available from Fortinet), or a lab FortiManager. The same code works against FortiManager Cloud.

Authentication Workflow

import requests
import urllib3
from contextlib import contextmanager

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


class FortiManagerAPI:
    """Simple FortiManager JSON-RPC client."""

    def __init__(self, host, username, password, verify=False):
        self.url = f"https://{host}/jsonrpc"
        self.session_key = None
        self._http = requests.Session()
        self._http.verify = verify
        self._http.headers.update({"Content-Type": "application/json"})
        self._username = username
        self._password = password

    def login(self):
        """Authenticate and store session key."""
        resp = self._http.post(self.url, json={
            "method": "exec",
            "params": [{"url": "/sys/login/user", "data": {
                "user": self._username,
                "passwd": self._password,
            }}],
            "id": 1,
        }, timeout=15)
        resp.raise_for_status()
        result = resp.json()

        if result["result"][0]["status"]["code"] != 0:
            raise ValueError(
                f"Login failed: {result['result'][0]['status']['message']}"
            )
        self.session_key = result["session"]

    def logout(self):
        """Close the session."""
        if self.session_key:
            self._http.post(self.url, json={
                "method": "exec",
                "params": [{"url": "/sys/logout"}],
                "session": self.session_key,
                "id": 1,
            }, timeout=10)
            self.session_key = None

    def call(self, method, url, data=None, params=None):
        """Make an authenticated API call.
        
        Returns the data portion of the result, or raises ValueError on error.
        """
        param = {"url": url}
        if data:
            param["data"] = data
        if params:
            param.update(params)

        resp = self._http.post(self.url, json={
            "method": method,
            "params": [param],
            "session": self.session_key,
            "id": 1,
        }, timeout=30)
        resp.raise_for_status()
        result = resp.json()

        status = result["result"][0]["status"]
        if status["code"] != 0:
            raise ValueError(f"API error {status['code']}: {status['message']}")

        return result["result"][0].get("data", {})

    def __enter__(self):
        self.login()
        return self

    def __exit__(self, *args):
        self.logout()

Using it as a context manager ensures logout always happens:

import os
from dotenv import load_dotenv

load_dotenv()

FMG_HOST = os.environ["FMG_HOST"]
FMG_USER = os.environ["FMG_USERNAME"]
FMG_PASS = os.environ["FMG_PASSWORD"]

with FortiManagerAPI(FMG_HOST, FMG_USER, FMG_PASS) as fmg:
    # List all managed devices
    devices = fmg.call("get", "/dvmdb/device")

    print(f"\n{'Name':<20} {'IP':<16} {'Platform':<20} {'Status'}")
    print("─" * 70)
    for device in devices:
        name     = device.get("name", "—")
        ip       = device.get("ip", "—")
        platform = device.get("os_type", "—")
        conn     = device.get("conn_status", "—")
        print(f"{name:<20} {ip:<16} {platform:<20} {conn}")

Common FortiManager API Calls

with FortiManagerAPI(FMG_HOST, FMG_USER, FMG_PASS) as fmg:

    # Get all ADOMs (administrative domains)
    adoms = fmg.call("get", "/dvmdb/adom")

    # Get policy packages in an ADOM
    packages = fmg.call("get", "/pm/pkg/adom/root")

    # Get all firewall policies in a package
    policies = fmg.call("get", "/pm/config/adom/root/pkg/default/firewall/policy")

    # Get device groups
    groups = fmg.call("get", "/dvmdb/group")

    # Get interface list for a specific device
    interfaces = fmg.call(
        "get",
        "/dvmdb/device/my-fortigate/vdom/root",
    )

Pagination in FortiManager

FortiManager uses range for pagination — a two-element list [offset, limit]:

def fmg_get_all(fmg, url, page_size=100):
    """Retrieve all records from a FortiManager endpoint, paginating automatically."""
    offset = 0
    while True:
        data = fmg.call("get", url, params={"range": [offset, page_size]})
        items = data if isinstance(data, list) else []
        yield from items
        if len(items) < page_size:
            break
        offset += page_size

with FortiManagerAPI(FMG_HOST, FMG_USER, FMG_PASS) as fmg:
    all_devices = list(fmg_get_all(fmg, "/dvmdb/device"))
    print(f"Total managed devices: {len(all_devices)}")

---

httpx for Async

httpx is a drop-in alternative to requests that supports both synchronous and asynchronous HTTP. The synchronous API is nearly identical to requests, so switching is low-friction:

uv pip install httpx

import httpx

with httpx.Client(verify=False) as client:
    response = client.get("https://192.168.100.10/command-api", auth=("admin", "netauto"))
    response.raise_for_status()

The async API becomes relevant when you’re querying many devices in parallel without a framework like Nornir:

import asyncio
import httpx

async def get_version(client, host):
    response = await client.post(
        f"https://{host}/command-api",
        json={"jsonrpc": "2.0", "method": "runCmds",
              "params": {"version": 1, "cmds": ["show version"], "format": "json"}, "id": 1},
        auth=("admin", "netauto"),
    )
    return host, response.json()["result"][0]["version"]

async def main():
    hosts = ["192.168.100.10", "192.168.100.11", "192.168.100.12"]
    async with httpx.AsyncClient(verify=False) as client:
        tasks = [get_version(client, h) for h in hosts]
        results = await asyncio.gather(*tasks)
        for host, version in results:
            print(f"{host}: {version}")

asyncio.run(main())

All three devices are queried simultaneously — total time is the slowest single device, not the sum of all devices. Part 11 (Nornir) handles this pattern for you automatically with a cleaner interface; the async code above is useful when you need the control or aren’t using Nornir.

---

Capstone: eAPI Inventory Collector

A complete script that collects version, interface status, and routing summary from all three lab devices via eAPI — no SSH, no TextFSM, structured data throughout.

"""Collect device state from Arista cEOS lab via eAPI."""

import json
import sys
import urllib3
from datetime import datetime, timezone
from pathlib import Path

import requests
from requests.exceptions import ConnectionError, Timeout, HTTPError

from inventory_loader import load_inventory

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

OUTPUT_FILE = "eapi_collected.json"


def eapi_call(host, commands, username, password):
    """Send commands to a device via eAPI. Returns list of result dicts."""
    url = f"https://{host}/command-api"
    payload = {
        "jsonrpc": "2.0",
        "method": "runCmds",
        "params": {"version": 1, "cmds": commands, "format": "json"},
        "id": 1,
    }
    response = requests.post(
        url, json=payload,
        auth=(username, password),
        verify=False,
        timeout=(5, 30),
    )
    response.raise_for_status()
    result = response.json()
    if "error" in result:
        raise ValueError(f"eAPI error: {result['error']['message']}")
    return result["result"]


def collect_device(device):
    """Collect version + interface state from one device. Returns a dict."""
    results = eapi_call(
        device.host,
        ["show version", "show interfaces status", "show ip route summary"],
        device.username,
        device.password,
    )
    version   = results[0]
    intfs     = results[1]["interfaceStatuses"]
    routes    = results[2]["vrfs"]["default"]

    up_intfs = [n for n, d in intfs.items() if d["linkStatus"] == "connected"]

    return {
        "name":          device.name,
        "host":          device.host,
        "collected_at":  datetime.now(timezone.utc).isoformat(),
        "eos_version":   version["version"],
        "model":         version["modelName"],
        "uptime_secs":   version["uptime"],
        "interfaces": {
            "total":     len(intfs),
            "connected": len(up_intfs),
            "up_list":   up_intfs,
        },
        "routes": {
            "total":    routes["totalRoutes"],
        },
    }


def main():
    devices = load_inventory("inventory.yaml")
    print(f"\nCollecting from {len(devices)} device(s) via eAPI...\n")

    results, failures = [], []

    for device in devices:
        print(f"  {device.name} ({device.host})... ", end="", flush=True)
        try:
            data = collect_device(device)
            results.append(data)
            print(
                f"OK — EOS {data['eos_version']}, "
                f"{data['interfaces']['connected']}/{data['interfaces']['total']} interfaces up, "
                f"{data['routes']['total']} routes"
            )
        except (ConnectionError, Timeout) as e:
            failures.append(device.name)
            print(f"FAILED — {e}")
        except (HTTPError, ValueError) as e:
            failures.append(device.name)
            print(f"FAILED — {e}")

    output = {
        "run_at":    datetime.now(timezone.utc).isoformat(),
        "collected": len(results),
        "failed":    failures,
        "devices":   results,
    }

    Path(OUTPUT_FILE).write_text(json.dumps(output, indent=2), encoding="utf-8")
    print(f"\nSaved to {OUTPUT_FILE}")

    if failures:
        sys.exit(1)


if __name__ == "__main__":
    main()

Run it with the lab up:

python3 eapi_collect.py

Collecting from 3 device(s) via eAPI...

  spine-01 (192.168.100.10)... OK — EOS 4.32.0F, 2/5 interfaces up, 3 routes
  leaf-01  (192.168.100.11)... OK — EOS 4.32.0F, 3/6 interfaces up, 4 routes
  leaf-02  (192.168.100.12)... OK — EOS 4.32.0F, 3/6 interfaces up, 4 routes

Saved to eapi_collected.json

The collected JSON contains richer, more structured data than the Netmiko approach with the same amount of Python. No TextFSM. No regex. The device did the parsing.

---

Netmiko vs REST API — When to Use Which

Factor	Netmiko (SSH)	REST API
Device support	Any SSH-accessible device	Requires API support
Output format	String (needs parsing)	Structured JSON
Multi-command	Sequential	Batch in one request
Config push	send_config_set()	PUT/PATCH/POST
Speed	One command at a time	Multiple commands per call
Availability	Universal	Platform-dependent

In practice: use REST when the platform supports it. Fall back to Netmiko for devices without a usable API, older kit, or when you need to run interactive commands the API doesn’t expose.

---

What’s Coming in Part 10

Part 10 covers NAPALM — a vendor-agnostic Python library that sits above Netmiko and REST APIs and provides a uniform interface for reading state and pushing configuration, regardless of vendor. The same Python code runs against Arista, Cisco, Juniper, and Fortinet without modification.

---

This post uses the same Containerlab topology as Parts 5–8. No new lab infrastructure required. The FortiManager section requires a separate FortiManager instance — evaluation licences are available from Fortinet.