bighub 0.2.4

Official Python SDK for governing AI agent execution with the BIGHUB intelligence and control layer.

BIGHUB - AI Agent Control Plane

BIGHUB is the execution control plane for AI agents in production.

It sits between agent reasoning and real-world execution, validating every action against enforceable policies before it reaches production systems.

As AI agents move from suggestion to execution, risk becomes structural. BIGHUB makes autonomy enforceable.

BIGHUB provides:

• Runtime execution validation
• Policy enforcement and approvals
• Decision audit trails
• Future Memory (execution learning)
• Safe policy recommendations
• Deterministic rule mutation with atomic locking
• Webhook export for compliance and SIEM

This Python package is the official client for the BIGHUB control plane API.

• OpenAI adapter: bighub-openai

Install

pip install bighub

Requires Python 3.9+.

Quickstart (Sync)

import os
from bighub import BighubClient

client = BighubClient(api_key=os.getenv("BIGHUB_API_KEY"))

result = client.actions.submit(
    action="update_price",
    value=150.0,
    domain="financial_actions",
    actor="AI_AGENT_001",
)

print(result["allowed"], result["risk_score"])
client.close()

Quickstart (Async)

import asyncio
import os
from bighub import AsyncBighubClient


async def main() -> None:
    async with AsyncBighubClient(api_key=os.getenv("BIGHUB_API_KEY")) as client:
        result = await client.actions.submit(
            action="update_price",
            value=150.0,
            domain="financial_actions",
            actor="AI_AGENT_001",
        )
        print(result["allowed"], result["risk_score"])


asyncio.run(main())

Execution Domains

Use these domain values in SDK calls:

• financial_actions
• operational_systems
• infrastructure_devops
• customer_transactions
• data_modifications
• custom

Why BIGHUB?

• Agents now act directly in production
• Guardrails are not enforcement
• Logging is not control
• Policy drift increases over time

BIGHUB enforces bounded autonomy at execution time.

Architecture

LLM / Agent Runtime
        ↓
Provider Adapter (e.g. bighub-openai)
        ↓
BIGHUB Control Plane API
        ↓
Rules + Approvals + Memory + Policy Intelligence

BIGHUB enforces bounded autonomy between AI reasoning and real-world execution. Agents keep planning and deciding, while BIGHUB enforces execution boundaries at runtime. Every governed action is evaluated against explicit policies, logged with full decision context, and exported through structured webhooks for operational and compliance systems.

Policy Intelligence

BIGHUB does not only validate actions. It learns from governed execution.

Future Memory creates a closed feedback loop:

• Observed runtime behavior
• Persisted execution memory
• Pattern detection
• Safe JSON Patch policy suggestions
• Preview-first apply workflow
• Atomic optimistic locking

BIGHUB never loosens autonomy automatically. Governance improves over time without increasing risk.

Provider Adapters

BIGHUB integrates with AI runtimes through provider-specific adapters:

• bighub-openai - OpenAI tool-calling governance
• (coming soon) bighub-anthropic
• (coming soon) bighub-perplexity

Install an adapter to enforce execution policies directly inside your agent runtime:

pip install bighub-openai

For streaming, approval loop helpers, provider resilience, and Responses API compatibility, see:

• Adapter package page: PyPI bighub-openai

Error Handling

import os
from bighub import BighubAPIError, BighubClient

client = BighubClient(api_key=os.getenv("BIGHUB_API_KEY"))

try:
    client.actions.submit(
        action="update_price",
        value=150.0,
        domain="financial_actions",
        actor="AI_AGENT_001",
    )
except BighubAPIError as e:
    print(e.status_code, e)
finally:
    client.close()

Versioning & Stability

This SDK follows Semantic Versioning. Until 1.0.0, breaking changes may occur but will be documented.

Future Memory (Execution Learning)

Note: Future Memory is experimental and may contain bugs.

Future Memory already exposes three endpoints in the API and the SDK:

• ingest_memory(...) -> stores governed execution events (tool outcomes + decision metadata)
• memory_context(...) -> returns learned context (blocked_rate, top reasons/tools, recent events)
• refresh_memory_aggregates(...) -> refreshes materialized aggregates for fast analytics
• memory_recommendations(...) -> returns policy recommendations from observed patterns (Pro/Enterprise only)

This gives you a closed feedback loop for agent governance: runtime decisions -> persisted memory -> analytics + prompt/context enrichment.

What was hardened

The memory pipeline is now hardened for production SaaS usage:

