sbn-sdk 0.12.12

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.pulse(
        "delivery_picked_up",
        score=0.9,
        actions=[{"action": "delivery_picked_up", "score": 0.9}],
        interactions=[{"action": "route_quality", "score": 0.94}],
        events={"friendly": 1},
        booking_id="bk_123",
        status="picked_up",
        metadata={"interaction_id": "rental-123"},
    )
    seal = stream.seal()
    print(seal.anchor_hash, seal.anchor_type)

    stream.pulse(
        "delivery_complete",
        score=1.0,
        actions=[{"action": "delivery_complete", "score": 1.0}],
        events={"friendly": 1},
        booking_id="bk_123",
        status="complete",
        metadata={"interaction_id": "rental-123"},
    )

close_receipt = stream.receipt

Close and finalize semantics

The wrapper now has three useful boundaries:

• seal() - compose and attest the current pulse while the slot stays open
• close() - close the slot and persist the final slot-summary receipt
• finalize() - optionally request an additional explicit attestation with a caller-provided snap_hash

That means proof can become real during the stream instead of waiting until the final close call.

Explicit attestation

When you need an extra billable / quorum attestation beyond progressive seals, 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.pulse(
        "delivery_complete",
        score=1.0,
        actions=[{"action": "delivery_complete", "score": 1.0}],
        events={"friendly": 1},
        booking_id="bk_123",
    )
    step_anchor = stream.seal()

final = stream.finalize(
    request_attestation=True,
    snap_hash=step_anchor.anchor_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
sbn atlas bindings history <app-id> --workflow pushup_session
sbn atlas bindings restore <app-id> --workflow pushup_session --index 2

Lattice-backed pgDAG execution

When a workflow lives in Lattice, the SDK can load it as a pgdag.DAGTemplate with the linked workflow_id already embedded in template metadata. That means pgDAG execution can emit proof-aware Lattice outcomes automatically as steps seal and publish.

from sbn import SbnClient
from pgdag import build, StepContext

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

layer = build(
    sbn_client=client,
    lattice_client=client.lattice,
    domain="kinetic.session",
)

template = client.lattice.build_pgdag_template("wf_kinetic_session_integrity_v3_ab12cd34")

async def collect_signal(ctx: StepContext) -> dict:
    return {"signal": "collected", "sample_id": ctx.inputs["sample_id"]}

async def seal_integrity(ctx: StepContext) -> dict:
    return {"integrity": "verified", "sample_id": ctx.inputs["sample_id"]}

layer.runner.register_steps(
    {
        "collect_signal": collect_signal,
        "seal_integrity": seal_integrity,
    }
)

result = await layer.runner.execute(
    template,
    inputs={"sample_id": "sess-42"},
    actor="kinetic-integrity-engine",
)

CLI helpers:

sbn lattice workflows create \
  --name "kinetic.pushup.amrap.v1" \
  --domain kinetic.pushup \
  --nodes-file nodes.json \
  --edges-file edges.json \
  --meta-file runtime_binding.json \
  --activate

sbn lattice workflows import workflow-export.json --activate
sbn lattice workflows import workflow-bundle.json
sbn lattice workflows template <workflow-id>
sbn lattice workflows template <workflow-id> --output kinetic_session_template.json
sbn lattice workflows runs <workflow-id>
sbn lattice workflows evidence-packet <workflow-id> <run-id> --output run-packet.json
sbn lattice workflows verify-packet --file run-packet.json

Reality review workflows can also be handled directly from the CLI:

sbn reality facts review <fact-id> --action accept
sbn reality facts review <fact-id> --action merge --target-fact-id <existing-fact-id> --note "duplicate extraction"