Metadata-Version: 2.4
Name: interven
Version: 0.5.3
Summary: Interven — Approvals + Audit for AI Agents. Scan tool calls through policy + risk scoring. Block, redact PII, or route to human approval via Slack.
Author-email: Interven Security <support@intervensecurity.com>
License: MIT
Project-URL: Homepage, https://intervensecurity.com
Project-URL: Documentation, https://intervensecurity.com/docs
Project-URL: Repository, https://github.com/intervensecurity/interven-python
Project-URL: Issues, https://github.com/intervensecurity/interven-python/issues
Project-URL: Changelog, https://github.com/intervensecurity/interven-python/blob/main/CHANGELOG.md
Keywords: interven,ai-agent,ai-firewall,security,guardrails,prompt-injection,tool-call,policy,langchain,crewai,openclaw,mcp
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Intended Audience :: Information Technology
Classifier: Topic :: Security
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Requires-Python: >=3.9
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: requests>=2.28.0
Provides-Extra: dev
Requires-Dist: pytest>=7.0; extra == "dev"
Requires-Dist: responses>=0.23; extra == "dev"
Dynamic: license-file

# interven

Python SDK for [Interven](https://intervensecurity.com) — **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.

```bash
pip install interven
```

## Quickstart

```python
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](https://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

```python
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

```python
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](https://github.com/boltyx0/langchain-reference)
for full examples including the Slack approval demo.

### CrewAI — step callback

```python
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

```python
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)

```python
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

```python
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.

```python
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](./LICENSE)

## Links

- [Interven docs](https://docs.intervensecurity.com)
- [API reference](https://docs.intervensecurity.com/docs/api)
- [LangChain reference repo](https://github.com/boltyx0/langchain-reference)
- [Get an API key](https://app.intervensecurity.com/signup)
- [GitHub](https://github.com/boltyx0/interven-mcp-guard)
