Alerts & Channels

Alerts fire when an agent crosses a configured threshold. Each alert is a (threshold, channel) pair. The SDK ships five channel types: email, Slack, PagerDuty, webhook, and a built-in kill switch.

How Alerts Work #

The threshold is a float in (0.0, 1.0]; 0.70 means “fire when usage hits 70% of the budget”. After every LLM call, the engine compares the new usage ratio to every configured threshold and dispatches alerts that just crossed their line.

Evaluable budget types

A threshold can apply to any of the dimensions the engine tracks. The SDK syncs the budget type with each alert so the backend evaluates the right axis.

ChannelType #

from agentkavach import AgentKavach, ChannelType

Configuring channels #

Build a list of ChannelConfig objects with the AgentKavach.channel() factory and pass it via the channels= argument:

from agentkavach import AgentKavach, Budget, ChannelType

guard = AgentKavach(
    provider="openai",
    api_key="ak_prod_...",     # your AgentKavach key
    llm_key="sk-...",          # your OpenAI key
    agent_name="research-bot",
    budget=Budget.daily(50),
    on_kill=lambda: None,
    channels=[
        AgentKavach.channel(ChannelType.EMAIL,     threshold=0.50, to="team@acme.com"),
        AgentKavach.channel(ChannelType.SLACK,     threshold=0.80, webhook_url="https://hooks.slack.com/..."),
        AgentKavach.channel(ChannelType.PAGERDUTY, threshold=0.95, routing_key="R0..."),
        AgentKavach.channel(ChannelType.KILL,      threshold=1.0),
    ],
)

Email #

Email is delivered through Resend. Two modes:

• Backend-dispatched (default): pass only to=. The hosted backend sends the message using its own Resend key. No SDK-side key required.
• SDK-dispatched: pass api_key= (or set RESEND_API_KEY in the environment). The SDK sends the message directly.

AgentKavach.channel(ChannelType.EMAIL, threshold=0.80, to="oncall@acme.com")

Slack #

Posts a Block Kit message to a Slack incoming webhook. Every message also includes a plain-text text fallback so legacy webhooks still render something useful.

Get a webhook URL #

1. Open api.slack.com/apps and click Create New App → From scratch.
2. Pick the workspace, then enable Incoming Webhooks in the left sidebar.
3. Click Add New Webhook to Workspace, choose a channel, and copy the URL.
4. Store the URL in .env as AGENTKAVACH_SLACK_WEBHOOK_URL and pass it via webhook_url=os.environ["AGENTKAVACH_SLACK_WEBHOOK_URL"].

Custom template

AgentKavach.channel(
    ChannelType.SLACK,
    threshold=0.70,
    webhook_url=os.environ["AGENTKAVACH_SLACK_WEBHOOK_URL"],
    template={
        "text": ":rotating_light: {agent_name} at {pct}% of {period} {budget_type} limit",
    },
)

PagerDuty #

Triggers an incident through the PagerDuty Events API v2. Severity is auto-mapped from the threshold: warning below 90%, error from 90% to 99%, critical at 100%.

Create the routing key #

1. In PagerDuty, go to Services → Service Directory → New Service.
2. Pick an escalation policy, then add an Events API v2 integration.
3. Copy the integration key from the integration row.
4. Store it as AGENTKAVACH_PAGERDUTY_ROUTING_KEY and pass it via routing_key=os.environ[...].

AgentKavach.channel(
    ChannelType.PAGERDUTY,
    threshold=0.95,
    routing_key=os.environ["AGENTKAVACH_PAGERDUTY_ROUTING_KEY"],
)

Webhook #

POSTs a JSON payload to any URL. Provide a secret to receive an HMAC-SHA256 signature in the X-AgentKavach-Signature header.

AgentKavach.channel(
    ChannelType.WEBHOOK,
    threshold=0.80,
    url="https://api.acme.com/budget-alerts",
    secret=os.environ["AGENTKAVACH_WEBHOOK_SECRET"],
)

Verifying the signature

import hmac
import hashlib

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

Kill switch #

The kill channel invokes the on_kill callback you registered on the AgentKavach instance, then sets an internal flag that causes every subsequent guard.create() call to raise BudgetExceededError immediately.

import sys
from agentkavach import AgentKavach, Budget, ChannelType

def emergency_stop():
    # Called once when the kill threshold is crossed.
    print("agent killed by budget")
    sys.exit(1)

guard = AgentKavach(
    provider="openai",
    api_key="ak_prod_...",
    llm_key="sk-...",
    agent_name="expensive-agent",
    budget=Budget.daily(10),
    on_kill=emergency_stop,
    channels=[
        AgentKavach.channel(ChannelType.KILL, threshold=1.0),
    ],
)

Dashboard Stop button #

The Stop button on the Agents page records a kill event on the backend. The next time the SDK syncs config (on instance construction), it picks up the kill state. If your process is already running, the SDK enforces the kill on its next post-flight cycle and future guard.create() calls raise BudgetExceededError.

Template variables #

Templates use Python str.format() substitution. Available variables:

Environment variables #

When you construct AgentKavach without an explicit channels= list, the SDK reads these environment variables and registers any channel whose credentials are present: