interven-openai-assistants 0.1.0

OpenAI Assistants API integration for Interven AI firewall. Scan tool calls the assistant decides to make before your code executes them.

interven-openai-assistants

OpenAI Assistants API integration for Interven. Scan every tool call the assistant decides to make before your code dispatches it.

Install

pip install interven-openai-assistants

How it fits in the Assistants loop

OpenAI Assistants runs in a loop: create thread → run assistant → if the assistant wants to call tools, the run transitions to requires_action → you dispatch the tools → submit outputs → assistant continues.

interven-openai-assistants inserts Interven between step 3 and step 4. You scan every tool_call the assistant returned; Interven tells you what to do with each.

Example

from openai import OpenAI
from interven_openai_assistants import scan_tool_calls

client = OpenAI()
run = client.beta.threads.runs.create_and_poll(thread_id=TID, assistant_id=AID)

if run.status == "requires_action":
    tool_calls = run.required_action.submit_tool_outputs.tool_calls
    decisions = scan_tool_calls(tool_calls, api_key="iv_live_...")

    outputs = []
    for scanned in decisions:
        call = scanned.tool_call
        scan = scanned.scan

        if scan.decision == "ALLOW":
            result = run_my_tool(call.function.name, scanned.effective_arguments)
            outputs.append({"tool_call_id": call.id, "output": result})

        elif scan.decision == "SANITIZE":
            # effective_arguments contains the redacted version
            result = run_my_tool(call.function.name, scanned.effective_arguments)
            outputs.append({"tool_call_id": call.id, "output": result})

        elif scan.decision == "DENY":
            outputs.append({
                "tool_call_id": call.id,
                "output": f"Tool blocked by Interven: {', '.join(scan.reason_codes)}",
            })

        elif scan.decision == "REQUIRE_APPROVAL":
            outputs.append({
                "tool_call_id": call.id,
                "output": f"Action requires human approval at {scan.raw.get('approval_url')}. Retry in a few minutes.",
            })

    client.beta.threads.runs.submit_tool_outputs(
        thread_id=TID, run_id=run.id, tool_outputs=outputs,
    )

What Interven understands

scan_tool_calls does best-effort mapping of common Assistants function names to real tool categories so policies fire correctly:

Function name contains	Mapped to
slack / post_message / webhook	Slack
github / pull_request / commit	GitHub
drive / gdrive / share_file	Google Drive
fetch / http / request	Generic HTTP
anything else	Generic tool with body scanning

For non-obvious function names, pass a url arg in your function schema and Interven will use it directly.

Options

scan_tool_calls(
    tool_calls,
    api_key="iv_live_...",   # or set INTERVEN_API_KEY env var
    gateway_url=None,         # defaults to https://api.intervensecurity.com
    timeout=30.0,
)

License

MIT