sbn-sdk 0.13.3

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.

For the internal pgdag-py execution wrapper, the package also exposes a dedicated HTTP client:

from sbn import PGDagPyClient

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

from pgdag import PGDAGRunner, DAGTemplate, DAGNode

Choose the surface by workload shape:

• telemetry or pulse streams -> slot event wrapper
• bounded exact-state work -> direct SmartBlock submission
• proof-gated workflows -> pgdag or pgdag-py
• SnapChore-only proof capture -> SnapChore client

Release/publish hygiene for this package is tracked in:

• docs/ops/sdk_publish_checklist.md

Which wrapper should I use?

For most client integrations, start with one of these two wrappers:

If your frontier feels like...	Use	Why
a live stream of events over time	client.open_slot(...)	best for generation windows, BitBlock emission, progressive seal(), and final close()
a native object or branch that evolves	client.open_surface(...)	best for origin/branch/successor flows and consistent surface metadata

Keep client.snapchore separate in your mental model:

• use SnapChore when you want local-first capture / verify / seal
• use promote() only when you explicitly want to hand that local proof into SBN later
• do not treat SnapChore as the third main wrapper for normal network lifecycle work

Install

pip install sbn-sdk
# optional if you need pgdag work claiming helpers
pip install "sbn-sdk[work-queue]"
# or from source
cd sdk/python && pip install -e .

Quick start

For most event-heavy product work, the safest default is the slot event wrapper. If your product instead revolves around a canonical object surface, skip to Agnostic surface wrapper. The broad quick start below shows the main client namespaces.

from sbn import SbnClient

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

# SnapChore — capture, verify, seal
captured = client.snapchore.capture({"event": "signup", "user": "u-42"})
client.snapchore.verify(captured["hash"], {"event": "signup", "user": "u-42"})
sealed = client.snapchore.seal({"event": "signup", "user": "u-42"}, domain="auth")

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

# 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,
)

Hello World: SnapChore

This is the smallest honest SnapChore flow:

1. capture a canonical hash
2. verify it
3. seal a local SmartBlock
4. inspect the local block
5. optionally promote it into SBN later

from sbn import SbnClient

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

payload = {"event": "hello_world", "message": "Hello, SnapChore"}

captured = client.snapchore.capture(payload)
print("hash:", captured["hash"])

verified = client.snapchore.verify(captured["hash"], payload)
print("valid:", verified["valid"])

sealed = client.snapchore.seal(payload, domain="hello.world")
print("block:", sealed["id"])
print("record:", sealed["smartblock_record_id"])

detail = client.snapchore.get_block(sealed["smartblock_record_id"])
print("lineage role:", detail.get("lineage_identity", {}).get("surface_role"))

# Optional later handoff into the SBN attestation lane
promotion = client.snapchore.promote(
    smartblock_record_id=sealed["smartblock_record_id"]
)
print("promotion status:", promotion["status"])

Important:

• seal() is local-first
• the block exists before any receipt exists
• promote() is the explicit SBN handoff boundary

Artifact descriptor helper

When you want to attest an external file without bloating the payload model, compute a small canonical artifact descriptor from the exact file bytes:

from sbn import SbnClient, artifact_descriptor_from_file

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

artifact = artifact_descriptor_from_file(
    "docs/EXACT_CLOSURE_PATENT_SUPPORTING_ARTIFACT_MAP.md",
    artifact_uri="repo://docs/EXACT_CLOSURE_PATENT_SUPPORTING_ARTIFACT_MAP.md",
)

captured = client.snapchore.capture({"artifact": artifact})
print(captured["hash"])

This gives you:

• exact file-byte identity via artifact_sha256
• a stable descriptor SnapChore hash
• a descriptor you can embed into a later SmartBlock payload before attestation

Golden path: hash local file bytes -> seal local proof block

If what you really want is the normal builder flow:

1. hash the local file bytes
2. wrap them in a standard proof payload
3. seal a local SmartBlock with moment auth

use the SnapChore convenience wrapper directly:

from sbn import SbnClient

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

sealed = client.snapchore.seal_artifact_file(
    "docs/EXACT_CLOSURE_PATENT_SUPPORTING_ARTIFACT_MAP.md",
    artifact_uri="repo://docs/EXACT_CLOSURE_PATENT_SUPPORTING_ARTIFACT_MAP.md",
    summary="exact closure support map ready for review",
    subject_ref="exact-closure-milestone",
    claim={"milestone_ref": "patent-support-map-ready"},
)

print(sealed["snapchore_hash"])
print(sealed["smartblock_record_id"])

This is the intended SnapChore-first golden path:

• file bytes are hashed first
• the sealed block carries the artifact descriptor
• moment auth is attached by the SnapChore seal path
• promotion into SBN still stays explicit and separate

You can also inspect the result in one stable local-proof shape before any promotion:

