luveo-rules-sdk

SDK for integrating Luveo services with the Luveo Rules Engine.

Installation

Install the shared packages from the same release:

pip install \
  --extra-index-url https://<your-package-index>/simple \
  luveo-json-logic==0.1.0 \
  luveo-rules-contracts==0.1.0 \
  luveo-rules-sdk==0.1.0

If you are consuming tagged GitHub release artifacts instead of a package index, install the wheels for all three packages from the same release tag.

Main API Surface

from luveo_rules_sdk import (
    EngineClient,
    InventoryReceiptEvent,
    SyncGuardMiddleware,
    WebhookSubscriptionClient,
    create_delivery_claims_dependency,
    ensure_rule_execution_delivery,
    handle_rule_execution_delivery,
)

• SyncGuardMiddleware: fail-closed pre-commit blocking for sync-critical rules
• EngineClient: async client for committed or blocked event submission
• canonical event models: shared contracts re-exported by the SDK
• WebhookSubscriptionClient: typed webhook admin API client
• ensure_rule_execution_delivery(...): idempotent webhook setup helper
• create_delivery_claims_dependency(...): FastAPI bearer-token verifier for callback routes
• handle_rule_execution_delivery(...): validate + dispatch helper for accepted callback batches

Canonical Event Example

from datetime import datetime, UTC

from luveo_rules_sdk import EngineClient, InventoryReceiptEvent

client = EngineClient(
    engine_url="https://rules-engine.internal",
    bearer_token="<engine-api-token>",
)

payload = InventoryReceiptEvent(
    tenant_id="00000000-0000-0000-0000-000000000001",
    facility_id="11111111-1111-1111-1111-111111111111",
    user_id="22222222-2222-2222-2222-222222222222",
    user_role="buyer",
    event_timestamp=datetime.now(UTC),
    drug_name_or_ndc="12345678901",
    drug_schedule="Schedule_II",
    supplier_name="Demo Supplier",
    supplier_dea_number="AB1234567",
    quantity_received="10",
    unit_of_measure="vials",
    receipt_date=datetime.now(UTC),
)

await client.submit_event(payload.model_dump(mode="json"))

Webhook Setup Example

from luveo_rules_sdk import WebhookSubscriptionClient, ensure_rule_execution_delivery

client = WebhookSubscriptionClient(
    engine_url="https://rules-engine.internal",
    bearer_token="<engine-admin-token>",
)

ensured = await ensure_rule_execution_delivery(
    client=client,
    tenant_id="tenant-123",
    target_url="https://luveo.internal/api/v1/rule-executions",
    headers={"X-Luveo-Service": "backend"},
    signature_mode="bearer_jwt",
)

Callback Receiver Example

from fastapi import Depends, FastAPI, Request

from luveo_rules_sdk import (
    DeliveryAuthSettings,
    RuleExecutionDeliveryRequest,
    RuleExecutionDeliveryResponse,
    create_delivery_claims_dependency,
    handle_rule_execution_delivery,
)

app = FastAPI()


def get_delivery_auth_settings(request: Request) -> DeliveryAuthSettings:
    return DeliveryAuthSettings(
        secret=request.app.state.delivery_secret,
        issuer=request.app.state.delivery_issuer,
        audience=request.app.state.delivery_audience,
    )


delivery_claims = create_delivery_claims_dependency(get_delivery_auth_settings)


@app.post("/api/v1/rule-executions", response_model=RuleExecutionDeliveryResponse)
async def receive_rule_executions(
    payload: RuleExecutionDeliveryRequest,
    claims: dict[str, object] = Depends(delivery_claims),
) -> RuleExecutionDeliveryResponse:
    return await handle_rule_execution_delivery(
        payload=payload,
        claims=claims,
        handler=process_rule_execution_batch,
    )

Release Model

This repository publishes luveo-json-logic, luveo-rules-contracts, and luveo-rules-sdk together on version tags so Luveo integrations always consume compatible schemas and transport helpers.