sbn-sdk 0.12.2

Python SDK for the SmartBlocks Network — Atlas frontier provisioning, GEC compute, SnapChore integrity, governance, and more.

SmartBlocks Network Python SDK

Canonical Python client for the SBN infrastructure. Covers the full network surface — gateway, SnapChore, console, and control plane.

The same package also ships the shared pgdag execution kernel used by Tower products and workflow runtimes:

from pgdag import PGDAGRunner, DAGTemplate, DAGNode

Install

pip install sbn-sdk
# or from source
cd sdk/python && pip install -e .

Quick start

from sbn import SbnClient

client = SbnClient(base_url="https://api.smartblocks.network")
client.authenticate_api_key("sbn_live_abc123")

# SnapChore — capture, verify, seal
block = client.snapchore.capture({"event": "signup", "user": "u-42"})
client.snapchore.verify(block["snapchore_hash"], {"event": "signup", "user": "u-42"})
client.snapchore.seal(block["snapchore_hash"], {"event": "signup", "user": "u-42"})

# Gateway — slots, receipts, attestations
slot = client.gateway.create_slot(worker_id="w-1", task_type="classify")
receipt = client.gateway.fetch_receipt(slot.receipt_id)

# Console — API keys, usage, billing
keys = client.console.list_api_keys("proj-123")
usage = client.console.get_usage("proj-123")

# Control plane — rate plans, tenants, validators
plans = client.control_plane.list_rate_plans()
client.control_plane.create_tenant(
    name="Acme Corp",
    contact_email="[e-mail]",
    aggregator_endpoint="https://agg.acme.co",
    rate_plan_id=plans[0].id,
)

Event streaming wrapper

For telemetry-style frontiers, use the slot session wrapper instead of manually wiring slot open, BitBlock submission, and close on every project.

from sbn import SbnClient

client = SbnClient(base_url="https://api.smartblocks.network")
client.authenticate_api_key("sbn_live_abc123")
client.set_tenant("tower-agents")

with client.open_slot(
    worker_id="tower-agents",
    task_type="grid.atomic_rentals",
    domain="grid.atomic_rentals.streaming",
    generation_interval=1800,  # 30 minutes
    metadata={"frontier_hint": "grid.atomic_rentals.streaming"},
) as stream:
    stream.event(
        {
            "event_type": "delivery_picked_up",
            "booking_id": "bk_123",
            "status": "picked_up",
        },
        metadata={"interaction_id": "rental-123"},
    )
    stream.event(
        {
            "event_type": "delivery_complete",
            "booking_id": "bk_123",
            "status": "complete",
        },
        metadata={"interaction_id": "rental-123"},
    )

close_receipt = stream.receipt

Explicit attestation

Slot closure and validator-facing attestation are modeled separately on purpose. When you need billable / quorum attestation, request it explicitly with the hash you want routed through the attestation pipeline.

from sbn import SbnClient

client = SbnClient(base_url="https://api.smartblocks.network")
client.authenticate_api_key("sbn_live_abc123")
client.set_tenant("tower-agents")

with client.open_slot(
    worker_id="tower-agents",
    task_type="grid.atomic_rentals",
    domain="grid.atomic_rentals.streaming",
) as stream:
    stream.event({"event_type": "delivery_complete", "booking_id": "bk_123"})

final = stream.finalize(
    request_attestation=True,
    snap_hash="sha256:your-snap-hash",
    frontier_id="grid.atomic_rentals.streaming",
    metadata={
        "type": "LaborBlock",
        "domain": "grid",
        "event_type": "delivery_complete",
    },
)

print(final.close.receipt_id)
print(final.attestation)

Auth methods

# API key (most common for external devs)
client.authenticate_api_key("sbn_live_...")

# Bearer token (console sessions, service-to-service)
client.authenticate_bearer("eyJ...")

# Ed25519 signing key (auto-refreshing JWTs for agents)
from sbn import SigningKey
key = SigningKey.from_pem("/path/to/key.pem", issuer="my-svc", audience="sbn")
client.authenticate_signing_key(key, scopes=["attest.write", "snapchore.seal"])

Sub-clients

Property	Domain	Key operations
client.gateway	Slots & receipts	create_slot, close_slot, fetch_receipt, request_attestation
client.snapchore	Hash capture	capture, verify, seal, create_chain, append_to_chain
client.console	Developer console	list_api_keys, create_api_key, get_usage, get_billing_status
client.control_plane	Multi-tenancy	list_rate_plans, create_tenant, register_validator

Legacy compatibility

The original sbn_gateway.py single-file SDK is preserved for backward compatibility. New integrations should use from sbn import SbnClient.

Atlas workflows

The Python SDK exposes Atlas app registration, key binding, and workflow binding operations through client.atlas.

from sbn import SbnClient

client = SbnClient(base_url="https://api.smartblocks.network")
client.authenticate_api_key("sbn_live_abc123")

app = client.atlas.register_app(
    app_name="pt-kinetic",
    app_family="kinetic.sessions",
    declared_workflows=["pushup_session", "session_integrity"],
    workflow_specs=[
        {
            "workflow": "pushup_session",
            "frontier_id": "kinetic.pushup.amrap.v1",
            "pulse_type": "slot_stream",
            "event_types": ["rep", "set_complete"],
        },
        {
            "workflow": "session_integrity",
            "frontier_id": "kinetic.session.integrity.v1",
            "pulse_type": "pgdag",
            "event_types": ["seal_complete"],
        },
    ],
)

minted = client.atlas.mint_bound_key(app["app_id"], label="pt-kinetic network key")
client.atlas.upsert_binding(
    app["app_id"],
    workflow="pushup_session",
    frontier_id="kinetic.pushup.amrap.v1",
)

The CLI mirrors the same surface:

sbn atlas register \
  --name "pt-kinetic" \
  --family kinetic.sessions \
  --workflow pushup_session \
  --workflow session_integrity \
  --bind pushup_session=kinetic.pushup.amrap.v1 \
  --bind session_integrity=kinetic.session.integrity.v1 \
  --pulse pushup_session=slot_stream \
  --pulse session_integrity=pgdag

sbn atlas keys mint <app-id> --label "pt-kinetic network key"
sbn atlas keys bind <app-id> <api-key-id>
sbn atlas bindings set <app-id> --workflow pushup_session --frontier kinetic.pushup.amrap.v1