feat: native notification hook with three backend options

[GangGreenTemperTatum]

[HOOKS] notification hook for local and remote backends

Key Changes:

• implements a native notification hook in the sdk with three backend options

1. TerminalNotificationBackend (default) - prints to stderr
2. LogNotificationBackend - uses loguru logger
3. WebhookNotificationBackend - HTTP POST to webhook

Added:

• implements a native notification hook in the sdk with three backend options and tests

Usgae example:

from dreadnode.agent import Agent
from dreadnode.agent.events import ToolStart, AgentEnd
from dreadnode.agent.hooks import notify, WebhookNotificationBackend, LogNotificationBackend

agent = Agent(
    name="scanner",
    hooks=[
        # Terminal (default) - visible immediately
        notify(ToolStart, lambda e: f"Starting: {e.tool_name}"),

        # Webhook - send to external service
        notify(
            ToolStart,
            lambda e: f"Tool {e.tool_name} started",
            backend=WebhookNotificationBackend(
                "https://example.com/webhook",
                headers={"Authorization": "Bearer token123"}
            ),
        ),

        # Logger - via loguru
        notify(
            AgentEnd,
            "Agent completed",
            backend=LogNotificationBackend(),
        ),
    ],
)

---

Generated Summary:

• Introduced a new Notification module with multiple backend implementations.
• Added LogNotificationBackend, TerminalNotificationBackend, and WebhookNotificationBackend for sending notifications.
• The notify function allows for flexible notification sending based on event types and messages.
• The new NotificationBackend class is implemented as an abstract base class, enhancing extensibility.
• Included tests for each notification backend to ensure functionality and robustness.
• Documentation for the notify function includes usage examples to assist developers.
• Updated pyproject.toml to include the new notification file for linting purposes.

This summary was generated with ❤️ by rigging

[rdheekonda]

Great work, @GangGreenTemperTatum . The current design requires users toi wrap every notification via notify() call, which doesn't align with the patterns in our sdk or the broad agent patterns (crewai).

1. It's repetitive and verbose.
   Ex.,

agent = Agent(name="scanner", hooks=[notify(ToolStart, ...), notify(ToolEnd, ...)])

2. inconsistent with our sdk pattern.

Looking at TaskAgent, we already auto inject hooks and tools here

We could consider leveraging the same pattern for notifications.

Recommended Design:

COnsider adding a top-level notifications param to Agent that auto-inject the notifications hook.

something like below

class Agent(Model):
    # ... existing fields ...
    
    notifications: bool | NotificationBackend | None = None
    """
    Enable notifications. 
    - True: Uses TerminalNotificationBackend (stderr output)
    - NotificationBackend instance: Uses custom backend
    - None/False: Disabled
    """
    
    notification_events: list[type[AgentEvent]] | Literal["all"] = "all"
    """Which event types to notify on. Defaults to all events."""
    
    notification_formatter: Callable[[AgentEvent], str] | None = None
    """Custom formatter for notification messages. If None, uses event's default representation."""
    
    def model_post_init(self, context: t.Any) -> None:
        super().model_post_init(context)
        
        # Auto-inject notification hook if enabled
        if self.notifications:
            backend = (
                self.notifications 
                if isinstance(self.notifications, NotificationBackend) 
                else TerminalNotificationBackend()
            )
            
            self.hooks.append(
                self._create_notification_hook(
                    backend, 
                    self.notification_events,
                    self.notification_formatter,
                )
            )

In that way: 80-90% of users can consume like below

agent = Agent(
    name="scanner",
    notifications=True,
)

Could filter specific events:

agent = Agent(
    name="scanner",
    notifications=True,
    notification_events=[ToolStart, ToolEnd, AgentEnd],
)

Power users can still customize:

agent = Agent(
    name="scanner",
    notifications=WebhookNotificationBackend(
        url="https:/....",
        headers={"Authorization": "Bearer token"},
    ),
)

