1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
|
import json
import os
import sys
import urllib.error
import urllib.parse
import urllib.request
NTFY_SERVER = os.environ.get("NTFY_SERVER", "https://ntfy.sh").rstrip("/")
NTFY_TOPIC = os.environ.get("NTFY_TOPIC", "CPU")
NTFY_TAGS = os.environ.get("NTFY_TAGS", "computer")
NTFY_TOKEN = os.environ.get("NTFY_TOKEN", "").strip()
def shorten(text: str, limit: int) -> str:
text = " ".join(text.split())
if len(text) <= limit:
return text
return text[: limit - 3] + "..."
def header_safe(text: str, fallback: str) -> str:
try:
text.encode("latin-1")
return text
except UnicodeEncodeError:
return fallback
def main() -> int:
if len(sys.argv) < 2:
return 0
notification = json.loads(sys.argv[1])
if notification.get("type") != "agent-turn-complete":
return 0
last_msg = notification.get("last-assistant-message") or "Turn complete"
inputs = notification.get("input-messages") or []
cwd = notification.get("cwd") or ""
thread_id = notification.get("thread-id") or ""
raw_title = f"Codex: {shorten(last_msg, 72)}"
title = header_safe(raw_title, "Codex notification")
lines = [f"Assistant: {last_msg}"]
if inputs:
lines.append(f"User: {shorten(' '.join(inputs), 300)}")
if cwd:
lines.append(f"CWD: {cwd}")
if thread_id:
lines.append(f"Thread: {thread_id}")
body = "\n".join(lines).encode("utf-8")
url = f"{NTFY_SERVER}/{urllib.parse.quote(NTFY_TOPIC, safe='')}"
headers = {
"Title": title,
"Tags": NTFY_TAGS,
}
if NTFY_TOKEN:
headers["Authorization"] = f"Bearer {NTFY_TOKEN}"
request = urllib.request.Request(url, data=body, headers=headers, method="POST")
try:
with urllib.request.urlopen(request, timeout=10):
return 0
except urllib.error.URLError as error:
print(f"ntfy notify failed: {error}", file=sys.stderr)
return 0
if __name__ == "__main__":
raise SystemExit(main())
|