Skip to content

Repository files navigation

MISTAPI - Python Package for Mist API

PyPI versionPython versionsLicense: MIT

A comprehensive Python package to interact with the Mist Cloud APIs, built from the official Mist OpenAPI specifications.


Table of Contents


Features

Supported Mist Clouds

Support for all Mist cloud instances worldwide:

  • APAC: api.ac5.mist.com, api.gc5.mist.com, api.gc7.mist.com
  • EMEA: api.eu.mist.com, api.gc3.mist.com, api.ac6.mist.com, api.gc6.mist.com
  • Global: api.mist.com, api.gc1.mist.com, api.ac2.mist.com, api.gc2.mist.com, api.gc4.mist.com

Authentication

  • API token and username/password authentication (with 2FA support)
  • Environment variable configuration (.env file support)
  • HashiCorp Vault integration for secure credential storage
  • System keyring integration (macOS Keychain, Windows Credential Locker, etc.)
  • Interactive CLI prompts for credentials when needed

Core Features

  • Complete API Coverage: Auto-generated from OpenAPI specs
  • Async Support: Run any API call asynchronously with mistapi.arun() — no changes to existing code
  • Automatic Pagination: Built-in support for paginated responses
  • WebSocket Streaming: Real-time event streaming for devices, clients, and location data
  • Device Diagnostics: High-level, non-blocking utilities for ping, traceroute, ARP, BGP, OSPF, and more
  • Error Handling: Detailed error responses and logging
  • Proxy Support: HTTP/HTTPS proxy configuration
  • Log Sanitization: Automatic redaction of sensitive data in logs

Installation

Basic Installation

# Linux/macOS
python3 -m pip install mistapi
# Windows
py -m pip install mistapi

Upgrade to Latest Version

# Linux/macOS
python3 -m pip install --upgrade mistapi
# Windows
py -m pip install --upgrade mistapi

Installation with uv

uv is a fast Python package manager:

# Install in current project
uv add mistapi
# Or run directly without installing
uv run --with mistapi python my_script.py

Development Installation

# With pip
pip install -e ".[dev]"# With uv
uv sync

Requirements

  • Python 3.10 or higher
  • Dependencies: requests, python-dotenv, tabulate, deprecation, hvac, keyring, websocket-client

Quick Start

importmistapi# Initialize sessionapisession=mistapi.APISession()
# Authenticate (interactive prompt if credentials not configured)apisession.login()
# Use the API - Get device modelsdevice_models=mistapi.api.v1.const.device_models.listDeviceModels(apisession)
print(f"Found {len(device_models.data)} device models")
# Interactive organization selectionorg_id=mistapi.cli.select_org(apisession)[0]
# Get organization informationorg_info=mistapi.api.v1.orgs.orgs.getOrg(apisession, org_id)
print(f"Organization: {org_info.data['name']}")

Configuration

Configuration is optional - you can pass all parameters directly to APISession. However, using an .env file simplifies credential management.

Using Environment File

importmistapiapisession=mistapi.APISession(env_file="~/.mist_env")

Environment Variables

Create a .env file with your credentials:

MIST_HOST=api.mist.com
MIST_APITOKEN=your_api_token_here
# Alternative to API token# MIST_USER=your_email@example.com# MIST_PASSWORD=your_password# Proxy configuration# HTTPS_PROXY=http://user:password@myproxy.com:3128# Logging configuration# CONSOLE_LOG_LEVEL=20 # 0=Disabled, 10=Debug, 20=Info, 30=Warning, 40=Error, 50=Critical# LOGGING_LOG_LEVEL=10

Configuration Options

Environment VariableAPISession ParameterTypeDefaultDescription
MIST_HOSThoststringNoneMist Cloud API endpoint (e.g., api.mist.com)
MIST_APITOKENapitokenstringNoneAPI Token for authentication (recommended)
MIST_USERemailstringNoneUsername/email for authentication
MIST_PASSWORDpasswordstringNonePassword for authentication
MIST_KEYRING_SERVICEkeyring_servicestringNoneSystem keyring service name
MIST_VAULT_URLvault_urlstringNoneHashiCorp Vault URL
MIST_VAULT_PATHvault_pathstringNonePath to secret in Vault
MIST_VAULT_MOUNT_POINTvault_mount_pointstringNoneVault mount point
MIST_VAULT_TOKENvault_tokenstringNoneVault authentication token
CONSOLE_LOG_LEVELconsole_log_levelint20Console log level (0-50)
LOGGING_LOG_LEVELlogging_log_levelint10File log level (0-50)
HTTPS_PROXYhttps_proxystringNoneHTTP/HTTPS proxy URL
env_filestrNonePath to .env file

