hansheinemann
084cfb0bb2
Adapters implemented: - adapters/llm/anthropic.py — Anthropic Claude SDK, capability-based model selection, max_tokens + temperature configurable via team.yaml, lazy SDK import - adapters/vcs/github.py — GitHub PR/branch operations via gh CLI - adapters/notify/openclaw.py — OpenClaw system event notifications - adapters/runtime/openclaw.py — OpenClaw sessions_spawn for agent execution - adapters/runtime/claude_code.py — Claude Code CLI for T4/T5 coding tasks All adapters follow the abstract base interfaces from Phase 1. Config-driven model selection via capability_map in team.yaml.
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)
|