bighub 0.1.0

Official Python SDK for integrating AI agents, workflows, and production systems with the BIGHUB execution control plane.

Official BIGHUB Python SDK

The official client library for integrating AI agents, workflows, and production systems with the BIGHUB execution control plane.

Install

pip install bighub

Requires Python 3.9+.

Error Handling

from bighub import BighubAPIError, BighubClient

client = BighubClient(api_key="bhk_...")

try:
    client.actions.submit(
        action="update_price",
        value=150.0,
        domain="pricing",
        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.

Quickstart (Sync)

from bighub import BighubClient

client = BighubClient(api_key="bhk_...")

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

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

Quickstart (Async)

import asyncio
from bighub import AsyncBighubClient


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


asyncio.run(main())

Supported Domains

• actions: submit, submit_v2, dry_run, verify_validation, observer_stats, dashboard_summary, status
• auth: signup, login, refresh, logout
• rules: create, list, get, update, delete, pause, resume, 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="bhk_...")

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

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

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

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

from bighub import BighubClient

client = BighubClient(api_key="bhk_...")

created = client.webhooks.create(
    {
        "url": "https://example.com/bighub-webhook",
        "label": "prod-endpoint",
        "events": ["signal.new", "rule.updated"],
    }
)
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="signal.new")

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()

Webhooks (Async)

import asyncio
from bighub import AsyncBighubClient


async def main() -> None:
    async with AsyncBighubClient(api_key="bhk_...") as client:
        created = await client.webhooks.create(
            {
                "url": "https://example.com/bighub-webhook",
                "label": "prod-endpoint",
                "events": ["signal.new", "rule.updated"],
            }
        )
        webhook_id = created["webhook_id"]

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


asyncio.run(main())

Auth + Governance Examples

from bighub import BighubClient

client = BighubClient()

tokens = client.auth.login({"email": "[e-mail]", "password": "pass1234"})
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
from bighub import AsyncBighubClient


async def main() -> None:
    async with AsyncBighubClient() as client:
        await client.auth.login({"email": "[e-mail]", "password": "pass1234"})
        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

from bighub import BighubClient

client = BighubClient(api_key="bhk_...")

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
from bighub import AsyncBighubClient


async def main() -> None:
    async with AsyncBighubClient(api_key="bhk_...") 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