Nukez

Docs · MCP

MCP server

15 tools

Examples

JSON-RPC example

OverviewExamplesTools

Executed MCP client sequence.

These examples mirror the canonical usage guide and the executed MCP notebook: direct JSON-RPC, local signing, external payment, signed storage, verification, memory, provenance, and diff.

Setup

Install and define the direct MCP client

Use the canonical raw JSON-RPC transport and local signer setup from the executed notebook.

python
pip install httpx pynacl base58 solders import json, hashlib, os, time, base64, pathlibimport httpxfrom nacl.signing import SigningKeyimport base58 BASE = "https://nukez-mcp-538908875166.us-central1.run.app/mcp"HEADERS = {"Content-Type": "application/json", "Accept": "application/json, text/event-stream"}KEYPAIR_PATH = pathlib.Path("/path/to/svm_key.json")RPC_URL = "https://mainnet.helius-rpc.com/?api-key=<API_KEY>" _kp_bytes = json.loads(KEYPAIR_PATH.read_text())_sk = SigningKey(bytes(_kp_bytes[:32]))PUBKEY = base58.b58encode(_sk.verify_key.encode()).decode()_req_id = 0 def rpc(method: str, params: dict = None):    global _req_id    _req_id += 1    payload = {"jsonrpc": "2.0", "id": _req_id, "method": method}    if params:        payload["params"] = params    resp = httpx.post(BASE, json=payload, headers=HEADERS, timeout=120)    ct = resp.headers.get("content-type", "")    if "text/event-stream" in ct:        for line in resp.text.splitlines():            if line.startswith("data: "):                data = json.loads(line[6:])                return data.get("result") or data.get("error")    body = resp.json()    return body.get("result") or body.get("error") def call(tool: str, args: dict = None):    params = {"name": tool}    if args:        params["arguments"] = args    result = rpc("tools/call", params)    if result and "content" in result:        texts = [c["text"] for c in result["content"] if c.get("type") == "text"]        if texts:            try:                return json.loads(texts[0])            except json.JSONDecodeError:                return texts[0]    return result

Code notes

This is the canonical setup from the usage guide and executed notebook: install direct-client dependencies, configure the hosted endpoint, load a local keypair, create JSON-RPC helpers, build signed envelopes, and provide an explicit solders-based Solana transfer helper.

The full guide also defines compute_locker_id(), build_envelope(), and solana_transfer(); those helpers are used by the following examples.

Signing

Build envelopes and keep payment local

Canonical JSON envelope signing and solders-based payment execution replace removed PyNukez helper methods.

python
def compute_locker_id(receipt_id: str) -> str:    return "locker_" + hashlib.sha256(receipt_id.encode()).hexdigest()[:12] def build_envelope(receipt_id, method, path, ops=None, body=None, ttl=300):    locker_id = compute_locker_id(receipt_id)    now = int(time.time())    canonical_body = None    if body is not None:        canonical_body = json.dumps(body, separators=(",", ":"), sort_keys=True)        body_hash = hashlib.sha256(canonical_body.encode()).hexdigest()    else:        body_hash = hashlib.sha256(b"").hexdigest()    envelope = {        "v": 1, "locker_id": locker_id, "receipt_id": receipt_id,        "nonce": os.urandom(16).hex(), "iat": now, "exp": now + ttl,        "ops": ops or [], "method": method.upper(), "path": path,        "body_sha256": body_hash, "sig_alg": "ed25519",    }    envelope_json = json.dumps(envelope, separators=(",", ":"), sort_keys=True)    sig = base58.b58encode(_sk.sign(envelope_json.encode()).signature).decode()    env_b64 = base64.urlsafe_b64encode(envelope_json.encode()).decode().rstrip("=")    result = {"headers": {"X-Nukez-Envelope": env_b64, "X-Nukez-Signature": sig}}    if canonical_body is not None:        result["body"] = canonical_body    return result from solders.keypair import Keypair as SoldersKeypairfrom solders.pubkey import Pubkeyfrom solders.system_program import transfer, TransferParamsfrom solders.transaction import Transactionfrom solders.message import Messagefrom solders.hash import Hash _solders_kp = SoldersKeypair.from_bytes(bytes(_kp_bytes[:64])) def solana_transfer(to_address: str, lamports: int) -> str:    bh = httpx.post(RPC_URL, json={        "jsonrpc": "2.0",        "id": 1,        "method": "getLatestBlockhash",        "params": [{"commitment": "finalized"}],    }, timeout=30).json()    blockhash = Hash.from_string(bh["result"]["value"]["blockhash"])    ix = transfer(TransferParams(        from_pubkey=_solders_kp.pubkey(),        to_pubkey=Pubkey.from_string(to_address),        lamports=lamports,    ))    msg = Message.new_with_blockhash([ix], _solders_kp.pubkey(), blockhash)    tx = Transaction.new_unsigned(msg)    tx.sign([_solders_kp], blockhash)    resp = httpx.post(RPC_URL, json={        "jsonrpc": "2.0",        "id": 1,        "method": "sendTransaction",        "params": [            base64.b64encode(bytes(tx)).decode(),            {"encoding": "base64", "skipPreflight": False},        ],    }, timeout=30).json()    if "error" in resp:        raise RuntimeError(f"sendTransaction failed: {resp['error']}")    return resp["result"]

