Streaming Events

Polling the API for events can be time-consuming and is only as reliable as your polling script. As an alternative, you can use our streaming event API:

This API uses server-sent events to push events to you without using client-side polling. Streaming events are also more easily consumed programmatically, with less effort and overhead than polling solutions.

For example, the following shows a stream that begins with a user disabling and then enabling a port, and then proceeds to display errors related to the port metrics (the port used in this example was a lab port named “boop”):

Animated gif of the stream python script

The example above used the following script:

import asyncio
import datetime
import requests
import sys


LOGIN = f"{BASE_URL}/v2/auth/login"
LOGOUT = f"{BASE_URL}/v2/auth/logout"
EVENTS_BASE = f"{BASE_URL}/v2/events"

def login(username, password):
    print("Logging in...")
    resp =, json={"login": username, "password": password})
    print(f"Login response: {resp.json()}")
    return resp.json()["token"]

def logout(token):
    headers = {"Authorization": f"Bearer {token}"}
    requests.get(LOGOUT, headers=headers)
    print("Logged out")

def create_subscription(token):
    url = EVENTS_BASE
    headers = {"Authorization": f"Bearer {token}"}
    body = {
        "streams": [
           # {
           #     "type": "vc",
                # "vcs": ["PF-BC-MIA-LAS-1234567-PF&PF-AP-LAS1-987654"],
                # "events": ["metrics"]
           # },
                "type": "customer",
                # "events": ["auth", "physical_interface"]
                 "type": "port",
            #     "events": ["etherstats"],
                 "ifds": ["PF-AP-LAB6-2201365"]

    resp =, json=body, headers=headers)
        return resp.json()["subscription_uuid"]
    except KeyError:
        print("Error creating subscription")

async def consume_simple(token, subscription_uuid):
    headers = {"Authorization": f"Bearer {token}", "Accept": "text/event-stream"}
    url = f"{EVENTS_BASE}/{subscription_uuid}"
    r = requests.get(url, stream=True, headers=headers)
    for line in r.iter_lines():
        decoded_line = line.decode('utf-8')

if __name__ == "__main__":
    username = ""
    password = "ILoveKn0wledgebases!!"
    token = login(username, password)
    subscription_uuid = create_subscription(token)
    print(f"Subscription UUID: {subscription_uuid}")
    _ = input("Continue?")
        print(f"Starting stream at {datetime.datetime.utcnow().isoformat()}"), subscription_uuid))
        print(f"Ending stream at {datetime.datetime.utcnow().isoformat()}")