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-sdkEvery design decision optimized for discoverability, safety, and minimal surprise.
Navigate the entire API through autocomplete: client.alerts.list(), client.scim.users.create()
Just iterate. No page loops, no offset math. Works across all list endpoints with lazy fetching.
Pydantic v2 models for every response. Full py.typed support for mypy and pyright.
Exponential backoff with jitter. Respects Retry-After headers. Configurable per client.
Choose NetskopeClient for scripts or AsyncNetskopeClient for high-throughput.
Specific exception types with request IDs. Catch RateLimitError, NotFoundError precisely.
pip install netskope-py-sdkThe SDK has only two dependencies: httpx and pydantic. Python 3.11 or newer is required.
# Clone and install with test dependencies
git clone https://github.com/netSkopeoss/netskope-py-sdk.git
cd netskope-py-sdk
pip install -e ".[dev]"import netskope
print(netskope.__version__) # "1.1.0"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()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 closedfrom 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()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")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)}")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}")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())| Parameter | Type | Default | Description |
|---|---|---|---|
tenant | str | None | NETSKOPE_TENANT | Tenant hostname, e.g. "mycompany.goskope.com". Falls back to the NETSKOPE_TENANT environment variable. |
api_token | str | None | NETSKOPE_API_TOKEN | REST API v2 token from your Netskope admin console. Falls back to the NETSKOPE_API_TOKEN environment variable. |
timeout | float | 30.0 | HTTP request timeout in seconds. |
max_retries | int | 3 | Maximum automatic retries for transient errors (429, 5xx). |
backoff_factor | float | 0.5 | Exponential backoff base multiplier. Sleep = factor * 2^attempt, capped at 60s, plus jitter. |
retry_on_status | frozenset[int] | None | {429,500,502,503,504} | HTTP status codes that trigger automatic retry. |
Credentials resolve in priority order (inspired by boto3):
tenant=, api_token=)NETSKOPE_TENANT, NETSKOPE_API_TOKENIf neither source provides a value, a ValidationError is raised at construction time (fail fast).
# 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# 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 = NetskopeClient(tenant="mycompany.goskope.com", api_token="...")
client.version # "1.1.0"
client.tenant # "mycompany.goskope.com"
client.base_url # "https://mycompany.goskope.com"The verify constructor option controls TLS verification. It accepts:
| Value | Behavior |
|---|---|
True (default) | Verify against the system trust store |
False | Disable 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 conventionThe 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())NetskopeClient is also available on AsyncNetskopeClient with the same signature. The only difference is await for single-resource methods and async for for iterators.
client.alerts — Query and retrieve security alerts via /api/v2/events/datasearch/alert
.list() — List Alerts| Parameter | Type | Default | Description |
|---|---|---|---|
query | str | None | None | JQL filter expression, e.g. 'alert_type eq "DLP"' |
fields | list[str] | None | None | Specific fields to return (reduces response size) |
start_time | datetime | int | None | None | Start of time range (datetime object or Unix epoch int) |
end_time | datetime | int | None | None | End of time range |
group_by | str | None | None | Field to aggregate results by |
order_by | str | None | None | Field to sort by |
descending | bool | True | Sort order (True = newest first) |
page_size | int | 100 | Results 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 AlertRetrieve 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")| Field | Type | Description |
|---|---|---|
id | str | None | Alert unique identifier (API field: _id) |
alert_name | str | None | Human-readable alert name |
alert_type | str | None | Alert category (DLP, malware, anomaly, Security Assessment, etc.) |
severity | str | None | Severity level (critical, high, medium, low) |
user | str | None | Affected user email |
app | str | None | Application name (e.g. "Workday", "Slack") |
activity | str | None | Activity that triggered the alert |
policy_name | str | None | Policy that generated the alert |
action | str | None | Action taken (alert, block, etc.) |
category | str | None | URL/app category |
cci | int | None | Cloud Confidence Index score |
ccl | str | None | Cloud Confidence Level |
access_method | str | None | How traffic was accessed (Client, API, etc.) |
traffic_type | str | None | Traffic classification |
timestamp | datetime | None | When the alert occurred (auto-parsed from epoch) |
client.events — Query security events across 10 event types via /api/v2/events/datasearch/{type}
.list(event_type, ...) — List EventsSame 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)The SDK automatically returns the right model subclass based on event type:
| Event Type | Model | Extra Fields |
|---|---|---|
"network" | NetworkEvent | src_ip, dst_ip, src_port, dst_port, protocol, num_bytes, domain |
"page" | PageEvent | url, domain, page_id, page_duration, referer, browser, os, device |
"audit" | AuditEvent | audit_log_event, audit_category, supporting_data, organization_unit |
| All others | Event | Base fields: user, app, activity, action, site, category, severity, timestamp, etc. |
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()client.url_lists.deploy(). This matches the Netskope admin console behavior — create, update, or delete, then deploy all changes at once.
| Field | Type | Description |
|---|---|---|
id | int | None | Numeric identifier |
name | str | None | Human-readable name |
type | str | None | Matching strategy: "exact" or "regex" |
urls | list[str] | The URL entries |
pending | bool | None | Whether changes await deployment |
modify_by | str | None | Last modified by (admin email) |
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)| Field | Type | Description |
|---|---|---|
publisher_id | int | None | Numeric identifier |
publisher_name | str | None | Human-readable name |
status | str | None | "connected" or "not_connected" |
apps_count | int | None | Number of apps assigned to this publisher |
registered | bool | None | Registration status |
sticky_ip_enabled | bool | None | Whether sticky IP is enabled |
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)| Field | Type | Description |
|---|---|---|
app_id | int | None | Numeric identifier |
app_name | str | None | Application display name |
host | str | None | Target host (IP or hostname) |
port | str | None | Target port(s) |
protocols | list | None | Protocol configuration |
publishers | list[dict] | None | Assigned publishers |
use_publisher_dns | bool | None | Use publisher DNS resolution |
clientless_access | bool | None | Browser-based access enabled |
client.scim — SCIM 2.0 user and group provisioning via /api/v2/scim/Users and /api/v2/scim/Groups
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)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)| Field | Type | Description |
|---|---|---|
id | str | None | SCIM unique identifier |
user_name | str | None | Username (typically email) |
display_name | str | None | Display name |
active | bool | None | Account status |
emails | list[ScimEmail] | Email addresses (each has value, primary, type) |
groups | list[dict] | Group memberships |
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")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)client.npa — Zero Trust Network Access policy rules, groups, publisher upgrade profiles, and local brokers via /api/v2/policy/npa/* and /api/v2/infrastructure/*
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)client.npa.policy.groupsfor 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)# 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")| Namespace | Methods | Endpoint |
|---|---|---|
npa.policy.rules | list(), get(), create(), update(), delete() | /api/v2/policy/npa/rules |
npa.policy.groups | list(), get(), create(), update(), delete() | /api/v2/policy/npa/policygroups |
npa.upgrade_profiles | list(), get(), create(), update(), delete(), assign() | /api/v2/infrastructure/publisherupgradeprofiles |
npa.local_brokers | list(), get(), create(), update(), delete(), get_config(), update_config(), create_registration_token() | /api/v2/infrastructure/lbrokers |
npa | validate_name(), search() | /api/v2/infrastructure/npa/* |
| Field | Type | Description |
|---|---|---|
rule_id | int | None | Numeric rule identifier |
rule_name | str | None | Human-readable rule name |
enabled | str | None | Enabled state (API returns "1"/"0") |
group_id | int | str | None | Owning policy group ID |
group_name | str | None | Owning policy group name |
action | str | None | Rule action (allow, block, etc.) |
rule_data | dict | None | Full rule payload (conditions, entities) |
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.
# 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")# 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()client.dem.usersfrom 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)| Field | Type | Description |
|---|---|---|
id | str | None | Alert identifier (API field: _id) |
alert_category | str | None | Category (API field: alertCategory) |
alert_type | str | None | Alert type (API field: alertType) |
severity | str | None | Severity level |
status | str | None | Open / closed status |
open_time | int | None | When 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.
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")users namespace is read-only. For provisioning (create/update/delete) use the SCIM namespace (client.scim.users).
| Field | Type | Description |
|---|---|---|
id | str | None | User identifier |
given_name | str | None | First name (API field: givenName) |
family_name | str | None | Last name (API field: familyName) |
emails | list | Email addresses |
accounts | list[UmUserAccount] | Provisioned accounts (each has user_name, active, parent_groups, etc.) |
primary_email | str | None | Computed property — the primary email address |
client.rbac — Role-based access control: admin roles via /api/v2/rbac/roles and administrators via /api/v2/platform/administration/scim/Users
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)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()| Field | Type | Description |
|---|---|---|
id | int | None | Role identifier (API field: roleId) |
name | str | None | Role name (API field: roleName) |
description | str | None | Role description (API field: roleDescription) |
type | int | None | Role type code |
user_count | int | None | Assigned admin count (API field: userCount) |
api_groups | list[RbacRoleApiGroup] | API-group permissions (API field: apiGroups) |
scopes | list[RbacRoleScope] | Object/functional scopes |
last_edited | datetime | None | Last modified time (API field: lastEdited) |
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()| Field | Type | Description |
|---|---|---|
id | str | None | Token identifier |
name | str | None | Token display name |
expires | int | None | Expiry as Unix epoch seconds |
endpoints | list[ApiTokenEndpoint] | Scoped endpoints (each has endpoint + permissions "r"/"rw") |
token | str | None | The secret value — only returned by create() and reissue() |
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)| Field | Type | Description |
|---|---|---|
device_id | str | None | Device identifier |
host_name | str | None | Hostname (API field: hostname) |
os | str | None | Operating system |
os_version | str | None | OS version |
client_version | str | None | Netskope client version |
users | list[dict] | Users associated with the device |
last_event | dict | None | Most recent device event |
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)client.dns.deploy().
| Field | Type | Description |
|---|---|---|
id | str | int | None | Profile identifier |
name | str | None | Profile name |
description | str | None | Description |
log_traffic | str | bool | None | Whether DNS traffic is logged |
client.cci — Cloud Confidence Index app lookups and custom app tags via /api/v2/services/cci
lookup_app(app_name, *, category=, ccl=, tag=, connector=, discovered=, limit=, offset=) — look up CCI metadata for an appcci.tags.list(), create(), update(), delete(), list_rules(), supported_attributes() — manage app tags# 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()client.enrollment — Device enrollment token sets via /api/v2/enrollment/tokenset
list_token_sets(*, limit=, offset=) — list enrollment token setscreate_token_set(name, *, max_devices=) — create a token setupdate_token_set(token_id, *, token_type=, valid_until=, enforce_status=)delete_token_set(token_id), delete_token_type(token_id, token_type)for ts in client.enrollment.list_token_sets():
print(ts.name)
ts = client.enrollment.create_token_set("contractors", max_devices=50)client.notifications — End-user notification templates & delivery settings via /api/v2/notifications/user/*
list_templates(*, limit=, offset=), get_template(template_id)create_template(name, *, title, message, ...), update_template(template_id, ...), delete_template(template_id)get_delivery_settings() — retrieve delivery configurationfor 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.",
)client.ips — Intrusion Prevention System status, signatures, allowlists, and overrides via /api/v2/ips
status(), update_status(*, web=, nonweb=, npa=)list_signatures(), search_signatures(...), list_signature_overrides(), update_signature_overrides(sig_ids, override), delete_signature_overrides(sig_ids)list_allowlist(), update_allowlist(...), get_alert_only_mode(), set_alert_only_mode(enabled)get_notification_template(), update_notification_template(...), get_threat_hunting_config(), update_threat_hunting_config(...)# 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)client.atp — Advanced Threat Protection sandbox file & URL scanning via /api/v2/atp
scan_url(url), get_url_report(submission_id), list_url_artifacts(submission_id)scan_file(filename, content, *, scan_type="sandbox"), scan_file_path(path, *, scan_type="sandbox")get_report(job_id), get_scan_result(submission_id), get_submission_report(submission_id)# 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")client.nsiq — URL/IOC intelligence, recategorization, and false-positive reporting via /api/v2/nsiq
url_lookup(urls, *, disable_dns_lookup=, category=) — category lookup for one or many URLsrecategorize(url, suggested_categories, ...), list_recategorizations(...), get_recategorization(task_id)lookup_iocs(hashes), get_ioc(sample_hash), get_ioc_report(sample_hash)report_url_false_positive(...), report_malware_false_positive(...), report_ips_false_positive(...), validate_user_email(user_email)# Categorize URLs
result = client.nsiq.url_lookup(["http://example.com", "http://intranet"])
# Look up IOC reputation by hash
iocs = client.nsiq.lookup_iocs(["44d88612fea8a8f36de82e1278abb02f"])client.rbi — Remote Browser Isolation templates, CDR, and cloud-storage config via /api/v2/rbi
list_templates(...), get_template(template_id), get_default_template(), create_template(data), update_template(id, data), delete_template(id)deploy_templates(ids, *, deploy_all=, note=), revert_templates(ids), restore_template(id)get_cdr()/update_cdr()/list_cdr_vendors()/test_cdr_config(), get_cloud_storage()/update_cloud_storage()list_applications(), list_supported_browsers(), list_default_categories()for tpl in client.rbi.list_templates()["templates"]:
print(tpl["name"])
client.rbi.deploy_templates(deploy_all=True, note="rollout")client.dspm — Data Security Posture Management resources, analytics, and scans via /api/v2/dspm
list_resources(resource_type, *, filter_expr=, sort_by=, sort_order=, limit=, offset=)analytics(metric_type) — posture analytics for a metricconnect_datastores(ids), scan_datastores(ids)# List discovered databases and kick off a scan
stores = client.dspm.list_resources("databases")
client.dspm.scan_datastores(["store-id-1"])client.spm — SaaS Security Posture Management app inventory & posture scoring via /api/v2/spm
list_apps(), get_app(app_name), inventory(*, filter=)posture_score(), list_policy_rules(), recent_changes()# List SaaS apps and get the overall posture score
apps = client.spm.list_apps()
score = client.spm.posture_score()Every .list() method returns a SyncPaginatedResponse (or AsyncPaginatedResponse) that handles pagination transparently. You never write page loops.
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 objectUse .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).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").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")| Field | Type | Description |
|---|---|---|
items | list[T] | Model instances on this page |
total | int | None | Total items across all pages (if provided by API) |
offset | int | Offset used for this page |
limit | int | Page size |
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 exceededAll API errors carry rich context for debugging and support escalation:
| Attribute | Type | Description |
|---|---|---|
status_code | int | HTTP status code (e.g. 401, 404, 429) |
request_id | str | None | Server-assigned request ID — use this for Netskope support tickets |
message | str | Human-readable error message |
body | dict | None | Parsed JSON response body |
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 errorfrom 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}")The SDK automatically retries transient failures with exponential backoff and jitter. No configuration required for sensible defaults.
429, 500, 502, 503, 5040.5 * 2^attempt (capped at 60s) + random jitterRetry-After headers from the server# 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,
)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)Netskope's JSON Query Language (JQL) is used in alerts.list(query=...), events.list(query=...), and incidents.list(query=...) to filter results server-side.
| Operator | Meaning | Example |
|---|---|---|
eq | Equals | severity eq "high" |
ne | Not equals | alert_type ne "policy" |
gt / lt | Greater/less than | timestamp gt 1709913600 |
ge / le | Greater/less or equal | cci ge 50 |
in | In list | activity in ("Upload", "Download") |
contains | Substring match | alert_name contains "DLP" |
AND / OR | Logical operators | severity eq "high" AND app eq "Slack" |
# 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")')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}")Alert — Security alertsEvent — Generic eventsNetworkEvent — Network-layer eventsPageEvent — Web page eventsAuditEvent — Audit trail eventsIncident — Security incidentsAnomaly — UBA anomaliesUserConfidenceIndex — UCI risk scoresDemAlert — DEM alertsPublisher — Private access publishersPrivateApp — ZTNA applicationsNpaPolicyRule — NPA policy rulesNpaPolicyGroup — NPA policy groupsLocalBroker — Local brokersPublisherUpgradeProfile — Upgrade profilesPop — Points of PresenceIPSecTunnel — VPN tunnelsDnsProfile — DNS profilesScimUser — SCIM usersScimGroup — SCIM groupsUmUser — Directory usersUmGroup — Directory groupsRbacRole — Admin rolesApiToken — API tokensDevice — Managed devicesUrlList — URL allow/block listsNotificationTemplate — Notification templatesfrom 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"All 24 resource namespaces and their key methods at a glance:
| Namespace | Methods | API Endpoint |
|---|---|---|
client.alerts | list(), get(id) | /api/v2/events/datasearch/alert |
client.events | list(type) | /api/v2/events/datasearch/{type} |
client.url_lists | list(), get(), create(), update(), delete(), deploy() | /api/v2/policy/urllist |
client.publishers | list(), get(), create(), update(), delete() | /api/v2/infrastructure/publishers |
client.private_apps | list(), get(), create(), update(), delete() | /api/v2/steering/apps/private |
client.scim.users | list(), get(), create(), update(), delete() | /api/v2/scim/Users |
client.scim.groups | list(), get(), create(), update(), delete() | /api/v2/scim/Groups |
client.incidents | list(), update(), get_uci(), get_anomalies(), get_forensics() | /api/v2/events/datasearch/incident |
client.steering | get_config(), update_config(), list_pops(), list_tunnels(), get_tunnel() | /api/v2/steering/* |
client.npa | validate_name(), search(), policy.rules.*, policy.groups.*, upgrade_profiles.*, local_brokers.* | /api/v2/policy/npa/*, /api/v2/infrastructure/* |
client.dem | probes.*, network_probes.*, alert_rules.*, alerts.*, query.*, apps.list(), users.* | /api/v2/dem/*, /api/v2/adem/* |
client.dns | list(), get(), create(), update(), delete(), deploy(), inheritance_groups.* | /api/v2/profiles/dns |
client.users | list(), get(), groups.list(), groups.get(), groups.members() | /api/v2/users/getusers, /getgroups |
client.rbac | roles.list(), roles.get(), roles.create(), roles.update(), roles.delete(), admins.list() | /api/v2/rbac/roles |
client.tokens | list(), get(), create(), update(), reissue(), revoke()/delete() | /api/v2/auth/tokens |
client.devices | list(), supported_os(), tags.list(), tags.create(), tags.update(), tags.delete() | /api/v2/steering/devices |
client.enrollment | list_token_sets(), create_token_set(), update_token_set(), delete_token_set() | /api/v2/enrollment/tokenset |
client.notifications | list_templates(), get_template(), create_template(), update_template(), delete_template(), get_delivery_settings() | /api/v2/notifications/user/* |
client.cci | lookup_app(), tags.list(), tags.create(), tags.update(), tags.delete() | /api/v2/services/cci |
client.ips | status(), search_signatures(), list_allowlist(), set_alert_only_mode(), signature overrides | /api/v2/ips |
client.atp | scan_url(), scan_file(), scan_file_path(), get_report(), get_url_report() | /api/v2/atp |
client.nsiq | url_lookup(), recategorize(), lookup_iocs(), get_ioc(), false-positive reporting | /api/v2/nsiq |
client.rbi | list_templates(), create_template(), deploy_templates(), CDR & cloud-storage config | /api/v2/rbi |
client.dspm | list_resources(), analytics(), connect_datastores(), scan_datastores() | /api/v2/dspm |
client.spm | list_apps(), get_app(), inventory(), posture_score(), list_policy_rules() | /api/v2/spm |