Authentication

The login() function must be called to authenticate. The package supports multiple authentication methods.

1. Interactive Authentication

If credentials are not configured, you'll be prompted interactively:

Cloud Selection:

----------------------------- Mist Cloud Selection -----------------------------
0) APAC 01 (host: api.ac5.mist.com)
1) EMEA 01 (host: api.eu.mist.com)
2) Global 01 (host: api.mist.com)
...
Select a Cloud (0 to 10, or q to exit):

Credential Prompt:

--------------------------- Login/Pwd authentication ---------------------------
Login: user@example.com
Password: [ INFO ] Authentication successful!
Two Factor Authentication code required: 123456
[ INFO ] 2FA authentication succeeded
-------------------------------- Authenticated ---------------------------------
Welcome Thomas Munzer!

2. Environment File Authentication

importmistapiapisession=mistapi.APISession(env_file="~/.mist_env")
apisession.login()
# Output:# -------------------------------- Authenticated ---------------------------------# Welcome Thomas Munzer!

3. HashiCorp Vault Authentication

importmistapiapisession=mistapi.APISession(
vault_url="https://vault.mycompany.com:8200",
vault_path="secret/data/mist/credentials",
vault_token="s.xxxxxxx"
)
apisession.login()

4. System Keyring Authentication

importmistapiapisession=mistapi.APISession(keyring_service="my_mist_service")
apisession.login()

Note: The keyring must contain: MIST_HOST, MIST_APITOKEN (or MIST_USER and MIST_PASSWORD)

5. Direct Parameter Authentication

importmistapiapisession=mistapi.APISession(
host="api.mist.com",
apitoken="your_token_here"
)
apisession.login()

API Requests Usage

Basic API Calls

# Get device models (constants)response=mistapi.api.v1.const.device_models.listDeviceModels(apisession)
print(f"Status: {response.status_code}")
print(f"Data: {len(response.data)} models")
# Get organization informationorg_info=mistapi.api.v1.orgs.orgs.getOrg(apisession, org_id)
print(f"Organization: {org_info.data['name']}")
# Get organization statisticsorg_stats=mistapi.api.v1.orgs.stats.getOrgStats(apisession, org_id)
print(f"Organization has {org_stats.data['num_sites']} sites")
# Search for devicesdevices=mistapi.api.v1.orgs.devices.searchOrgDevices(apisession, org_id, type="ap")
print(f"Found {len(devices.data['results'])} access points")

Error Handling

# Check response statusresponse=mistapi.api.v1.orgs.orgs.listOrgs(apisession)
ifresponse.status_code==200:
print(f"Success: {len(response.data)} organizations")
else:
print(f"Error {response.status_code}: {response.data}")
# Exception handlingtry:
org_info=mistapi.api.v1.orgs.orgs.getOrg(apisession, "invalid-org-id")
exceptExceptionase:
print(f"API Error: {e}")

Log Sanitization

The package automatically sanitizes sensitive data in logs:

importloggingfrommistapi.__loggerimportLogSanitizer# Configure loggingLOG_FILE="./app.log"logging.basicConfig(filename=LOG_FILE, filemode="w")
LOGGER=logging.getLogger(__name__)
LOGGER.setLevel(logging.DEBUG)
# Add sanitization filterLOGGER.addFilter(LogSanitizer())
# Sensitive data is automatically redactedLOGGER.debug({"user": "john", "apitoken": "secret123", "password": "pass456"})
# Output: {"user": "john", "apitoken": "****", "password": "****"}

Getting Help

# Get detailed help on any API functionhelp(mistapi.api.v1.orgs.stats.getOrgStats)

CLI Helper Functions

Interactive functions for selecting organizations and sites.

