Python 3.11+ • Typed • Sync + Async • Pydantic v2

The Netskope
Python SDK

A modern, typed, and intuitive interface to the Netskope REST API v2. Auto-pagination, retries, and rich models — so you can focus on security, not HTTP plumbing.

pip install netskope-py-sdk
Get Started →

Built for Developers & AI Agents

Every design decision optimized for discoverability, safety, and minimal surprise.

NS

Hierarchical Namespaces

Navigate the entire API through autocomplete: client.alerts.list(), client.scim.users.create()

AP

Automatic Pagination

Just iterate. No page loops, no offset math. Works across all list endpoints with lazy fetching.

TS

Full Type Safety

Pydantic v2 models for every response. Full py.typed support for mypy and pyright.

RT

Automatic Retries

Exponential backoff with jitter. Respects Retry-After headers. Configurable per client.

SA

Sync + Async

Choose NetskopeClient for scripts or AsyncNetskopeClient for high-throughput.

EX

Rich Exceptions

Specific exception types with request IDs. Catch RateLimitError, NotFoundError precisely.

Documentation

Installation

pip install netskope-py-sdk

The SDK has only two dependencies: httpx and pydantic. Python 3.11 or newer is required.

Development Installation

# Clone and install with test dependencies git clone https://github.com/netSkopeoss/netskope-py-sdk.git cd netskope-py-sdk pip install -e ".[dev]"

Verify Installation

import netskope print(netskope.__version__) # "1.1.0"

Quick Start

from netskope import NetskopeClient # Initialize — reads NETSKOPE_TENANT and NETSKOPE_API_TOKEN env vars client = NetskopeClient() # Or pass credentials explicitly client = NetskopeClient( tenant="mycompany.goskope.com", api_token="your-v2-api-token", ) # List alerts — pagination is completely automatic for alert in client.alerts.list(): print(f"{alert.alert_name} — {alert.user} — {alert.app}") # Query network events with JQL filtering for event in client.events.list("application", page_size=50): print(f"{event.user} → {event.app} ({event.activity})") # Manage URL allow/block lists blocklist = client.url_lists.create("threat-iocs", ["malware.example.com"]) client.url_lists.deploy() # List publishers and their status for pub in client.publishers.list(): print(f"{pub.publisher_name}: {pub.status} ({pub.apps_count} apps)") # SCIM user provisioning for user in client.scim.users.list(): print(f"{user.user_name} active={user.active}") # Always close when done (or use context manager) client.close()

Context Manager (Recommended)

with NetskopeClient(tenant="mycompany.goskope.com", api_token="...") as client: alerts = client.alerts.list().to_list(max_items=100) print(f"Found {len(alerts)} alerts") # Connection pool automatically closed

Real-World Examples

SIEM Integration — Ship Events to Splunk/Sentinel

from netskope import NetskopeClient from datetime import datetime, timedelta client = NetskopeClient() end = datetime.now() start = end - timedelta(hours=1) # Stream all event types from the last hour for event_type in ["application", "network", "page", "alert"]: for event in client.events.list(event_type, start_time=start, end_time=end): send_to_siem(event.model_dump(exclude_none=True)) client.close()

Automated Threat Response — Block IOCs from Feed

from netskope import NetskopeClient client = NetskopeClient() # Get IOCs from your threat intel feed malicious_urls = fetch_threat_intel() # → ["malware.com", "phish.org", ...] # Create or update the blocklist try: blocklist = client.url_lists.create( name="auto-threat-intel", urls=malicious_urls, list_type="exact", ) except Exception: # List already exists — find and update it for ul in client.url_lists.list(): if ul.name == "auto-threat-intel": client.url_lists.update(ul.id, urls=malicious_urls) break # Deploy immediately client.url_lists.deploy() print(f"Blocked {len(malicious_urls)} malicious URLs")

Infrastructure Health Check

from netskope import NetskopeClient with NetskopeClient() as client: publishers = client.publishers.list().to_list() connected = [p for p in publishers if p.status == "connected"] down = [p for p in publishers if p.status != "connected"] print(f"Publishers: {len(connected)} connected, {len(down)} down") for p in down: print(f" ⚠ {p.publisher_name} is {p.status}") apps = client.private_apps.list().to_list() print(f"Private apps: {len(apps)}")

User Lifecycle Automation

from netskope import NetskopeClient client = NetskopeClient() # Onboard new employees from HR system new_hires = [ {"email": "alice@company.com", "name": "Alice Smith"}, {"email": "bob@company.com", "name": "Bob Jones"}, ] for hire in new_hires: user = client.scim.users.create( user_name=hire["email"], email=hire["email"], display_name=hire["name"], ) print(f"Provisioned {user.user_name} (id={user.id})") # Offboard departing employees for email in departing_employees: users = client.scim.users.list( filter_expr=f'userName eq "{email}"' ).to_list() for user in users: client.scim.users.update(user.id, active=False) print(f"Deactivated {user.user_name}")

High-Throughput Async Processing

import asyncio from netskope import AsyncNetskopeClient async def monitor_all_event_types(): async with AsyncNetskopeClient() as client: event_types = ["application", "network", "page", "alert"] # Fetch all event types concurrently tasks = [ client.events.list(et, page_size=100).to_list(max_items=500) for et in event_types ] results = await asyncio.gather(*tasks) for et, events in zip(event_types, results): print(f"{et}: {len(events)} events") asyncio.run(monitor_all_event_types())

Configuration

NetskopeClient Constructor