status = client.snapchore.local_proof_status(sealed)
print(status["status"])                # sealed_local
print(status["snapchore_hash"])
print(status["moment_auth"]["present"])

And when you do promote later, normalize the handoff the same way:

promotion = client.snapchore.promote(
    smartblock_record_id=sealed["smartblock_record_id"],
    frontier_id="artifact.review",
)
handoff = client.snapchore.promotion_status(
    promotion,
    local_status=status,
)
print(handoff["status"])  # submitted | queued_for_attestation | attested

If the attested receipt should land in a specific SBN project, target it explicitly at promotion time:

promotion = client.snapchore.promote_to_project(
    smartblock_record_id=sealed["smartblock_record_id"],
    target_project_id="tenant-123",
    frontier_id="artifact.review",
)
print(promotion["target_project_id"])

And if you need to create the target SBN project first:

result = client.snapchore.create_project_and_promote(
    project_name="Artifact Lane",
    contact_email="[e-mail]",
    aggregator_endpoint="https://agg.example.com",
    rate_plan_id="plan-sandbox",
    smartblock_record_id=sealed["smartblock_record_id"],
    frontier_id="artifact.review",
)
print(result["target_project_id"])
print(result["promotion"]["promotion_status"])

If you want one copy-paste task-oriented example for this same lane, run:

py -3 scripts/examples/snapchore_artifact_task_flow.py docs/EXACT_CLOSURE_PATENT_SUPPORTING_ARTIFACT_MAP.md --api-key <sbn_live_key> --task-ref exact-closure-milestone --target-project-id <target_sbn_project_uuid>

Canonical task -> artifact -> audit flow

If you want one infrastructure-level proof lane that builders can reuse across products, use this shape:

1. TaskBlock records task_started
2. SnapChore seals the completed artifact
3. explicit promotion yields the attested receipt anchor
4. AuditBlock records reconciled

Minimal payload convention:

• TaskBlock
  • task_ref
  • task_state
  • recommended: workflow_ref, operator_ref, subject_ref
• SnapChore proof claim
  • task_ref
  • task_state = artifact_completed
  • task_block_id
• AuditBlock
  • subject_ref
  • evidence_ref
  • reviewer_class
  • recommended: review_state, reconciliation_status, task_ref

Recommended shared task_state ladder:

• task_declared
• task_started
• task_in_progress
• artifact_expected
• artifact_completed
• task_closed

Keep reconciled on the AuditBlock, not the TaskBlock.

Default allowed transitions:

• task_declared -> task_started
• task_started -> task_in_progress
• task_started -> artifact_expected
• task_in_progress -> artifact_expected
• task_in_progress -> artifact_completed
• artifact_expected -> artifact_completed
• artifact_completed -> task_closed

Reasonable shortcuts:

• task_declared -> artifact_expected
• task_started -> artifact_completed

Runnable example:

py -3 scripts/examples/task_artifact_audit_flow.py docs/EXACT_CLOSURE_PATENT_SUPPORTING_ARTIFACT_MAP.md --api-key <sbn_live_key> --task-ref task-42 --target-project-id <target_sbn_project_uuid>

It writes the latest builder report to:

• artifacts/examples/task_artifact_audit_latest.json

That example:

• creates a direct TaskBlock
• seals the artifact through SnapChore
• promotes into SBN
• reads back the receipt anchor
• emits an AuditBlock tied to that receipt
• returns a normalized receipt_anchor payload for downstream app use
• if the public receipt catalogue has to degrade malformed rows in place, ops can inspect /health/degraded for aggregator.receipt_summary
• polls queue state long enough to emit a realistic proof_ladder.final_status and queue summary for builder reports

The shared contract for this lane lives in:

• docs/SBN_TASK_ARTIFACT_AUDIT_FLOW.md

Builder helper imports:

from sbn import (
    validate_finance_block_metadata,
    validate_task_block_metadata,
    validate_task_state_transition,
    validate_audit_block_metadata,
    validate_audit_review_transition,
)

validate_task_block_metadata(
    {"task_ref": "task-42", "task_state": "artifact_completed"},
    previous_state="task_started",
)
validate_audit_block_metadata(
    {
        "subject_ref": "task-42",
        "evidence_ref": "receipt:abc123",
        "reviewer_class": "ops",
        "review_state": "reconciled",
    },
    previous_state="review_in_progress",
)
validate_finance_block_metadata(
    {
        "lifecycle_state": "settlement_observed",
        "instrument_ref": "payable-100",
        "position_ref": "position-100",
        "counterparty_ref": "merchant-1",
        "financial_interop": {
            "standard": "iso20022",
            "business_domain": "payments",
        },
    },
    previous_state="obligation_attached",
)

If you want the reusable cross-project smoke for operators, run:

py -3 scripts/smoke/snapchore_target_project_smoke.py --api-key <sbn_live_key> --target-project-id <target_sbn_project_uuid> --service-api-key <service_client_secret>

