43 lines
1.7 KiB
Python
43 lines
1.7 KiB
Python
"""External notification tool (#168) — Discord, Telegram, SMTP, webhook.
|
|
|
|
``notify_external`` is WRITE (external side effect → confirmation card in the
|
|
UI, propose/apply over MCP). Delivery itself lives in :mod:`backend.notify`.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from backend import notify as _notify
|
|
from backend.tools.context import ToolContext, ToolError, ToolRisk
|
|
from backend.tools.registry import tool
|
|
from backend.tools.schemas import NotifyExternalInput
|
|
|
|
|
|
@tool(
|
|
name="notify_external",
|
|
description="Send a notification through external channels (Discord, Telegram, SMTP, webhook).",
|
|
input_model=NotifyExternalInput,
|
|
risk=ToolRisk.WRITE,
|
|
)
|
|
def notify_external(ctx: ToolContext, params: NotifyExternalInput) -> dict[str, Any]:
|
|
"""Broadcast to the trigger scope, or target a single channel id."""
|
|
try:
|
|
if params.channel_id:
|
|
channel = next(
|
|
(c for c in _notify._read_channels() if c.get("id") == params.channel_id),
|
|
None,
|
|
)
|
|
if channel is None:
|
|
raise ToolError(f"Unknown channel: {params.channel_id}", code="not_found")
|
|
if not channel.get("enabled", True):
|
|
raise ToolError(f"Channel disabled: {params.channel_id}", code="invalid_arguments")
|
|
_notify.send_via_channel(channel, params.title, params.message, params.trigger)
|
|
return {"ok": True, "channel_id": params.channel_id}
|
|
results = _notify.broadcast(params.trigger, params.title, params.message)
|
|
return {"ok": True, "deliveries": results}
|
|
except ToolError:
|
|
raise
|
|
except Exception as e:
|
|
raise ToolError(f"Notification failed: {e}", code="notify_failed") from e
|