Notifications

# .env
# Slack — Create at https://api.slack.com/messaging/webhooks
SLACK_WEBHOOK_URL=https://hooks.slack.com/services/T00000000/B00000000/XXXXXXXXXXXXXXXXXXXXXXXX

# Discord — Server Settings → Integrations → Webhooks → New Webhook → Copy URL
DISCORD_WEBHOOK_URL=https://discord.com/api/webhooks/123456789/abcdefghijklmnop

# Telegram — Message @BotFather → /newbot → copy token, then get chat_id
TELEGRAM_BOT_TOKEN=123456:ABC-DEF1234ghIkl-zyx57W2v1u123ew11
TELEGRAM_CHAT_ID=-1001234567890

from shipit_agent import Agent
from shipit_agent.notifications import SlackNotifier, NotificationManager

slack = SlackNotifier(webhook_url="https://hooks.slack.com/services/T.../B.../xxx")
manager = NotificationManager([slack])

agent = Agent.with_builtins(llm=llm, hooks=manager.as_hooks("my-agent"))
result = agent.run("Analyze the codebase")
# Slack receives: "my-agent started", "my-agent completed in 12.3s"

import os
from dotenv import load_dotenv
from shipit_agent import Agent
from shipit_agent.notifications import (
    NotificationManager, SlackNotifier, DiscordNotifier, TelegramNotifier,
)
from shipit_agent.costs import CostTracker, Budget

load_dotenv()

# Build notifiers from environment variables
notifiers = []
if os.getenv("SLACK_WEBHOOK_URL"):
    notifiers.append(SlackNotifier(webhook_url=os.environ["SLACK_WEBHOOK_URL"]))
if os.getenv("DISCORD_WEBHOOK_URL"):
    notifiers.append(DiscordNotifier(webhook_url=os.environ["DISCORD_WEBHOOK_URL"]))
if os.getenv("TELEGRAM_BOT_TOKEN"):
    notifiers.append(TelegramNotifier(
        bot_token=os.environ["TELEGRAM_BOT_TOKEN"],
        chat_id=os.environ["TELEGRAM_CHAT_ID"],
    ))

manager = NotificationManager(
    notifiers=notifiers,
    min_severity="info",            # only info+ gets sent
    events=["run_completed", "tool_failed", "cost_alert"],  # filter events
)

# Cost tracker with budget alert
tracker = CostTracker(budget=Budget(max_dollars=5.00, warn_at=0.80))

# Wire everything into the agent
agent = Agent.with_builtins(
    llm=llm,
    prompt="You are a security auditor.",
    hooks=manager.as_hooks("security-auditor"),
)

result = agent.run("Audit the authentication module")
# Slack/Discord/Telegram all receive status updates automatically

from shipit_agent.notifications import Notification

note = Notification(
    event="run_completed",
    title="researcher -- Run Completed",
    message="researcher completed in 4.2s | Cost: 1832 tokens",
    severity="info",
    metadata={"agent": "researcher", "duration": "4.2s", "cost": "1832 tokens"},
)

# Serialize for logging or storage
d = note.to_dict()

from shipit_agent.notifications import SlackNotifier

slack = SlackNotifier(
    webhook_url="https://hooks.slack.com/services/T.../B.../xxx",
    channel="#agent-alerts",    # optional channel override (legacy webhooks only)
    username="ShipIt Agent",   # bot display name
)

# Synchronous send (no event loop needed)
slack.send_sync(note)

# Async send (inside an async context)
ok = await slack.send(note)

# Batch send
results = await slack.send_batch([note1, note2, note3])

from shipit_agent.notifications import DiscordNotifier

discord = DiscordNotifier(
    webhook_url="https://discord.com/api/webhooks/123/abc...",
    username="ShipIt Agent",
    avatar_url="https://example.com/bot-avatar.png",  # optional
)

ok = await discord.send(note)
discord.send_sync(note)

from shipit_agent.notifications import TelegramNotifier

telegram = TelegramNotifier(
    bot_token="123456:ABC-DEF1234ghIkl-zyx57W2v1u123ew11",
    chat_id="-1001234567890",
)

<emoji> *Title*
--------------------
Message body

*key:* value
*key:* value

_2026-04-13 14:30:00 UTC | info_

ok = await telegram.send(note)
telegram.send_sync(note)

from shipit_agent.notifications import (
    NotificationManager,
    SlackNotifier,
    DiscordNotifier,
    TelegramNotifier,
)

manager = NotificationManager(
    notifiers=[SlackNotifier(webhook_url="https://hooks.slack.com/..."),
        DiscordNotifier(webhook_url="https://discord.com/api/webhooks/..."),
        TelegramNotifier(bot_token="...", chat_id="..."),],
    min_severity="info",     # minimum severity to dispatch
    events=None,             # None = all events; or ["run_completed", "tool_failed"]
)

from shipit_agent.notifications import Notification

note = Notification(
    event="deployment",
    title="Deployment Complete",
    message="v2.1.0 deployed to production",
    severity="info",
    metadata={"version": "2.1.0", "env": "production"},
)

# Async
results = await manager.notify(note)
# {"slack": True, "discord": True, "telegram": False}

