watchfor (Python)

Official Python SDK + CLI for WatchFor — uptime & infrastructure monitoring (monitors, alert rules, incidents, maintenance windows) over the REST API. Zero dependencies (standard library only).

There is also a TypeScript SDK, an MCP server and an A2A agent for AI agents.

Install

pip install watchfor

Quick start

from watchfor import WatchFor

wf = WatchFor(api_key="wf_live_...")  # create keys in Settings → API keys

# One-call org snapshot
print(wf.summary())

# List monitors
for m in wf.monitors.list()["data"]:
    print(m["name"], m["status"])

# Create a monitor (idempotency key optional, for safe retries)
mon = wf.monitors.create(
    {
        "name": "example.com",
        "type": "http",
        "target": "https://example.com",
        "interval": 300,
        "locations": ["<location-id>"],  # from wf.locations()
    },
    idempotency_key="create-example-1",
)

# Diagnose: firing incidents right now
for inc in wf.incidents.list(status="firing")["data"]:
    print(inc["message"], inc["severity"])

# False positive? Keep the resolved incident on record but drop it from
# uptime, SLA, reports and the status page (include() undoes it)
wf.incidents.exclude(19351, reason="Probe-side DNS hiccup")

# Irreversible: wipe a monitor's checks and incidents; uptime restarts "since reset"
wf.monitors.reset(mon["id"])

# Page integrity: after a deploy, check the pages and accept what that check saw
wf.monitors.accept_baseline(integrity_monitor_id, fresh=True)

# Browser checks: try a Playwright script once (nothing saved), then read the result
run = wf.playwright.test_run(script=script, target="https://shop.example.com")
wf.playwright.get_test_run(run["id"])  # status: queued | running | done | error
wf.variables.put("LOGIN_PASSWORD", value=password, secret=True)

# The same org secrets in HTTP/API/MCP checks: {{NAME}} in auth, a header or the body
wf.variables.put("API_TOKEN", value=token, secret=True)
wf.monitors.update(api_monitor_id, {"config": {"requestHeaders": [
    {"key": "Authorization", "value": "Bearer {{API_TOKEN}}"}]}})

# Older checks: pass next_cursor back as cursor (keep the same hours/success)
page = wf.monitors.checks(mon["id"], hours=24, limit=100)
if page["next_cursor"]:
    wf.monitors.checks(mon["id"], hours=24, limit=100, cursor=page["next_cursor"])

# Delete a contact group that alert rules still use (409 without force)
wf.contact_groups.delete("<group-id>", force=True)

# What the alerting engine sees right now: alerted, recovering 1/3, confirming 2 of 3…
state = wf.monitors.get(mon["id"])["alerting_state"]

# Response times on a real time axis (5-min buckets for 24h, 15 min for 7d)
up = wf.monitors.uptime(mon["id"], "24h", series=True)
for b in up.get("response_time_series", []):
    print(b["at"], b["avg_ms"])

# Pause up to 100 monitors at once; post a status-page announcement
wf.monitors.bulk("pause", ["<id-1>", "<id-2>"])
notice = wf.status_pages.create_announcement(
    "<page-id>", {"title": "Checkout errors", "body": "We are looking into it."}
)
wf.status_pages.update_announcement(
    "<page-id>", notice["id"], {"body": "Fixed.", "status": "resolved"}
)

Live diagnostics

Run any of 24 checks from WatchFor's probe fleet against any public target, monitored or not — this measures right now, rather than reading what WatchFor recorded:

# What can I run, and how much budget is left?
catalog = wf.diagnostics.list()

# One check, optionally from a location you choose (plan-gated)
dns = wf.diagnostics.run(
    "dns-lookup", "example.com", options={"recordType": "A", "resolver": "8.8.8.8"}
)
# A closed port / NXDOMAIN / failed handshake does NOT raise:
if not dns["success"]:
    print("finding:", dns["error"])
print("runs left this hour:", dns["remaining"])

