interven 0.5.3

Interven — Approvals + Audit for AI Agents. Scan tool calls through policy + risk scoring. Block, redact PII, or route to human approval via Slack.

interven

Python SDK for Interven — Approvals + Audit for AI Agents. Add human-in-the-loop approval workflows, policy enforcement, and compliance audit trails to any AI agent in 5 lines of code.

pip install interven

Quickstart

from interven import Client

client = Client(api_key="iv_live_...")  # or set INTERVEN_API_KEY env

result = client.scan(
    method="POST",
    url="https://api.bank.com/v1/wire-transfer",
    body={"amount": 25000, "to": "ACME Logistics", "memo": "Q2 invoice"},
)

if result.decision == "ALLOW":
    execute_transfer(original_body)
elif result.decision == "SANITIZE":
    execute_transfer(result.sanitized_body)     # PII redacted
elif result.decision == "REQUIRE_APPROVAL":
    # Agent pauses — Slack notification sent — human approves
    status = client.wait_for_approval(result.approval_id)
    if status.is_approved:
        execute_transfer(original_body)         # approved, proceed
else:
    log_blocked(result.reason_codes)            # DENY

Get an API key at intervensecurity.com (free tier: 1,000 scans/month).

What Interven does

Interven sits between your AI agent and the tools/APIs it calls. Every outbound action is scanned through policy + risk scoring before it executes:

Decision	What to do	Property
ALLOW	Forward the original request	result.allowed
DENY	Block. reason_codes explain why.	result.blocked
SANITIZE	Forward result.sanitized_body — PII/secrets redacted	result.needs_sanitization
REQUIRE_APPROVAL	Pause. Human approves via Slack or Console. Agent resumes.	result.needs_approval

REQUIRE_APPROVAL is Interven's unique capability — no other tool supports end-to-end human-in-the-loop with Slack buttons and automatic agent resumption.

Approval workflow

result = client.scan(method="POST", url="...", body={...})

if result.needs_approval:
    print(f"Waiting for approval: {result.approval_id}")

    # Blocks until analyst clicks Approve in Slack or Console
    status = client.wait_for_approval(
        result.approval_id,
        poll_interval=5.0,    # check every 5s
        max_wait=600.0,       # timeout after 10 min
    )

    if status.is_approved:
        print("Approved! Proceeding...")
    elif status.status == "denied":
        print("Denied by analyst.")
    elif status.status == "expired":
        print("Approval expired.")

Configuration

Argument	Env var	Default
api_key	INTERVEN_API_KEY	— (required)
gateway_url	INTERVEN_GATEWAY_URL	https://api.intervensecurity.com
timeout	—	30.0
agent_id	—	unset (server uses default)
runtime_type	—	"python"

Framework integrations

LangChain / LangGraph — callback handler

from interven_langchain import InterventCallback

agent = create_react_agent(model, tools=[...])
agent.invoke(
    {"messages": [HumanMessage("...")]},
    config={"callbacks": [InterventCallback(api_key="iv_live_...")]},
)

Three lines. Every tool call scanned. DENY/REQUIRE_APPROVAL handled automatically.

See the LangChain reference repo for full examples including the Slack approval demo.

CrewAI — step callback

from interven import Client
from crewai import Agent

interven = Client(runtime_type="crewai")

def step_guard(step):
    for call in step.tool_calls:
        result = interven.scan(
            method="POST",
            url=call.tool_url,
            body=call.payload,
        )
        if result.blocked:
            raise RuntimeError(f"Interven blocked: {result.reason_codes}")

agent = Agent(role="...", goal="...", step_callback=step_guard)

MCP server — middleware

from interven import Client

interven = Client(runtime_type="mcp")

@server.tool_middleware
async def scan_before_call(tool_name, params, next_handler):
    result = interven.scan(
        method="POST",
        url=f"mcp://{tool_name}",
        body=params,
    )
    if result.blocked:
        raise RuntimeError(f"Blocked: {result.reason_codes}")
    return await next_handler(tool_name, params)

Generic agent (AutoGen, OpenAI Assistants, custom)

import requests
from interven import Client

interven = Client()

def safe_post(url, json=None):
    r = interven.scan(method="POST", url=url, body=json or {})
    if r.blocked:
        raise RuntimeError(f"Blocked: {r.reason_codes}")
    if r.needs_approval:
        status = interven.wait_for_approval(r.approval_id)
        if not status.is_approved:
            raise RuntimeError("Approval denied or expired")
    body = r.sanitized_body if r.needs_sanitization else json
    return requests.post(url, json=body)

Errors

from interven import (
    AuthenticationError,      # bad / revoked API key
    GatewayError,             # network or 5xx
    PayloadTooLargeError,     # >256KB body
    ApprovalDeniedError,      # analyst denied
    ApprovalExpiredError,     # approval timed out
    ApprovalTimeoutError,     # poll max_wait exceeded
)

Legacy: HMAC AifClient

The original HMAC-signed /invoke flow is still supported for existing deployments. New integrations should prefer Client — fewer fields, no shared secret.

from interven import AifClient, InvokeParams

client = AifClient(
    gateway_url="http://localhost:4000",
    agent_id="...",
    agent_name="release-bot",
    agent_secret="...",
)

result = client.invoke(InvokeParams(
    tool_name="github",
    method="PUT",
    url_path="/repos/acme/main-app/collaborators/external-user",
    credential_type="pat",
    credential_token="ghp_...",
    scopes=["repo"],
))

License

MIT

Links

• Interven docs
• API reference
• LangChain reference repo
• Get an API key
• GitHub