Raw runtime states collapse into the normalized ladder like this:

• sealed_only / local_only -> sealed_local
• promotion_requested / submitted_to_sbn -> submitted
• handoff_prepared / queued_for_attestation -> queued_for_attestation
• any result carrying receipt_hash or receipt_id -> attested

For receipt linkage, prefer the shared resolver:

receipt_anchor = client.snapchore.resolve_receipt_anchor(
    handoff=handoff,
    promotion_result=promotion,
    block_detail=block_detail,
    receipt_lookup=receipt_lookup,
    target_project_id="your-project-id",
)
print(receipt_anchor["evidence_ref"])

If you want a tiny runnable ladder example instead of copying snippets, use:

py -3 scripts/examples/snapchore_proof_ladder.py docs/EXACT_CLOSURE_PATENT_SUPPORTING_ARTIFACT_MAP.md --api-key <sbn_live_key> --promote --frontier-id artifact.review

It writes:

• artifacts/examples/snapchore_proof_ladder_latest.json

If you want the reusable post-deploy smoke for this same handoff lane, run:

py -3 scripts/smoke/snapchore_attestation_smoke.py --api-key <sbn_live_key>

It writes:

• artifacts/smoke/snapchore_attestation_latest.json

and proves the normalized ladder all the way through:

• sealed_local
• submitted / queued_for_attestation
• attested

End-to-end artifact attestation example

from sbn import SbnClient, artifact_descriptor_from_file

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

artifact = artifact_descriptor_from_file(
    "docs/EXACT_CLOSURE_PATENT_SUPPORTING_ARTIFACT_MAP.md",
    artifact_uri="repo://docs/EXACT_CLOSURE_PATENT_SUPPORTING_ARTIFACT_MAP.md",
)

block = client.blocks.create(
    domain="audit.core",
    payload={
        "type": "AuditBlock",
        "domain": "audit.core",
        "metadata": {
            "review_state": "reconciled",
            "subject_ref": "exact-closure-milestone",
            "evidence_ref": "artifact:exact-closure-map",
            "reviewer_class": "ops",
        },
        "artifact": artifact,
        "claim": {
            "milestone_ref": "patent-support-map-ready",
            "summary": "supporting artifact map assembled for counsel handoff",
        },
    },
)

detail = client.blocks.get(block["data"]["id"])
receipt_hash = detail.get("hash") or detail.get("canonical_proof_hash")

if receipt_hash:
    receipt = client.receipts.get_by_hash(
        receipt_hash,
        project_id="your-project-id",
    )
    print(receipt["items"][0]["native_receipt"]["subject_ref"])

Hello World: SBN

This is the smallest native SBN flow using the agnostic surface wrapper:

1. open an origin surface
2. emit one event
3. seal canonical local state
4. close the stream
5. inspect the resulting proof/receipt surfaces

from sbn import SbnClient

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

with client.open_surface(
    worker_id="hello-agent",
    surface_family="hello.world",
    surface_id="hello-world-001",
    metadata={"frontier_hint": "hello.world"},
) as surface:
    surface.pulse(
        "hello_event",
        score=1.0,
        actions=[{"action": "hello_event", "score": 1.0}],
        message="Hello, SBN",
    )

    seal = surface.seal()
    print("anchor hash:", seal.anchor_hash)

close_receipt = surface.receipt
print("canonical proof hash:", close_receipt.canonical_proof_hash)
print("canonical view:", close_receipt.canonical_view)
print("provenance view:", close_receipt.provenance_view)

Important:

• the native scalar law stays r = Y / X, g = r / c_max
• frontier contracts declare what Y, X, and c_max mean for that ecology
• X is customizable
• timed generation windows are a strong live-telemetry default, not a universal framework rule

Use this path when you want:

• canonical object surfaces
• branch and successor behavior later
• proof-bearing close/seal lifecycle
• receipts as first-class network outputs

pgdag-py wrapper

Use PGDagPyClient when you want to talk directly to the services/pgdag_py HTTP wrapper instead of embedding pgdag in-process.

from sbn import PGDagPyClient

client = PGDagPyClient(base_url="http://localhost:8083")
health = client.health_live()

result = client.run(
    template={
        "name": "kinetic_integrity",
        "version": "1.0",
        "domain": "kinetic.session.integrity.v1",
        "nodes": [
            {
                "id": "integrity_eval",
                "label": "Integrity eval",
                "metadata": {"builtin": "pass_inputs"},
            }
        ],
        "edges": [],
    },
    inputs={"session_id": "sess-42"},
    actor="tower-pgdag-py",
)

print(health["status"])
print(result["slot_id"], result["slot_receipt_id"])

When the pgdag-py service itself has SBN_BASE_URL plus valid SBN credentials, that run publishes the standard open -> BitBlock -> seal -> close lifecycle through SBN.

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. This is the preferred developer surface when:

