Pipelines

The hardware API is just WebSocket JSON — anything the IDE can do, a script can do. That makes Nuclei’s kernel usable as a batch driver for parameter sweeps, calibration curves, and simulator-vs-hardware comparisons.

Prerequisites: the client examples (connection + message-draining patterns) and the replay-tested simulator session (the exact submit → status → results message shapes this page builds on).

The scripted loop

Per job, the protocol sequence is always:

1. hardware_submit with the source code (not a circuit object) → hardware_job_submitted with a JobHandle.
2. Poll hardware_results until data carries measurements (or error). hardware_status only refreshes queue position — it will never tell you a real-hardware job finished.
3. Handle both failure shapes: an error envelope (no job exists) and a handle with status: "failed" (the convention).

Worked example: parameter sweep

sweep_driver.py drives an RY(θ) rotation sweep — θ over a linspace, one hardware_submit per point against the always-available simulator provider, polled to completion, printed as a θ → P(1) table. The expected curve is P(1) = sin²(θ/2).

This exact file is executed by the kernel test suite against a live server on every CI run (kernel/tests/test_docs_fixtures.py), so it speaks the current protocol.

sweep_driver.py

#!/usr/bin/env python3
"""Batch parameter sweep over the Nuclei hardware API: an RY(theta) curve.

Submits one single-qubit circuit per sweep point through `hardware_submit`
to the local simulator provider, polls each job to completion, and prints a
theta -> P(1) table.

Usage: python sweep_driver.py [ws://localhost:9742] [shots] [points]
Requires: pip install websockets
Auth: reads the token from $NUCLEI_KERNEL_TOKEN or ~/.nuclei/runtime/kernel-<port>.json

This file is replay-tested: kernel/tests/test_docs_fixtures.py runs it as a
subprocess against a live kernel server (3 points, 64 shots) and asserts
the table output.
"""
import asyncio
import hashlib
import hmac
import json
import os
import math
import secrets
import sys
from urllib.parse import urlparse

import websockets

PROVIDER = "simulator"
BACKEND = "sim_qasm"

CIRCUIT_TEMPLATE = """from qiskit import QuantumCircuit

qc = QuantumCircuit(1, 1)
qc.ry({theta}, 0)
qc.measure(0, 0)
"""


def load_token(url: str | None = None) -> str:
    """Token for this kernel launch: $NUCLEI_KERNEL_TOKEN, else the user-only
    runtime file the kernel writes at startup (like a Jupyter runtime file)."""
    token = os.environ.get("NUCLEI_KERNEL_TOKEN")
    if token:
        return token
    runtime_dir = os.environ.get("NUCLEI_RUNTIME_DIR") or os.path.join(
        os.path.expanduser("~"), ".nuclei", "runtime"
    )
    port = urlparse(url).port if url else None
    # Several kernels can run at once: prefer this port's own file; kernel.json
    # is the newest-kernel pointer and is only used when it names the same port.
    for name in ([f"kernel-{port}.json"] if port else []) + ["kernel.json"]:
        try:
            with open(os.path.join(runtime_dir, name)) as handle:
                info = json.load(handle)
        except OSError:
            continue
        if not port or info.get("port") == port:
            return info["token"]
    raise RuntimeError(f"no token for port {port}: set NUCLEI_KERNEL_TOKEN or start a kernel")


async def authenticate(ws, url: str | None = None) -> None:
    """First frame on every connection: present the token and a random nonce,
    then check the kernel's proof (HMAC-SHA256(token, nonce)) in the ack."""
    token = load_token(url)
    nonce = secrets.token_hex(32)
    await ws.send(json.dumps({"type": "authenticate", "token": token, "nonce": nonce}))
    reply = json.loads(await ws.recv())
    if reply.get("type") != "authenticated":
        raise RuntimeError("kernel rejected the authentication token")
    expected = hmac.new(token.encode(), nonce.encode(), hashlib.sha256).hexdigest()
    if not hmac.compare_digest(str(reply.get("proof", "")), expected):
        raise RuntimeError("peer did not prove it is the Nuclei kernel; refusing to continue")


async def request(ws, payload: dict, reply_type: str) -> dict:
    """Send one hardware request and wait for its single reply.

    Every hardware request gets exactly one reply: the named response type
    or an `error` envelope — which this raises so the sweep fails loudly.
    """
    await ws.send(json.dumps(payload))
    while True:
        message = json.loads(await ws.recv())
        if message["type"] == "error":
            raise RuntimeError(message["message"])
        if message["type"] == reply_type:
            return message


