feat: native notification hook with three backend options

Author: GangGreenTemperTatum

dreadnode/agent/hooks/__init__.py

@@ -4,12 +4,24 @@
     retry_with_feedback,
 )
 from dreadnode.agent.hooks.metrics import tool_metrics
+from dreadnode.agent.hooks.notification import (
+    LogNotificationBackend,
+    NotificationBackend,
+    TerminalNotificationBackend,
+    WebhookNotificationBackend,
+    notify,
+)
 from dreadnode.agent.hooks.summarize import summarize_when_long
 
 __all__ = [
     "Hook",
+    "LogNotificationBackend",
+    "NotificationBackend",
+    "TerminalNotificationBackend",
+    "WebhookNotificationBackend",
     "backoff_on_error",
     "backoff_on_ratelimit",
+    "notify",
     "retry_with_feedback",
     "summarize_when_long",
     "tool_metrics",

dreadnode/agent/hooks/notification.py

@@ -0,0 +1,106 @@
+import typing as t
+from abc import ABC, abstractmethod
+
+from loguru import logger
+
+if t.TYPE_CHECKING:
+    from dreadnode.agent.events import AgentEvent
+    from dreadnode.agent.reactions import Reaction
+
+
+class NotificationBackend(ABC):
+    @abstractmethod
+    async def send(self, event: "AgentEvent", message: str) -> None:
+        """Send a notification for the given event."""
+
+
+class LogNotificationBackend(NotificationBackend):
+    async def send(self, event: "AgentEvent", message: str) -> None:
+        logger.info(f"[{event.agent.name}] {message}")
+
+
+class TerminalNotificationBackend(NotificationBackend):
+    async def send(self, event: "AgentEvent", message: str) -> None:
+        import sys
+
+        print(f"[{event.agent.name}] {message}", file=sys.stderr)
+
+
+class WebhookNotificationBackend(NotificationBackend):
+    def __init__(self, url: str, headers: dict[str, str] | None = None):
+        self.url = url
+        self.headers = headers or {}
+
+    async def send(self, event: "AgentEvent", message: str) -> None:
+        import httpx
+
+        payload = {
+            "agent": event.agent.name,
+            "event": event.__class__.__name__,
+            "message": message,
+            "timestamp": event.timestamp.isoformat(),
+        }
+
+        async with httpx.AsyncClient() as client:
+            await client.post(self.url, json=payload, headers=self.headers)
+
+
+def notify(
+    event_type: "type[AgentEvent] | t.Callable[[AgentEvent], bool]",
+    message: str | t.Callable[["AgentEvent"], str],
+    backend: NotificationBackend | None = None,
+) -> t.Callable[["AgentEvent"], t.Awaitable["Reaction | None"]]:
+    """
+    Create a notification hook that sends notifications when events occur.
+
+    Unlike other hooks, notification hooks don't affect agent execution - they return
+    None (no reaction) and run asynchronously to deliver notifications.
+
+    Args:
+        event_type: Event type to trigger on, or predicate function
+        message: Static message or callable that generates message from event
+        backend: Notification backend (defaults to terminal output)
+
+    Returns:
+        Hook that sends notifications
+
+    Example:
+        ```python
+        from dreadnode.agent import Agent
+        from dreadnode.agent.events import ToolStart
+        from dreadnode.agent.hooks.notification import notify
+
+        agent = Agent(
+            name="analyzer",
+            hooks=[
+                notify(
+                    ToolStart,
+                    lambda e: f"Starting tool: {e.tool_name}",
+                ),
+            ],
+        )
+        ```
+    """
+    notification_backend = backend or TerminalNotificationBackend()
+
+    async def notification_hook(event: "AgentEvent") -> "Reaction | None":
+        should_notify = False
+
+        if isinstance(event_type, type):
+            should_notify = isinstance(event, event_type)
+        elif callable(event_type):
+            should_notify = event_type(event)
+
+        if not should_notify:
+            return None
+
+        msg = message(event) if callable(message) else message
+
+        try:
+            await notification_backend.send(event, msg)
+        except Exception:  # noqa: BLE001
+            logger.exception("Notification hook failed")
+
+        return None
+
+    return notification_hook

pyproject.toml

@@ -192,3 +192,9 @@ skip-magic-trailing-comma = false
 "dreadnode/transforms/language.py" = [
     "RUF001",  # intentional use of ambiguous unicode characters for airt
 ]
+"dreadnode/agent/tools/interaction.py" = [
+    "T201",  # print required for user interaction
+]
+"dreadnode/agent/hooks/notification.py" = [
+    "T201",  # print required for terminal notifications
+]

tests/test_notification_hook.py

@@ -0,0 +1,161 @@
+import typing as t
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from dreadnode.agent.events import AgentEvent, ToolStart
+from dreadnode.agent.hooks.notification import (
+    LogNotificationBackend,
+    NotificationBackend,
+    TerminalNotificationBackend,
+    WebhookNotificationBackend,
+    notify,
+)
+
+
+class MockEvent(AgentEvent):
+    pass
+
+
+@pytest.fixture
+def mock_event() -> AgentEvent:
+    agent = MagicMock()
+    agent.name = "test_agent"
+    thread = MagicMock()
+    messages: list[t.Any] = []
+    events: list[AgentEvent] = []
+
+    return MockEvent(
+        session_id=MagicMock(),
+        agent=agent,
+        thread=thread,
+        messages=messages,
+        events=events,
+    )
+
+
+async def test_log_notification_backend(mock_event: AgentEvent) -> None:
+    from unittest.mock import patch
+
+    backend = LogNotificationBackend()
+
+    with patch("dreadnode.agent.hooks.notification.logger.info") as mock_logger:
+        await backend.send(mock_event, "Test notification")
+
+    mock_logger.assert_called_once()
+    call_args = mock_logger.call_args[0][0]
+    assert "Test notification" in call_args
+    assert "test_agent" in call_args
+
+
+async def test_terminal_notification_backend(mock_event: AgentEvent) -> None:
+    from io import StringIO
+    from unittest.mock import patch
+
+    backend = TerminalNotificationBackend()
+
+    stderr_capture = StringIO()
+    with patch("sys.stderr", stderr_capture):
+        await backend.send(mock_event, "Test notification")
+
+    output = stderr_capture.getvalue()
+    assert "Test notification" in output
+    assert "test_agent" in output
+
+
+async def test_webhook_notification_backend(mock_event: AgentEvent) -> None:
+    mock_client = AsyncMock()
+    mock_post = AsyncMock()
+    mock_client.__aenter__.return_value.post = mock_post
+
+    backend = WebhookNotificationBackend("https://example.com/webhook")
+
+    from unittest.mock import patch
+
+    import httpx
+
+    with patch.object(httpx, "AsyncClient", return_value=mock_client):
+        await backend.send(mock_event, "Test notification")
+
+    mock_post.assert_called_once()
+    call_kwargs = mock_post.call_args.kwargs
+    assert call_kwargs["json"]["message"] == "Test notification"
+    assert call_kwargs["json"]["agent"] == "test_agent"
+
+
+async def test_notify_hook_with_event_type(mock_event: AgentEvent) -> None:
+    backend = MagicMock(spec=NotificationBackend)
+    backend.send = AsyncMock()
+
+    hook = notify(MockEvent, "Test message", backend=backend)
+
+    reaction = await hook(mock_event)
+
+    assert reaction is None
+    backend.send.assert_called_once_with(mock_event, "Test message")
+
+
+async def test_notify_hook_uses_terminal_by_default(mock_event: AgentEvent) -> None:
+    from io import StringIO
+    from unittest.mock import patch
+
+    hook = notify(MockEvent, "Default notification")
+
+    stderr_capture = StringIO()
+    with patch("sys.stderr", stderr_capture):
+        reaction = await hook(mock_event)
+
+    assert reaction is None
+    output = stderr_capture.getvalue()
+    assert "Default notification" in output
+
+
+async def test_notify_hook_with_callable_message(mock_event: AgentEvent) -> None:
+    backend = MagicMock(spec=NotificationBackend)
+    backend.send = AsyncMock()
+
+    hook = notify(MockEvent, lambda e: f"Event from {e.agent.name}", backend=backend)
+
+    reaction = await hook(mock_event)
+
+    assert reaction is None
+    backend.send.assert_called_once_with(mock_event, "Event from test_agent")
+
+
+async def test_notify_hook_with_predicate(mock_event: AgentEvent) -> None:
+    backend = MagicMock(spec=NotificationBackend)
+    backend.send = AsyncMock()
+
+    hook = notify(lambda e: e.agent.name == "test_agent", "Matched!", backend=backend)
+
+    reaction = await hook(mock_event)
+
+    assert reaction is None
+    backend.send.assert_called_once()
+
+
+async def test_notify_hook_no_match(mock_event: AgentEvent) -> None:
+    backend = MagicMock(spec=NotificationBackend)
+    backend.send = AsyncMock()
+
+    hook = notify(ToolStart, "Should not send", backend=backend)
+
+    reaction = await hook(mock_event)
+
+    assert reaction is None
+    backend.send.assert_not_called()
+
+
+async def test_notify_hook_handles_backend_failure(mock_event: AgentEvent) -> None:
+    from unittest.mock import patch
+
+    backend = MagicMock(spec=NotificationBackend)
+    backend.send = AsyncMock(side_effect=Exception("Backend failed"))
+
+    hook = notify(MockEvent, "Test message", backend=backend)
+
+    with patch("dreadnode.agent.hooks.notification.logger.exception") as mock_logger:
+        reaction = await hook(mock_event)
+
+    assert reaction is None
+    mock_logger.assert_called_once_with("Notification hook failed")