# Graded checks (tls-grade, http-headers, cwv-check, mcp-check, email-health,
# mail-server-check, cdn-check, smart-audit, dns-health) also return the
# dashboard's WatchFor Grade: letter, rubric id and what holds it down
headers = wf.diagnostics.run("http-headers", "https://example.com")
grade = headers.get("grade") or {}
if grade.get("letter"):
    print(f"grade {grade['letter']} ({grade['rubric']})")
    for cap in grade["capped_by"]:
        print(f"  max {cap['cap']}: {cap['title']} — fix: {cap['fix']}")
else:
    print("not graded:", grade.get("not_graded_reason"))

# The whole picture in one call: DNS + propagation + TLS + HTTP + ping
# from up to 3 regions, aggregated into one verdict
report = wf.diagnostics.diagnose_target("example.com")
print(report["verdict"]["status"], report["verdict"]["summary"])

Runs need a write key (a probe sends real traffic from WatchFor's IPs) and spend the same per-plan hourly allowance as the dashboard Toolbox. Details: https://watchfor.io/docs/api/diagnostics

Resource namespaces: wf.monitors, wf.alert_rules, wf.incidents, wf.hosts, wf.status_pages, wf.on_call, wf.notification_channels, wf.maintenance_windows, wf.contacts, wf.contact_groups, wf.diagnostics, wf.playwright, wf.variables. Top-level: wf.summary(), wf.plan(), wf.me(), wf.locations(), wf.monitor_types(), wf.incident_stats(period=...), wf.report(...), wf.notifications(...), wf.activity(...), wf.batch([...]), wf.metrics(...).

Errors raise WatchForError with .status, .code and .message.

Organizations

An API key belongs to one organization. An OAuth access token acts as the user who approved it, so it reaches every organization that user belongs to; choose one with organization — the SDK sends it as the X-WatchFor-Organization header on every request:

acme = WatchFor(api_key=oauth_access_token, organization="<organization-id>")
acme.me()  # {"organization": {"id", "name", "slug"}, "scope": ...}
other = acme.with_organization("<another-organization-id>")  # new client, same token
other.monitors.list()

• Without organization, an OAuth token acts on the user's oldest owned organization (else their oldest membership).
• The user must be a member of that organization (otherwise 403 forbidden), and their role there applies: members read, acknowledge incidents and run diagnostics; changes need owner or admin.
• With an API key leave it unset; it may only repeat the key's own organization id, and any other id is 403 forbidden (a key never switches organization).
• with_organization(None) drops the header again.

Details: https://watchfor.io/docs/api/authentication#how-oauth-maps-to-your-data

Prometheus metrics (Pro plan and above)

text = wf.metrics(tag="production")  # Prometheus text format

Usually you point Prometheus at https://watchfor.io/api/v1/metrics directly; this is for scripts and one-off checks. See https://watchfor.io/docs/api/prometheus.

CLI

export WATCHFOR_API_KEY=wf_live_...
watchfor summary
watchfor monitors
watchfor incidents --status firing
watchfor checks <monitor_id> [--cursor <next_cursor>]
watchfor report --period 30d
watchfor me                       # which organization + scope

# OAuth token: pick the organization (before the command) or via the environment
export WATCHFOR_API_KEY=<oauth-access-token>
watchfor --organization <organization-id> monitors
WATCHFOR_ORGANIZATION=<organization-id> watchfor summary

Auth & scopes

Keys carry a scope: read (all GET endpoints) or write (read plus create/update/delete). See authentication. The API is also reachable with OAuth 2.1 access tokens (pass one as api_key; see Organizations).

Reference

• OpenAPI spec: https://watchfor.io/openapi.json
• Guides: https://watchfor.io/docs/api

MIT License.

Releasing

./publish.sh          # build, check, confirm, upload
./publish.sh --build  # build and check only

Debian and Ubuntu mark the system Python "externally managed" (PEP 668), so pip install build twine is refused and python3 -m build reports No module named build. The script keeps its own .venv-publish instead of touching the system Python. Credentials are read from ~/.pypirc as usual.