Nukez

Docs · NukezAgent

NukezAgent

Examples

Delegation examples

Run the hosted Agent flow end to end.

These examples mirror the Agent usage guide and notebook: set up signed requests, activate service access, delegate writes, recall bytes, verify evidence, and inspect operational endpoints without placing private keys in the hosted service.

signed_headers

Builds the canonical request signature and X-Nukez request headers.

complete_if_signing

Finishes write operations by locally signing returned envelopes.

solana_transfer

Represents client-side payment execution. The agent never submits this transfer.

Example setup

Request signature helper

Prepare the exact request-signing primitive used by the rest of the examples.

python
pip install httpx pynacl base58 solders import base58, hashlib, httpx, json, os, timefrom nacl.signing import SigningKey FRONT_DOOR = "https://agent.nukez.xyz"_sk = SigningKey(bytes(json.load(open("/path/to/keypair.json"))[:32]))PUBKEY = base58.b58encode(_sk.verify_key.encode()).decode() def canon_body(body: dict | None) -> bytes:    if body is None:        return b""    return json.dumps(body, separators=(",", ":"), sort_keys=True).encode() def signed_headers(method: str, path: str, body: dict | None = None) -> dict:    nonce = os.urandom(32).hex()    timestamp = str(int(time.time()))    digest = hashlib.sha256(canon_body(body)).hexdigest()    msg = "\n".join([        "nukez-request:v1", f"method={method}", f"path={path}",        f"identity={PUBKEY}", f"nonce={nonce}", f"timestamp={timestamp}",        f"body_sha256={digest}",    ])    sig = base58.b58encode(_sk.sign(msg.encode()).signature).decode()    return {        "X-Nukez-Identity": PUBKEY,        "X-Nukez-Nonce": nonce,        "X-Nukez-Timestamp": timestamp,        "X-Nukez-Signature": sig,    }

Code notes

The canonical notebook starts with a thin direct client: httpx for transport, PyNaCl/base58 for request signing, and solders for the optional local Solana transfer helper. Every private endpoint uses the same signed_headers helper.