Implementation: Delegating Formatting to resp[ective Events

delegate formatting to the events themselves. This is scalable and maintainable:

# In events.py - add a method to the base class
@dataclass
class AgentEvent:
    # ... existing fields ...
    
    def format_notification(self) -> str:
        """
        Format this event as a human-readable notification message.
        Override in subclasses for custom formatting.
        """
        return f"{self.__class__.__name__}"


@dataclass
class ToolStart(AgentEventInStep):
    tool_call: rg.tools.ToolCall
    
    def format_notification(self) -> str:
        return f"Starting tool: {self.tool_call.name}"


@dataclass
class ToolEnd(AgentEventInStep):
    tool_call: rg.tools.ToolCall
    message: rg.Message
    stop: bool
    
    def format_notification(self) -> str:
        status = " (requesting stop)" if self.stop else ""
        return f"Finished tool: {self.tool_call.name}{status}"


@dataclass
class GenerationEnd(AgentEventInStep):
    message: rg.Message
    usage: "rg.generator.Usage | None"
    
    def format_notification(self) -> str:
        tokens = self.usage.total_tokens if self.usage else "unknown"
        return f"Generation complete ({tokens} tokens)"


@dataclass
class AgentEnd(AgentEvent):
    stop_reason: "AgentStopReason"
    result: "AgentResult"
    
    def format_notification(self) -> str:
        status = "❌ Failed" if self.result.failed else "✅ Finished"
        return f"{status}: {self.stop_reason} (steps: {self.result.steps}, tokens: {self.result.usage.total_tokens})"


@dataclass
class AgentError(AgentEventInStep):
    error: BaseException
    
    def format_notification(self) -> str:
        return f"Error: {self.error.__class__.__name__}: {str(self.error)}"


@dataclass
class AgentStalled(AgentEventInStep):
    def format_notification(self) -> str:
        return "Agent stalled: no tool calls and no stop conditions met"

Then the noitifcation hook becomes simple something like below:

def _create_notification_hook(
    self,
    backend: NotificationBackend,
    events: list[type[AgentEvent]] | Literal["all"],
    formatter: Callable[[AgentEvent], str] | None,
) -> Hook:
    """Create a notification hook that delegates formatting to events."""
    
    async def notification_hook(event: AgentEvent) -> None:
        # Filter events
        if events != "all" and not any(isinstance(event, et) for et in events):
            return None
        
        # Use custom formatter if provided, otherwise delegate to event
        message = formatter(event) if formatter else event.format_notification()
        
        # Fire and forget - don't block agent execution
        asyncio.create_task(_safe_send(backend, event, message))
        
        return None
    
    return notification_hook


async def _safe_send(backend: NotificationBackend, event: AgentEvent, message: str) -> None:
    """Send notification with error handling."""
    try:
        await backend.send(event, message)
    except Exception:
        logger.exception(f"Notification failed for {event.__class__.__name__}")

With this:

1. Adding new events is easy, just add a format_notifications method to the event class
2. no central formatter to maintain
3. Consistent with AgentEvent.format_as_panel

[rdheekonda]

We are creating a new client per notification, consider creating in the init once and leverage withc context managers something like below:

class WebhookNotificationBackend(NotificationBackend):
    def __init__(self, url: str, headers: dict[str, str] | None = None, timeout: float = 5.0):
        self.url = url
        self.headers = headers or {}
        self.timeout = timeout
        self._client: httpx.AsyncClient | None = None
    
    async def __aenter__(self):
        self._client = httpx.AsyncClient(timeout=self.timeout)
        return self
    
    async def __aexit__(self, *args):
        if self._client:
            await self._client.aclose()
    
    async def send(self, event: "AgentEvent", message: str) -> None:
        if not self._client:
            self._client = httpx.AsyncClient(timeout=self.timeout)
        
        payload = self._build_payload(event, message)
        await self._client.post(self.url, json=payload, headers=self.headers)
    
    def _build_payload(self, event: "AgentEvent", message: str) -> dict:
        """Override this to customize webhook payload."""
        return {
            "agent": event.agent.name,
            "event": event.__class__.__name__,
            "message": message,
            "timestamp": event.timestamp.isoformat(),
        }

[rdheekonda]

Are we returning Reaction, based on the function should be None?

[GangGreenTemperTatum]

ty so much for the review @rdheekonda ! i just refactored my approach and hoping you could take a look (no rush) when you get a second if OK pls? new changes reflect httpx already being a dependency in the SDK (version >=0.28.0,<1.0.0)