Metadata-Version: 2.4
Name: sbn-sdk
Version: 0.12.11
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)

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

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

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