Code notes

Post-provisioning operations require canonical JSON envelopes signed with the local Ed25519 key. The Solana payment helper is explicit client code: it obtains a blockhash, builds a system transfer, signs with solders, and submits the transaction through the configured RPC.

Envelope ops must match the action: locker:provision, locker:write, locker:list, or locker:read.

Provision

Initialize, quote, pay, and activate the locker

Follow the exact two-step provisioning flow after recording the external payment transaction.

python
init = rpc("initialize", {    "protocolVersion": "2025-03-26",    "capabilities": {},    "clientInfo": {"name": "mcp-client", "version": "1.0"},})tools = rpc("tools/list")assert len(tools["tools"]) == 15 status = call("nukez_status")assert status["capabilities"]["payment_rails"] == ["solana-mainnet", "monad-mainnet"] quote = call("nukez_quote", {"units": 1, "provider": "gcs"})sol_opt = next(o for o in quote["payment_options"] if o["pay_asset"] == "SOL")pay_req_id = quote["pay_req_id"] tx_sig = solana_transfer(sol_opt["pay_to_address"], int(sol_opt["amount"]))pay = call("nukez_pay", {"pay_req_id": pay_req_id, "tx_sig": tx_sig, "chain": "solana"}) confirm = call("nukez_provision", {"pay_req_id": pay_req_id, "tx_sig": tx_sig})assert confirm.get("action_required") == "sign_provision_envelope"receipt_id = confirm["receipt_id"]locker_id = confirm["locker_id"] provision_env = build_envelope(    receipt_id, "POST", "/v1/storage/signed_provision",    ops=["locker:provision"], body={"receipt_id": receipt_id},)provision = call("nukez_provision", {"receipt_id": receipt_id, "envelope": provision_env})

Code notes

The notebook starts by initializing the MCP server and confirming exactly 15 tools. It checks pre-setup status, quotes storage, executes a local SOL transfer, records that transaction with nukez_pay, then provisions through the two-step envelope flow.

The expected confirm result includes ok: true, status: payment_confirmed, requires_envelope: true, and envelopes_needed.

Storage

Store inline bytes or staged large files

Use the right byte path for the size and execution environment.

python
body = {"filename": "smoke.txt", "content_type": "text/plain", "ttl_min": 30}env = build_envelope(receipt_id, "POST", f"/v1/lockers/{locker_id}/files", ops=["locker:write"], body=body)store = call("nukez_store", {    "receipt_id": receipt_id,    "envelope": env,    "files": [{"name": "smoke.txt", "data_b64": base64.b64encode(b"smoke test\n").decode()}],}) large_path = pathlib.Path("/path/to/NewConstruction.kml")large_bytes = large_path.read_bytes()large_sha256 = hashlib.sha256(large_bytes).hexdigest()plan = call("nukez_store", {    "receipt_id": receipt_id,    "files": [{"name": large_path.name, "source": "staged", "size_bytes": len(large_bytes), "sha256": large_sha256}],})assert plan.get("action_required") == "stage_file_bytes" start_resp = httpx.post(f"{BASE}/blob/start", json={    "receipt_id": receipt_id,    "filename": large_path.name,    "content_type": "application/vnd.google-earth.kml+xml",    "sha256": large_sha256,    "size_bytes": len(large_bytes),}, timeout=30).json()data_token = start_resp["data_token"]CHUNK_SIZE = 16 * 1024 * 1024parts = [large_bytes[i:i + CHUNK_SIZE] for i in range(0, len(large_bytes), CHUNK_SIZE)] for i, part in enumerate(parts):    is_last = i == len(parts) - 1    part_resp = httpx.put(        f"{BASE}/blob/{data_token}/parts/{i}",        params={"is_last": str(is_last).lower()},        headers={            "Content-Type": "application/octet-stream",            "X-Nukez-Part-SHA256": hashlib.sha256(part).hexdigest(),        },        content=part,        timeout=300,    ).json()    assert part_resp.get("ok"), part_respif part_resp.get("status") != "ready":    part_resp = httpx.post(f"{BASE}/blob/{data_token}/complete", timeout=60).json() spec = part_resp["envelopes_needed"][0]env = build_envelope(receipt_id, spec["method"], spec["path"], ops=spec["ops"], body=spec["body"])store = call("nukez_store", {    "receipt_id": receipt_id,    "envelope": env,    "files": [{"name": large_path.name, "content_type": "application/vnd.google-earth.kml+xml", "data_token": data_token}],})