ParameterTypeDefaultDescription
tenantstr | NoneNETSKOPE_TENANTTenant hostname, e.g. "mycompany.goskope.com". Falls back to the NETSKOPE_TENANT environment variable.
api_tokenstr | NoneNETSKOPE_API_TOKENREST API v2 token from your Netskope admin console. Falls back to the NETSKOPE_API_TOKEN environment variable.
timeoutfloat30.0HTTP request timeout in seconds.
max_retriesint3Maximum automatic retries for transient errors (429, 5xx).
backoff_factorfloat0.5Exponential backoff base multiplier. Sleep = factor * 2^attempt, capped at 60s, plus jitter.
retry_on_statusfrozenset[int] | None{429,500,502,503,504}HTTP status codes that trigger automatic retry.

Credential Resolution Chain

Credentials resolve in priority order (inspired by boto3):

1Explicit constructor parameters (tenant=, api_token=)
2Environment variables: NETSKOPE_TENANT, NETSKOPE_API_TOKEN

If neither source provides a value, a ValidationError is raised at construction time (fail fast).

Environment Variables

# Set credentials via environment export NETSKOPE_TENANT="mycompany.goskope.com" export NETSKOPE_API_TOKEN="your-v2-token" # Then just: client = NetskopeClient() # reads from env automatically

Multiple Tenants

# Manage multiple tenants simultaneously prod = NetskopeClient(tenant="prod.goskope.com", api_token=prod_token) staging = NetskopeClient(tenant="staging.goskope.com", api_token=staging_token) # Compare publisher counts across environments prod_pubs = prod.publishers.list().to_list() staging_pubs = staging.publishers.list().to_list() print(f"Production: {len(prod_pubs)} publishers") print(f"Staging: {len(staging_pubs)} publishers")

Client Properties

client = NetskopeClient(tenant="mycompany.goskope.com", api_token="...") client.version # "1.1.0" client.tenant # "mycompany.goskope.com" client.base_url # "https://mycompany.goskope.com"

SSL / Custom CA

The verify constructor option controls TLS verification. It accepts:

ValueBehavior
True (default)Verify against the system trust store
FalseDisable verification (not recommended)
"path/to/ca.pem"Verify against a custom CA bundle file

When your traffic passes through Netskope SSL inspection, the certificate chain is re-signed by the Netskope client CA. Use find_netskope_ca_cert() to locate the installed CA bundle automatically (it checks the well-known Netskope agent install paths on macOS, Linux, and Windows and returns the first existing path, or None):

from netskope import NetskopeClient, find_netskope_ca_cert # Trust the Netskope SSL-inspection CA if present, else fall back to system store client = NetskopeClient(verify=find_netskope_ca_cert() or True) # Or point at an explicit bundle client = NetskopeClient(verify="/etc/ssl/certs/corporate-ca.pem")

If verify is left at its default, the SDK also honors the following CA-bundle environment variables, checked in order:

# CA bundle env-var resolution chain (first match wins) export NETSKOPE_CA_BUNDLE="/path/to/netskope-ca.pem" # 1. Netskope-specific export REQUESTS_CA_BUNDLE="/path/to/ca.pem" # 2. requests convention export SSL_CERT_FILE="/path/to/ca.pem" # 3. OpenSSL convention export CURL_CA_BUNDLE="/path/to/ca.pem" # 4. curl convention

Async Usage

The AsyncNetskopeClient provides the exact same API as the sync client, but all methods are coroutines and list endpoints return async iterators.

import asyncio from netskope import AsyncNetskopeClient async def main(): async with AsyncNetskopeClient(tenant="mycompany.goskope.com", api_token="...") as client: # Async iteration async for alert in client.alerts.list(page_size=50): print(alert.alert_name) # Async to_list publishers = await client.publishers.list().to_list(max_items=100) # Async first first_alert = await client.alerts.list().first() # Async page-level access async for page in client.alerts.list().pages(): print(f"Page: {len(page.items)} items") # Async CRUD url_list = await client.url_lists.create("blocklist", ["bad.com"]) await client.url_lists.deploy() asyncio.run(main())
Note: Every method available on NetskopeClient is also available on AsyncNetskopeClient with the same signature. The only difference is await for single-resource methods and async for for iterators.

Alerts

client.alerts — Query and retrieve security alerts via /api/v2/events/datasearch/alert

.list() — List Alerts

ParameterTypeDefaultDescription
querystr | NoneNoneJQL filter expression, e.g. 'alert_type eq "DLP"'
fieldslist[str] | NoneNoneSpecific fields to return (reduces response size)
start_timedatetime | int | NoneNoneStart of time range (datetime object or Unix epoch int)
end_timedatetime | int | NoneNoneEnd of time range
group_bystr | NoneNoneField to aggregate results by
order_bystr | NoneNoneField to sort by
descendingboolTrueSort order (True = newest first)
page_sizeint100Results per API call

Returns: SyncPaginatedResponse[Alert] — a lazy iterator of Alert objects.

from datetime import datetime, timedelta # List all alerts (auto-paginated) for alert in client.alerts.list(): print(f"{alert.alert_name} — {alert.alert_type} — {alert.app}") # Filter by JQL query for alert in client.alerts.list(query='alert_type eq "DLP"'): print(alert.alert_name) # Time-range query (last 7 days) alerts = client.alerts.list( start_time=datetime.now() - timedelta(days=7), end_time=datetime.now(), page_size=200, ).to_list(max_items=1000) print(f"Found {len(alerts)} alerts in the last 7 days") # Get just the first alert first = client.alerts.list().first() # → Alert(alert_name="Password should be changed every 90 days", ...)

.get(alert_id) — Get Single Alert

Retrieve a single alert by its _id. Raises NotFoundError if not found.

from netskope.exceptions import NotFoundError try: alert = client.alerts.get("44d8af1bcc1d7bd0fc8aa698") print(f"Name: {alert.alert_name}") print(f"Type: {alert.alert_type}") print(f"App: {alert.app}") except NotFoundError: print("Alert not found")

