langchain-kavachos

KavachOS DAN gate callback for LangChain agents.

Intercepts every tool call through the AEGIS KAVACH gate before execution. Zero agent code changes — add the callback and every tool invocation is governed.

Install

pip install langchain-kavachos

Quick start

from langchain_kavachos import KavachGateCallback

callback = KavachGateCallback(
    base_url="http://localhost:4850",   # AEGIS server
    on_block="raise",                   # raise KavachGateError on DAN-3/4
    dry_run=False,
)

# LangChain agent — pass callback in config
result = agent.invoke(
    {"input": "summarise the quarterly report"},
    config={"callbacks": [callback]},
)

# Or attach to a single tool:
result = my_tool.invoke("drop table users", config={"callbacks": [callback]})

KavachGateCallback parameters

Direct client

from langchain_kavachos import AegisClient

client = AegisClient(base_url="http://localhost:4850")

# Pre-flight budget check
state = client.state()
if state["budget"]["breached"]:
    raise RuntimeError("Daily budget breached — halt")

# Manual gate call
result = client.gate(command="rm -rf /var/postgres", tool_name="my-agent")
print(result)  # {"allow": false, "level": 4, "reason": "DAN-4 catastrophic..."}

# Audit query
records = client.audit(session_id="lc-abc123", status="stop", limit=20)

How it works

KavachGateCallback.on_tool_start() fires before any tool execution. It POSTs to POST /api/v1/kavach/gate on the AEGIS server.

• DAN-1/2: allowed immediately, logged.
• DAN-3: notify approver via Telegram/WhatsApp, wait for ALLOW/STOP.
• DAN-4: blocked immediately, KavachGateError raised.

All policy is in AEGIS — the callback is a thin HTTP relay.

AEGIS server

Run with: bun /root/aegis/src/dashboard/server.ts
Default port: 4850
Gate endpoint: POST /api/v1/kavach/gate

License

AGPL-3.0 — see LICENSE.