Guardrails

Guardrails are safety limits that prevent runaway agents from consuming excessive resources. They complement budgets by enforcing token counts, call counts, runtime duration, and detecting infinite loops. When a guardrail fires, the agent is stopped immediately with a specific exception.

ℹ️ Fail-open design

Guardrails only raise their specific errors. All internal SDK errors are caught and swallowed so your agent keeps running. Only guardrail violations and budget violations propagate.

Token Limits #

Limit the total number of tokens (input + output) an agent can consume in a single run. The check happens after each LLM call (post-flight), meaning the call that crosses the threshold succeeds but the next call is blocked.

Parameter	Type	Required	Default	Description
`max_tokens_per_run`	`int`	No	None	Maximum total tokens (input + output) allowed per run. None disables the limit.

⚠️ Post-flight check

The LLM call that crosses the token threshold will succeed and you will be billed for it. The limit is enforced on the next call. This ensures you always get a complete response rather than a partial one.

python

from agentkavach import AgentKavach
from agentkavach.exceptions import TokenLimitError

guard = AgentKavach(
    agent_name="summarizer",
    api_key="ak_prod_...",
    max_tokens_per_run=10_000,  # stop after ~10k tokens
)

try:
    for chunk in work_items:
        response = guard.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": chunk}],
        )
        print(response.choices[0].message.content)
except TokenLimitError as e:
    print(f"Token limit reached: {e}")
    # gracefully wrap up the agent run

Call Count Limits #

Limit the number of LLM calls an agent can make in a single run. The check happens before each call (pre-flight), so you never pay for the blocked call.

Parameter	Type	Required	Default	Description
`max_calls_per_run`	`int`	No	None	Maximum number of LLM calls allowed per run. None disables the limit.

ℹ️ Pre-flight check

Unlike token limits, call count limits are checked before the call is made. The excess call is blocked and you are never billed for it.

python

from agentkavach import AgentKavach
from agentkavach.exceptions import CallLimitError

guard = AgentKavach(
    agent_name="researcher",
    api_key="ak_prod_...",
    max_calls_per_run=50,  # max 50 LLM calls per run
)

try:
    while has_more_questions():
        response = guard.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": next_question()}],
        )
        process(response)
except CallLimitError as e:
    print(f"Call limit reached after {e.call_count} calls")
    # save partial results and exit gracefully

Runtime Limits #

Limit how long an agent can run. The clock starts on the first LLM call and is checked before each subsequent call. If the elapsed time exceeds the limit, a RuntimeLimitError is raised.

Parameter	Type	Required	Default	Description
`max_runtime_seconds`	`float`	No	None	Maximum wall-clock seconds from the first LLM call. None disables the limit.

python

from agentkavach import AgentKavach
from agentkavach.exceptions import RuntimeLimitError

guard = AgentKavach(
    agent_name="deep-researcher",
    api_key="ak_prod_...",
    max_runtime_seconds=300.0,  # 5-minute timeout
)

try:
    while not done:
        response = guard.create(
            model="gpt-4o",
            messages=build_messages(),
        )
        done = evaluate(response)
except RuntimeLimitError as e:
    print(f"Runtime limit exceeded: ran for {e.elapsed:.1f}s")
    # save progress for later resumption

Loop Detection #

Detects when an agent gets stuck in a repeating pattern of calls. The detector tracks (model, tool_name) pairs across the last 20 calls and looks for repeating sequences of length 2 to 5. When a sequence repeats more than loop_threshold times, a LoopDetectedError is raised.

Parameter	Type	Required	Default	Description
`detect_loops`	`bool`	No	False	Enable loop detection. When False, no loop checking is performed.
`loop_threshold`	`int`	No	3	Number of times a sequence must repeat before triggering. Minimum value is 2.

🚨 Infinite loops can be expensive

Without loop detection, an agent stuck calling the same tool in a cycle can burn through your entire budget in seconds. Enable this guardrail for any agent that uses tool calling.

python

from agentkavach import AgentKavach
from agentkavach.exceptions import LoopDetectedError

guard = AgentKavach(
    agent_name="tool-agent",
    api_key="ak_prod_...",
    detect_loops=True,
    loop_threshold=3,  # fire after 3 repetitions of same pattern
)

try:
    while not done:
        response = guard.create(
            model="gpt-4o",
            messages=messages,
            tools=tool_definitions,
        )
        # process tool calls...
except LoopDetectedError as e:
    print(f"Loop detected: {e.pattern} repeated {e.count} times")
    # break the loop — maybe inject a different prompt

Exception Hierarchy #

All guardrail exceptions inherit from GuardrailError, which is separate from BudgetExceededError. This lets you catch budget and guardrail violations independently.

text

Exception
├── BudgetExceededError          # budget limits (daily/monthly/total)
└── GuardrailError               # all guardrail violations
    ├── TokenLimitError          # max_tokens_per_run exceeded
    ├── CallLimitError           # max_calls_per_run exceeded
    ├── RuntimeLimitError        # max_runtime_seconds exceeded
    └── LoopDetectedError        # repeating call pattern detected