Alert Model Fields

FieldTypeDescription
idstr | NoneAlert unique identifier (API field: _id)
alert_namestr | NoneHuman-readable alert name
alert_typestr | NoneAlert category (DLP, malware, anomaly, Security Assessment, etc.)
severitystr | NoneSeverity level (critical, high, medium, low)
userstr | NoneAffected user email
appstr | NoneApplication name (e.g. "Workday", "Slack")
activitystr | NoneActivity that triggered the alert
policy_namestr | NonePolicy that generated the alert
actionstr | NoneAction taken (alert, block, etc.)
categorystr | NoneURL/app category
cciint | NoneCloud Confidence Index score
cclstr | NoneCloud Confidence Level
access_methodstr | NoneHow traffic was accessed (Client, API, etc.)
traffic_typestr | NoneTraffic classification
timestampdatetime | NoneWhen the alert occurred (auto-parsed from epoch)

Events

client.events — Query security events across 10 event types via /api/v2/events/datasearch/{type}

Supported Event Types

application
network
page
alert
incident
audit
infrastructure
clientstatus
epdlp
transaction

.list(event_type, ...) — List Events

Same parameters as alerts.list(), plus the required event_type first argument. Returns type-specific models for network, page, and audit.

from netskope.models.events import EventType # Application events — SaaS/cloud app access for event in client.events.list(EventType.APPLICATION, page_size=50): print(f"{event.user} → {event.app} ({event.activity})") # → onasr+web@netskope.com → Amazon Systems Manager (View) # Network events — returns NetworkEvent with src_ip, dst_ip, protocol for event in client.events.list("network", page_size=20): print(f"{event.src_ip} → {event.dst_ip} ({event.protocol})") # Page events — returns PageEvent with url, domain, browser for event in client.events.list("page"): print(f"{event.url} — {event.domain}") # Filter by JQL and time range from datetime import datetime, timedelta events = client.events.list( "application", query='user eq "alice@example.com"', start_time=datetime.now() - timedelta(days=1), end_time=datetime.now(), ).to_list(max_items=500)

Type-Specific Models

The SDK automatically returns the right model subclass based on event type:

Event TypeModelExtra Fields
"network"NetworkEventsrc_ip, dst_ip, src_port, dst_port, protocol, num_bytes, domain
"page"PageEventurl, domain, page_id, page_duration, referer, browser, os, device
"audit"AuditEventaudit_log_event, audit_category, supporting_data, organization_unit
All othersEventBase fields: user, app, activity, action, site, category, severity, timestamp, etc.

URL Lists

client.url_lists — Full CRUD for URL allow/block lists and policy deployment via /api/v2/policy/urllist

# List all URL lists for url_list in client.url_lists.list(): print(f"[{url_list.id}] {url_list.name}: {len(url_list.urls)} URLs ({url_list.type})") # → [1] Exceptions to prohibited sites: 5 URLs (exact) # → [3] YouTube Allowed: 2 URLs (exact) # → [5] malwareDomains: 26874 URLs (exact) # Get a specific URL list by ID url_list = client.url_lists.get(1) # Create a new URL list new_list = client.url_lists.create( name="threat-intel-iocs", urls=["malware.example.com", "phishing.bad.org"], list_type="exact", # or "regex" ) print(f"Created: id={new_list.id}, name={new_list.name}") # Update — pass only the fields you want to change. # update() GETs the current list and merges your changes over the existing # name/urls/type, since the API requires all three on every PUT. updated = client.url_lists.update( new_list.id, urls=["malware.example.com", "phishing.bad.org", "new-threat.evil.net"], ) # Deploy all pending policy changes (required after create/update/delete) client.url_lists.deploy() # Delete client.url_lists.delete(new_list.id) client.url_lists.deploy()
Important: Changes to URL lists are staged until you call client.url_lists.deploy(). This matches the Netskope admin console behavior — create, update, or delete, then deploy all changes at once.

UrlList Model Fields

FieldTypeDescription
idint | NoneNumeric identifier
namestr | NoneHuman-readable name
typestr | NoneMatching strategy: "exact" or "regex"
urlslist[str]The URL entries
pendingbool | NoneWhether changes await deployment
modify_bystr | NoneLast modified by (admin email)

Publishers

client.publishers — Manage private-access gateway publishers via /api/v2/infrastructure/publishers

# List all publishers for pub in client.publishers.list(): print(f"{pub.publisher_name}: {pub.status} ({pub.apps_count} apps)") # → flonkerton-sedemo-publisher: connected (6 apps) # → NPA-Unified-Pub-US-EAST-1: connected (18 apps) # → NPA-Unified-Pub-US-WEST-1: connected (7 apps) # Get by ID pub = client.publishers.get(109) print(f"{pub.publisher_name} — registered={pub.registered}") # Create a new publisher new_pub = client.publishers.create(name="aws-us-east-1-publisher") # Update client.publishers.update(new_pub.publisher_id, name="aws-us-east-1-primary") # Delete client.publishers.delete(new_pub.publisher_id)

Publisher Model Fields

FieldTypeDescription
publisher_idint | NoneNumeric identifier
publisher_namestr | NoneHuman-readable name
statusstr | None"connected" or "not_connected"
apps_countint | NoneNumber of apps assigned to this publisher
registeredbool | NoneRegistration status
sticky_ip_enabledbool | NoneWhether sticky IP is enabled

Private Apps (ZTNA)

client.private_apps — Manage zero-trust private applications via /api/v2/steering/apps/private

