Alerts & Channels

An alert is a (threshold, channel) pair. The SDK fires the channel when an agent crosses the threshold. Four notification channels ship in the box: email, Slack, PagerDuty, and webhook. A separate kill switch can halt the agent outright, covered at the end.

How alerts work #

A threshold is a fraction in (0.0, 1.0], so 0.70 fires when usage reaches 70% of the budget. After every LLM call the engine recomputes the usage ratio and dispatches any threshold that was just crossed.

A threshold does not have to track cost. The engine measures three dimensions, and each alert is bound to one of them so the backend evaluates the right axis. Cost comes from your Budget (daily, monthly, or total), total token usage comes from max_tokens_per_run, and elapsed runtime comes from max_runtime_seconds.

Because a channel is bound to a single dimension, a channel watching cost will not fire on token usage. To alert on both, register the channel twice, once per dimension. To keep a busy agent from flooding you, each (budget, threshold, channel) combination fires at most once every five minutes within a budget period.

ChannelType #

from agentkavach import AgentKavach, ChannelType

Configuring channels #

from agentkavach import AgentKavach, Budget, ChannelType

guard = AgentKavach(
    provider="openai",
    api_key="ak_prod_...",     # AgentKavach key
    llm_key="sk-...",          # OpenAI key
    agent_name="research-bot",
    budget=Budget.daily(50),
    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..."),
    ],
)

Build a list of ChannelConfig objects with AgentKavach.channel() and pass them to the channels= argument.

Email #

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

You provide only the recipient address. The AgentKavach backend sends the message through Resend using its own key, so there is nothing else to configure and no email credential of your own to manage.

Slack #

The channel posts a Block Kit message to a Slack incoming webhook. Each message includes a plain-text text fallback so legacy webhooks render something useful.

Get a webhook URL #

1. Open api.slack.com/apps and click Create New App, then 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 #

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

The channel triggers an incident through the PagerDuty Events API v2. Severity maps from the threshold: warning below 90%, error from 90% to 99%, critical at 100%.

Create the routing key #

1. In PagerDuty, open Services, Service Directory, then 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[...].

Webhook #

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

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

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)

Template variables #

By default each channel formats its own message. When you want different wording, pass a template to the channel (as shown in the Slack example above) and reference any of the variables below with Python str.format() syntax, for example {agent_name} hit {pct}% of its {period} budget. Unknown names are left untouched, so a typo never crashes an alert.

Credentials #

AgentKavach never reads channel credentials from the environment. Give each channel its credential explicitly when you build the client, or define it in your YAML config. Reading from os.environ in your own code is encouraged, for example webhook_url=os.environ["AGENTKAVACH_SLACK_WEBHOOK_URL"]; the SDK simply will not do it for you.

Kill switch #

The kill switch is not a notification channel. The four channels above tell someone that spend is climbing; the kill switch acts on it by stopping the agent. You arm it by giving the client an on_kill callback and a kill threshold, usually 1.0. When usage reaches that point the SDK invokes your callback and then refuses to run again: every later guard.create() on that instance raises BudgetExceededError immediately.

import sys
from agentkavach import AgentKavach, Budget, ChannelType

def emergency_stop():
    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 Kill button

The Kill button on the Agents page records a kill on the backend. The SDK picks it up on its next config sync, and on the agent's next guard.create() it invokes the agent's on_kill callback and then raises BudgetExceededError. This is the same stop signal as a budget kill, which is why it runs the same callback and surfaces the same exception.