• your app emits event pulses over time
• generation windows matter
• you want progressive seal() boundaries
• you want to avoid repeated auth/lifecycle mistakes across integrations

Mental model:

• one open slot = one live operational window
• BitBlocks capture what happened during that window
• seal() gives you progressive proof anchors while work continues
• close() gives you the final slot-summary receipt
• explicit extra attestation is optional and separate

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"},
        operation_id="tower-outbox-123",
        idempotency_key="sha256:stable-provider-command",
    )
    seal = stream.seal()
    print(seal.anchor_hash, seal.anchor_type)
    print(seal.proof_chain)
    print(seal.operation_refs)

    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
print(close_receipt.operation_id)
print(close_receipt.idempotency_key)
print(close_receipt.operation_refs)
print(close_receipt.canonical_proof_hash)
print(close_receipt.native_proof_chain)
print(close_receipt.lifecycle_contract)
print(close_receipt.canonical_view)
print(close_receipt.provenance_view)

Operation identity is evidence correlation only. It does not prove that SBN executed an external side effect or that the external provider enforces the idempotency key. Use provider receipt retrieval and verification before reconciling an uncertain action as successful.

Finance example

Use a FinanceBlock when the app wants to record a financial lifecycle state, not when it wants SBN to settle or custody value.

from sbn import SbnClient

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

finance_block = client.blocks.create(
    domain="finance.core",
    payload={
        "type": "FinanceBlock",
        "domain": "finance.core",
        "metadata": {
            "lifecycle_state": "settlement_observed",
            "instrument_ref": "payable-100",
            "external_settlement_ref": "sonicpay-8841",
            "observed_at": "2026-06-07T10:30:00Z",
        },
    },
)

detail = client.blocks.get(finance_block["data"]["id"])
print(detail["block_type"])  # finance
print(detail["block_type_profile"]["scalar_emphasis"])  # muted
print(detail["scalar_gec"])  # None on public read surfaces

Audit example

Use an AuditBlock when Ops, NUMA, or an SBN cadence process wants to record review or reconciliation state.

from sbn import SbnClient

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

audit_block = client.blocks.create(
    domain="audit.core",
    payload={
        "type": "AuditBlock",
        "domain": "audit.core",
        "metadata": {
            "review_state": "reconciled",
            "subject_ref": "finance-surface-demo-1",
            "evidence_ref": "receipt:abc123",
            "reviewer_class": "ops",
            "reconciliation_status": "reconciled",
            "reviewed_at": "2026-06-07T10:45:00Z",
        },
    },
)

detail = client.blocks.get(audit_block["data"]["id"])
print(detail["block_type"])  # audit
print(detail["block_type_profile"]["authority_posture"])
print(detail["block_type_profile"]["minimal_fields"])

App-oriented slot pattern

One common app pattern is:

1. submit several labor blocks into an open slot over a work window
2. submit one finance block describing the resulting financial state transition
3. close the slot

from sbn import SbnClient

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

with client.open_slot(
    worker_id="delivery-agent",
    task_type="delivery.window",
    domain="gig.delivery",
    generation_interval=1800,
) as stream:
    for delivery_id in ("d1", "d2", "d3", "d4"):
        client.blocks.submit(
            slot_id=stream.slot_id,
            domain="gig.delivery",
            payload={
                "type": "LaborBlock",
                "domain": "gig.delivery",
                "metadata": {
                    "delivery_id": delivery_id,
                    "completion_state": "completed",
                    "window_ref": "delivery-window-2026-06-07",
                },
            },
        )

    client.blocks.submit(
        slot_id=stream.slot_id,
        domain="finance.core",
        payload={
            "type": "FinanceBlock",
            "domain": "finance.core",
            "metadata": {
                "lifecycle_state": "settlement_observed",
                "instrument_ref": "delivery-window-2026-06-07",
                "external_settlement_ref": "sonicpay-cashout-1",
                "observed_amount": 100,
            },
        },
    )
close_receipt = stream.receipt
print(close_receipt.canonical_proof_hash)

This keeps the contract clean:

• the app decides when the labor window is complete
• the app decides when the financial state transition was observed
• SBN attests the lifecycle and groups the resulting blocks under one slot close

Paired finance -> audit follow-up

Use this when a finance state later needs authoritative review.

from sbn import SbnClient

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

finance_block = client.blocks.create(
    domain="finance.core",
    payload={
        "type": "FinanceBlock",
        "domain": "finance.core",
        "metadata": {
            "lifecycle_state": "settlement_observed",
            "instrument_ref": "delivery-window-2026-06-07",
            "external_settlement_ref": "sonicpay-cashout-1",
        },
    },
)

audit_block = client.blocks.create(
    domain="audit.core",
    payload={
        "type": "AuditBlock",
        "domain": "audit.core",
        "metadata": {
            "review_state": "reconciled",
            "subject_ref": finance_block["data"]["id"],
            "evidence_ref": "receipt:abc123",
            "reviewer_class": "ops",
            "reconciliation_status": "reconciled",
        },
    },
)

