Metadata-Version: 2.4
Name: sbn-sdk
Version: 0.12.22
Summary: Python SDK for the SmartBlocks Network — Atlas frontier provisioning, GEC compute, SnapChore integrity, governance, and more.
License: MIT
Keywords: smartblocks,sbn,snapchore,gec,attestation,integrity,governance,workflow,lattice
Author: SmartBlocks Team
Author-email: devrel@smartblocks.network
Requires-Python: >=3.10,<4.0
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Dist: PyJWT (>=2.8)
Requires-Dist: cryptography (>=41.0)
Requires-Dist: httpx (>=0.27.0,<0.28.0)
Requires-Dist: rich (>=13.0)
Requires-Dist: typer (>=0.9)
Project-URL: Documentation, https://smartblocks.network/docs/sdk
Project-URL: Homepage, https://smartblocks.network
Project-URL: Repository, https://github.com/smartblocks-network/sbn-sdk
Description-Content-Type: text/markdown

# 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:

```python
from pgdag import PGDAGRunner, DAGTemplate, DAGNode
```

## Install

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

## Quick start

```python
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)
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="ops@acme.co",
    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.

```python
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)
    print(seal.proof_chain)

    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.native_proof_chain)
print(close_receipt.lifecycle_contract)
print(close_receipt.canonical_view)
print(close_receipt.provenance_view)
```

### 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:

```python
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)
```

### 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.

```python
from sbn import SbnClient, GecOperationContract

contract = GecOperationContract(
    app_name="pt_web",
    gec_domain="kinetic",
    frontier="pushup",
    constraint_tag="amrap",
    phase="execution",
    x_mode="elapsed_time",
    action_weight_map={"rep_complete": 1.0},
    interaction_weight_map={"form_quality": 0.25},
    event_polarity_weights={"friendly": 1.0, "neutral": 0.1, "hostile": -1.0},
)

with client.open_slot(
    worker_id="tower-agents",
    task_type="kinetic.pushup.amrap",
    domain="kinetic.pushup",
    generation_interval=60,
    gec_contract=contract,
) as stream:
    ...
```

The same contract can be rendered for Atlas registration with:

```bash
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 and attest the current pulse 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.

```python
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:

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

## Auth methods

```python
# 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`.

```python
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:

```bash
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.

```python
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:

```bash
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:

```bash
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"
```