# Sync (creates event loop if needed, safe from any context)
results = manager.notify_sync(note)

hooks = manager.as_hooks(agent_name="researcher")
agent = Agent.with_builtins(llm=llm, hooks=hooks)

# Only errors and critical -- skip info and warning
manager = NotificationManager(
    notifiers=[slack],
    min_severity="error",
)

# Only notify on failures and cost alerts
manager = NotificationManager(
    notifiers=[slack],
    events=["tool_failed", "cost_alert"],
)

manager = NotificationManager(
    notifiers=[slack],
    templates={
        "run_started": "Agent `{agent}` is starting: {prompt_preview}",
        "run_completed": "Agent `{agent}` finished in {duration}. Cost: {cost}",
        "tool_failed": "ALERT: `{agent}` tool `{tool}` failed -- {error}",
        "cost_alert": "BUDGET: {agent} spent {spent} of {budget} ({percent}%)",
    },
)

from shipit_agent.costs import CostTracker, Budget
from shipit_agent.notifications import NotificationManager, SlackNotifier, Notification

slack = SlackNotifier(webhook_url="https://hooks.slack.com/...")
manager = NotificationManager([slack], min_severity="warning")

def on_budget_warning(spent: float, limit: float) -> None:
    note = Notification(
        event="cost_alert",
        title="Budget Warning",
        message=f"Agent has spent ${spent:.2f} of ${limit:.2f} budget",
        severity="warning",
        metadata={"spent": f"${spent:.2f}", "budget": f"${limit:.2f}"},
    )
    manager.notify_sync(note)

tracker = CostTracker(
    budget=Budget(max_dollars=5.00, warn_at=0.80),
    on_cost_alert=on_budget_warning,
)

agent = Agent.with_builtins(llm=llm, hooks=tracker.as_hooks())

from shipit_agent import Agent
from shipit_agent.costs import CostTracker, Budget
from shipit_agent.notifications import (
    NotificationManager,
    SlackNotifier,
    DiscordNotifier,
    Notification,
)

# Channels
slack = SlackNotifier(
    webhook_url=os.environ["SLACK_WEBHOOK"],
    username="Production Agent",
)
discord = DiscordNotifier(
    webhook_url=os.environ["DISCORD_WEBHOOK"],
)

# Manager: only errors+ to Discord, everything to Slack
slack_mgr = NotificationManager([slack], min_severity="info")
discord_mgr = NotificationManager([discord], min_severity="error")

# Cost tracking with budget alert -> Slack
def alert(spent, limit):
    slack_mgr.notify_sync(Notification(
        event="cost_alert",
        title="Budget Alert",
        message=f"${spent:.2f} of ${limit:.2f}",
        severity="warning",
    ))

tracker = CostTracker(
    budget=Budget(max_dollars=10.00, warn_at=0.80),
    on_cost_alert=alert,
)

# Merge hooks from both systems
agent = Agent.with_builtins(
    llm=llm,
    hooks=slack_mgr.as_hooks("prod-agent"),
)

result = agent.run("Analyze production logs for anomalies")
print(f"Cost: ${tracker.total_cost:.4f}")

from shipit_agent import Agent

slack = SlackNotifier(webhook_url="https://hooks.slack.com/...")
manager = NotificationManager([slack])

# Plain Agent — no built-in tools
agent = Agent(
    llm=llm,
    prompt="You analyze code quality.",
    hooks=manager.as_hooks("code-analyzer"),
)

result = agent.run("What are common Python anti-patterns?")
# Slack gets: "code-analyzer started" and "code-analyzer completed"

from shipit_agent.deep import DeepAgent

manager = NotificationManager([slack, discord])

deep = DeepAgent.with_builtins(
    llm=llm,
    verify=True,
    reflect=True,
    hooks=manager.as_hooks("deep-researcher"),
)

# Streaming with notifications — events are sent to Slack/Discord
# while you also get them locally
for event in deep.stream("Research AI safety frameworks"):
    print(f"[{event.type}] {event.message[:80]}")

from shipit_agent.deep.ship_crew import ShipCrew, ShipAgent, ShipTask

# Wire notifications into crew agents
researcher = ShipAgent(
    name="researcher",
    agent=Agent(llm=llm, prompt="You research.", hooks=manager.as_hooks("researcher")),
    role="Researcher",
)

writer = ShipAgent(
    name="writer",
    agent=Agent(llm=llm, prompt="You write.", hooks=manager.as_hooks("writer")),
    role="Writer",
)

crew = ShipCrew(
    name="notified-crew",
    coordinator_llm=llm,
    agents=[researcher, writer],
    tasks=[ShipTask(name="research", description="Research {topic}", agent="researcher", output_key="findings"),
        ShipTask(name="write", description="Write about {findings}", agent="writer", depends_on=["research"]),],
)

# Every agent in the crew sends notifications independently
result = crew.run(topic="microservices patterns")

from shipit_agent.notifications.base import Notification, Notifier

class PagerDutyNotifier:
    name: str = "pagerduty"

    async def send(self, notification: Notification) -> bool:
        # POST to PagerDuty Events API v2
        ...
        return True

    async def send_batch(self, notifications: list[Notification]) -> list[bool]:
        return [await self.send(n) for n in notifications]