Integration

Python

Know when your Python agent or job silently stops.

from nedwatch import Ned                       # pip install ned-watch, and set NED_CALLBACK_URL
@Ned().deadman("nightly-sync", every="1h")     # Ned fires if this hasn't finished cleanly for an hour
def sync(): ...

ned-watch has no dependencies (it uses urllib). If httpx is installed, the async calls use it. Python 3.9 and up.

Setup

pip install ned-watch
export NED_CALLBACK_URL=https://your-agent.example/hooks/ned

That's all. The first time a watch is used, the client creates your agent, saves the agent key to ~/.config/ned-watch/state.json (mode 0600, never printed), registers the watch with condition.arm so the clock starts right away.

Deadman: "fire if this stops succeeding"

from nedwatch import Ned

ned = Ned()

@ned.deadman("nightly-sync", every="1h")        # every: "90s", "20m", "1h", "1h30m", "1d", or seconds
def sync():
    ...                                         # a return checks in; an exception doesn't

Or check in yourself, for example at the end of each loop of a long-running agent:

heartbeat = ned.deadman("research-agent", every="10m")
while True:
    do_one_task()
    heartbeat.checkin()

grace= sets a different deadline from every (for example every="1h", grace="90m").

Overrun: "fire if this is still running"

with ned.run("etl", max="20m"):
    load_everything()

run() starts an overrun run on entry and finishes it on a clean exit. If the block is still running 20 minutes after it started, Ned fires once; if it finishes late, he sends clear with late: true and the runtime.

Add every= to get both in one go: the overrun catches hangs, the deadman catches failures and missed schedules.

with ned.run("etl", max="20m", every="1d"):
    load_everything()

What happens when the block raises:

ned.run(...) also works as a decorator (a new run per call) and with async with.

Async

The decorator and run() work on async def functions and with async with. Direct calls have a versions: await ned.acheckin(...), await ned.astart(...), await ned.afinish(...), await ned.aget(...).

@ned.deadman("poller", every="5m")
async def poll(): ...

async with ned.run("crawl", max="1h"):
    await crawl()

Direct calls

ned.checkin("nightly-sync")                    # by name (registered by this client), or by id "w_..."
ned.checkin("w_0123456789ab", signing_secret=...)
ned.start(overrun, run_id="2026-10-05")        # run_id makes retries safe
ned.finish(overrun, run_id="2026-10-05")       # failed=True, error="..." reports a crash
ned.get("w_...")  ned.watches()  ned.cancel("w_...")  ned.balance()
ned.register("http", target="https://example.com/health", interval_s=300)   # any watch type, as is

Errors are NedError (with .status and .detail); a missing callback URL or secret is NedConfigError. The decorator and run() never raise for a Ned problem unless you made the client with Ned(strict=True).

Command line

nedwatch setup --name nightly-sync --every 1h [--max 30m]   # make the watch(es) and do the first check-in
nedwatch checkin nightly-sync                               # put at the end of a job
ned-run --name backup --max 30m --every 1d -- ./backup.sh  # wrap any command (see cron)
nedwatch doctor [--create]                                  # key, connectivity, a real test callback
nedwatch status

uvx ned-watch <command> runs the same CLI without installing anything. Secrets stay in the state file; setup --show-secrets prints them as env lines if you need them elsewhere.

Frameworks

• LangGraph / LangChain: NedCallbackHandler
• CrewAI: kickoff(crew, ...)
• OpenAI Agents SDK: NedRunHooks
• Claude Agent SDK: NedHooks

Ask before you call

No key, no account. Ned publishes what he sees of agents' MCP servers, the AI APIs and the services agents depend on at ned.watch/w/; this asks it by host.

from nedwatch import is_up, up_state

if up_state("api.openai.com") == "down":      # "up" | "degraded" | "down" | "reachable" | "unknown"
    ...fail over, or wait...
r = is_up("status.github.com")                # dict: state, since, reason, watches, last_change, ask_again_s

degraded means it answered but not as expected (for a vendor's status feed, their own page reports an incident). unknown means the host is not on The Watch, never that it is down; the answer then says how to add it. A network problem returns unknown with an error field rather than raising, so a pre-flight check never breaks the call it guards.

Verifying callbacks

from nedwatch import verify_signature
if not verify_signature(signing_secret, request.headers, await request.body()):
    return Response(status_code=401)