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 fires any threshold that was just crossed.

A threshold can watch any budget dimension, not just cost: cost, tokens_total, or duration. Each alert binds to one dimension so the backend evaluates the right axis. See Limits & budgets for where each dimension comes from.

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

The AgentKavach cloud aggregates your spend across every agent, process, and host, then delivers each alert channel itself — email, Slack, PagerDuty, and webhooks. You provide the target for each channel; there is no extra setup and nothing to deliver from your own side.

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 your Slack URL. Each message includes a plain-text text fallback so simpler Slack apps still render something useful.

Get a Slack 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 Slack URL it gives you.
4. Store it 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 on every threshold breach. Use HTTPS in production (plain HTTP is allowed only for local development). Provide a secret and, when you set a secret, signs each request so your receiver can verify it. See Verifying requests below.

The payload

A single JSON object, sent with Content-Type: application/json. The default body is below; override the shape with a template using any of the template variables.

{
  "event": "threshold.breach",
  "level": "WARNING",
  "agent": "research-bot",
  "budget_type": "cost",
  "pct": "80",
  "current": "$40.0600",
  "limit": "$50.0000",
  "remaining": "$9.9400",
  "threshold_pct": 0.8
}

threshold_pct is the configured trigger ratio; pct is the actual usage percentage at the time of the breach. The body carries no timestamp — use the X-AgentKavach-Timestamp header (below) for request freshness.

Signature headers

When a secret is set, every request carries two headers:

• X-AgentKavach-Signature: sha256=<hex> — HMAC-SHA256 of the raw request body, keyed with your secret.
• X-AgentKavach-Timestamp: <unix-seconds> — when we sent the request. Reject anything older than a few minutes to block replays.

Verifying requests

Compute the HMAC over the raw bytes you received — do not re-serialize the parsed JSON first, or the signature won't match (we sign compact, key-sorted JSON). Always compare in constant time, and reject stale timestamps. A complete receiver:

import hashlib
import hmac
import os
import time

from flask import Flask, abort, request

app = Flask(__name__)
WEBHOOK_SECRET = os.environ["AGENTKAVACH_WEBHOOK_SECRET"]
MAX_SKEW_SECONDS = 300  # reject requests older than 5 minutes


@app.post("/budget-alerts")
def budget_alerts():
    body = request.get_data()  # RAW bytes — sign over THESE, not the parsed JSON
    signature = request.headers.get("X-AgentKavach-Signature", "")
    timestamp = request.headers.get("X-AgentKavach-Timestamp", "")

    # 1. Replay protection: timestamp must be present and recent.
    if not timestamp.isdigit() or abs(time.time() - int(timestamp)) > MAX_SKEW_SECONDS:
        abort(401, "stale or missing timestamp")

    # 2. Verify the HMAC-SHA256 signature in constant time.
    expected = hmac.new(WEBHOOK_SECRET.encode(), body, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(f"sha256={expected}", signature):
        abort(401, "bad signature")

    # 3. Trusted — act on the alert.
    alert = request.get_json()
    print(f"{alert['agent']} hit {alert['pct']}% of its "
          f"{alert['budget_type']} budget ({alert['current']} of {alert['limit']})")
    return "", 200

Delivery #

The AgentKavach cloud both evaluates and delivers every alert. Evaluation runs centrally because only the cloud has your full spend across every process and host — a single SDK instance sees only its own traffic. Delivery then happens from the cloud too: it makes the HTTP call to each channel's target — email through AgentKavach's mail provider, Slack to your incoming-webhook URL, PagerDuty to your routing key, and webhooks to your URL.

You provide the target for each channel and the cloud handles the rest. Every alert is evaluated, recorded for your dashboard history, and delivered exactly once.

channels:
  ops_slack:
    type: slack
    webhook_url: https://hooks.slack.com/services/T000/B000/xxxx
  budget_hook:
    type: webhook
    url: https://hooks.example.com/budget-alerts
    secret: ${ALERT_WEBHOOK_SECRET}

Webhooks sign their requests when you set a secret — verify the X-AgentKavach-Signature header on your receiver; see Verifying requests above.

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(reason):
    # reason tells you WHY: "cost" for a budget kill, "tier_agent_limit"
    # etc. for a backend stop — see SDK reference › on_kill kill reasons.
    print(f"agent killed: {reason}")
    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),
    ],
)

The reason parameter is optional — a zero-argument callback (def emergency_stop():) keeps working unchanged. Declare the parameter to receive the kill reason; the full vocabulary is in the SDK reference.

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 a stop exception. It runs the same callback as a budget kill, but the callback's reason parameter and the exception type (IngestRejectedError for backend stops, a BudgetExceededError subclass) tell the two apart.