print(audit_block["data"]["id"])

Receipt lookup after finance -> audit

Use this when you want to inspect the read surface produced by the finance and audit flow.

from sbn import SbnClient

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

lookup = client.receipts.get_by_hash(
    "receipt-hash",
    project_id="project-id",
)
item = (lookup.get("items") or [{}])[0]

print(item.get("block_type_profile", {}).get("block_type"))
print(item.get("trust_summary", {}).get("trust_class"))
print(item.get("cdna_state", {}).get("closure_regime"))
print(item.get("mutation_summary", {}).get("decision_divergence"))

First law example from the same base

law is still maturing, but this is the intended follow-up shape when a finance or audit state needs an explicit obligation or constraint record.

from sbn import SbnClient

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

law_block = client.blocks.create(
    domain="law.core",
    payload={
        "type": "LawBlock",
        "domain": "law.core",
        "metadata": {
            "law_state": "bound",
            "subject_ref": "finance-or-audit-subject-id",
            "constraint_ref": "derivative-exercise-window-1",
            "issuer_class": "compliance_service",
            "obligation_code": "exercise_window_active",
            "binding_scope": "derivative_position",
        },
    },
)

print(law_block["data"]["id"])

Law lifecycle proposal

The smallest honest lifecycle for law right now is:

declared -> bound -> active -> satisfied | expired

Use it as:

• declared for a newly stated rule or obligation
• bound when attached to a subject or position
• active while in force
• satisfied when fulfilled
• expired when no longer in force

Narrative pattern: labor -> finance -> audit -> law

One clean app-level narrative is:

1. submit labor blocks for the productive window
2. submit a FinanceBlock for the observed value-state transition
3. submit an AuditBlock for reconciliation
4. submit a LawBlock for the resulting obligation or constraint

Derivative-specific example: finance -> audit -> law

from sbn import SbnClient

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

finance_block = client.blocks.create(
    domain="finance.core",
    payload={
        "type": "FinanceBlock",
        "domain": "finance.core",
        "metadata": {
            "lifecycle_state": "obligation_attached",
            "instrument_ref": "eth-call-option-2026-q3",
            "position_ref": "position-eth-call-1",
            "counterparty_ref": "desk-a",
        },
    },
)

audit_block = client.blocks.create(
    domain="audit.core",
    payload={
        "type": "AuditBlock",
        "domain": "audit.core",
        "metadata": {
            "review_state": "reconciled",
            "subject_ref": finance_block["data"]["id"],
            "evidence_ref": "receipt:derivative-book-1",
            "reviewer_class": "ops",
            "reconciliation_status": "reconciled",
        },
    },
)

law_block = client.blocks.create(
    domain="law.core",
    payload={
        "type": "LawBlock",
        "domain": "law.core",
        "metadata": {
            "law_state": "active",
            "subject_ref": finance_block["data"]["id"],
            "constraint_ref": "exercise-window-2026-q3",
            "issuer_class": "compliance_service",
            "obligation_code": "cash_settlement_if_exercised",
            "binding_scope": "derivative_position",
            "effective_at": "2026-07-01T00:00:00Z",
            "expires_at": "2026-09-30T23:59:59Z",
        },
    },
)

print(audit_block["data"]["id"], law_block["data"]["id"])

Shared derivative payload convention

Prefer this shared reference vocabulary across the derivative lifecycle:

• instrument_ref
• position_ref
• counterparty_ref
• constraint_ref
• evidence_ref

Recommended by type:

1. FinanceBlock
   • instrument_ref
   • position_ref
   • counterparty_ref
2. AuditBlock
   • subject_ref
   • evidence_ref
3. LawBlock
   • subject_ref
   • constraint_ref
   • effective_at
   • expires_at

Receipt surfaces

Gateway receipts now follow the same three-surface model used across block and receipt reads:

• canonical_view - default proof truth
• provenance_view - source ids and compatibility lineage
• forensic_view - raw receipt body and transport-era detail when explicitly needed

As a rule, SDK examples should read canonical view first and inspect forensic view only when debugging or auditing.

Read versions

Gateway read methods now support an explicit compatibility seam:

• read_version=2 (default) returns the canonical-first contract
• read_version=1 restores older top-level compatibility aliases for legacy consumers

Treat read_version=2 as the stable/public contract for new integrations. Use read_version=1 only when migrating older consumers or when internal ops/debug tooling still needs the legacy top-level aliases.

Examples:

block = client.gateway.fetch_block(block_id, read_version=2)
receipt = client.gateway.fetch_receipt(receipt_id, read_version=2)
legacy_receipt = client.gateway.fetch_receipt(receipt_id, read_version=1)

Stream profile examples

Use stream_profile when you want the slot wrapper to stamp a deterministic frontier contract and canonical block-type posture without hand-building the entire contract object first.

labor_v1

