Notification Hub: Slack, Discord & Telegram

from pathlib import Path
import sys

ROOT = (
    Path.cwd().resolve().parent
    if Path.cwd().name == "notebooks"
    else Path.cwd().resolve()
)
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

import json
from shipit_agent.notifications import (
    Notification,
    NotificationManager,
    SlackNotifier,
    DiscordNotifier,
    TelegramNotifier,
)
from shipit_agent.notifications.base import SEVERITY_ORDER
from shipit_agent.notifications.templates import DEFAULT_TEMPLATES, render_template
from IPython.display import display, Markdown

note = Notification(
    event="run_completed",
    title="Security Auditor — Run Completed",
    message="Audit completed in 4.2s. Found 3 critical, 5 high vulnerabilities.",
    severity="warning",
    metadata={
        "agent": "security-auditor",
        "duration": "4.2s",
        "cost": "$0.47",
        "findings": "3 critical, 5 high",
    },
)

print(f"Event:     {note.event}")
print(f"Title:     {note.title}")
print(f"Message:   {note.message}")
print(f"Severity:  {note.severity}")
print(f"Metadata:  {note.metadata}")
print(f"Timestamp: {note.timestamp}")

print(f"\nJSON:\n{json.dumps(note.to_dict(), indent=2)}")

print("Severity levels (lowest → highest):")
for level, order in sorted(SEVERITY_ORDER.items(), key=lambda x: x[1]):
    emoji = {"info": "ℹ️", "warning": "⚠️", "error": "❌", "critical": "🚨"}[level]
    print(f"  {emoji} {level:10s} = {order}")

print("\nDefault templates:")
for event, template in DEFAULT_TEMPLATES.items():
    print(f"  {event:20s} → {template}")

# Safe rendering — missing vars stay as {var}
rendered = render_template(
    DEFAULT_TEMPLATES["run_completed"],
    agent="my-agent",
    duration="3.1s",
    cost="$0.23",
    output_preview="Done!",
)
print(f"\nRendered: {rendered}")

partial = render_template(DEFAULT_TEMPLATES["run_completed"], agent="my-agent")
print(f"Partial:  {partial}")

slack = SlackNotifier(
    webhook_url="https://hooks.slack.com/services/d//",
    username="ShipIt Agent",
    channel="#shipit-agent-notifications",
)

payload = slack._build_payload(note)
print("Slack Block Kit Payload:")
print(json.dumps(payload, indent=2))
print(f"\nSeverity color: {payload['attachments'][0]['color']}")
print(f"Block types: {[b['type'] for b in payload['attachments'][0]['blocks']]}")

discord = DiscordNotifier(
    webhook_url="https://discord.com/api/webhooks/dd/dsdsd",
    username="ShipIt Agent",
    avatar_url="https://shipiit.com/logo.png",
)

payload = discord._build_payload(note)
print("Discord Embed Payload:")
print(json.dumps(payload, indent=2))
embed = payload["embeds"][0]
print(f"\nColor: {embed['color']} (hex: {hex(embed['color'])})")
print(f"Fields: {[f['name'] for f in embed.get('fields', [])]}")

telegram = TelegramNotifier(bot_token="", chat_id="-1001237890")

formatted = telegram._format_markdown(note)
print("Telegram MarkdownV2 Message:")
print(formatted)

severities = ["info", "warning", "error", "critical"]
events = ["run_started", "cost_alert", "tool_failed", "run_completed"]

print("=== Severity Comparison ===\n")
for sev in severities:
    n = Notification(
        event="test",
        title=f"Test {sev.upper()}",
        message=f"This is a {sev} notification.",
        severity=sev,
        metadata={"agent": "test-agent"},
    )

    slack_color = slack._build_payload(n)["attachments"][0]["color"]
    discord_color = discord._build_embed(n)["color"]
    # tg_msg = telegram._format_markdown(n).split("\n")[0]  # first line has emoji

    print(
        f"  {sev:10s} | Slack: {slack_color} | Discord: {hex(discord_color):10s} | Telegram: "
    )

production_notifications = [
    Notification(
        event="run_started",
        title="Security Auditor — Started",
        message="Scanning /src for OWASP Top 10 vulnerabilities",
        severity="info",
        metadata={"agent": "security-auditor", "project": "my-api"},
    ),
    Notification(
        event="tool_failed",
        title="Bash Tool Failed",
        message="Permission denied: /etc/shadow",
        severity="error",
        metadata={"agent": "security-auditor", "tool": "bash", "error": "EACCES"},
    ),
    Notification(
        event="cost_alert",
        title="Budget Warning",
        message="Agent has spent $4.12 of $5.00 budget (82%)",
        severity="warning",
        metadata={"agent": "deep-researcher", "spent": "$4.12", "budget": "$5.00"},
    ),
    Notification(
        event="run_completed",
        title="Research Crew Complete",
        message="3 tasks completed in 45.2s. All passed.",
        severity="info",
        metadata={
            "crew": "research-crew",
            "tasks": "3/3",
            "duration": "45.2s",
            "cost": "$1.23",
        },
    ),
    Notification(
        event="crew_started",
        title="Deploy Pipeline Started",
        message="Crew 'deploy-pipe' started with 4 agents, 6 tasks",
        severity="info",
        metadata={"crew": "deploy-pipe", "agents": 4, "tasks": 6},
    ),
]