• Idempotency: per-event event_id support to avoid duplicate ingest on retries
• Schema versioning: schema_version and source_version to keep event contracts evolvable
• Privacy controls: redact + redaction_policy support on ingest
• Tenant safety: context is scoped server-side by authenticated org_id
• Aggregate safety: refresh endpoint is admin-only and rate-limited on server side

SDK Example

You can ingest governed adapter/runtime events and query memory context for policy-aware execution:

import os
from bighub import BighubClient

client = BighubClient(api_key=os.getenv("BIGHUB_API_KEY"))

client.actions.ingest_memory(
    source="openai_adapter",
    source_version="bighub-openai@0.2.3",
    actor="AI_AGENT_001",
    domain="customer_transactions",
    model="gpt-4.1",
    redact=True,
    redaction_policy="default",
    events=[
        {
            "event_id": "6f1aa3f4-4a86-4f95-b65e-2a37642b90ad",
            "schema_version": 1,
            "source_version": "bighub-openai@0.2.3",
            "seq": 0,
            "tool": "refund_payment",
            "status": "blocked",
            "decision": {"allowed": False, "result": "blocked", "blocked_by": "max_value", "risk_score": 0.91},
            "arguments": {"order_id": "ord_123", "amount": 2500.0},
        }
    ],
)

context = client.actions.memory_context(window_hours=24, tool="refund_payment")
print(context["blocked_rate"], context["top_block_reasons"])

# Operational endpoint (admin-scoped server-side)
client.actions.refresh_memory_aggregates(concurrent=True, window_hours=24)

# Pro/Enterprise: policy recommendations generated from memory patterns
recommendations = client.actions.memory_recommendations(
    window_hours=24,
    scope={"domain": "customer_transactions", "tool": "refund_payment", "actor": "AI_AGENT_001"},
    min_events=20,
    auto_apply=False,
)
print(recommendations["recommendations"])

Recommendations include:

• explicit scope and time_window_hours
• impact_estimate heuristics for business value
• suggested_policy_patch as JSON Patch
• is_safe_patch + risk_increasing_patch=False
• requires_human_review=True by default
• target_rule_id / rule_selector for deterministic patch targeting
• optional apply_endpoint for DX (/rules/{id}/apply_patch)

One-click safe apply pattern (preview + optimistic apply):

rec = recommendations["recommendations"][0]
preview = client.rules.apply_patch(
    rec["target_rule_id"],
    patch=rec["suggested_policy_patch"],
    preview=True,
)
applied = client.rules.apply_patch(
    rec["target_rule_id"],
    patch=rec["suggested_policy_patch"],
    preview=False,
    if_match_version=preview["after"]["version"],
)

Supported Domains

• actions: submit, submit_v2, dry_run, verify_validation, observer_stats, dashboard_summary, status, ingest_memory, memory_context, refresh_memory_aggregates, memory_recommendations (Pro/Enterprise)
• auth: signup, login, refresh, logout
• rules: create, list, get, update, delete, pause, resume, apply_patch (admin workflow), dry_run, validate, validate_dry_run, domains, versions, purge_idempotency
• kill_switch: status, activate, deactivate
• events: list, stats
• approvals: list, resolve
• api_keys: create, list, delete/revoke, rotate, validate, scopes
• webhooks: create, list, get, update, delete, deliveries, test, list_events, verify_signature, replay_failed_delivery

Ergonomic Models (Dataclass)

You can pass typed models instead of raw dictionaries for key endpoints.

from bighub import (
    BighubClient,
    ActionSubmitV2Model,
    RuleCreateModel,
    RuleUpdateModel,
    WebhookUpdateModel,
)

client = BighubClient(api_key=os.getenv("BIGHUB_API_KEY"))

decision = client.actions.submit_v2(
    payload=ActionSubmitV2Model(
        action="update_price",
        value=150.0,
        target="product_123",
        domain="financial_actions",
    )
)

rule = client.rules.create(
    RuleCreateModel(
        name="Pricing Safety Rule",
        domain="financial_actions",
        max_per_day=100,
        max_value=1000,
        require_approval_above=500,
    )
)

client.rules.update(
    rule["rule_id"],
    RuleUpdateModel(
        require_approval_above=650,
        tags=["ops", "financial"],
    ),
)

client.webhooks.update(
    "wh_123",
    WebhookUpdateModel(
        label="prod-endpoint-v2",
        is_active=True,
    ),
)

client.close()

Webhooks Management Examples