# List all private apps for app in client.private_apps.list(): print(f"{app.app_name} → {app.host}:{app.port}") # → [Portal Test] → npaportal.mynetskopedemo.com:443 # → [Finance MyNetskopeDemo] → finance.mynetskopedemo.com:443 # Get by ID app = client.private_apps.get(110) # Create a new private app app = client.private_apps.create( name="internal-dashboard", host="10.0.0.5", port="443", protocols=["TCP"], publisher_ids=[109, 126], ) # Update client.private_apps.update(app.app_id, app_name="internal-dashboard-v2") # Delete client.private_apps.delete(app.app_id)

PrivateApp Model Fields

FieldTypeDescription
app_idint | NoneNumeric identifier
app_namestr | NoneApplication display name
hoststr | NoneTarget host (IP or hostname)
portstr | NoneTarget port(s)
protocolslist | NoneProtocol configuration
publisherslist[dict] | NoneAssigned publishers
use_publisher_dnsbool | NoneUse publisher DNS resolution
clientless_accessbool | NoneBrowser-based access enabled

SCIM Users & Groups

client.scim — SCIM 2.0 user and group provisioning via /api/v2/scim/Users and /api/v2/scim/Groups

Users — client.scim.users

# List all SCIM users for user in client.scim.users.list(): print(f"{user.user_name} active={user.active} display={user.display_name}") # Filter users with SCIM filter syntax admins = client.scim.users.list( filter_expr='userName co "admin"', ).to_list() # Get a specific user user = client.scim.users.get("user-uuid-here") # Create a new user user = client.scim.users.create( user_name="alice@company.com", email="alice@company.com", display_name="Alice Smith", given_name="Alice", family_name="Smith", active=True, ) print(f"Created user: {user.id}") # Update (PATCH) — only send changed fields client.scim.users.update(user.id, active=False) # Delete client.scim.users.delete(user.id)

Groups — client.scim.groups

# List all groups for group in client.scim.groups.list(): print(f"{group.display_name}: {len(group.members)} members") # Create a group with members group = client.scim.groups.create( display_name="Engineering", member_ids=["user-id-1", "user-id-2"], ) # Update group (PUT — full replacement) client.scim.groups.update(group.id, display_name="Platform Engineering") # Delete client.scim.groups.delete(group.id)

ScimUser Model Fields

FieldTypeDescription
idstr | NoneSCIM unique identifier
user_namestr | NoneUsername (typically email)
display_namestr | NoneDisplay name
activebool | NoneAccount status
emailslist[ScimEmail]Email addresses (each has value, primary, type)
groupslist[dict]Group memberships

Incidents

client.incidents — View incidents, update status, get risk scores, DLP forensics, and UBA anomalies

# List incidents with JQL filtering for incident in client.incidents.list(query='severity eq "critical"'): print(f"{incident.incident_id} — {incident.severity} — {incident.status}") # Update an incident (with optimistic locking) client.incidents.update( incident_id="INC-001", field="status", old_value="open", new_value="in_progress", user="analyst@company.com", ) # Get User Confidence Index (risk score) uci = client.incidents.get_uci("bob@company.com") print(f"Risk score: {uci.score}, severity: {uci.severity}") # Get UBA anomalies for specific users anomalies = client.incidents.get_anomalies( users=["bob@company.com", "alice@company.com"], timeframe=30, # days severity="high", ) for a in anomalies: print(f"{a.user}: {a.anomaly_type} — {a.risk_level}") # Get DLP forensics for an incident forensics = client.incidents.get_forensics("dlp-incident-id")

Steering & Infrastructure

client.steering — Steering configuration, PoPs, and IPSec tunnels

# Get NPA steering configuration config = client.steering.get_config("npa") print(config.data) # Update steering configuration client.steering.update_config("npa", some_setting="value") # List Points of Presence (PoPs) for pop in client.steering.list_pops(): print(f"{pop.name} — {pop.region} ({pop.country})") # List IPSec tunnels for tunnel in client.steering.list_tunnels(): print(f"{tunnel.name}: {tunnel.status}") # Get specific tunnel tunnel = client.steering.get_tunnel(42)

NPA Policy & Infrastructure