from sbn import SbnClient

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

with client.open_slot(
    worker_id="grid-worker-1",
    task_type="gig.delivery.window",
    domain="gig.delivery",
    generation_interval=30,
    stream_profile="labor_v1",
    metadata={
        "frontier_hint": "grid.atomic_rentals",
        "proof_intent": "delivery_window",
    },
) as stream:
    stream.yield_pulse(
        "delivery_complete",
        score=1.0,
        actions=[{"action": "delivery_complete", "score": 1.0}],
        events={"friendly": 1},
        state_hint="delivery_complete",
        delivery_state="completed",
    )
    stream.burden_pulse(
        "late_delivery",
        burden_hint=0.35,
        score=0.2,
        actions=[{"action": "late_delivery", "score": 0.2}],
        events={"hostile": 1},
        severity="warning",
        state_hint="late_delivery",
        delivery_state="late",
    )
    seal = stream.seal()

final = stream.finalize(
    request_attestation=True,
    snap_hash=str(seal.anchor_hash or ""),
    frontier_id="grid.atomic_rentals",
    metadata={"type": "LaborBlock", "domain": "gig.delivery"},
)

Expected posture:

• composed block resolves as labor
• public block read exposes scalar_gec
• receipt and block both surface bb_contribution_summary

ops_audit_v1

with client.open_slot(
    worker_id="ops-review-1",
    task_type="ops.audit.review.window",
    domain="audit.core",
    generation_interval=30,
    stream_profile="ops_audit_v1",
    metadata={
        "frontier_hint": "ops.audit.review",
        "proof_intent": "scheduled_review",
    },
) as stream:
    stream.yield_pulse(
        "cadence_snapshot",
        score=1.0,
        actions=[{"action": "cadence_snapshot", "score": 1.0}],
        events={"neutral": 1},
        state_hint="cadence_snapshot",
        review_state="cadence_snapshot",
        subject_ref="ops.audit.review",
        evidence_ref="frontier:ops.audit.review",
        reviewer_class="sbn_timer",
    )
    stream.burden_pulse(
        "review_in_progress",
        burden_hint=0.5,
        score=0.2,
        actions=[{"action": "review_in_progress", "score": 0.2}],
        events={"hostile": 1},
        severity="warning",
        state_hint="review_in_progress",
        review_state="review_in_progress",
        subject_ref="ops.audit.review",
        evidence_ref="frontier:ops.audit.review:delta",
        reviewer_class="ops",
    )

Expected posture:

• composed block resolves as audit
• block_type_profile.grouping_pattern = "scheduled_or_exception_review"
• block_type_profile.allowed_emitters includes ops, numa, and sbn_timer

finance_state_v1

with client.open_slot(
    worker_id="finance-state-1",
    task_type="finance.state.window",
    domain="finance.core",
    generation_interval=30,
    stream_profile="finance_state_v1",
    metadata={
        "frontier_hint": "finance.state.window",
        "proof_intent": "financial_lifecycle_state",
    },
) as stream:
    stream.yield_pulse(
        "allocation",
        score=1.0,
        actions=[{"action": "allocation", "score": 1.0}],
        events={"friendly": 1},
        state_hint="allocation",
        lifecycle_state="allocation",
        instrument_ref="position-1:instrument",
        position_ref="position-1",
        counterparty_ref="venue.alpha",
    )
    stream.yield_pulse(
        "obligation_attached",
        score=0.9,
        actions=[{"action": "obligation_attached", "score": 0.9}],
        events={"friendly": 1},
        state_hint="obligation_attached",
        lifecycle_state="obligation_attached",
        instrument_ref="position-1:instrument",
        position_ref="position-1",
        counterparty_ref="venue.alpha",
        obligation_ref="position-1:obligation",
    )
    stream.burden_pulse(
        "settlement_observed",
        burden_hint=0.25,
        score=0.4,
        actions=[{"action": "settlement_observed", "score": 0.4}],
        events={"hostile": 1},
        severity="warning",
        state_hint="settlement_observed",
        lifecycle_state="settlement_observed",
        instrument_ref="position-1:instrument",
        position_ref="position-1",
        counterparty_ref="venue.alpha",
        settlement_ref="position-1:settlement",
        external_system_ref="sonicpay",
    )

Expected posture:

• composed block resolves as finance
• public block read mutes scalar_gec
• block_type_profile.settlement_posture = "observe_reference_reconcile"
• block_type_profile.external_system_role = "adjacent_non_settlement"

Metric taxonomy

SBN now treats block-side quantitative state in four classes:

• local descriptive metrics
  • yield, entropy, compression, trust, reflex, local metrics.gec
• canonical reduction inputs
  • lifecycle-supporting evidence used to derive carrier state
• canonical lifecycle state
  • carrier_state, carrier_level, branch/surface/root state
• parent-facing widening context
  • widening and related sufficiency context