client.webhooks.* calls the BIGHUB API (https://api.bighub.io by default). The "url" value below is your own public endpoint where BIGHUB delivers events (for example, https://api.yourapp.com/webhooks/bighub).

import os
from bighub import BighubClient

client = BighubClient(api_key=os.getenv("BIGHUB_API_KEY"))

created = client.webhooks.create(
    {
        "url": "https://example.com/bighub-webhook",
        "label": "prod-endpoint",
        "events": ["decision_event.created", "rule.patch_applied", "approval.created"],
    }
)
webhook_id = created["webhook_id"]

all_webhooks = client.webhooks.list(include_inactive=True)
webhook = client.webhooks.get(webhook_id)
deliveries = client.webhooks.deliveries(webhook_id, limit=20)
test_result = client.webhooks.test(webhook_id, event_type="decision_event.created")

updated = client.webhooks.update(webhook_id, {"label": "prod-endpoint-v2"})
replayed = client.webhooks.replay_failed_delivery(webhook_id, delivery_id=123)
deleted = client.webhooks.delete(webhook_id)

client.close()

Use client.webhooks.list_events() as the source of truth for subscribable public event names in your environment.

Webhooks (Async)

import asyncio
import os
from bighub import AsyncBighubClient


async def main() -> None:
    async with AsyncBighubClient(api_key=os.getenv("BIGHUB_API_KEY")) as client:
        created = await client.webhooks.create(
            {
                "url": "https://example.com/bighub-webhook",
                "label": "prod-endpoint",
                "events": ["decision_event.created", "rule.patch_applied", "approval.created"],
            }
        )
        webhook_id = created["webhook_id"]

        await client.webhooks.test(webhook_id, event_type="decision_event.created")
        await client.webhooks.deliveries(webhook_id, limit=20)
        await client.webhooks.delete(webhook_id)


asyncio.run(main())

Auth + Governance Examples

import os
from bighub import BighubClient

client = BighubClient()

tokens = client.auth.login({"email": "[e-mail]", "password": os.getenv("BIGHUB_PASSWORD")})
events = client.events.list(event_type="rule.updated", limit=20)
pending = client.approvals.list(status_filter="pending", limit=50)

if pending:
    client.approvals.resolve(
        pending[0]["request_id"],
        resolution="approved",
        comment="approved by on-call",
    )

client.close()

Auth + Governance (Async)

import asyncio
import os
from bighub import AsyncBighubClient


async def main() -> None:
    async with AsyncBighubClient() as client:
        await client.auth.login({"email": "[e-mail]", "password": os.getenv("BIGHUB_PASSWORD")})
        await client.events.list(event_type="rule.updated", limit=20)

        pending = await client.approvals.list(status_filter="pending", limit=50)
        if pending:
            await client.approvals.resolve(
                pending[0]["request_id"],
                resolution="approved",
                comment="approved by on-call",
            )


asyncio.run(main())

API Keys Management Examples

import os
from bighub import BighubClient

client = BighubClient(api_key=os.getenv("BIGHUB_API_KEY"))

created = client.api_keys.create(
    {
        "label": "production-key",
        "scopes": ["actions:validate", "rules:read"],
        "expires_in_days": 90,
    }
)
key_id = created["key_id"]

all_keys = client.api_keys.list(include_revoked=False)
validated = client.api_keys.validate(created["key"])
scopes = client.api_keys.scopes()

rotated = client.api_keys.rotate(key_id)
revoked = client.api_keys.delete(key_id, reason="rotated")

client.close()

API Keys (Async)

import asyncio
import os
from bighub import AsyncBighubClient


async def main() -> None:
    async with AsyncBighubClient(api_key=os.getenv("BIGHUB_API_KEY")) as client:
        created = await client.api_keys.create(
            {
                "label": "production-key",
                "scopes": ["actions:validate", "rules:read"],
                "expires_in_days": 90,
            }
        )
        key_id = created["key_id"]

        await client.api_keys.list(include_revoked=False)
        await client.api_keys.scopes()
        await client.api_keys.rotate(key_id)


asyncio.run(main())

Auth

Use one of:

• api_key="..." (X-API-Key)
• bearer_token="..." (Authorization: Bearer)

If both are provided, X-API-Key is preferred by default.

Reliability Features

• Configurable timeout
• Retry with exponential backoff for transient failures (429/5xx/network)
• Idempotency-Key support for mutable endpoints
• Typed exception hierarchy with request/response metadata