Alerts & Channels

from agentkavach.alerts import AlertRule, ChannelType

from agentkavach.alerts import AlertRule, ChannelType

# Basic email alert at 80% spend
email_alert = AlertRule(
    channel_type=ChannelType.EMAIL,
    threshold=0.80,
    to="platform-team@example.com",
)

# Email with custom template
email_custom = AlertRule(
    channel_type=ChannelType.EMAIL,
    threshold=0.90,
    to="oncall@example.com,cto@example.com",
    template="""
    <h2>AgentKavach Alert: {agent_name}</h2>
    <p>Agent <strong>{agent_name}</strong> has used {pct}% of its {period} budget.</p>
    <p>Spent: ${spent} / ${budget} (${remaining} remaining)</p>
    <p>Severity: {severity}</p>
    <p><a href="{dashboard_url}">View Dashboard</a></p>
    """,
)

from agentkavach.alerts import AlertRule, ChannelType

slack_alert = AlertRule(
    channel_type=ChannelType.SLACK,
    threshold=0.70,
    webhook_url="https://hooks.slack.com/services/T00000000/B00000000/XXXXXXXXXXXXXXXXXXXXXXXX",
)

from agentkavach.alerts import AlertRule, ChannelType

pagerduty_alert = AlertRule(
    channel_type=ChannelType.PAGERDUTY,
    threshold=0.95,
    routing_key="e93facc04764012d7bfb002500d5d1a6",
)

from agentkavach.alerts import AlertRule, ChannelType

webhook_alert = AlertRule(
    channel_type=ChannelType.WEBHOOK,
    threshold=0.80,
    url="https://api.example.com/agentkavach-alerts",
    secret="whsec_your_signing_secret_here",
)

{
  "event": "budget.threshold_crossed",
  "agent_name": "research-bot",
  "threshold": 0.80,
  "pct": 82.5,
  "spent": 4.13,
  "budget": 5.00,
  "remaining": 0.87,
  "period": "daily",
  "severity": "warning",
  "timestamp": "2025-01-15T14:30:00Z"
}

import hmac
import hashlib

def verify_signature(body: bytes, signature: str, secret: str) -> bool:
    expected = hmac.new(
        secret.encode(),
        body,
        hashlib.sha256,
    ).hexdigest()
    return hmac.compare_digest(f"sha256={expected}", signature)

from agentkavach import AgentKavach, Budget
from agentkavach.alerts import AlertRule, ChannelType

def emergency_stop(agent_name: str, spent: float, budget: float):
    """Called when the kill switch fires."""
    print(f"KILLED: {agent_name} spent {spent:.2f} of {budget:.2f}")
    # Clean up resources, save state, notify team, etc.

guard = AgentKavach(
    agent_name="expensive-agent",
    api_key="cg_...",
    budget=Budget(daily=10.00),
    channels=[
        AlertRule(channel_type=ChannelType.KILL, threshold=1.0),
    ],
    on_kill=emergency_stop,
)

def emergency_stop(agent_name: str, spent: float, budget: float):
    """Called when the kill switch fires via the dashboard Stop button."""
    print(f"KILLED: {agent_name} spent {spent:.2f} of {budget:.2f}")
    # Save state, release resources, notify team
    agent.save_checkpoint("killed_by_dashboard")
    sys.exit(1)

guard = AgentKavach(
    agent_name="my-agent",
    api_key="cg_...",
    budget=Budget.daily(50),
    on_kill=emergency_stop,  # Required for Stop button to halt the process
)

from agentkavach import AgentKavach, Budget
from agentkavach.alerts import AlertRule, ChannelType

def on_kill(agent_name, spent, budget):
    print(f"Agent {agent_name} killed. Spent {spent:.2f}/{budget:.2f}")

guard = AgentKavach(
    agent_name="production-agent",
    api_key="cg_...",
    budget=Budget(daily=50.00, monthly=1000.00),
    channels=[
        # 50% — early warning via email
        AlertRule(
            channel_type=ChannelType.EMAIL,
            threshold=0.50,
            to="platform-team@example.com",
        ),

        # 70% — escalate to Slack
        AlertRule(
            channel_type=ChannelType.SLACK,
            threshold=0.70,
            webhook_url="https://hooks.slack.com/services/T.../B.../xxx",
        ),

        # 80% — dual alert: email + Slack
        AlertRule(
            channel_type=ChannelType.EMAIL,
            threshold=0.80,
            to="oncall@example.com",
        ),
        AlertRule(
            channel_type=ChannelType.SLACK,
            threshold=0.80,
            webhook_url="https://hooks.slack.com/services/T.../B.../xxx",
        ),

        # 95% — page the on-call engineer
        AlertRule(
            channel_type=ChannelType.PAGERDUTY,
            threshold=0.95,
            routing_key="e93facc04764012d7bfb002500d5d1a6",
        ),

        # 100% — kill the agent
        AlertRule(
            channel_type=ChannelType.KILL,
            threshold=1.0,
        ),
    ],
    on_kill=on_kill,
)