python

from agentkavach.exceptions import (
    BudgetExceededError,
    GuardrailError,
    TokenLimitError,
    CallLimitError,
    RuntimeLimitError,
    LoopDetectedError,
)

try:
    response = guard.create(model="gpt-4o", messages=msgs)
except BudgetExceededError:
    print("Budget limit hit — agent stopped")
except GuardrailError as e:
    print(f"Guardrail triggered: {type(e).__name__}: {e}")

You can also catch specific guardrail types for fine-grained control:

python

try:
    response = guard.create(model="gpt-4o", messages=msgs)
except TokenLimitError:
    save_partial_results()
except CallLimitError:
    log_and_summarize()
except RuntimeLimitError:
    checkpoint_state()
except LoopDetectedError:
    inject_new_prompt()
except BudgetExceededError:
    alert_team()

Combining Guardrails #

All guardrails can be used together alongside budgets and alert channels. Each guardrail is evaluated independently — the first one to trigger stops the agent.

python

from agentkavach import AgentKavach, Budget, ChannelType

guard = AgentKavach(
    provider="openai",
    agent_name="production-agent",
    api_key="ak_prod_...",
    llm_key="sk-...",

    # Cost budget
    budget=Budget.daily(5.00),

    # Guardrails
    max_tokens_per_run=50_000,
    max_calls_per_run=200,
    max_runtime_seconds=600.0,  # 10 minutes
    detect_loops=True,
    loop_threshold=3,

    # Alert channels
    channels=[
        AgentKavach.channel(ChannelType.SLACK, threshold=0.70,
            webhook_url="https://hooks.slack.com/services/T.../B.../xxx"),
        AgentKavach.channel(ChannelType.EMAIL, threshold=0.80,
            to="team@example.com"),
        AgentKavach.channel(ChannelType.KILL, threshold=1.0),
    ],
    on_kill=lambda: print("agent killed"),
)

response = guard.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Analyze this dataset..."}],
)

ℹ️ Evaluation order

Pre-flight checks run before the LLM call: call count, runtime, and budget. Post-flight runs after: token count, loop detection, and budget update. The first guardrail or budget violation raises the corresponding exception and stops the run.

Fail on Error

By default, AgentKavach uses a fail-open design: if the SDK encounters an internal error (e.g., telemetry export failure, engine error), the LLM call still proceeds. Only budget and guardrail violations propagate.

Set fail_on_error=True for strict mode: any internal error will call on_kill (if configured) and raise the exception, stopping the agent immediately.

Parameter	Type	Required	Default	Description
`fail_on_error`	`bool`	No	False	When True, any internal SDK error (pre-flight, post-flight, export) will kill the agent and raise. When False (default), errors are logged and the call proceeds.

python

from agentkavach import AgentKavach, Budget

def stop_agent():
    print("Agent killed due to error!")
    # Cleanup, send notification, etc.

guard = AgentKavach(
    provider="openai",
    llm_key="sk-...",
    agent_name="strict-bot",
    budget=Budget.daily(50),
    on_kill=stop_agent,
    fail_on_error=True,  # Strict mode: errors kill the agent
)

# If the SDK encounters any internal error during this call,
# on_kill() will be called and the error will be raised.
response = guard.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello"}],
)

Save Prompts

By default, AgentKavach does not store the prompt text sent to LLM providers. Enable save_prompts=True to include prompt content in telemetry events. This is useful for debugging and auditing but increases storage usage.

Parameter	Type	Required	Default	Description
`save_prompts`	`bool`	No	False	When True, the prompt text is included in telemetry events and visible in the dashboard Event History. When False (default), prompts are not stored.

python

guard = AgentKavach(
    provider="openai",
    llm_key="sk-...",
    agent_name="audit-bot",
    budget=Budget.daily(50),
    save_prompts=True,  # Store prompt text in events
)

# Prompts will now appear in the dashboard Event History
response = guard.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Analyze this report..."}],
)

⚠️ Privacy considerations

Enabling save_prompts means prompt text is transmitted to the AgentKavach backend and stored in the database. Ensure this complies with your data handling policies. Do not enable this if prompts contain sensitive PII or credentials.

⚠️ PRIVACY MODE — opt-in required at every layer

save_prompts=True on the SDK is necessary but not sufficient. Four independent gates guard prompt storage:

SDK opt-in (default off): set save_prompts=True when constructing AgentKavach(...). The SDK emits a one-time warning so operators see the opt-in.
Org opt-in (default off): your organisation owner must flip allow_prompt_storage via PATCH /v1/settings. Until they do, the ingest endpoint silently strips event.prompt to NULL before writing — even if the SDK sent it.
2 kB length cap: stored prompts are truncated to 2048 characters with a ... [truncated] suffix. Enforced both client-side and server-side so direct API users can't bypass.
30-day retention: stored prompts are NULL'd from events.prompt automatically after 30 days. Event metadata (cost, tokens, model) is retained per your tier's analytics window.

Existing data is not retroactively scrubbed when you flip the org flag back to False — it ages out naturally via the 30-day retention sweep.