Code notes

Small and moderate files can be sent with data_b64 when the client can tolerate bytes in the tool argument. Large local files should use the canonical host byte staging path: ask nukez_store for a staging plan, move bytes through /blob routes, then store the resulting data_token.

nukez_upload_chunk exists as an advanced fallback, but it is not the canonical notebook path for large local files.

Proof

Retrieve, bootstrap, verify, and attest

Use signed list and read envelopes for production state and proof operations.

python
env_list = build_envelope(receipt_id, "GET", f"/v1/lockers/{locker_id}/files", ops=["locker:list"])files = call("nukez_retrieve", {"receipt_id": receipt_id, "envelope": env_list}) env_dl = build_envelope(receipt_id, "GET", f"/v1/lockers/{locker_id}/files/Anza.pdf", ops=["locker:read"])dl = call("nukez_retrieve", {"receipt_id": receipt_id, "envelope": env_dl, "filenames": ["Anza.pdf"]})downloaded = base64.b64decode(dl["files"][0]["content"]) status = call("nukez_status", {"receipt_id": receipt_id, "envelope": env_list})cached_manifest_hash = status.get("manifest_hash")cached_memory_version = status.get("memory", {}).get("version") fresh = call("nukez_status", {    "receipt_id": receipt_id,    "envelope": env_list,    "known_manifest_hash": cached_manifest_hash,    "known_memory_version": cached_memory_version,}) verify_pre = call("nukez_verify", {"receipt_id": receipt_id, "push": False})verify = call("nukez_verify", {"receipt_id": receipt_id, "push": True})proof = call("nukez_verify", {"receipt_id": receipt_id, "filename": "Anza.pdf", "push": False})

Code notes

The canonical flow uses signed locker:list envelopes for file listings and active status. It uses per-file locker:read envelopes for downloads, status for cached freshness markers, and nukez_verify for pre-attest checks, on-chain anchoring, and Merkle inclusion proof.

One per-file read envelope cannot authorize multiple different file paths.

Memory

Remember, recall, provenance, and diff

Use two-phase memory writes for session state, provenance records, retrieval fall-through, and cheap freshness checks.

python
plan = call("nukez_remember", {    "receipt_id": receipt_id,    "envelope": [],    "key": "mainnet_test_note_1",    "content": json.dumps({"test": "mcp mainnet log", "receipt_id": receipt_id}),    "summary": "MCP mainnet test note - memory round-trip verification",    "namespace": "test",    "tags": "mcp,mainnet,test",})assert plan.get("action_required") == "sign_envelopes" envs = [build_envelope(receipt_id, s["method"], s["path"], ops=s["ops"], body=s["body"]) for s in plan["envelopes_needed"]]remember = call("nukez_remember", {"receipt_id": receipt_id, "envelope": envs, "key": "mainnet_test_note_1"})recall = call("nukez_recall", {"receipt_id": receipt_id, "key": "mainnet_test_note_1"}) result = call("nukez_retrieve", {    "receipt_id": receipt_id,    "envelope": env_list,    "filenames": ["mainnet_test_note_1"],}) diff = call("nukez_diff", {    "cached_manifest_hash": cached_manifest_hash,    "receipt_id": receipt_id,    "envelope": env_list,})

Code notes

Memory writes are two-phase: plan with an empty envelope list, sign each returned spec, then execute. Recall does not need an envelope. Production provenance records returned as provenance_pending should be persisted with nukez_remember. nukez_diff provides zero-download freshness against cached manifests.

Recall hides superseded records by default. Pass include_archived when you need the full chain.

§ next

Reference

Tool reference

Every hosted MCP tool, resource URI, prompt, and canonical flow step.

Overview

MCP overview

Production model, envelope rules, byte paths, and session semantics.

Compare

HTTP route

The same gateway envelopes without the MCP tool wrapper.