interven 0.4.0

Interven AI firewall — Python SDK. Scan agent tool calls before they execute. Block malicious requests, redact PII/secrets, route risky actions to human approval.

interven

Python SDK for the Interven AI firewall. Scan agent tool calls before they execute — block malicious requests, redact PII and secrets, and route risky actions to human approval.

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://slack.com/api/chat.postMessage",
    body={"text": "Customer SSN 478-23-9156, email [e-mail]"},
)

if result.decision == "ALLOW":
    send_to_slack(original_body)
elif result.decision == "SANITIZE":
    send_to_slack(result.sanitized_body)        # PII redacted
elif result.decision == "REQUIRE_APPROVAL":
    poll_approval(result.approval_id)
else:
    log_blocked(result.reason_codes)

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

Decisions

Decision	What to do	Helper
ALLOW	Forward the original request	result.allowed
DENY	Block the call. reason_codes explain why.	result.blocked
SANITIZE	Forward result.sanitized_body instead of the original — secrets/PII redacted	result.needs_sanitization
REQUIRE_APPROVAL	Pause; poll /approvals/{id}/status until decided	result.needs_approval

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 recipes

LangChain — callback handler

from langchain_core.callbacks import BaseCallbackHandler
from interven import Client

interven = Client(runtime_type="langchain")

class IntervenCallback(BaseCallbackHandler):
    def on_tool_start(self, serialized, input_str, **kwargs):
        url = serialized.get("kwargs", {}).get("url")
        if not url:
            return
        result = interven.scan(method="GET", url=url)
        if result.decision == "DENY":
            raise RuntimeError(f"Blocked by Interven: {result.reason_codes}")

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
from your_mcp_server import MCPServer

interven = Client(runtime_type="mcp")
server = MCPServer()

@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

Wrap any outbound HTTP call. Works with AutoGen, OpenAI Assistants, custom agents.

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}")
    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
)

try:
    client.scan(method="POST", url="...", body={...})
except AuthenticationError:
    rotate_key()
except PayloadTooLargeError:
    chunk_payload()
except GatewayError as e:
    log_and_fail_open(e)

Legacy: HMAC AifClient

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

from interven import AifClient, InvokeParams

client = AifClient(
    gateway_url="http://localhost:4000",
    agent_id="00000000-0000-0000-0000-000000000010",
    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"],
))

The HMAC path supports approval polling (client.poll_approval, client.wait_for_approval, client.execute_approval) and the full event envelope. See the HMAC docs for details.

License

MIT © Interven Security

Links

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