client.npa — Zero Trust Network Access policy rules, groups, publisher upgrade profiles, and local brokers via /api/v2/policy/npa/* and /api/v2/infrastructure/*

Policy Rules — client.npa.policy.rules

# List NPA policy rules (auto-paginated) for rule in client.npa.policy.rules.list(): print(f"[{rule.rule_id}] {rule.rule_name} — enabled={rule.enabled}") # Filter with JQL and get one rule rules = client.npa.policy.rules.list(filter_expr='group_id eq "12"').to_list() rule = client.npa.policy.rules.get(101) # Create / update / delete rule = client.npa.policy.rules.create(rule_name="allow-finance", group_id=12, enabled=True) client.npa.policy.rules.update(rule.rule_id, enabled=False) client.npa.policy.rules.delete(rule.rule_id)

Policy Groups — client.npa.policy.groups

for group in client.npa.policy.groups.list(): print(f"{group.group_id}: {group.group_name}") group = client.npa.policy.groups.create("Finance Apps", order="after") client.npa.policy.groups.update(group.group_id, group_name="Finance") client.npa.policy.groups.delete(group.group_id)

Upgrade Profiles & Local Brokers

# Publisher upgrade profiles — client.npa.upgrade_profiles for profile in client.npa.upgrade_profiles.list(): print(profile.name) client.npa.upgrade_profiles.assign(profile_id, publisher_ids=[109, 126]) # Local brokers — client.npa.local_brokers brokers = client.npa.local_brokers.list() config = client.npa.local_brokers.get_config() token = client.npa.local_brokers.create_registration_token(broker_id) # Name validation & search helpers client.npa.validate_name("private_app", "internal-dashboard") client.npa.search("private_app", "dashboard")

Methods

NamespaceMethodsEndpoint
npa.policy.ruleslist(), get(), create(), update(), delete()/api/v2/policy/npa/rules
npa.policy.groupslist(), get(), create(), update(), delete()/api/v2/policy/npa/policygroups
npa.upgrade_profileslist(), get(), create(), update(), delete(), assign()/api/v2/infrastructure/publisherupgradeprofiles
npa.local_brokerslist(), get(), create(), update(), delete(), get_config(), update_config(), create_registration_token()/api/v2/infrastructure/lbrokers
npavalidate_name(), search()/api/v2/infrastructure/npa/*

NpaPolicyRule Model Fields

FieldTypeDescription
rule_idint | NoneNumeric rule identifier
rule_namestr | NoneHuman-readable rule name
enabledstr | NoneEnabled state (API returns "1"/"0")
group_idint | str | NoneOwning policy group ID
group_namestr | NoneOwning policy group name
actionstr | NoneRule action (allow, block, etc.)
rule_datadict | NoneFull rule payload (conditions, entities)

DEM / ADEM

client.dem — Digital Experience Management: synthetic probes, alert rules, alerts, ad-hoc query, and per-user ADEM telemetry via /api/v2/dem/* and /api/v2/adem/*

The dem namespace is organized into seven sub-namespaces: probes, network_probes, alert_rules, alerts, query, apps, and users.

Probes & Alert Rules

# Application probes — client.dem.probes probes = client.dem.probes.list() probe = client.dem.probes.create("portal-check", "https://portal.example.com", protocol="https") client.dem.probes.update(probe_id, {"interval": 60}) client.dem.probes.delete(probe_id) # Network probes (read/update/delete) — client.dem.network_probes client.dem.network_probes.list() # Alert rules — client.dem.alert_rules rule = client.dem.alert_rules.create("latency-high", metric="latency", threshold=200, severity="high")

Alerts & Ad-hoc Query

# Search DEM alerts — client.dem.alerts for alert in client.dem.alerts.search(severity=["high"], limit=50): print(f"{alert.alert_type} — {alert.severity} — {alert.status}") alert = client.dem.alerts.get("alert-id") entities = client.dem.alerts.entities("alert-id") # Ad-hoc query engine (privileged) — client.dem.query from datetime import datetime, timedelta data = client.dem.query.get_data( "page", select=["user", "exp_score"], begin=datetime.now() - timedelta(hours=1), end=datetime.now(), ) # Discovered apps — client.dem.apps apps = client.dem.apps.list()

Per-User Telemetry — client.dem.users

from datetime import datetime, timedelta start = datetime.now() - timedelta(hours=6) end = datetime.now() # Experience info & devices for a user info = client.dem.users.info("alice@example.com", start_time=start, end_time=end) print(f"{info.user}: exp_score={info.exp_score}") devices = client.dem.users.devices("alice@example.com", start_time=start, end_time=end) for d in devices: apps = client.dem.users.applications("alice@example.com", d.device_id, start_time=start, end_time=end) # One-shot RCA-style diagnosis (composite call) report = client.dem.users.diagnose("alice@example.com", start_time=start, end_time=end)

DemAlert Model Fields

FieldTypeDescription
idstr | NoneAlert identifier (API field: _id)
alert_categorystr | NoneCategory (API field: alertCategory)
alert_typestr | NoneAlert type (API field: alertType)
severitystr | NoneSeverity level
statusstr | NoneOpen / closed status
open_timeint | NoneWhen the alert opened (API field: openTime)

Per-user calls return AdemUserInfo (user, exp_score, last_known_location, organization_unit, user_group), plus AdemDevice and AdemApplication.

Users

client.users — Read-only User Management directory of users and groups via /api/v2/users/getusers and /api/v2/users/getgroups

# List directory users for user in client.users.list(limit=100): print(f"{user.primary_email} — {user.given_name} {user.family_name}") # Look up a single user by email or username user = client.users.get("alice@example.com", by="email") # Groups — client.users.groups for group in client.users.groups.list(): print(group.name) group = client.users.groups.get("Engineering") members = client.users.groups.members("Engineering")
Note: The users namespace is read-only. For provisioning (create/update/delete) use the SCIM namespace (client.scim.users).

UmUser Model Fields

FieldTypeDescription
idstr | NoneUser identifier
given_namestr | NoneFirst name (API field: givenName)
family_namestr | NoneLast name (API field: familyName)
emailslistEmail addresses
accountslist[UmUserAccount]Provisioned accounts (each has user_name, active, parent_groups, etc.)
primary_emailstr | NoneComputed property — the primary email address

RBAC

client.rbac — Role-based access control: admin roles via /api/v2/rbac/roles and administrators via /api/v2/platform/administration/scim/Users

Roles — client.rbac.roles

# List roles (optionally filter by type / scope / search) for role in client.rbac.roles.list(): print(f"[{role.id}] {role.name} — users={role.user_count}") role = client.rbac.roles.get(7) # Create a scoped role with API-group permissions role = client.rbac.roles.create( "SOC Analyst", description="Read-only incident access", api_groups=[{"name": "Incidents", "permission": "read"}], ) client.rbac.roles.update(role.id, description="Updated") client.rbac.roles.delete(role.id)

Administrators — client.rbac.admins

# List admins (SCIM-paginated ScimUser records) for admin in client.rbac.admins.list(): print(f"{admin.user_name} active={admin.active}") # Filter with SCIM syntax admins = client.rbac.admins.list(filter_expr='userName co "@example.com"').to_list()

RbacRole Model Fields

FieldTypeDescription
idint | NoneRole identifier (API field: roleId)
namestr | NoneRole name (API field: roleName)
descriptionstr | NoneRole description (API field: roleDescription)
typeint | NoneRole type code
user_countint | NoneAssigned admin count (API field: userCount)
api_groupslist[RbacRoleApiGroup]API-group permissions (API field: apiGroups)
scopeslist[RbacRoleScope]Object/functional scopes
last_editeddatetime | NoneLast modified time (API field: lastEdited)

API Tokens

client.tokens — Manage REST API v2 tokens and their endpoint scopes via /api/v2/auth/tokens

from datetime import datetime, timedelta # List tokens for token in client.tokens.list(): print(f"[{token.id}] {token.name} — expires {token.expires}") # Create a scoped token (returns the secret once, in token.token) token = client.tokens.create( "ci-automation", endpoints=[{"endpoint": "/api/v2/events/dataexport", "permissions": "r"}], expires=datetime.now() + timedelta(days=90), ) print("Secret:", token.token) # Update, reissue (rotate the secret), and revoke client.tokens.update(token.id, name="ci-automation-v2") rotated = client.tokens.reissue(token.id) client.tokens.revoke(token.id) # alias of .delete()

ApiToken Model Fields

FieldTypeDescription
idstr | NoneToken identifier
namestr | NoneToken display name
expiresint | NoneExpiry as Unix epoch seconds
endpointslist[ApiTokenEndpoint]Scoped endpoints (each has endpoint + permissions "r"/"rw")
tokenstr | NoneThe secret value — only returned by create() and reissue()

Devices

client.devices — Managed device inventory and device tags via /api/v2/steering/devices and /api/v2/devices/*

# List managed devices (auto-paginated) for device in client.devices.list(): print(f"{device.host_name} — {device.os} {device.os_version} (client {device.client_version})") # Supported operating systems oses = client.devices.supported_os() # Device tags — client.devices.tags for tag in client.devices.tags.list(): print(tag.name) tag = client.devices.tags.create("kiosk", description="Shared kiosk devices") client.devices.tags.update(tag.id, description="Lobby kiosks") client.devices.tags.delete(tag.id)

Device Model Fields

FieldTypeDescription
device_idstr | NoneDevice identifier
host_namestr | NoneHostname (API field: hostname)
osstr | NoneOperating system
os_versionstr | NoneOS version
client_versionstr | NoneNetskope client version
userslist[dict]Users associated with the device
last_eventdict | NoneMost recent device event

DNS Profiles

client.dns — DNS security profiles, inheritance groups, and deployment via /api/v2/profiles/dns

# List DNS profiles (auto-paginated) for profile in client.dns.list(): print(f"[{profile.id}] {profile.name} — log_traffic={profile.log_traffic}") profile = client.dns.create("corp-dns") client.dns.update(profile.id, description="Corporate resolver", log_traffic=True) client.dns.delete(profile.id) # Changes are staged — deploy them (all or by id) client.dns.deploy(all=True, change_note="quarterly update") # Reference data client.dns.list_domain_categories() client.dns.list_record_types() client.dns.list_tunnels() # Inheritance groups — client.dns.inheritance_groups for grp in client.dns.inheritance_groups.list(): print(grp.name) grp = client.dns.inheritance_groups.create("emea-inherit") client.dns.inheritance_groups.deploy(all=True)
Deploy required: Like URL lists, DNS profile create/update/delete changes are staged until you call client.dns.deploy().

DnsProfile Model Fields

FieldTypeDescription
idstr | int | NoneProfile identifier
namestr | NoneProfile name
descriptionstr | NoneDescription
log_trafficstr | bool | NoneWhether DNS traffic is logged

CCI

client.cci — Cloud Confidence Index app lookups and custom app tags via /api/v2/services/cci

Key Methods

# Look up an app's Cloud Confidence Index info = client.cci.lookup_app("Dropbox") # Tag apps — client.cci.tags client.cci.tags.create("sanctioned", apps=["Box", "Google Drive"]) tags = client.cci.tags.list()

Enrollment

client.enrollment — Device enrollment token sets via /api/v2/enrollment/tokenset

Key Methods

for ts in client.enrollment.list_token_sets(): print(ts.name) ts = client.enrollment.create_token_set("contractors", max_devices=50)

Notifications

client.notifications — End-user notification templates & delivery settings via /api/v2/notifications/user/*

Key Methods

for tpl in client.notifications.list_templates(): print(tpl.name) tpl = client.notifications.create_template( "block-page", title="Access Blocked", message="This site is not permitted.", )

IPS

client.ips — Intrusion Prevention System status, signatures, allowlists, and overrides via /api/v2/ips

Key Methods

# Check IPS status and enable alert-only mode print(client.ips.status()) client.ips.set_alert_only_mode(True) # Search signatures sigs = client.ips.search_signatures(cvss_severity=["critical"], limit=50)

ATP

client.atp — Advanced Threat Protection sandbox file & URL scanning via /api/v2/atp

Key Methods

# Submit a URL for sandbox analysis job = client.atp.scan_url("http://suspicious.example.com") report = client.atp.get_url_report(job["submission_id"]) # Scan a file from disk result = client.atp.scan_file_path("/tmp/sample.exe")

SkopeIQ (nsiq)

client.nsiq — URL/IOC intelligence, recategorization, and false-positive reporting via /api/v2/nsiq

Key Methods

# Categorize URLs result = client.nsiq.url_lookup(["http://example.com", "http://intranet"]) # Look up IOC reputation by hash iocs = client.nsiq.lookup_iocs(["44d88612fea8a8f36de82e1278abb02f"])

RBI

client.rbi — Remote Browser Isolation templates, CDR, and cloud-storage config via /api/v2/rbi

Key Methods

for tpl in client.rbi.list_templates()["templates"]: print(tpl["name"]) client.rbi.deploy_templates(deploy_all=True, note="rollout")

DSPM

client.dspm — Data Security Posture Management resources, analytics, and scans via /api/v2/dspm

Key Methods

# List discovered databases and kick off a scan stores = client.dspm.list_resources("databases") client.dspm.scan_datastores(["store-id-1"])

SSPM (spm)

client.spm — SaaS Security Posture Management app inventory & posture scoring via /api/v2/spm

Key Methods

# List SaaS apps and get the overall posture score apps = client.spm.list_apps() score = client.spm.posture_score()

Pagination

Every .list() method returns a SyncPaginatedResponse (or AsyncPaginatedResponse) that handles pagination transparently. You never write page loops.

Iterate Items (Lazy)

Pages are fetched on demand as you iterate. Memory-efficient for large result sets.

# Fetches pages automatically as needed for alert in client.alerts.list(page_size=100): process(alert) # Each alert is a typed Alert object

Page-Level Access

Use .pages() when you need page metadata (total count, page size).

for page in client.alerts.list(page_size=50).pages(): print(f"Page at offset {page.offset}: {len(page.items)} items (total={page.total})") for alert in page.items: process(alert)

Collect All (.to_list())

Fetch everything into a list. Always specify max_items as a safety limit.

# Collect up to 5000 alerts all_alerts = client.alerts.list().to_list(max_items=5000) print(f"Got {len(all_alerts)} alerts")

Get First (.first())

Fetch just the first result without iterating further. Returns None for empty results.

latest = client.alerts.list(order_by="timestamp").first() if latest: print(f"Latest: {latest.alert_name}") else: print("No alerts found")

Page Object Fields

FieldTypeDescription
itemslist[T]Model instances on this page
totalint | NoneTotal items across all pages (if provided by API)
offsetintOffset used for this page
limitintPage size

Error Handling

Exception Hierarchy

NetskopeError # Base — catch everything ├── APIError # Any non-2xx HTTP response │ ├── AuthenticationError # 401 — invalid or expired token │ ├── ForbiddenError # 403 — token lacks required scope │ ├── NotFoundError # 404 — resource does not exist │ ├── ConflictError # 409 — duplicate resource │ ├── RateLimitError # 429 — rate limit exceeded (.retry_after available) │ └── ServerError # 5xx — Netskope server-side failure ├── ValidationError # Bad input caught before sending request ├── ConnectionError # Network-level failure └── TimeoutError # Request timeout exceeded

APIError Attributes

All API errors carry rich context for debugging and support escalation:

AttributeTypeDescription
status_codeintHTTP status code (e.g. 401, 404, 429)
request_idstr | NoneServer-assigned request ID — use this for Netskope support tickets
messagestrHuman-readable error message
bodydict | NoneParsed JSON response body

Catching Specific Errors

from netskope.exceptions import ( NetskopeError, AuthenticationError, ForbiddenError, NotFoundError, RateLimitError, ServerError, ) try: alert = client.alerts.get("nonexistent-id") except NotFoundError as e: print(f"Not found: {e.message}") print(f"Request ID: {e.request_id}") # for support escalation except AuthenticationError: print("Token is invalid or expired — check NETSKOPE_API_TOKEN") except ForbiddenError: print("Token lacks the required scope for this endpoint") except RateLimitError as e: print(f"Rate limited — retry after {e.retry_after} seconds") except ServerError as e: print(f"Netskope server error ({e.status_code}): {e.message}") except NetskopeError as e: print(f"SDK error: {e}") # catch-all for any SDK error

Catching All API Errors

from netskope.exceptions import APIError try: result = client.publishers.create(name="test") except APIError as e: print(f"API returned HTTP {e.status_code}: {e.message}") if e.request_id: print(f"Include this in support tickets: {e.request_id}")

Retries & Backoff

The SDK automatically retries transient failures with exponential backoff and jitter. No configuration required for sensible defaults.

Default Behavior

Custom Configuration

# Aggressive retry for batch jobs client = NetskopeClient( tenant="mycompany.goskope.com", api_token="...", max_retries=5, backoff_factor=1.0, # longer waits: 1s, 2s, 4s, 8s, 16s timeout=60.0, # longer timeout for slow queries ) # No retries (fail fast) client = NetskopeClient( tenant="mycompany.goskope.com", api_token="...", max_retries=0, )

Logging

The SDK uses Python's standard logging module under the "netskope" logger. Tokens are never logged.

import logging # See request/response at INFO level logging.getLogger("netskope").setLevel(logging.INFO) # → ← GET https://mycompany.goskope.com/api/v2/events/datasearch/alert → 200 (request_id=abc123) # Full debug output (request URLs, retry decisions) logging.getLogger("netskope").setLevel(logging.DEBUG) # → → GET /api/v2/events/datasearch/alert # → Netskope API returned 429, retrying in 1.2s (attempt 1/3) # Silence the SDK entirely logging.getLogger("netskope").setLevel(logging.CRITICAL)

JQL Query Syntax

Netskope's JSON Query Language (JQL) is used in alerts.list(query=...), events.list(query=...), and incidents.list(query=...) to filter results server-side.

Operators

OperatorMeaningExample
eqEqualsseverity eq "high"
neNot equalsalert_type ne "policy"
gt / ltGreater/less thantimestamp gt 1709913600
ge / leGreater/less or equalcci ge 50
inIn listactivity in ("Upload", "Download")
containsSubstring matchalert_name contains "DLP"
AND / ORLogical operatorsseverity eq "high" AND app eq "Slack"

Examples

# Simple equality client.alerts.list(query='alert_type eq "DLP"') # Compound query client.alerts.list(query='severity eq "high" AND app eq "Slack"') # User-specific events client.events.list("application", query='user eq "alice@company.com"') # Activity filter client.events.list("application", query='activity in ("Upload", "Download", "Delete")')

Model Serialization

All response models are Pydantic v2 objects with full serialization support. They are frozen (immutable) and forward-compatible (unknown API fields are silently ignored, so new fields added by the API won't break existing code).

alert = client.alerts.list().first() # Access fields with dot notation alert.alert_name # "Sensitive data should be masked in output" alert.app # "Workday" alert.alert_type # "Security Assessment" # Serialize to dict d = alert.model_dump() # → {'id': '44d8af1b...', 'alert_name': 'Sensitive data...', 'app': 'Workday', ...} # Serialize to JSON string json_str = alert.model_dump_json() # → '{"id": "44d8af1b...", "alert_name": "Sensitive data...", ...}' # Exclude None fields clean = alert.model_dump(exclude_none=True) # Import models directly for type annotations from netskope.models import Alert, Publisher, UrlList, ScimUser def process_alert(alert: Alert) -> None: print(f"{alert.alert_name} — {alert.severity}")

All Available Models

Security

  • Alert — Security alerts
  • Event — Generic events
  • NetworkEvent — Network-layer events
  • PageEvent — Web page events
  • AuditEvent — Audit trail events
  • Incident — Security incidents
  • Anomaly — UBA anomalies
  • UserConfidenceIndex — UCI risk scores
  • DemAlert — DEM alerts

ZTNA & Infrastructure

  • Publisher — Private access publishers
  • PrivateApp — ZTNA applications
  • NpaPolicyRule — NPA policy rules
  • NpaPolicyGroup — NPA policy groups
  • LocalBroker — Local brokers
  • PublisherUpgradeProfile — Upgrade profiles
  • Pop — Points of Presence
  • IPSecTunnel — VPN tunnels
  • DnsProfile — DNS profiles

Identity, Devices & Config

  • ScimUser — SCIM users
  • ScimGroup — SCIM groups
  • UmUser — Directory users
  • UmGroup — Directory groups
  • RbacRole — Admin roles
  • ApiToken — API tokens
  • Device — Managed devices
  • UrlList — URL allow/block lists
  • NotificationTemplate — Notification templates

Enums

from netskope.models.alerts import AlertSeverity, AlertType from netskope.models.events import EventType from netskope.models.incidents import IncidentStatus from netskope.models.publishers import PublisherStatus from netskope.models.url_lists import UrlListType AlertSeverity.CRITICAL # "critical" AlertSeverity.HIGH # "high" EventType.NETWORK # "network" EventType.APPLICATION # "application" IncidentStatus.OPEN # "open" PublisherStatus.CONNECTED # "connected" UrlListType.EXACT # "exact" UrlListType.REGEX # "regex"

Complete API Map

All 24 resource namespaces and their key methods at a glance:

NamespaceMethodsAPI Endpoint
client.alertslist(), get(id)/api/v2/events/datasearch/alert
client.eventslist(type)/api/v2/events/datasearch/{type}
client.url_listslist(), get(), create(), update(), delete(), deploy()/api/v2/policy/urllist
client.publisherslist(), get(), create(), update(), delete()/api/v2/infrastructure/publishers
client.private_appslist(), get(), create(), update(), delete()/api/v2/steering/apps/private
client.scim.userslist(), get(), create(), update(), delete()/api/v2/scim/Users
client.scim.groupslist(), get(), create(), update(), delete()/api/v2/scim/Groups
client.incidentslist(), update(), get_uci(), get_anomalies(), get_forensics()/api/v2/events/datasearch/incident
client.steeringget_config(), update_config(), list_pops(), list_tunnels(), get_tunnel()/api/v2/steering/*
client.npavalidate_name(), search(), policy.rules.*, policy.groups.*, upgrade_profiles.*, local_brokers.*/api/v2/policy/npa/*, /api/v2/infrastructure/*
client.demprobes.*, network_probes.*, alert_rules.*, alerts.*, query.*, apps.list(), users.*/api/v2/dem/*, /api/v2/adem/*
client.dnslist(), get(), create(), update(), delete(), deploy(), inheritance_groups.*/api/v2/profiles/dns
client.userslist(), get(), groups.list(), groups.get(), groups.members()/api/v2/users/getusers, /getgroups
client.rbacroles.list(), roles.get(), roles.create(), roles.update(), roles.delete(), admins.list()/api/v2/rbac/roles
client.tokenslist(), get(), create(), update(), reissue(), revoke()/delete()/api/v2/auth/tokens
client.deviceslist(), supported_os(), tags.list(), tags.create(), tags.update(), tags.delete()/api/v2/steering/devices
client.enrollmentlist_token_sets(), create_token_set(), update_token_set(), delete_token_set()/api/v2/enrollment/tokenset
client.notificationslist_templates(), get_template(), create_template(), update_template(), delete_template(), get_delivery_settings()/api/v2/notifications/user/*
client.ccilookup_app(), tags.list(), tags.create(), tags.update(), tags.delete()/api/v2/services/cci
client.ipsstatus(), search_signatures(), list_allowlist(), set_alert_only_mode(), signature overrides/api/v2/ips
client.atpscan_url(), scan_file(), scan_file_path(), get_report(), get_url_report()/api/v2/atp
client.nsiqurl_lookup(), recategorize(), lookup_iocs(), get_ioc(), false-positive reporting/api/v2/nsiq
client.rbilist_templates(), create_template(), deploy_templates(), CDR & cloud-storage config/api/v2/rbi
client.dspmlist_resources(), analytics(), connect_datastores(), scan_datastores()/api/v2/dspm
client.spmlist_apps(), get_app(), inventory(), posture_score(), list_policy_rules()/api/v2/spm