Python Integration

The Python SDK (hybridb-sdk) is coming soon. Until then, use the REST API directly.

Setup

pip install httpx python-dotenv

Client wrapper

# stellrai.py
import os
import httpx
from typing import Any

HDB_BASE_URL = os.environ.get("HDB_BASE_URL", "https://hybridb.stellrai.com")
HDB_API_KEY  = os.environ["HDB_API_KEY"]

_headers = {
    "Authorization": f"Bearer {HDB_API_KEY}",
    "Content-Type":  "application/json",
}

def _raise_for_error(data: dict) -> None:
    if not data.get("success"):
        err = data.get("error", {})
        raise Exception(f"[{err.get('code')}] {err.get('message')}")


def request_decision(
    decision_type: str,
    action: str,
    input_data: dict[str, Any],
    idempotency_key: str | None = None,
) -> dict:
    payload = {
        "decisionType": decision_type,
        "action":       action,
        "inputData":    input_data,
    }
    if idempotency_key:
        payload["idempotencyKey"] = idempotency_key

    r = httpx.post(
        f"{HDB_BASE_URL}/api/v1/decisions",
        json=payload,
        headers=_headers,
    )
    data = r.json()
    _raise_for_error(data)
    return data["data"]


def trigger_pipeline(
    pipeline_name: str,
    input_data: dict[str, Any],
    decision_id: str | None = None,
    idempotency_key: str | None = None,
) -> dict:
    payload: dict[str, Any] = {"input": input_data}
    if decision_id:
        payload["decisionId"] = decision_id
    if idempotency_key:
        payload["idempotencyKey"] = idempotency_key

    r = httpx.post(
        f"{HDB_BASE_URL}/api/v1/pipelines/{pipeline_name}/execute",
        json=payload,
        headers=_headers,
    )
    data = r.json()
    _raise_for_error(data)
    return data["data"]


def rollback(execution_id: str, reason: str) -> dict:
    r = httpx.post(
        f"{HDB_BASE_URL}/api/v1/executions/{execution_id}/rollback",
        json={"reason": reason},
        headers=_headers,
    )
    data = r.json()
    _raise_for_error(data)
    return data["data"]

Usage

# payment_service.py
from stellrai import request_decision, trigger_pipeline, rollback

def process_payment(user_id: str, amount: int, order_id: str) -> dict:
    # 1. Request decision
    decision = request_decision(
        decision_type    = "payment.authorize",
        action           = "payment.initiate",
        input_data       = {"actorId": f"user:{user_id}", "amount": amount},
        idempotency_key  = f"decision-{order_id}",
    )

    if decision["outcome"] == "blocked":
        return {"status": "blocked", "reason": decision["rationale"]}

    # 2. Execute pipeline
    execution = trigger_pipeline(
        pipeline_name    = "payment.execute.standard",
        input_data       = {"amount": amount, "to": "acct_9k2x"},
        decision_id      = decision["decisionId"],
        idempotency_key  = f"exec-{order_id}",
    )

    return {"status": "success", "execution_id": execution["id"]}


def reverse_payment(execution_id: str) -> dict:
    result = rollback(execution_id, reason="user_request")
    return {"reversed": result["status"] == "rolled_back"}

Async (httpx)

import httpx
import asyncio

async def request_decision_async(decision_type: str, action: str, input_data: dict) -> dict:
    async with httpx.AsyncClient() as client:
        r = await client.post(
            f"{HDB_BASE_URL}/api/v1/decisions",
            json={"decisionType": decision_type, "action": action, "inputData": input_data},
            headers=_headers,
        )
    data = r.json()
    _raise_for_error(data)
    return data["data"]

Python SDK coming soon

A native hybridb-sdk Python package with full type hints and async support is in development. This page will be updated when it ships.