Organization Selection

# Select single organizationorg_id=mistapi.cli.select_org(apisession)[0]
# Select multiple organizationsorg_ids=mistapi.cli.select_org(apisession, allow_many=True)

Output:

Available organizations:
0) Acme Corp (id: 203d3d02-xxxx-xxxx-xxxx-76896a3330f4)
1) Demo Lab (id: 6374a757-xxxx-xxxx-xxxx-361e45b2d4ac)
Select an Org (0 to 1, or q to exit): 0

Site Selection

# Select site within an organizationsite_id=mistapi.cli.select_site(apisession, org_id=org_id)[0]

Output:

Available sites:
0) Headquarters (id: f5fcbee5-xxxx-xxxx-xxxx-1619ede87879)
1) Branch Office (id: a8b2c3d4-xxxx-xxxx-xxxx-987654321abc)
Select a Site (0 to 1, or q to exit): 0

Pagination Support

Get Next Page

# Get first pageresponse=mistapi.api.v1.orgs.clients.searchOrgClientsEvents(
apisession, org_id, duration="1d"
)
print(f"First page: {len(response.data['results'])} results")
# Get next pageifresponse.next:
response_2=mistapi.get_next(apisession, response)
print(f"Second page: {len(response_2.data['results'])} results")

Get All Pages Automatically

# Get all pages with a single callresponse=mistapi.api.v1.orgs.clients.searchOrgClientsEvents(
apisession, org_id, duration="1d"
)
print(f"First page: {len(response.data['results'])} results")
# Retrieve all remaining pagesall_data=mistapi.get_all(apisession, response)
print(f"Total results across all pages: {len(all_data)}")

Examples

Comprehensive examples are available in the Mist Library repository.

Device Management

# List all devices in an organizationdevices=mistapi.api.v1.orgs.devices.listOrgDevices(apisession, org_id)
# Get specific device detailsdevice=mistapi.api.v1.orgs.devices.getOrgDevice(
apisession, org_id, device_id
)
# Update device configurationupdate_data= {"name": "New Device Name"}
result=mistapi.api.v1.orgs.devices.updateOrgDevice(
apisession, device.org_id, device.id, body=update_data
)

Site Management

# Create a new sitesite_data= {
"name": "New Branch Office",
"country_code": "US",
"timezone": "America/New_York"
}
new_site=mistapi.api.v1.orgs.sites.createOrgSite(
apisession, org_id, body=site_data
)
# Get site statisticssite_stats=mistapi.api.v1.sites.stats.getSiteStats(apisession, new_site.id)

Client Analytics

# Search for wireless clientsclients=mistapi.api.v1.orgs.clients.searchOrgWirelessClients(
apisession, org_id, duration="1d",
limit=100
)
# Get client eventsevents=mistapi.api.v1.orgs.clients.searchOrgClientsEvents(
apisession, org_id,
duration="1h",
client_mac="aabbccddeeff"
)

Async Usage

All API functions in mistapi.api.v1 are synchronous by default. To use them in an asyncio context (e.g., FastAPI, aiohttp, or any async application) without blocking the event loop, use mistapi.arun().

arun() wraps any sync mistapi function in asyncio.to_thread(), running the blocking HTTP request in a thread pool while the event loop continues. No changes are needed to the existing API functions.

Running API Calls Asynchronously

importasyncioimportmistapifrommistapi.api.v1.sitesimportdevicesapisession=mistapi.APISession(env_file="~/.mist_env")
apisession.login()
asyncdefmain():
# Wrap any sync API call with mistapi.arun()response=awaitmistapi.arun(
devices.listSiteDevices, apisession, site_id
)
print(response.data)
asyncio.run(main())

Concurrent API Calls

Use asyncio.gather() to run multiple API calls concurrently:

importasyncioimportmistapifrommistapi.api.v1.orgsimportorgsfrommistapi.api.v1.sitesimportdevicesasyncdefmain():
org_info, site_devices=awaitasyncio.gather(
mistapi.arun(orgs.getOrg, apisession, org_id),
mistapi.arun(devices.listSiteDevices, apisession, site_id),
)
print(f"Org: {org_info.data['name']}")
print(f"Devices: {len(site_devices.data)}")
asyncio.run(main())

