Hans Heinemann
c88c4309ac
Sends notifications via `openclaw system event --text <msg> --mode now`. - Always logs locally (info/warning/error) regardless of CLI availability - Gracefully handles FileNotFoundError (openclaw not on PATH) and TimeoutExpired; notifications are best-effort and never crash the pipeline - OPENCLAW_SIGNAL_NUMBER env var stored for future direct-signal support Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
94 lines
3.1 KiB
Python
94 lines
3.1 KiB
Python
"""
|
|
adapters/notify/openclaw.py
|
|
OpenClaw notification adapter — Phase 2 implementation.
|
|
|
|
Sends notifications by shelling out to the ``openclaw`` CLI::
|
|
|
|
openclaw system event --text "<message>" --mode now
|
|
|
|
If the binary is not on PATH the method logs a warning and returns without
|
|
raising — notifications are best-effort and should never crash the pipeline.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import os
|
|
import subprocess
|
|
|
|
from adapters.base.notify import NotifyAdapter
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class OpenClawNotifyAdapter(NotifyAdapter):
|
|
"""
|
|
Notification adapter that dispatches messages via the ``openclaw`` CLI.
|
|
|
|
Environment variables
|
|
---------------------
|
|
OPENCLAW_SIGNAL_NUMBER : Optional. Direct signal target for OpenClaw sends.
|
|
"""
|
|
|
|
def __init__(self, config: dict) -> None:
|
|
"""
|
|
Initialise the OpenClaw notification adapter.
|
|
|
|
Parameters
|
|
----------
|
|
config : Loaded team.yaml config dict (reserved for future options).
|
|
"""
|
|
self._config = config
|
|
self._signal_number: str = os.environ.get("OPENCLAW_SIGNAL_NUMBER", "")
|
|
|
|
def send(self, message: str, context: dict) -> None:
|
|
"""
|
|
Send a notification via ``openclaw system event``.
|
|
|
|
Parameters
|
|
----------
|
|
message : Human-readable notification text.
|
|
context : Optional metadata. Recognised keys:
|
|
level (str) — "info" | "warning" | "error"; logged locally.
|
|
run_id (str) — included in the local log record.
|
|
brief_id (str) — included in the local log record.
|
|
|
|
Notes
|
|
-----
|
|
If the ``openclaw`` binary is not present on PATH, the method logs a
|
|
warning and returns silently. Notifications are best-effort.
|
|
"""
|
|
level: str = context.get("level", "info")
|
|
run_id: str = context.get("run_id", "")
|
|
brief_id: str = context.get("brief_id", "")
|
|
|
|
# Always log locally regardless of CLI availability.
|
|
log_msg = "[notify:%s] %s (run=%s brief=%s)" % (level, message, run_id, brief_id)
|
|
if level == "error":
|
|
logger.error(log_msg)
|
|
elif level == "warning":
|
|
logger.warning(log_msg)
|
|
else:
|
|
logger.info(log_msg)
|
|
|
|
cmd = ["openclaw", "system", "event", "--text", message, "--mode", "now"]
|
|
try:
|
|
result = subprocess.run(
|
|
cmd,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=30,
|
|
)
|
|
if result.returncode != 0:
|
|
logger.warning(
|
|
"openclaw event returned non-zero exit %d: %s",
|
|
result.returncode,
|
|
result.stderr.strip(),
|
|
)
|
|
except FileNotFoundError:
|
|
logger.warning(
|
|
"openclaw CLI not found on PATH; notification not delivered: %s",
|
|
message,
|
|
)
|
|
except subprocess.TimeoutExpired:
|
|
logger.warning("openclaw event timed out for message: %s", message)
|