Unsigned endpoints include /health, /v1/card, /.well-known/*, /v1/service/request, /v1/provision/challenge, and /v1/provision/verify.

Example setup

Signed transport and envelope completion

Keep HTTP transport and operation envelope signing in one reusable client layer.

python
def agent_get(path: str):    r = httpx.get(FRONT_DOOR + path, headers=signed_headers("GET", path), timeout=60)    r.raise_for_status()    return r.json() def agent_post(path: str, body: dict | None = None, extra_headers: dict | None = None):    headers = signed_headers("POST", path, body)    if extra_headers:        headers.update(extra_headers)    r = httpx.post(FRONT_DOOR + path, json=body, headers=headers, timeout=120)    r.raise_for_status()    return r.json() def sign_envelopes(envelopes: list[dict]) -> list[str]:    signatures = []    for envelope in envelopes:        payload = envelope.get("envelope_json") or json.dumps(            envelope["envelope"], separators=(",", ":"), sort_keys=True        )        sig = base58.b58encode(_sk.sign(payload.encode()).signature).decode()        signatures.append(sig)    return signatures def complete_if_signing(result: dict) -> dict:    while result.get("status") == "signing_needed":        result = agent_post("/v1/delegate", {            "signing_request_id": result["signing_request_id"],            "signatures": sign_envelopes(result["envelopes"]),        })    return result

Code notes

These helpers are the common base for the rest of the notebook. They sign each request, preserve extra headers like X402-TX, and loop through signing_needed responses by signing returned envelopes locally.

Request signatures identify the caller. Envelope signatures authorize a specific storage operation.

Subscription

Quote, pay, confirm

Activate the owner once before storage tasks are delegated.

python
health = httpx.get(FRONT_DOOR + "/health").json()card = httpx.get(FRONT_DOOR + "/v1/card").json() quote = httpx.get(FRONT_DOOR + "/v1/service/request").json()sol = next(o for o in quote["payment_options"] if o["network"] == "solana-mainnet") tx_sig = solana_transfer(sol["pay_to_address"], int(sol["amount"])) confirm = agent_post(    "/v1/service/confirm",    extra_headers={"X402-TX": tx_sig},) status = agent_get("/v1/status")assert status["identity"] == PUBKEY

Code notes

New owners request a subscription quote, choose a payment rail, execute payment with their own wallet or helper, and confirm the transaction with a signed request. The Agent guide uses a Solana helper in the notebook, but the agent itself does not move funds.

Returning active owners skip the quote/pay/confirm path and go directly to signed delegate calls.

Write path

Persist and replace files

Exercise the storage flow and make overwrite intent explicit.

python
filename = f"agent-demo-{int(time.time())}.json"payload_b64 = "eyJtZXNzYWdlIjoiSGVsbG8gTnVrZXpBZ2VudCJ9" stored = complete_if_signing(agent_post("/v1/delegate", {    "task": f"Store this file named {filename}. Bytes (base64): {payload_b64}"})) repeat = complete_if_signing(agent_post("/v1/delegate", {    "task": f"Store this file named {filename}. Bytes (base64): {payload_b64}"})) conflict = agent_post("/v1/delegate", {    "task": f"Store this file named {filename}. Bytes (base64): eyJkaWZmZXJlbnQiOnRydWV9"}) replace = complete_if_signing(agent_post("/v1/delegate", {    "task": f"Replace the file named {filename}. Bytes (base64): {payload_b64}"}))

Code notes

Net-new writes normally return signing_needed. Same-content repeats may no-op, while different content for an existing filename is refused unless the task explicitly says overwrite or replace.

The term update is intentionally treated as ambiguous and may trigger a clarification rather than a silent overwrite.

Read path

Recall, verify, list

Retrieve stored bytes, inspect verification evidence, and list owner files.

python
recall = agent_post("/v1/delegate", {    "task": f"Recall the file named {filename}"}) if recall.get("download_token"):    r = httpx.get(FRONT_DOOR + f"/v1/download/{recall['download_token']}", timeout=60)    r.raise_for_status()    recalled_bytes = r.contentelse:    recalled_bytes = recall["payload"].encode() verified = agent_post("/v1/delegate", {    "task": f"Verify storage for {filename}"}) assert verified["summary"].startswith(("Verified", "Verification failed", "Attestation in flight"))print(verified.get("content_hash"), verified.get("merkle_root"), verified.get("tx_signature")) files = agent_post("/v1/delegate", {"task": "List my stored files"})

Code notes

The read flow stays inside signed delegation. The agent can return inline payloads for small files or a download_token for a one-time retrieval. Verification returns concise text plus structured evidence fields.

A tx_signature of null can be normal if attestation has not finalized yet.

Operations

Expand capacity and inspect portal data

Round out the production workflow with capacity and UI-supporting endpoints.

python
expansion = agent_get("/v1/service/expand?units=2")option = next(o for o in expansion["payment_options"] if o["network"] == "solana-mainnet") expand_tx = solana_transfer(option["pay_to_address"], int(option["amount"])) expanded = agent_post(    "/v1/service/expand/confirm",    {"units": 2, "tx_signature": expand_tx},    extra_headers={"X402-TX": expand_tx},) portal_files = agent_get("/v1/portal/files")portal_ops = agent_get("/v1/portal/ops-log")portal_usage = agent_get("/v1/portal/usage") receipt_check = agent_post("/v1/delegate", {    "task": f"Verify receipt id {stored.get('receipt_id')}"})

Code notes

Capacity expansion uses a signed quote, external payment, and signed confirmation. Portal endpoints support browser dashboards, while receipt verification can be requested through the agent until direct audit forwarding is complete.

If all lockers are full, the overflow guard should lead the user to purchase additional capacity rather than failing opaquely.

For a product-facing worked example, see the NukezAgent flow.

§ next