Combining with Device Utilities

Device utility functions are already non-blocking and return a UtilResponse that supports await. You can mix arun() for API calls and await for device utilities:

importasyncioimportmistapifrommistapi.api.v1.sitesimportdevicesfrommistapi.device_utilsimportexasyncdefmain():
# Start device utility — returns immediately, collects data in a background threadresponse=ex.retrieveArpTable(apisession, site_id, device_id)
# Meanwhile, run an API call via arun() — both execute concurrentlydevice_info=awaitmistapi.arun(
devices.getSiteDevice, apisession, site_id, device_id
)
print(f"Device: {device_info.data['name']}")
# Wait for the device utility background thread to finishawaitresponseprint(f"ARP entries: {len(response.ws_data)}")
asyncio.run(main())

WebSocket Streaming

The package provides a WebSocket client for real-time event streaming from the Mist API (wss://{host}/api-ws/v1/stream). Authentication is handled automatically using the same session credentials (API token or login/password).

Connection Parameters

All channel classes accept the following optional keyword arguments:

ParameterTypeDefaultDescription
ping_intervalint60Seconds between automatic ping frames. Set to 0 to disable pings.
ping_timeoutint | NoneNoneSeconds to wait for a pong response before treating the connection as dead. Defaults to min(45, ping_interval - 1) when pings are enabled, or 45 when ping_interval=0 (unused since pings are disabled). When ping_interval > 0, this must be lower than ping_interval.
auto_reconnectboolFalseAutomatically reconnect on transient failures using exponential backoff.
max_reconnect_attemptsint5Maximum number of reconnect attempts before giving up.
reconnect_backofffloat2.0Base backoff delay in seconds. Doubles after each failed attempt (2s, 4s, 8s, ...). Resets once the connection is fully established and all requested subscriptions are acknowledged.
queue_maxsizeint0Maximum messages buffered in the internal queues used for both receive() and callback delivery. 0 means unbounded. When set, incoming messages are dropped with a warning when either queue is full, preventing memory growth on high-frequency streams.
subscription_watchdog_timeoutfloat10.0Maximum time to wait for all channel_subscribed acknowledgements after connect. On timeout, the error is reported to on_error and the connection is closed; with auto_reconnect=True this triggers a clean reconnect.
rate_limit_backofffloat30.0Minimum reconnect delay after a 429 rate-limit response.
throughput_log_intervalint100Logs queue depth and processed counts every N messages. Set to 0 to disable periodic throughput logs.
ws=mistapi.websockets.sites.DeviceStatsEvents(
apisession,
site_ids=["<site_id>"],
ping_interval=60, # ping every 60 sping_timeout=45, # wait up to 45 s for pongauto_reconnect=True, # reconnect on transient failures
)
ws.connect()

Methods

MethodSignatureDescription
ws.on_open(cb)cb()Register callback for connection established
ws.on_message(cb)cb(data: dict)Register callback for incoming messages. Mutually exclusive with receive().
ws.on_error(cb)cb(error: Exception)Register callback for WebSocket errors
ws.on_close(cb)cb(code: int | None, msg: str | None)Register callback for connection close. Safe to call connect() from within.
ws.on_ping(cb)cb(message: str | bytes | None)Register callback for received ping frames.
ws.on_pong(cb)cb(message: str | bytes | None)Register callback for received pong frames.
ws.connect(run_in_background)Open the connection. True (default) runs in a daemon thread; False blocks.
ws.disconnect(wait, timeout)Close the connection. wait=True blocks until the background thread finishes.
ws.receive()-> Generator[dict]Blocking generator yielding messages. Mutually exclusive with on_message.
ws.ready()-> boolReturns True if the connection is open and ready

Available Channels

Organization Channels

ClassChannelDescription
mistapi.websockets.orgs.InsightsEvents/orgs/{org_id}/insights/summaryReal-time insights events for an organization
mistapi.websockets.orgs.MxEdgesStatsEvents/orgs/{org_id}/stats/mxedgesReal-time MX edges stats for an organization
mistapi.websockets.orgs.MxEdgesEvents/orgs/{org_id}/mxedgesReal-time MX edges events for an organization

Site Channels

ClassChannelDescription
mistapi.websockets.sites.ClientsStatsEvents/sites/{site_id}/stats/clientsReal-time clients stats for a site
mistapi.websockets.sites.DeviceCmdEvents/sites/{site_id}/devices/{device_id}/cmdReal-time device command events for a site
mistapi.websockets.sites.DeviceStatsEvents/sites/{site_id}/stats/devicesReal-time device stats for a site
mistapi.websockets.sites.DeviceEvents/sites/{site_id}/devicesReal-time device events for a site
mistapi.websockets.sites.MxEdgesStatsEvents/sites/{site_id}/stats/mxedgesReal-time MX edges stats for a site
mistapi.websockets.sites.PcapEvents/sites/{site_id}/pcapReal-time PCAP events for a site

Location Channels

ClassChannelDescription
mistapi.websockets.location.BleAssetsEvents/sites/{site_id}/stats/maps/{map_id}/assetsReal-time BLE assets location events
mistapi.websockets.location.ConnectedClientsEvents/sites/{site_id}/stats/maps/{map_id}/clientsReal-time connected clients location events
mistapi.websockets.location.SdkClientsEvents/sites/{site_id}/stats/maps/{map_id}/sdkclientsReal-time SDK clients location events
mistapi.websockets.location.UnconnectedClientsEvents/sites/{site_id}/stats/maps/{map_id}/unconnected_clientsReal-time unconnected clients location events
mistapi.websockets.location.DiscoveredBleAssetsEvents/sites/{site_id}/stats/maps/{map_id}/discovered_assetsReal-time discovered BLE assets location events

Session Channels

ClassChannelDescription
mistapi.websockets.session.SessionWithUrlCustom URLConnect to a custom WebSocket channel URL

Usage Patterns

Callback style (recommended)

connect() defaults to run_in_background=True and returns immediately. The WebSocket runs in a daemon thread, so your program must stay alive (e.g., with input() or an event loop). Messages are delivered to the registered callback in the background thread.

importmistapiapisession=mistapi.APISession(env_file="~/.mist_env")
apisession.login()
ws=mistapi.websockets.sites.DeviceStatsEvents(apisession, site_ids=["<site_id>"])
ws.on_message(lambdadata: print(data))
ws.connect() # non-blockinginput("Press Enter to stop")
ws.disconnect()

Generator style

Iterate over incoming messages as a blocking generator. Useful when you want to process messages sequentially in a loop.

ws=mistapi.websockets.sites.DeviceStatsEvents(apisession, site_ids=["<site_id>"])
ws.connect(run_in_background=True)
formsginws.receive(): # blocks, yields each message as a dictprint(msg)
ifsome_condition:
ws.disconnect() # stops the generator cleanly

Blocking style

connect(run_in_background=False) blocks the calling thread until the connection closes. Useful for simple scripts.

ws=mistapi.websockets.sites.DeviceStatsEvents(apisession, site_ids=["<site_id>"])
ws.on_message(lambdadata: print(data))
ws.connect(run_in_background=False) # blocks until disconnected

Context manager

disconnect() is called automatically on exit, even if an exception is raised.

importtimewithmistapi.websockets.sites.DeviceStatsEvents(apisession, site_ids=["<site_id>"]) asws:
ws.on_message(lambdadata: print(data))
ws.connect()
time.sleep(60)
# ws.disconnect() called automatically here

Device Utilities

mistapi.device_utils provides high-level utilities for running diagnostic commands on Mist-managed devices. Each function triggers a REST API call and streams the results back via WebSocket. The library handles the connection plumbing — you just call the function and get back a UtilResponse object.

Supported Devices

ModuleDevice TypeFunctions
device_utils.apMist Access Pointsping, traceroute, retrieveArpTable
device_utils.exJuniper EX Switchesping, monitorTraffic, topCommand, interactiveShell, createShellSession, retrieveArpTable, retrieveBgpSummary, retrieveDhcpLeases, releaseDhcpLeases, retrieveMacTable, clearMacTable, clearLearnedMac, clearBpduError, clearDot1xSessions, clearHitCount, bouncePort, cableTest
device_utils.srxJuniper SRX Firewallsping, monitorTraffic, topCommand, interactiveShell, createShellSession, retrieveArpTable, retrieveBgpSummary, retrieveDhcpLeases, releaseDhcpLeases, retrieveOspfDatabase, retrieveOspfNeighbors, retrieveOspfInterfaces, retrieveOspfSummary, retrieveSessions, clearSessions, bouncePort, retrieveRoutes
device_utils.ssrJuniper SSR Routersping, retrieveArpTable, retrieveBgpSummary, retrieveDhcpLeases, releaseDhcpLeases, retrieveOspfDatabase, retrieveOspfNeighbors, retrieveOspfInterfaces, retrieveOspfSummary, retrieveSessions, clearSessions, bouncePort, retrieveRoutes, showServicePath

Device Utilities Usage

All device utility functions are non-blocking: they trigger the REST API call, start a WebSocket stream in the background, and return a UtilResponse immediately. Your script can continue processing while data streams in.

Callback style

Pass an on_message callback to process each result as it arrives:

frommistapi.device_utilsimportexdefhandle(msg):
print("Live:", msg)
response=ex.retrieveArpTable(apisession, site_id, device_id, on_message=handle)
# returns immediately — on_message fires for each message in the backgrounddo_other_work()
response.wait() # block until streaming is completeprint(response.ws_data) # all collected data

Generator style

Iterate over processed messages as they arrive, similar to _MistWebsocket.receive():

response=ex.retrieveMacTable(apisession, site_id, device_id)
formsginresponse.receive(): # blocking generator, yields each messageprint(msg, end="", flush=True)
# loop ends when the WebSocket closesprint(response.ws_data)

Context manager

disconnect() is called automatically when the context exits:

withex.cableTest(apisession, site_id, device_id, port_id="ge-0/0/0") asresponse:
formsginresponse.receive():
print(msg, end="", flush=True)
# WebSocket disconnected, data readyprint(response.ws_data)

Polling

Check response.done to avoid blocking:

response=ex.retrieveBgpSummary(apisession, site_id, device_id)
whilenotresponse.done:
do_other_work()
print(response.ws_data)

Cancel early

Stop a long-running stream before it completes:

response=ex.monitorTraffic(apisession, site_id, device_id, port_id="ge-0/0/0")
do_some_work()
response.disconnect() # stop the WebSocketprint(response.ws_data) # data collected so far

Async await

Works in asyncio contexts without blocking the event loop:

importasynciofrommistapi.device_utilsimportexasyncdefmain():
response=ex.retrieveArpTable(apisession, site_id, device_id)
awaitresponse# non-blocking awaitprint(response.ws_data)
asyncio.run(main())

UtilResponse Object

All device utility functions return a UtilResponse object:

Attributes

AttributeTypeDescription
trigger_api_responseAPIResponseThe initial REST API response that triggered the device command. Contains status_code, data, and headers from the trigger request.
ws_requiredboolTrue if the command required a WebSocket connection to stream results (most diagnostic commands do). False if the REST response alone was sufficient.
ws_datalist[str]Parsed result data extracted from the WebSocket stream. This list is live — it grows as messages arrive in the background, even before wait() is called.
ws_raw_eventslist[str]Raw, unprocessed WebSocket event payloads as received from the Mist API. Useful for debugging or custom parsing.

Properties and Methods

Method / PropertyReturnsDescription
doneboolTrue if data collection is complete (or no WS was needed).
wait(timeout=None)UtilResponseBlock until data collection is complete. Returns self.
receive()GeneratorBlocking generator that yields each processed message as it arrives. Exits when the WebSocket closes.
disconnect()NoneStop the WebSocket connection early.
await responseUtilResponseNon-blocking await for asyncio contexts.

UtilResponse also supports the context manager protocol (with statement).

Enums

  • ap.TracerouteProtocolICMP, UDP (for ap.traceroute())
  • srx.Node / ssr.NodeNODE0, NODE1 (for dual-node devices)

Interactive Shell

interactiveShell() and createShellSession() provide SSH-over-WebSocket access to EX and SRX devices. Unlike the diagnostic utilities above, the shell is bidirectional — you send keystrokes and receive terminal output in real time.

Interactive mode (human at the keyboard)

Takes over the terminal. Blocks until the connection closes (e.g. after typing exit on the device). On Linux/macOS the terminal runs in raw mode, so Ctrl+C is forwarded to the device; on Windows, Ctrl+C ends the session locally:

frommistapi.device_utilsimportexex.interactiveShell(apisession, site_id, device_id)

Requires an interactive terminal (TTY); raises RuntimeError if stdin is piped or redirected. No extra package is needed.

Programmatic mode

Use createShellSession() to get a ShellSession object for scripting:

frommistapi.device_utilsimportexwithex.createShellSession(apisession, site_id, device_id) assession:
session.send_commands(["configure","show | display set | no-more", "exit"])
whileTrue:
data=session.recv(timeout=0.5)
ifdataisNone:
breakprint(data.decode("utf-8", errors="replace"), end="")

ShellSession API

Method / PropertyReturnsDescription
connect()NoneOpen the WebSocket connection. Called automatically by createShellSession().
disconnect()NoneClose the WebSocket connection.
connectedboolTrue if the WebSocket is currently connected.
send(data)NoneSend raw bytes (keystrokes) to the device.
send_text(text)NoneSend a text string to the device (auto-prefixed with \x00).
send_commands(commands)NoneSend a list of commands to the device, each is automatically followed by a newline.
recv(timeout=0.1)bytes | NoneReceive output from the device. Returns None on timeout or if disconnected.
resize(rows, cols)NoneSend a terminal resize message.

ShellSession also supports the context manager protocol (with statement).


Development and Testing

Development Setup

# Clone the repository
git clone https://github.com/tmunzer/mistapi_python.git
cd mistapi_python
# With pip
pip install -e ".[dev]"# With uv
uv sync

Running Tests

# Run all tests
pytest
# or with uv
uv run pytest
# Run with coverage report
pytest --cov=src/mistapi --cov-report=html
# Run specific test file
pytest tests/unit/test_api_session.py
# Run linting
ruff check src/
# or with uv
uv run ruff check src/

Package Structure

src/mistapi/
├── __init__.py # Main package exports (lazy-loads api, cli, utils, websockets)
├── __api_session.py # Session management and authentication
├── __api_request.py # HTTP request handling
├── __api_response.py # Response parsing and pagination
├── __logger.py # Logging and sanitization
├── __pagination.py # Pagination utilities
├── cli.py # Interactive CLI functions
├── __models/ # Data models
│ ├── __init__.py
│ └── privilege.py
├── api/v1/ # Auto-generated API endpoints
│ ├── const/ # Constants and enums
│ ├── orgs/ # Organization-level APIs
│ ├── sites/ # Site-level APIs
│ ├── login/ # Authentication APIs
│ └── utils/ # Utility functions
├── device_utils/ # Device utility implementations
│ ├── ap.py # Access Point utilities
│ ├── ex.py # EX Switch utilities
│ ├── srx.py # SRX Firewall utilities
│ ├── ssr.py # Session Smart Router utilities
│ └── ... # Function-based modules (arp, bgp, dhcp, etc.)
└── websockets/ # Real-time WebSocket streaming
├── __ws_client.py # Base WebSocket client
├── orgs.py # Organization-level channels
├── sites.py # Site-level channels
├── location.py # Location/map channels
└── session.py # Custom URL session channel

Contributing

Contributions are welcome! Please follow these guidelines:

How to Contribute

  1. Fork the repository
  2. Create a feature branch
    git checkout -b feature/amazing-feature
  3. Commit your changes
    git commit -m 'Add amazing feature'
  4. Push to the branch
    git push origin feature/amazing-feature
  5. Open a Pull Request

Development Guidelines

  • Write tests for new features
  • Ensure all tests pass before submitting PR
  • Follow existing code style and conventions
  • Update documentation as needed
  • Add entries to CHANGELOG.md for significant changes

License

MIT License

Copyright (c) 2023 Thomas Munzer

Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.


Links

About

Python package to simplify the Mist System APIs usage

Topics

Resources

Stars

16 stars

Watchers

4 watching

Forks

Releases

Used by

Contributors

Languages