bighub-openai 0.2.3

OpenAI adapter for governing tool execution with BIGHUB.

bighub-openai - Production-Safe OpenAI Agents

bighub-openai makes OpenAI tool-calling agents production-safe.

Where it fits:

OpenAI Responses API  ->  bighub-openai  ->  BIGHUB Control Plane  ->  execute / block / approve

bighub-openai depends on both bighub and the official openai Python SDK. It targets the Responses API (client.responses.create) and requires openai>=2.0.0,<3.0.0.

Useful links:

• Core SDK package (PyPI): bighub
• OpenAI Responses API docs: platform.openai.com/docs/api-reference/responses

Before any registered tool executes, the adapter:

• Validates the action against BIGHUB policies
• Enforces execution boundaries
• Blocks or escalates risky decisions
• Ingests governed execution into Future Memory for pattern learning
• Sets store=false by default so tool-call data is not persisted on OpenAI servers

Decision enforcement outcomes:

• allowed -> execute tool
• blocked -> do not execute tool
• requires_approval -> do not execute tool; adapter output status is approval_required

Install

pip install bighub-openai

Requires Python 3.9+. Recommended: use a dedicated virtual environment to avoid dependency conflicts with other packages that pin different openai versions.

Quickstart (5 lines)

import os
from bighub_openai import GuardedOpenAI

def refund_payment(order_id: str, amount: float) -> dict:
    return {"ok": True, "order_id": order_id, "amount": amount}

guard = GuardedOpenAI(
    openai_api_key=os.getenv("OPENAI_API_KEY"),
    bighub_api_key=os.getenv("BIGHUB_API_KEY"),
    actor="AI_AGENT_001",
    domain="customer_transactions",
)

guard.tool("refund_payment", refund_payment, value_from_args=lambda a: a["amount"])

response = guard.run(
    messages=[
        {"role": "user", "content": "Refund order ord_123 for 199.99"},
    ],
    model="gpt-4.1",
)

print(response)

Async quickstart

import os
from bighub_openai import AsyncGuardedOpenAI

guard = AsyncGuardedOpenAI(
    openai_api_key=os.getenv("OPENAI_API_KEY"),
    bighub_api_key=os.getenv("BIGHUB_API_KEY"),
    actor="AI_AGENT_001",
    domain="customer_transactions",
)

guard.tool(...) auto-generates a strict JSON schema from your Python function signature. Provide parameters_schema=... only when you need custom schema constraints.

run(...) returns a structured payload with both model output and governance execution events:

{
  "llm_response": {...},
  "execution": {
    "events": [...],
    "last": {
      "tool": "refund_payment",
      "status": "executed" | "blocked" | "approval_required",
      "decision": {...}
    }
  }
}

run_stream(...) yields structured stream events mapped from the Responses API streaming protocol:

Adapter event type	Responses API event	Description
llm_delta	response.output_text.delta	Incremental text token
llm_text_done	response.output_text.done	Complete text segment
output_item_added	response.output_item.added	New output item started
function_call_args_delta	response.function_call_arguments.delta	Streaming function args
function_call_args_done	response.function_call_arguments.done	Complete function args
refusal_delta	response.refusal.delta	Model refusal chunk
response_done	response.completed	Response finished
response_failed	response.failed	Response error
execution_event	(adapter)	Governed tool decision/execution result
final_response	(adapter)	Final payload, same shape as run(...)

Example:

for event in guard.run_stream(
    messages=[{"role": "user", "content": "Refund order ord_123 for 199.99"}],
    model="gpt-4.1",
):
    if event["type"] == "llm_delta":
        print(event["delta"], end="")
    elif event["type"] == "execution_event":
        print("\n[tool]", event["event"]["tool"], event["event"]["status"])
    elif event["type"] == "final_response":
        print("\nDone:", event["response"]["output_text"])

Async example:

async for event in guard.run_stream(
    messages=[{"role": "user", "content": "Refund order ord_123 for 199.99"}],
    model="gpt-4.1",
):
    if event["type"] == "llm_delta":
        print(event["delta"], end="")
    elif event["type"] == "execution_event":
        print("\n[tool]", event["event"]["tool"], event["event"]["status"])
    elif event["type"] == "final_response":
        print("\nDone:", event["response"]["output_text"])

Decision modes

• decision_mode="submit" (default) -> calls client.actions.submit(...)
• decision_mode="submit_v2" -> calls client.actions.submit_v2(...)

You can set it globally on GuardedOpenAI(...) or per tool in register_tool(..., decision_mode="submit_v2").

Audit hook

Use on_decision to forward structured events to your observability stack:

