Push Notifications for Claude Code with ntfy + Hooks

Save to /usr/local/bin/ntfy-claude:

#!/usr/bin/env python3
"""Push notifications for Claude Code events via ntfy"""

import os
import sys
import json
import subprocess
import time
import hashlib

NTFY_TOPIC = "your-ntfy-topic"  # Change this!
NTFY_SERVER = "https://ntfy.sh"
RATE_LIMIT_SECONDS = 10

# Commands that are safe (don't notify)
SAFE_COMMANDS = ["git status", "git diff", "ls", "cat", "grep"]

def get_rate_limit_file(project):
    """Get path to rate limit file for this project"""
    safe_name = hashlib.md5(project.encode()).hexdigest()[:8]
    return f"/tmp/ntfy-claude-{safe_name}.last"

def should_rate_limit(project):
    """Check if we should skip notification due to rate limiting"""
    rate_file = get_rate_limit_file(project)
    try:
        if os.path.exists(rate_file):
            last_time = float(open(rate_file).read().strip())
            if time.time() - last_time < RATE_LIMIT_SECONDS:
                return True
    except:
        pass
    return False

def record_notification(project):
    """Record that we sent a notification for this project"""
    try:
        with open(get_rate_limit_file(project), 'w') as f:
            f.write(str(time.time()))
    except:
        pass

def is_terminal_focused():
    """Check if Terminal.app is frontmost"""
    try:
        result = subprocess.run(
            ["osascript", "-e",
             'tell app "System Events" to get name of first process whose frontmost is true'],
            capture_output=True, text=True
        )
        return "Terminal" in result.stdout
    except:
        return False

def is_system_idle(threshold=300):
    """Check if user has been idle for threshold seconds"""
    try:
        result = subprocess.run(
            ["ioreg", "-c", "IOHIDSystem"],
            capture_output=True, text=True
        )
        for line in result.stdout.split('\n'):
            if 'HIDIdleTime' in line:
                idle_ns = int(line.split('=')[1].strip())
                return (idle_ns / 1_000_000_000) > threshold
    except:
        pass
    return False

def send_notification(title, message, project, priority="default", tags=None):
    """Send notification via ntfy"""
    # Rate limit: skip if we just sent a notification for this project
    if should_rate_limit(project):
        return

    # Skip if user is actively using terminal
    if is_terminal_focused() and not is_system_idle():
        return

    cmd = [
        "curl", "-s",
        "-H", f"Title: {title}",
        "-H", f"Priority: {priority}",
    ]

    if tags:
        cmd.extend(["-H", f"Tags: {tags}"])

    cmd.extend(["-d", message, f"{NTFY_SERVER}/{NTFY_TOPIC}"])
    subprocess.run(cmd, capture_output=True)

    # Record that we sent a notification
    record_notification(project)

def format_ask_user_question(tool_input):
    """Format AskUserQuestion for rich notification"""
    questions = tool_input.get("questions", [])
    if not questions:
        return "Question from Claude"

    q = questions[0]
    header = q.get("header", "")
    question = q.get("question", "")[:120]
    options = q.get("options", [])

    parts = []
    if header:
        parts.append(f"**{header}**")
    if question:
        parts.append(question)
    if options:
        opts = " | ".join([o.get("label", "") for o in options[:4]])
        parts.append(f"→ {opts}")

    return "\n".join(parts)

def handle_pre_tool_use(data):
    """Handle PreToolUse events"""
    tool_name = data.get("tool_name", "")
    tool_input = data.get("tool_input", {})
    cwd = os.path.basename(data.get("cwd", ""))

    if tool_name == "AskUserQuestion":
        message = format_ask_user_question(tool_input)
        send_notification(
            f"❓ {cwd}",
            message,
            cwd,
            priority="high",
            tags="question"
        )
    elif tool_name == "Bash":
        command = tool_input.get("command", "")
        # Skip safe commands
        if any(command.startswith(safe) for safe in SAFE_COMMANDS):
            return
        # Notify for potentially dangerous commands
        send_notification(
            f"⚠️ {cwd}",
            f"Command: {command[:100]}",
            cwd,
            priority="default",
            tags="warning"
        )

def handle_stop(data):
    """Handle Stop events (task completed)"""
    cwd = os.path.basename(data.get("cwd", ""))
    send_notification(
        f"✅ {cwd}",
        "Task completed",
        cwd,
        priority="default",
        tags="white_check_mark"
    )

def handle_notification(data):
    """Handle Notification events"""
    message = data.get("message", "")
    cwd = os.path.basename(data.get("cwd", ""))

    if "waiting for" in message.lower():
        send_notification(
            f"⏳ {cwd}",
            "Waiting for input",
            cwd,
            priority="high",
            tags="hourglass"
        )

def main():
    if len(sys.argv) < 2:
        return

    event_type = sys.argv[1]
    data = json.loads(sys.stdin.read())

    handlers = {
        "PreToolUse": handle_pre_tool_use,
        "Stop": handle_stop,
        "Notification": handle_notification,
    }

    handler = handlers.get(event_type)
    if handler:
        handler(data)

if __name__ == "__main__":
    main()