async def run_point(ws, theta: float, shots: int) -> dict:
    """Submit one circuit and poll until its counts arrive."""
    submitted = await request(ws, {
        "type": "hardware_submit",
        "provider": PROVIDER,
        "backend": BACKEND,
        "shots": shots,
        "code": CIRCUIT_TEMPLATE.format(theta=repr(theta)),
        "language": "python",
    }, "hardware_job_submitted")
    job_id = submitted["job"]["id"]

    # Completion is discovered via hardware_results, NOT hardware_status
    # (status polling only refreshes queue position). The simulator
    # completes synchronously; real providers return {"status": "running"}
    # here until done — hence the loop.
    while True:
        result = await request(ws, {
            "type": "hardware_results", "job_id": job_id,
        }, "hardware_result")
        data = result["data"]
        if "error" in data:
            raise RuntimeError(f"job {job_id} failed: {data['error']}")
        if "measurements" in data:
            return data["measurements"]
        await asyncio.sleep(1.0)


async def main(url: str, shots: int, points: int) -> int:
    step = math.pi / (points - 1) if points > 1 else 0.0
    async with websockets.connect(url) as ws:
        await authenticate(ws, url)
        print(f"sweep: RY(theta) on {PROVIDER}/{BACKEND}, {shots} shots/point")
        for i in range(points):
            theta = i * step
            counts = await run_point(ws, theta, shots)
            p1 = counts.get("1", 0) / shots
            print(f"theta={theta:.4f}  P(1)={p1:.4f}  counts={counts}")
    return 0


if __name__ == "__main__":
    url = sys.argv[1] if len(sys.argv) > 1 else "ws://localhost:9742"
    shots = int(sys.argv[2]) if len(sys.argv) > 2 else 256
    points = int(sys.argv[3]) if len(sys.argv) > 3 else 9
    sys.exit(asyncio.run(main(url, shots, points)))

cd kernel && NUCLEI_KERNEL_TOKEN="$(python -c 'import secrets; print(secrets.token_urlsafe(32))')" python server.py   # terminal 1
python sweep_driver.py                 # terminal 2 — 9 points, 256 shots
python sweep_driver.py ws://localhost:9742 1024 17

Sample output, abridged (counts vary run to run):

sweep: RY(theta) on simulator/sim_qasm, 256 shots/point
theta=0.0000  P(1)=0.0000  counts={'0': 256}
theta=0.7854  P(1)=0.1484  counts={'0': 218, '1': 38}
theta=1.5708  P(1)=0.5039  counts={'1': 129, '0': 127}
...
theta=3.1416  P(1)=1.0000  counts={'1': 256}

Into analysis

Collect the table into a DataFrame and plot it (illustrative — adapt freely, not replay-tested):

import pandas as pd
import matplotlib.pyplot as plt
import numpy as np

# rows collected from run_point(): [(theta, counts), ...]
df = pd.DataFrame(
    {"theta": theta, "p1": counts.get("1", 0) / shots}
    for theta, counts in rows
)

ax = df.plot(x="theta", y="p1", marker="o", label="measured")
thetas = np.linspace(0, np.pi, 200)
ax.plot(thetas, np.sin(thetas / 2) ** 2, "--", label="sin²(θ/2)")
ax.set_xlabel("θ (rad)"); ax.set_ylabel("P(1)"); ax.legend()
plt.savefig("ry_sweep.png", dpi=150)

Remember the per-framework bitstring-key conventions before aggregating counts across frameworks — see schemas.

Simulator vs real hardware

The driver is provider-agnostic by design — switch PROVIDER/BACKEND (after a hardware_connect with credentials) and the same loop runs on real devices. Adjust expectations:

	Simulator	Real hardware
Submit	Completes synchronously — the handle arrives already complete.	Returns queued; minutes-to-hours in queue is normal.
Polling	First hardware_results has the data.	{"status": "running"} until done — poll with backoff (seconds, not the driver’s 1 s, for long queues).
Results	Full SimulationResult.	Counts-style dict only — no statevector, no Bloch coordinates.
Failure modes	data.error carries a KernelError.	SDK error strings; plus queue cancellations and per-job credit limits.
Restart safety	Cheap to re-run.	A kernel restart mid-sweep turns in-flight jobs stale — results unrecoverable through Nuclei. Keep sweeps short or checkpoint your own (point → counts) pairs as you go.

For sweeps on Azure Quantum specifically, start with the free targets listed in testing tips.