def log_decision(event: dict) -> None:
    # Avoid dumping full payloads/secrets to logs.
    minimal = {
        "trace_id": event.get("trace_id"),
        "request_id": event.get("request_id"),
        "event_id": event.get("event_id"),
        "allowed": event.get("allowed"),
    }
    print(minimal)

guard = GuardedOpenAI(..., on_decision=log_decision)

Event payload contract includes stable identifiers:

• trace_id (run correlation id)
• request_id (BIGHUB validation id when available)
• event_id (adapter event id)

Silent mode

When you want to evaluate governance without executing tools:

decision = guard.check_tool("refund_payment", {"order_id": "ord_123", "amount": 199.0})

Approval loop helper

Use HITL helpers when a tool decision returns requires_approval:

• run_with_approval(...) runs, captures pending approval, and can resume with a callback
• resume_after_approval(...) resolves one approval request then resumes tool execution

Approval callbacks should run server-side (not in the client) to avoid exposing approval credentials.

Example:

result = guard.run_with_approval(
    messages=[{"role": "user", "content": "Refund order ord_123 for 5000"}],
    model="gpt-4.1",
    on_approval_required=lambda ctx: {
        "resolution": "approved",
        "comment": "approved by on-call",
    },
)

print(result["approval_loop"])

Async example:

result = await guard.run_with_approval(
    messages=[{"role": "user", "content": "Refund order ord_123 for 5000"}],
    model="gpt-4.1",
    on_approval_required=lambda ctx: {
        "resolution": "approved",
        "comment": "approved by on-call",
    },
)

print(result["approval_loop"])

Future memory ingest

Note: Future Memory is experimental and may contain bugs.

By default, GuardedOpenAI ingests governed execution events to BIGHUB future memory:

guard = GuardedOpenAI(..., memory_enabled=True, memory_source="openai_adapter")

The adapter ingests memory in best-effort mode:

• short timeout (memory_ingest_timeout_ms, default 300)
• exceptions are swallowed
• governance execution path is never blocked by telemetry

Each event includes idempotency/versioning metadata for stable analytics:

• event_id (dedupe key)
• seq (position within run)
• schema_version (current: 1)
• source_version (for example bighub-openai@0.2.x)

This powers pattern learning and context endpoints such as client.actions.memory_context(...).

Fail Modes

• fail_mode="closed" (default): if policy check fails, tool execution is blocked.
• fail_mode="open": if policy check fails unexpectedly, tool execution proceeds.

Provider resilience knobs

GuardedOpenAI and AsyncGuardedOpenAI expose provider resilience settings:

• provider_timeout_seconds
• provider_max_retries
• provider_retry_backoff_seconds
• provider_retry_max_backoff_seconds
• provider_retry_jitter_seconds
• provider_circuit_breaker_failures (set >0 to enable)
• provider_circuit_breaker_reset_seconds

Retries only trigger on transient OpenAI errors (APIConnectionError, APITimeoutError, RateLimitError). Non-retryable errors (e.g. BadRequestError, AuthenticationError) fail immediately.

Example:

import os
guard = GuardedOpenAI(
    openai_api_key=os.getenv("OPENAI_API_KEY"),
    bighub_api_key=os.getenv("BIGHUB_API_KEY"),
    actor="AI_AGENT_001",
    domain="customer_transactions",
    provider_timeout_seconds=20,
    provider_max_retries=3,
    provider_retry_backoff_seconds=0.2,
    provider_retry_max_backoff_seconds=2.0,
    provider_retry_jitter_seconds=0.15,
    provider_circuit_breaker_failures=5,
    provider_circuit_breaker_reset_seconds=30,
)

Responses API compatibility

This adapter is built specifically for the OpenAI Responses API (client.responses.create). Key design choices:

• Tool schema: Uses flat {type: "function", name, parameters, strict} format (not the nested {type: "function", function: {...}} from Chat Completions).
• Multi-turn context: Uses previous_response_id for efficient multi-turn loops. Reasoning items from prior turns are preserved automatically by the API.
• store: false: Set by default on all provider calls. Override via extra_create_args={"store": True} if you need statefulness.
• Output parsing: Handles response.output_text natively and falls back to response.output[].content[].text extraction for edge cases.
• Function calls: Parses function_call items from response.output[] and ignores reasoning, message, and other item types safely.
• function_call_output: Returns results as {type: "function_call_output", call_id, output} per the Responses API specification.

Notes

• This adapter is intentionally provider-specific.
• Core policy and transport behavior remain in the bighub SDK.
• Compatible with openai>=2.0.0,<3.0.0 (Responses API).