print("Production notification examples:\n")
for n in production_notifications:
    print(f"  [{n.severity:8s}] {n.event:20s} — {n.title}")
    print(f"           {n.message}")
    print(f"           metadata: {n.metadata}")
    print()

manager = NotificationManager(notifiers=[slack, discord, telegram], min_severity="info")

test_notes = [
    Notification(
        event="run_started", title="Started", message="Starting audit", severity="info"
    ),
    Notification(
        event="tool_failed", title="Failed", message="bash failed", severity="error"
    ),
    Notification(
        event="run_completed", title="Done", message="Complete", severity="info"
    ),
]

print("Dispatch decisions (min_severity='info'):")
for n in test_notes:
    print(f"  [{n.severity:8s}] {n.event:20s} → {manager._should_notify(n)}")

# Error-only filter
error_mgr = NotificationManager(notifiers=[slack], min_severity="error")
print("\nWith min_severity='error':")
for n in test_notes:
    print(f"  [{n.severity:8s}] {n.event:20s} → {error_mgr._should_notify(n)}")

# Event filter
event_mgr = NotificationManager(
    notifiers=[slack], events=["run_completed", "tool_failed"]
)
print("\nWith events=['run_completed', 'tool_failed']:")
for n in test_notes:
    print(f"  [{n.severity:8s}] {n.event:20s} → {event_mgr._should_notify(n)}")

# Full production wiring pattern
from shipit_agent import Agent
from examples.run_multi_tool_agent import build_llm_from_env

llm = build_llm_from_env("bedrock")

# In production, use real URLs from environment:
# import os
# slack_real = SlackNotifier(webhook_url=os.environ["SLACK_WEBHOOK"])
# discord_real = DiscordNotifier(webhook_url=os.environ["DISCORD_WEBHOOK"])
# telegram_real = TelegramNotifier(
#     bot_token=os.environ["TELEGRAM_TOKEN"],
#     chat_id=os.environ["TELEGRAM_CHAT_ID"],
# )

# Wire notifications + cost tracking together
from shipit_agent.costs import CostTracker, Budget

tracker = CostTracker(budget=Budget(max_dollars=2.00))
notify_hooks = manager.as_hooks(agent_name="demo-agent")
cost_hooks = tracker.as_hooks()

# Merge both hook sets into a single AgentHooks object.
notify_hooks.before_llm.extend(cost_hooks.before_llm)
notify_hooks.after_llm.extend(cost_hooks.after_llm)
notify_hooks.before_tool.extend(cost_hooks.before_tool)
notify_hooks.after_tool.extend(cost_hooks.after_tool)

agent = Agent.with_builtins(
    llm=llm,
    prompt="You are a helpful assistant.",
    hooks=notify_hooks,
)

result = agent.run("Explain the CAP theorem in 3 sentences.")

# After the run, you'd send notifications:
completion_note = Notification(
    event="run_completed",
    title="Agent Complete",
    message=f"Cost: ${tracker.total_cost:.4f} | Tokens: {tracker.total_tokens['input_tokens'] + tracker.total_tokens['output_tokens']:,}",
    severity="info",
    metadata={
        "cost": f"${tracker.total_cost:.4f}",
        "input_tokens": tracker.total_tokens["input_tokens"],
        "output_tokens": tracker.total_tokens["output_tokens"],
    },
)

print("Would send this notification:")
print(f"  {completion_note.title}: {completion_note.message}")
print("\nSlack payload preview:")
print(json.dumps(slack._build_payload(completion_note), indent=2)[:500])

display(Markdown(result.output[:400]))

hooks = manager.as_hooks(agent_name="security-auditor")
print(f"Hooks type: {type(hooks).__name__}")
print("\nEvents that trigger notifications:")
print("  • on_before_llm → run_started (first call only)")
print("  • on_after_llm  → run_completed (final LLM response only)")
print("  • on_after_tool  → tool_failed (on error)")

custom_mgr = NotificationManager(
    notifiers=[slack],
    templates={
        "run_started": "🚀 {agent} is working on: {prompt_preview}",
        "run_completed": "✅ {agent} done! ({duration}) — Cost: {cost}",
        "tool_failed": "🔥 ALERT: {agent} → {tool} failed: {error}",
    },
)

rendered = render_template(
    "✅ {agent} done! ({duration}) — Cost: {cost}",
    agent="auditor",
    duration="4.2s",
    cost="$0.47",
)
print(f"Custom: {rendered}")

from shipit_agent.costs import CostTracker, Budget

cost_alert = Notification(
    event="cost_alert",
    title="Budget Warning",
    message=render_template(
        DEFAULT_TEMPLATES["cost_alert"],
        agent="deep-researcher",
        spent="$4.12",
        budget="$5.00",
        percent="82",
    ),
    severity="warning",
    metadata={"agent": "deep-researcher", "spent": "$4.12", "budget": "$5.00"},
)

print(f"Title:    {cost_alert.title}")
print(f"Message:  {cost_alert.message}")
print(f"Severity: {cost_alert.severity}")
print(
    f"\nSlack color: {slack._build_payload(cost_alert)['attachments'][0]['color']} (yellow = warning)"
)