antarraksha-langchain

Antarraksha AI Agent Enforcement SDK for LangChain.

Installation

pip install antarraksha-langchain

Quick Start

Attach Antarraksha as a callback handler on your LLM. Registration happens automatically on first use — no login, no API key, no signup required.

from antarraksha_langchain import AntarrakshaCallbackHandler
from langchain_anthropic import ChatAnthropic

handler = AntarrakshaCallbackHandler(agent_id="my-langchain-agent")

llm = ChatAnthropic(model="claude-sonnet-4-5", callbacks=[handler])
print(llm.invoke("Hello").content)

Run your agent normally — every LLM call, tool call, and chain call is now enforced against Antarraksha policy.

Tool Wrapping (optional)

Wrap individual LangChain tools for inline enforcement:

from langchain_community.tools import ShellTool
from antarraksha_langchain import AntarrakshaSafeTool, AntarrakshaClient

client = AntarrakshaClient(agent_id="my-langchain-agent")
client.register()

safe_shell = AntarrakshaSafeTool(wrapped_tool=ShellTool(), antarraksha_client=client)
safe_shell.run("ls")

Parameters

Enforcement Behavior

• on_llm_start fires before every LLM call. DENY raises PermissionError.
• on_tool_start fires before every tool invocation. DENY raises PermissionError.
• on_chain_start fires for every chain run (informational; no enforcement halt).
• Every call is logged and visible at https://antarraksha.ai/registry.

Human-in-the-loop Escalation (long-poll + auto-abandon)

When Antarraksha holds a tool call for 4-eyes review, the SDK exposes a long-poll helper that blocks until the operator decides — and automatically gives up (fires POST /sdk/escalation/:id/abandon) the moment the caller cancels. This collapses the 60-second server-side silence-sweep window to roughly one second, so the operator queue drains immediately when the SDK side walks away.

import threading
from antarraksha_langchain import AntarrakshaClient

client = AntarrakshaClient(agent_id="my-agent")
client.register()

cancel = threading.Event()
result = client.wait_escalation(
    "esc-123",
    timeout_ms=30_000,
    cancel_event=cancel,   # set this from your AbortController / Ctrl-C handler
)
print(result["status"], result.get("finalDecision"))

wait_escalation auto-fires POST /sdk/escalation/:id/abandon when any of these happen mid-wait:

• cancel_event.set() is called (your AbortController / request-timeout).
• The process receives SIGINT (Ctrl-C) or SIGTERM (only when called from the main thread; pass install_signal_handlers=False to opt out).
• An exception propagates out of the wait (KeyboardInterrupt, network error, anything) — abandon is fired in a finally before re-raising.
• The Python interpreter exits while the wait is still in flight (an atexit hook fires the abandon as a last-chance signal).

The helper is idempotent against the server (the row CAS-flips PENDING → ABANDONED + final_decision=BLOCK exactly once); calling client.abandon_escalation("esc-123") directly is safe at any time.

Corporate Network Note

If you're behind a corporate TLS-inspection proxy and see SSLCertVerificationError, install:

pip install pip-system-certs

This makes Python trust your Windows / macOS system certificate store.