← Files OneSignalARCHIVED FILE
skills/setup/scripts/onesignal_api.py
21.1 KB · Nov 15, 2026 · 06:13 UTC
#!/usr/bin/env python3
"""Deterministic OneSignal API probes for the setup / credentials / verify skills.
This encodes the fiddly, verified details that models keep getting wrong:
- the web sync probe is CDN-cached ~1h, so we always append ?fresh=<ts>
- android_params must be polled AFTER credentials upload (caller's job to
order; this just polls with a timeout and never primes an empty cache)
- response codes carry meaning: 409 = already configured (relay, don't retry),
404 = feature flag off (fall back to dashboard), 401 = wrong app key
- delivery stat `failed` counts UNSUBSCRIBED targets, not errors
- the view-app read decides credential presence from the non-secret fields
(apns_env, apns_key_id, apns_team_id, fcm_sender_id, chrome_web_origin),
never from the plaintext credential fields the server plans to remove
- the write-once credential upload echoes the stored credential fields back
in its 2xx body; `provision` drops that body and prints a verdict, so the
secret never reaches the agent transcript
Prefer the OneSignal MCP server's tools when connected (better auth story);
this script is the generic curl-equivalent fallback that needs no MCP.
Endpoint hosts are taken from references/api-reference.md and skills/verify/SKILL.md
(the subscriber-presence probe path and its Key→Basic auth fallback are specified
in verify/SKILL.md). Keys are read from --key or the environment; never printed back.
Usage:
onesignal_api.py web-probe <app_id>
onesignal_api.py android-params <app_id> [--timeout 60]
onesignal_api.py notification-stats <notif_id> <app_id> --key KEY
onesignal_api.py app <app_id> --key KEY
onesignal_api.py subscribers <app_id> --key KEY
onesignal_api.py provision <app_id> --fcm-json <path> --key KEY
onesignal_api.py provision <app_id> --apns-p8 <path> --apns-key-id ID --apns-team-id ID --apns-bundle-id ID --key KEY
onesignal_api.py provision <app_id> --web-origin https://example.com [--web-icon URL] --key KEY
Key sources (in order): --key, $ONESIGNAL_REST_API_KEY, $ONESIGNAL_SETUP_TOKEN
Exit: 0 ok (see JSON `status`), 2 usage error, 3 network error, 4 auth missing
"""
import argparse
import base64
import json
import os
import sys
import time
import urllib.error
import urllib.request
API = "https://api.onesignal.com"
def _get(url, key=None, timeout=20, scheme="Key"):
req = urllib.request.Request(url)
if key:
req.add_header("Authorization", f"{scheme} {key}")
req.add_header("Accept", "application/json")
try:
with urllib.request.urlopen(req, timeout=timeout) as r:
body = r.read().decode("utf-8", "replace")
return r.status, body
except urllib.error.HTTPError as e:
return e.code, e.read().decode("utf-8", "replace")
except Exception as e: # noqa: BLE001
sys.stderr.write(f"ERROR: network failure: {e}\n")
sys.exit(3)
def _post_json(url, payload, key, timeout=60):
data = json.dumps(payload).encode("utf-8")
req = urllib.request.Request(url, data=data, method="POST")
req.add_header("Authorization", f"Key {key}")
req.add_header("Content-Type", "application/json")
req.add_header("Accept", "application/json")
try:
with urllib.request.urlopen(req, timeout=timeout) as r:
return r.status, r.read().decode("utf-8", "replace")
except urllib.error.HTTPError as e:
return e.code, e.read().decode("utf-8", "replace")
except Exception as e: # noqa: BLE001
sys.stderr.write(f"ERROR: network failure: {e}\n")
sys.exit(3)
def _json(body):
try:
return json.loads(body)
except Exception:
return None
def _present(value):
# null, "" and an absent key (None here) all read as "not configured".
if value is None:
return False
if isinstance(value, str):
return value.strip() != ""
return bool(value)
def _resolve_key(args):
return (getattr(args, "key", None)
or os.environ.get("ONESIGNAL_REST_API_KEY")
or os.environ.get("ONESIGNAL_SETUP_TOKEN"))
def _require_key(args):
k = _resolve_key(args)
if not k:
sys.stderr.write(
"ERROR: no API key. Pass --key, or export ONESIGNAL_REST_API_KEY / "
"ONESIGNAL_SETUP_TOKEN. Never paste the key into committed files.\n"
)
sys.exit(4)
return k
def cmd_web_probe(args):
# Always cache-bust: the plain URL (and its code-2 error) is CDN-cached ~1h.
url = f"{API}/sync/{args.app_id}/web?fresh={int(time.time())}"
status, body = _get(url)
data = _json(body) or {}
if data.get("success") is True:
result = {"status": "provisioned", "detail": "Web platform is live."}
elif data.get("code") == 2:
result = {"status": "not_configured",
"detail": "Web push NOT provisioned. The dashboard web-platform "
"step (Site URL) never happened — most common web wall."}
elif data.get("code") == 1:
result = {"status": "no_such_app", "detail": "No app with that ID."}
else:
result = {"status": "unknown", "detail": f"HTTP {status}", "raw": data}
result.update({"probe": "web", "app_id": args.app_id, "http": status, "cache_busted": True})
print(json.dumps(result, indent=2))
def cmd_android_params(args):
# Poll only AFTER credentials upload. android_sender_id present => FCM live.
deadline = time.time() + args.timeout
url = f"{API}/apps/{args.app_id}/android_params.js"
attempt = 0
while True:
attempt += 1
status, body = _get(url)
data = _json(body) or {}
if data.get("android_sender_id"):
print(json.dumps({"probe": "android_params", "app_id": args.app_id,
"http": status, "status": "fcm_live", "attempts": attempt}, indent=2))
return
if time.time() >= deadline:
# Distinguish "endpoint answered, sender_id absent" (keep waiting) from
# "endpoint erroring" (bad app id / outage). Reporting a non-2xx as
# not_live sends the caller into an endless wait on a request failure.
if not (200 <= status < 300):
print(json.dumps({"probe": "android_params", "app_id": args.app_id,
"http": status, "status": "error",
"detail": f"HTTP {status} on every poll — not an FCM-not-live "
"signal but a request failure (bad app id, or the "
"route is unavailable). Do NOT keep waiting on this.",
"attempts": attempt}, indent=2))
return
print(json.dumps({"probe": "android_params", "app_id": args.app_id,
"http": status, "status": "not_live",
"detail": "android_sender_id absent — FCM v1 credential "
"not yet on the app (or still propagating).",
"attempts": attempt}, indent=2))
return
time.sleep(min(5, max(1, args.timeout // 12)))
def cmd_notification_stats(args):
key = _require_key(args)
url = f"{API}/notifications/{args.notif_id}?app_id={args.app_id}"
status, body = _get(url, key=key)
data = _json(body) or {}
if status in (401, 403):
print(json.dumps({"probe": "notification_stats", "app_id": args.app_id,
"notif_id": args.notif_id, "http": status, "status": "auth_error",
"detail": "Key does not belong to this app (or lacks access)."}, indent=2)); return
if not (200 <= status < 300):
# Never print null stat fields as if they were real numbers on an error.
print(json.dumps({"probe": "notification_stats", "app_id": args.app_id,
"notif_id": args.notif_id, "http": status, "status": "error",
"detail": f"HTTP {status} — no stats returned (bad notif/app id, or route "
"unavailable). Do NOT read the numbers below as real.",
"raw": data}, indent=2)); return
out = {
"probe": "notification_stats", "app_id": args.app_id, "notif_id": args.notif_id,
"http": status, "status": "ok",
"successful": data.get("successful"),
"errored": data.get("errored"),
"failed_unsubscribed": data.get("failed"), # NOT delivery errors
"converted_clicks": data.get("converted"),
"received_confirmed": data.get("received"),
"note": "`failed` = unsubscribed targets, not delivery failures. "
"`errored` = real delivery errors (dashboard 'Failed').",
}
print(json.dumps(out, indent=2))
def cmd_app(args):
key = _require_key(args)
status, body = _get(f"{API}/api/v1/apps/{args.app_id}", key=key)
data = _json(body)
if status in (401, 403):
print(json.dumps({"probe": "app", "app_id": args.app_id, "http": status, "status": "auth_error",
"detail": "Key does not belong to this app (or lacks access)."}, indent=2)); return
if status == 404:
print(json.dumps({"probe": "app", "app_id": args.app_id, "http": status, "status": "not_found",
"detail": "No app with that ID (or route unavailable)."}, indent=2)); return
if not (200 <= status < 300):
print(json.dumps({"probe": "app", "app_id": args.app_id, "http": status, "status": "error",
"detail": f"HTTP {status}", "raw": data}, indent=2)); return
if not isinstance(data, dict):
# Fail closed: a 2xx with a body that is not a JSON object (a proxy page,
# a truncated read) must not turn into a confident "not configured".
print(json.dumps({"probe": "app", "app_id": args.app_id, "http": status, "status": "error",
"detail": "Response body is not a JSON object — presence unknown, not zero. "
"Re-run, or fall back to the dashboard (Settings > Push Platforms)."},
indent=2)); return
platforms, cross_check = _presence(data)
out = {"probe": "app", "app_id": args.app_id, "http": status, "status": "ok",
"platforms": platforms,
"cross_check": cross_check,
"note": "Presence only, from non-secret fields (apns_env, apns_key_id, apns_team_id, "
"fcm_sender_id, chrome_web_origin). Not a validity check: the test send in "
"the verify skill proves the credential works."}
print(json.dumps(out, indent=2))
def _presence(data):
# Never read the plaintext credential fields: the server plans to remove them.
apns_configured = _present(data.get("apns_env"))
key_id_set = _present(data.get("apns_key_id"))
team_id_set = _present(data.get("apns_team_id"))
if key_id_set and team_id_set:
apns_auth = "p8"
elif apns_configured and not key_id_set and not team_id_set:
apns_auth = "p12"
else:
apns_auth = None
fcm_configured = _present(data.get("fcm_sender_id"))
web_configured = _present(data.get("chrome_web_origin"))
# channels.push.platforms is a cross-check only: it derives from the
# enabled-platform flags, not from credential presence, and the server
# marks it as temporary. It never changes the verdict above.
channels = data.get("channels")
push = channels.get("push") if isinstance(channels, dict) else None
listed = push.get("platforms") if isinstance(push, dict) else None
cross_check = None
if isinstance(listed, list):
apns_listed = "apns" in listed
gcm_listed = "gcm" in listed
cross_check = {"source": "channels.push.platforms",
"apns_listed": apns_listed, "fcm_listed": gcm_listed,
"mismatch": (apns_listed != apns_configured) or (gcm_listed != fcm_configured)}
platforms = {
"apns": {"configured": apns_configured, "auth_type": apns_auth},
"fcm": {"configured": fcm_configured},
"web": {"configured": web_configured},
}
return platforms, cross_check
def _read_base64(path, label):
try:
with open(path, "rb") as f:
raw = f.read()
except OSError as e:
sys.stderr.write(f"ERROR: cannot read the {label} file at the given path: {e.strerror}\n")
sys.exit(2)
if not raw.strip():
sys.stderr.write(f"ERROR: the {label} file is empty\n")
sys.exit(2)
return base64.b64encode(raw).decode("ascii")
def _drop_long_strings(value, limit=256):
# An error body can echo a request param. A base64 credential is longer
# than this limit (a .p8 is about 320 chars); an error string is shorter.
if isinstance(value, dict):
return {k: _drop_long_strings(v, limit) for k, v in value.items()}
if isinstance(value, list):
return [_drop_long_strings(v, limit) for v in value]
if isinstance(value, str) and len(value) > limit:
return f"<dropped {len(value)}-char value>"
return value
def cmd_provision(args):
key = _require_key(args)
apns_values = (args.apns_p8, args.apns_key_id, args.apns_team_id, args.apns_bundle_id)
apns = any(apns_values)
fcm = bool(args.fcm_json)
web = bool(args.web_origin or args.web_icon)
if sum((apns, fcm, web)) != 1:
sys.stderr.write("ERROR: pass exactly one platform set per call: APNs (--apns-p8 with "
"--apns-key-id, --apns-team-id, --apns-bundle-id), FCM (--fcm-json), "
"or web (--web-origin [--web-icon]).\n")
sys.exit(2)
payload = {}
if apns:
if not all(apns_values):
sys.stderr.write("ERROR: APNs needs all four: --apns-p8, --apns-key-id, --apns-team-id, "
"--apns-bundle-id.\n")
sys.exit(2)
payload["apns_p8"] = _read_base64(args.apns_p8, ".p8")
payload["apns_key_id"] = args.apns_key_id
payload["apns_team_id"] = args.apns_team_id
payload["apns_bundle_id"] = args.apns_bundle_id
platform = "apns"
elif fcm:
payload["fcm_v1_service_account_json"] = _read_base64(args.fcm_json, "service-account JSON")
platform = "fcm"
else:
if not args.web_origin:
sys.stderr.write("ERROR: --web-icon needs --web-origin.\n")
sys.exit(2)
if not args.web_origin.startswith("https://"):
sys.stderr.write("ERROR: --web-origin must be an https:// origin.\n")
sys.exit(2)
payload["chrome_web_origin"] = args.web_origin
if args.web_icon:
payload["chrome_web_default_notification_icon"] = args.web_icon
platform = "web"
status, body = _post_json(f"{API}/api/v1/apps/{args.app_id}/credentials", payload, key)
data = _json(body)
out = {"probe": "provision", "app_id": args.app_id, "platform": platform, "http": status}
if 200 <= status < 300:
# The 2xx body is the app object and still echoes the plaintext
# credential fields. Print the presence verdict from the non-secret
# fields and nothing else from the body.
out["status"] = "provisioned"
# Only an app object for this app can give a presence verdict; any
# other 2xx body would print `configured: false` next to `provisioned`.
if isinstance(data, dict) and data.get("id") == args.app_id:
platforms, _ = _presence(data)
out["platforms"] = platforms
else:
out["presence"] = ("unknown: the 2xx body is not the app object, so no verdict; "
f"run `onesignal_api.py app {args.app_id}` to confirm.")
out["note"] = ("Stored and validated server-side. The response body echoed the stored "
"credential fields; this script dropped them. That echo is the server's reply "
"to the caller and adds nothing the request did not already send — no rotation "
"is needed because of it. A 2xx is configuration success, not proof of "
"delivery: the test send in the verify skill proves that.")
print(json.dumps(out, indent=2))
return
if status in (401, 403):
out.update({"status": "auth_error",
"detail": "Key does not belong to this app (or lacks access)."})
elif status == 404:
out.update({"status": "not_found",
"detail": "The provisioning route is off for this app (feature flag), or the App ID "
"is wrong. Fall back to the dashboard upload (Settings > Push Platforms); "
"do not retry."})
elif status == 409:
out.update({"status": "already_configured",
"detail": "Write-once: this platform already has credentials. Nothing was written "
"by this request. Relay the message verbatim; do not retry."})
elif status == 400:
out.update({"status": "validation_error",
"detail": "The server rejected the credential. Relay the message verbatim."})
else:
out.update({"status": "error", "detail": f"HTTP {status}"})
if isinstance(data, dict):
# The length cap keeps a base64 credential out of the output if a 400
# ever echoes the input; the relayable 409 text is well under it.
for field in ("message", "errors", "platforms"):
if field in data:
out[field] = _drop_long_strings(data[field])
elif body.strip():
out["raw"] = _drop_long_strings(body.strip())
print(json.dumps(out, indent=2))
def cmd_subscribers(args):
key = _require_key(args)
# Presence pattern: limit 1, non-empty => a subscriber exists.
# Path + auth per verify/SKILL.md: /players is the legacy Devices API; try the
# current `Key` scheme first and fall back to legacy `Basic` if it is rejected,
# so a Basic-only app is not misdiagnosed as a wrong key.
url = f"{API}/players?app_id={args.app_id}&limit=1"
status, body = _get(url, key=key)
if status in (401, 403):
status, body = _get(url, key=key, scheme="Basic")
data = _json(body) or {}
if status in (401, 403):
print(json.dumps({"probe": "subscribers", "app_id": args.app_id, "http": status,
"status": "auth_error",
"detail": "Rejected under both Key and Basic auth — the key does not "
"belong to this app, or lacks Devices-API access."}, indent=2)); return
if not (200 <= status < 300):
# A non-2xx here must NOT read as "no subscriber" — that sends verify into an
# endless "keep waiting" loop. Report the error and let the caller fall back.
print(json.dumps({"probe": "subscribers", "app_id": args.app_id, "http": status,
"status": "error", "has_subscriber": None,
"detail": f"HTTP {status} — presence unknown, not zero. Fall back to the "
"dashboard (Audience → Subscriptions) to confirm.",
"raw": data}, indent=2)); return
total = data.get("total_count")
players = data.get("players") or []
nt = players[0].get("notification_types") if players else None
out = {"probe": "subscribers", "app_id": args.app_id, "http": status, "status": "ok",
"has_subscriber": bool(players) or bool(total),
"total_count": total,
"first_subscription_id": players[0].get("id") if players else None,
"notification_types": nt,
"first_subscription_opted_in": bool(nt is not None and nt >= 1),
"note": "notification_types >= 1 = opted in (the activation-ladder bar); "
"< 1 = registered but not opted in (pre-permission or unsubscribed); "
"do not test-send to it yet."}
print(json.dumps(out, indent=2))
def main():
ap = argparse.ArgumentParser(description="OneSignal API probes (generic curl fallback; prefer MCP if connected).")
sub = ap.add_subparsers(dest="cmd", required=True)
p = sub.add_parser("web-probe"); p.add_argument("app_id"); p.set_defaults(fn=cmd_web_probe)
p = sub.add_parser("android-params"); p.add_argument("app_id"); p.add_argument("--timeout", type=int, default=60); p.set_defaults(fn=cmd_android_params)
p = sub.add_parser("notification-stats"); p.add_argument("notif_id"); p.add_argument("app_id"); p.add_argument("--key"); p.set_defaults(fn=cmd_notification_stats)
p = sub.add_parser("app"); p.add_argument("app_id"); p.add_argument("--key"); p.set_defaults(fn=cmd_app)
p = sub.add_parser("subscribers"); p.add_argument("app_id"); p.add_argument("--key"); p.set_defaults(fn=cmd_subscribers)
p = sub.add_parser("provision", help="write-once credential upload; one platform set per call")
p.add_argument("app_id"); p.add_argument("--key")
p.add_argument("--apns-p8", metavar="PATH"); p.add_argument("--apns-key-id"); p.add_argument("--apns-team-id"); p.add_argument("--apns-bundle-id")
p.add_argument("--fcm-json", metavar="PATH")
p.add_argument("--web-origin"); p.add_argument("--web-icon")
p.set_defaults(fn=cmd_provision)
args = ap.parse_args()
args.fn(args)
if __name__ == "__main__":
main()
SHA-256: 85104dcbdf915621a38bead0de8217d0d4c1b28aae19d32e2f67fc2244261cf8