bighub-openai 0.1.0

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 tool call -> bighub-openai -> BIGHUB policies -> execute/block/approve

bighub-openai depends on both bighub and the official openai Python SDK. It is built for OpenAI Python SDK v1+ (openai>=1.0.0,<2.0.0).

Before any registered tool executes, the adapter:

• Validates the action via BIGHUB
• Enforces policy boundaries
• Blocks or escalates risky decisions
• Ingests governed execution into Future Memory

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+.

Quickstart (5 lines)

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="sk-...",
    bighub_api_key="bhk_...",
    actor="AI_AGENT_001",
    domain="payments",
)

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

from bighub_openai import AsyncGuardedOpenAI

guard = AsyncGuardedOpenAI(
    openai_api_key="sk-...",
    bighub_api_key="bhk_...",
    actor="AI_AGENT_001",
    domain="payments",
)

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:

• llm_delta (provider text delta)
• execution_event (governed tool decision/execution result)
• final_response (same payload 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:

guard = GuardedOpenAI(..., on_decision=lambda event: print(event))

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

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.1.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

Example:

guard = GuardedOpenAI(
    openai_api_key="sk-...",
    bighub_api_key="bhk_...",
    actor="AI_AGENT_001",
    domain="payments",
    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,
)

Notes

• This adapter is intentionally provider-specific.
• Core policy and transport behavior remain in the bighub SDK.