That means not every numeric field on a block should be read as canonical lifecycle truth. Local metrics describe; carrier state carries; widening qualifies upstream sufficiency.

Shared GEC operation contract

When you want SBN to derive default y and x from the same operation metadata everywhere, pass a shared contract into the slot wrapper.

If you want a more builder-friendly SmartBlock-native path, define the frontier contract first and hand the result into either open_slot(...) or open_surface(...). Atlas can later catalogue the same contract, but the helper itself is not Atlas-owned.

from sbn import FrontierContractBuilder, SbnClient

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

contract = (
    FrontierContractBuilder(
        frontier="assistant",
        gec_domain="ai",
        app_name="assistant-demo",
    )
    .archetype("stream_efficiency")
    .native_defaults()
    .canonical_cost_profile("ai_assistant_v1")
    .actions("prompt_submitted", "tool_call_completed", "response_delivered", weights={
        "prompt_submitted": 0.25,
        "tool_call_completed": 0.6,
        "response_delivered": 1.0,
    })
    .events("assistant_turn", weights={
        "friendly": 1.0,
        "neutral": 0.0,
        "hostile": -1.0,
    })
    .strategy("weighted_mean")
    .yield_rule("completed_turns", unit="assistant_turn")
    .cost_rule(
        "blended",
        composition_rule="weighted_sum_v1",
        required_components=["compute"],
        optional_components=["time", "resource", "intervention", "reversal_risk"],
    )
    .cmax_rule("frontier_registry")
    .interpretation(
        entropy_mode="structural_entropy_v1",
        compression_mode="canonical_compression_v1",
        trust_mode="provenance_and_coherence",
        reflex_mode="trajectory_delta",
    )
    .build()
)

with client.open_surface(
    worker_id="assistant-demo",
    surface_family="ai.assistant",
    surface_id="conversation-001",
    gec_contract=contract,
) as surface:
    surface.event({"event_type": "prompt_submitted", "score": 0.25})

from sbn import SbnClient, GecOperationContract

contract = GecOperationContract(
    app_name="assistant-demo",
    gec_domain="ai",
    frontier="assistant",
    frontier_archetype="stream_efficiency",
    phase="response_generation",
    x_mode="elapsed_time",
    action_weight_map={"response_delivered": 1.0},
    interaction_weight_map={"tool_call_completed": 0.6},
    event_polarity_weights={"friendly": 1.0, "neutral": 0.1, "hostile": -1.0},
    native_metric_profile="native_gec_v1",
    yield_mode="completed_turns",
    yield_unit="assistant_turn",
    cost_mode="blended",
    cost_composition_rule="weighted_sum_v1",
    cost_components={
        "time": 0.5,
        "compute": 1.0,
        "intervention": 0.75,
        "reversal_risk": 0.25,
    },
    cost_required_components=["compute"],
    cost_optional_components=["time", "resource", "intervention", "reversal_risk"],
    c_max_mode="frontier_registry",
    entropy_mode="structural_entropy_v1",
    compression_mode="canonical_compression_v1",
    trust_mode="provenance_and_coherence",
    reflex_mode="trajectory_delta",
    native_notes={"canonical_cost_profile": "ai_assistant_v1"},
)

with client.open_slot(
    worker_id="assistant-demo",
    task_type="ai.assistant.response_generation",
    domain="ai.assistant",
    generation_interval=60,
    gec_contract=contract,
) as stream:
    ...

When a live telemetry frontier uses generation_interval as its base X coordinate, read that as a frontier declaration:

• X = elapsed generation window
• Y = persisted useful output inside that window

That keeps interval comparisons stable and easy to interpret. Other frontiers may instead declare another lawful extensive X basis, such as labor, compute, tokens, or capital.

You can also compose an observed burden packet against the contract directly:

observed_cost = contract.compose_cost_observation(
    {
        "time": 2.0,
        "compute": 1.5,
        "intervention": 0.5,
    }
)

print(observed_cost["rule"])   # weighted_sum_v1
print(observed_cost["total"])  # 2.875
print(observed_cost["valid"])  # True

For a generic binary approval frontier, use the public-safe fixed-burden profile:

approval_contract = client.define_frontier(
    frontier="approval",
    gec_domain="workflow",
    frontier_archetype="binary_phase",
    apply_native_defaults=True,
    canonical_cost_profile="workflow_approval_v1",
)

Agnostic surface wrapper

When you want developers to think in terms of a canonical object surface instead of raw slot metadata, use open_surface(...).

This keeps the same transport under the hood, but it writes the native surface/branch metadata consistently for you:

• origin surfaces omit fractal_parent_block_id
• branch surfaces set fractal_parent_block_id
• branch surfaces default to fractal_auto_rollup=True
• surface_family becomes the default task_type / domain

Mental model:

• slots stay the operational transport
• surfaces give that transport a native object identity
• origin/branch/successor structure becomes explicit
• seal() and close() still preserve the same proof-bearing lifecycle underneath

