JSON-RPC MCP endpoint hosted on Cloud Run.
Docs · MCP
MCP
Live · Cloud Run
Overview
Nukez MCP Server
The Nukez Model Context Protocol Server.
The canonical MCP flow uses the hosted streamable HTTP JSON-RPC endpoint, raw client-side tool calls, local Ed25519 envelope signing, and externally executed payment transactions. The MCP server coordinates Nukez gateway operations without custodying a wallet or client signing key.
Client signs and submits the selected chain payment locally.
Protected locker operations pass signed Ed25519 envelopes.
Model
Hosted MCP, local payment, local envelopes
The server wraps the gateway; clients keep custody of payment keys and protected signing.
MCP endpoint: https://nukez-mcp-538908875166.us-central1.run.app/mcpGateway: https://api.nukez.xyzNetwork: solana-mainnetTool surface: 15 thick MCP tools Payment flow:nukez_quote -> external chain transfer -> nukez_pay -> nukez_provision Post-provisioning auth:client-signed Ed25519 envelopes passed as envelope
Code notes
The production MCP server is a streamable HTTP JSON-RPC server. It wraps the Nukez gateway, but it does not hold a server wallet or signing key. Clients execute payment externally, sign operation envelopes locally, and pass those envelopes through MCP tool arguments.
PyNukez is not required for the canonical MCP client flow. The guide uses raw JSON-RPC, local Ed25519 envelope signing, and an explicit Solana transfer helper built with solders.
Setup
Install and configure the direct client runtime
Use raw JSON-RPC plus local signing dependencies; no PyNukez package is required for MCP.
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")_kp_bytes = json.loads(KEYPAIR_PATH.read_text())_sk = SigningKey(bytes(_kp_bytes[:32]))PUBKEY = base58.b58encode(_sk.verify_key.encode()).decode() RPC_URL = "https://mainnet.helius-rpc.com/?api-key=<API_KEY>"
Code notes
Run the shared setup once before the examples. Replace the keypair path and RPC URL with the client environment's values. The client needs httpx for transport, PyNaCl/base58 for envelope signing, and solders only when the example executes a Solana transfer.
Never send keypair bytes to the gateway or MCP server. The public key and transaction signatures are the only payment-side values the hosted tools need.
Transport
Call tools and build signed envelopes
Use tools/call for MCP operations and canonical signed envelopes for protected locker actions.
_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 def build_envelope(receipt_id, method, path, ops=None, body=None, ttl=300): canonical_body = json.dumps(body, separators=(",", ":"), sort_keys=True) if body else None body_hash = hashlib.sha256((canonical_body or "").encode()).hexdigest() envelope = { "v": 1, "locker_id": compute_locker_id(receipt_id), "receipt_id": receipt_id, "nonce": os.urandom(16).hex(), "iat": int(time.time()), "exp": int(time.time()) + 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
Code notes
The canonical client uses a small rpc helper, a tools/call wrapper, and a canonical JSON envelope builder. Authenticated locker operations bind method, path, receipt id, locker id, nonce, expiry, operation scope, and body hash into the signed payload.
Use locker:provision for provisioning, locker:write for create/delete/write flows, locker:list for file listings, and locker:read for per-file reads.
Lifecycle
Initialize, quote, pay, confirm, and provision
Follow the canonical two-step provisioning sequence from the executed notebook.
init = rpc("initialize", { "protocolVersion": "2025-03-26", "capabilities": {}, "clientInfo": {"name": "mcp-client", "version": "1.0"},})tools = rpc("tools/list") status = call("nukez_status")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})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
Provisioning is intentionally two-step. First, confirm the externally executed payment with nukez_provision and receive action_required: sign_provision_envelope. Then sign the provided provision operation and call nukez_provision again with receipt_id and envelope.
Save receipt_id permanently after payment confirmation. It is the durable handle for future sessions.
Storage
Store files through the right byte path
Inline small content, moderate data_b64, or host-byte staging with data_token for large files.
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\n").decode()}],}) plan = call("nukez_store", { "receipt_id": receipt_id, "files": [{ "name": large_filename, "source": "staged", "content_type": large_content_type, "size_bytes": large_size, "sha256": large_sha256, }],}) start_resp = httpx.post(f"{BASE}/blob/start", json={ "receipt_id": receipt_id, "filename": large_filename, "content_type": large_content_type, "sha256": large_sha256, "size_bytes": large_size,}, timeout=30).json()data_token = start_resp["data_token"] # PUT parts to /blob/{data_token}/parts/{part_no}, then complete if needed.blob_resp = httpx.post(f"{BASE}/blob/{data_token}/complete", timeout=60).json()spec = blob_resp["envelopes_needed"][0]env = build_envelope(receipt_id, spec["method"], spec["path"], ops=spec["ops"], body=spec["body"])store_large = call("nukez_store", { "receipt_id": receipt_id, "envelope": env, "files": [{"name": large_filename, "content_type": large_content_type, "data_token": data_token}],})
Code notes
Post-provisioning writes require a signed locker:write envelope for POST /v1/lockers/{locker_id}/files. Use inline data_b64 for small or moderate controlled clients. For large local files, use the canonical host byte staging flow and store the resulting data_token.
The /blob routes are not MCP tools. Reserve nukez_upload_chunk for advanced fallback cases where host byte staging is unavailable.
Proof
Bootstrap, verify, attest, and resume
Use signed status for active sessions and cached freshness markers for returning sessions.
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})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
For production bootstrap, pass a signed locker:list envelope to nukez_status so it can include files, manifest_hash, memory.version, and trust_check. Use verify with push=false before attestation, push=true to anchor, and filename for Merkle inclusion proof.
Re-run nukez_verify(push=True) after storing or deleting files to refresh the on-chain attestation.
Memory
Remember, recall, provenance, and diff
Use two-phase memory writes and lightweight diff checks for agent session continuity.
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",})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
Production memory writes use a two-phase plan/execute flow. Recall reads by key or query without an envelope, unified retrieval can fall through to memory, and nukez_diff gives a lightweight changed/unchanged signal without downloading file bytes.
For provenance in production envelope-passthrough mode, persist provenance_pending records manually with nukez_remember.