from sbn import SbnClient

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

with client.open_surface(
    worker_id="grid-agent",
    surface_family="grid.atomic_rentals",
    surface_id="rental-001",
) as surface:
    surface.event({"event_type": "rental_created", "score": 1.0})
    root_seal = surface.seal()

with client.open_surface(
    worker_id="grid-agent",
    surface_family="grid.atomic_rentals",
    surface_id="rental-001.active",
    parent_block_id=root_seal.smartblock_id,
    branch_kind="active_rental",
) as branch:
    branch.event({"event_type": "vehicle_checkout", "score": 0.92, "weight": 1.0})
    branch_seal = branch.seal()

This is intentionally transport-agnostic:

• slots remain the operational wrapper
• BitBlocks remain the atomic evidence path
• successor surfaces remain the canonical continuation target
• public attestation still belongs on canonical surfaces, not raw branch snapshots

cDNA mutation lineage

When your runtime needs explicit mutation or adaptation history, use the public sbn.cdna helper path instead of reaching into the internal pgdag package structure.

from sbn.cdna import (
    EvolutionaryMutator,
    LineageManager,
    MutationStrategy,
    MutationTrigger,
)

trigger = MutationTrigger(
    strategy=MutationStrategy.ENTROPY_YIELD,
    yield_score=0.31,
    entropy_score=0.82,
    reasoning="low yield under chaotic conditions",
)

mutator = EvolutionaryMutator(
    yield_threshold=0.5,
    entropy_threshold=0.65,
)

# Your product supplies the concrete chain manager / proof writer.
lineage = LineageManager(
    chain_manager=my_chain_manager,
    writer=my_proof_writer,
)

record = lineage.seed("agent-42", "watcher:performance", spec_version="0.1.0")
proof = lineage.check_and_mutate(
    "agent-42",
    yield_score=trigger.yield_score,
    entropy_score=trigger.entropy_score,
    reasoning=trigger.reasoning,
)

if proof is not None:
    print(proof.before_cdna, "->", proof.after_cdna)

This maps directly to the existing core lifecycle:

• agent JWTs can carry a cdna claim
• SmartBlock creation injects c_dna
• cDNA may mutate once pre-sign during metrics computation
• lineage material is preserved in downstream BitBlock and fusion state

The same contract can be rendered for Atlas registration with:

sbn frontier workflow-spec \
  --workflow pushup_session \
  --app-name pt_web \
  --gec-domain kinetic \
  --frontier pushup \
  --constraint-tag amrap \
  --phase execution

Close and finalize semantics

The wrapper now has three useful boundaries:

• seal() - compose the current pulse into a proof-bearing anchor while the slot stays open
• close() - close the slot and persist the final slot-summary receipt while preserving native lifecycle context
• 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.

The wrapper remains operational only. The native lifecycle truth still lives underneath it, and the close/seal results preserve that structure:

• close().native_proof_chain
• close().successor_surface_ids
• close().lifecycle_contract
• seal().proof_chain

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. The close-time receipt remains available even if the explicit attestation path fails. In that case final.attestation_error explains the failure while final.close still carries the successful slot close result.

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.close.native_proof_chain)
print(final.close.lifecycle_contract)
print(final.attestation)
print(final.attestation_error)

If the explicit attestation is required to hard-fail the call, opt into strict mode:

final = stream.finalize(
    request_attestation=True,
    snap_hash=step_anchor.anchor_hash,
    frontier_id="grid.atomic_rentals.streaming",
    strict_attestation=True,
)

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

Atlas frontier cards from the CLI

Atlas also exposes the card-first workflow directly from the CLI, including round-tripping card JSON for external authoring tools such as RI.

# inspect the private catalogue
sbn atlas cards list
sbn atlas cards show kinetic.pushup.amrap.v1

# register from flags
sbn atlas cards register \
  --domain ri.kinetic.session.integrity \
  --app-family ri.ecology \
  --family-class digital_information \
  --authored-by "Mycal Brooks" \
  --label "RI Kinetic Session Integrity" \
  --workflow session_integrity \
  --pulse-type pgdag

# import/export raw card JSON
sbn atlas cards import frontier-card.json
sbn atlas cards import frontier-card.json --supersede-existing --change-note "RI refinement"
sbn atlas cards export ri.kinetic.session.integrity --out ri-card-export.json

# attach a card to an app workflow and optionally bind an existing key
sbn atlas cards attach ri.kinetic.session.integrity \
  --app-id <app-id> \
  --workflow session_integrity \
  --key-id <existing-api-key-id>

cards import accepts either:

• an SBN-exported frontier card read model from cards export
• or a raw RI-authored JSON card payload centered on frontier_card

That makes it practical to author cards in another system, then bring them into SBN for inspection, testing, simulation, and later workflow binding.

A neutral starter template is available at:

• ATLAS_FRONTIER_CARD_TEMPLATE.json

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"