Kipu Quantum Hub Service SDK

At a glance

• Purpose: Programmatic client for Managed Services on Kipu Quantum Hub — submit executions, poll status, fetch results and logs, download result files, and attach short-lived data pool grants.
• Packages: qhub-service (Python 3.9+, installed via pip or uv) · @quantum-hub/qhub-service (TypeScript/JavaScript, installed via npm, yarn, or pnpm)
• Entry point: HubServiceClient (Python and TypeScript share the same surface)
• Use it when: you want to drive a Managed Service from a script, notebook, or application — or automate an end-to-end pipeline around one.
• Pairs with: qhubctl login stores the personal access token the Python client auto-resolves for data pool grants. Service endpoints and Access Keys are configured in your Service Settings.
• Quick reference: HubServiceClient(endpoint, accessKeyId, secretAccessKey).run(request) → ServiceExecution.result() / .logs() / .cancel()
• New here? See the References Overview.

The Kipu Quantum Hub Service SDK lets you interact programmatically with managed services on the Kipu Quantum Hub. It provides idiomatic clients in Python and TypeScript that wrap the underlying REST API to:

• Submit service executions to a managed service.
• Track execution status and wait for a final state.
• Retrieve results, download result files, and stream logs.
• Cancel pending or running executions.
• Inject short-lived Data Pool access grants into a request.

The SDK ships in two flavours that expose the same conceptual surface:

Language	Package
Python	qhub-service
TypeScript	@quantum-hub/qhub-service

Installation

Python requires >= 3.9. The TypeScript package is published as an ES module.

Install the SDK using pip or uv:

uv add qhub-service
# OR with native pip
pip install --upgrade qhub-service

Concepts

The SDK models the following resources:

• Service — a managed application published on the Kipu Quantum Hub marketplace (e.g. Rimay - Quantum Feature Extraction - Simulator). A service exposes a gateway endpoint that clients call.
• Application — a client-side container in your organization. An application owns access keys and groups subscriptions.
• Subscription — a link between an application and a service. A subscription provides the gatewayEndpoint used for execution requests.
• Access key — an accessKeyId / secretAccessKey pair issued per application. Used with the OAuth 2.0 client-credentials flow to obtain short-lived bearer tokens.
• Service execution — a single invocation of a service. Identified by a UUID, progresses through a lifecycle of statuses (PENDING → RUNNING → SUCCEEDED / FAILED / CANCELLED).
• Data Pool — a managed storage resource. Services that read from or write to a data pool require a short-lived JWT access grant scoped to (datapoolId, applicationId, permission).

Authentication

The SDK uses two distinct authentication mechanisms:

Service gateway (OAuth 2.0 client credentials)

Calls to the service gateway are authenticated with a bearer token obtained via the OAuth 2.0 client-credentials flow using an application's accessKeyId and secretAccessKey.

• The default token endpoint is https://gateway.hub.kipu-quantum.com/token.
• The SDK caches the token and transparently refreshes it before it expires (Python refreshes ~120 seconds before expiry; TypeScript checks AccessToken.expired() before every request).
• If both accessKeyId and secretAccessKey are omitted, the client falls back to a random token, which is useful for local gateways that do not enforce auth.

Platform API (personal access token)

Calls to the Hub Platform API (used for data pool grants and service discovery) are authenticated with a personal access token (API key).

• In Python, credentials and the active organization are resolved automatically via DefaultCredentialsProvider and ContextResolver. If you are already logged in through the qhubctl CLI, no parameters need to be supplied.
• In TypeScript, you must pass a HubPlatformClient with an explicit apiKey.

Quickstart

import os
from qhub.service.client import HubServiceClient

client = HubServiceClient(
    service_endpoint="https://gateway.hub.kipu-quantum.com/acme/qft-simulator/1.0.0",
    access_key_id=os.environ["ACCESS_KEY_ID"],
    secret_access_key=os.environ["SECRET_ACCESS_KEY"],
)

# Start a new execution
execution = client.run(request={"values": [2], "shots": 100})

# Block until the execution finishes
execution.wait_for_final_state(timeout=300)

print(f"Finished at {execution.ended_at} with status {execution.status}")

# Fetch the result
result = execution.result()

# Download every result file into the current working directory
for file_name in execution.result_files():
    execution.download_result_file(file_name, os.getcwd())

HubServiceClient

HubServiceClient is the primary entry point. It manages authentication, exposes the low-level REST API via api, and offers high-level helpers for submitting executions and acquiring data pool grants.

Constructor

HubServiceClient(
    service_endpoint: str,
    access_key_id: str | None,
    secret_access_key: str | None,
    token_endpoint: str = "https://gateway.hub.kipu-quantum.com/token",
    platform_client: HubPlatformClient | None = None,
)

Parameter	Required	Description
service_endpoint	yes	Gateway URL of the subscribed service.
access_key_id	no*	Application access key id. Omit for anonymous/local gateways.
secret_access_key	no*	Application secret access key. Omit for anonymous/local gateways.
token_endpoint	no	OAuth 2.0 token endpoint. Defaults to the Kipu Quantum Hub gateway token endpoint.
platform_client	no	Pre-configured HubPlatformClient. Required only when using data pool grants.

* Both access key parameters must either be supplied together or both be omitted.

In Python, if platform_client is not supplied the constructor attempts to create one from the ambient credentials (via DefaultCredentialsProvider). If no credentials are available the field is left as None and attempts to use grants will raise.

api — low-level REST client

api returns the generated REST client (ServiceApiClient) that exposes every endpoint of the service gateway directly. Use it when you need fine-grained control that the high-level helpers do not provide.

client.api.get_service_executions()
client.api.start_execution(request={"shots": 100})
client.api.get_status(id="...")
client.api.get_result(id="...")
client.api.get_result_file(id="...", file="output.json")  # returns Iterator[bytes]
client.api.get_logs(id="...")
client.api.cancel_execution(id="...")

run — submit a new execution

Starts a new service execution. In Python, run returns a rich HubServiceExecution wrapper with polling and result helpers. In TypeScript, run returns the raw ServiceExecution DTO.

run(
    request: dict[str, Any],
    secrets: dict[str, Any] | None = None,
    tags: list[str] | None = None,
    grants: list[DataPoolGrant] | None = None,
) -> HubServiceExecution

Behaviour:

• tags are injected under the reserved key $tags (unless the caller already set it).
• secrets are injected under the reserved key $secrets (unless the caller already set it).
• grants trigger acquisition of short-lived JWTs against the Platform API and inject DataPoolReference objects under the name of each grant (see Data Pool Access Grants). Requires a configured HubPlatformClient.
• Keys the caller already placed in the request are never overwritten by $tags or $secrets. Grants, by design, overwrite any pre-existing key with the same name so that the injected JWT always takes effect.

Listing and fetching executions

# Fetch a single execution by its ID (returns a HubServiceExecution wrapper)
execution = client.get_service_execution("0030737b-35cb-46a8-88c2-f59d4885484d")

# List every execution visible to the authenticated application
for execution in client.get_service_executions():
    print(execution.id, execution.status, execution.created_at)

Service Executions

In Python, HubServiceExecution wraps a ServiceExecution DTO and provides convenience operations (refreshing status, waiting for a final state, downloading files). In TypeScript these helpers are not wrapped — use the raw ServiceExecution and the api() client directly.

Lifecycle

PENDING ──▶ RUNNING ──▶ SUCCEEDED
                │
                ├───────▶ FAILED
                │
                └───────▶ CANCELLED

UNKNOWN is used when the gateway cannot report a status (generally only on transient errors). The final states are SUCCEEDED, FAILED, and CANCELLED.

HubServiceExecution members (Python)

Member	Description
id	UUID of the execution.
status	One of UNKNOWN, PENDING, RUNNING, SUCCEEDED, CANCELLED, FAILED.
created_at	ISO-8601 timestamp of when the execution was created.
started_at	ISO-8601 timestamp of when the execution transitioned to RUNNING. None until set.
ended_at	ISO-8601 timestamp of when the execution reached a final state. None until set.
has_finished	True once the status is a final state. Refreshes the status on every access.
refresh()	Re-fetch the status from the gateway.
wait_for_final_state(timeout, wait)	Poll until a final state is reached. Raises TimeoutError when timeout is exceeded.
result()	Wait for the final state, then fetch and return the ResultResponse. Retries with exponential backoff.
result_files()	List the names of downloadable result files (excludes the HAL status and self links).
result_file_stream(name)	Return an iterator of bytes for the named result file.
download_result_file(name, target_path)	Stream the named result file into the directory target_path.
cancel()	Cancel the execution if still pending or running.
logs()	Return the list of LogEntry items attached to the execution.

Waiting for a final state

wait_for_final_state blocks until the execution transitions to a final state or timeout seconds have elapsed.

# Wait indefinitely with a 5 second polling interval
execution.wait_for_final_state()

# Wait at most 5 minutes, polling every 10 seconds
execution.wait_for_final_state(timeout=300, wait=10)

Raises TimeoutError if the deadline is reached before a final state.

Downloading result files

result_files returns the list of downloadable file names excluding the HAL navigational links (status, self).

import os

for name in execution.result_files():
    execution.download_result_file(name, os.getcwd())

download_result_file streams the file into the provided directory. The target must exist and be a directory; otherwise the method raises ValueError.

For manual streaming, use result_file_stream which returns an iterator of chunks:

with open("result.json", "wb") as f:
    for chunk in execution.result_file_stream("result.json"):
        f.write(chunk)

Reading logs

logs = execution.logs() or []
logs.sort(key=lambda entry: entry.timestamp)
for entry in logs:
    print(entry.severity, entry.timestamp, entry.message)

Cancelling an execution

execution.cancel()

Data Pool Access Grants

Services that read from or write to data pools require a short-lived JWT grant. The SDK can acquire these grants automatically by calling the Hub Platform API on your behalf.

Data model

• DataPoolPermission — VIEW or MODIFY.
• DataPoolReference — the in-request representation of a data pool. Has a stable ref field set to DATAPOOL, the id of the pool, and an optional grant JWT.
• DataPoolGrant — the grant specification passed to run:
  • name — the key under which the resolved DataPoolReference will be injected.
  • datapool_id / datapoolId — the id of the data pool to reference.
  • application_id / applicationId — the application under which the grant is requested.
  • permission — defaults to MODIFY.

Example

from qhub.service.client import HubServiceClient
from qhub.service.datapool import DataPoolGrant, DataPoolPermission

# Credentials and organization are resolved via qhubctl / the platform context.
client = HubServiceClient(service_endpoint, access_key_id, secret_access_key)

grants = [
    DataPoolGrant(
        name="input_data",
        datapool_id="dp-abc123",
        application_id="app-xyz789",
        permission=DataPoolPermission.VIEW,
    ),
    DataPoolGrant(
        name="output_data",
        datapool_id="dp-def456",
        application_id="app-xyz789",
        permission=DataPoolPermission.MODIFY,
    ),
]

execution = client.run(request={"shots": 100}, grants=grants)

To override the resolved credentials or organization pass an explicit HubPlatformClient:

from qhub.service.client import HubServiceClient, HubPlatformClient

platform_client = HubPlatformClient(api_key="...", organization_id="...")
client = HubServiceClient(
    service_endpoint,
    access_key_id,
    secret_access_key,
    platform_client=platform_client,
)

Caching

Grants are cached per (datapoolId, applicationId, permission) tuple until 30 seconds before the issued JWT's exp claim. Re-running an execution with the same grant re-uses the cached token, which keeps the number of platform calls bounded even when executions are issued in a tight loop.

If run is called with grants but no HubPlatformClient is available the call fails fast:

• Python raises ValueError: api_key is required for data pool grant acquisition.
• TypeScript throws Error: apiKey is required for data pool grant acquisition.

Manual references

If your workflow already owns a valid grant token you can bypass automatic acquisition and place a DataPoolReference directly in the request:

from qhub.service.datapool import DataPoolReference

request = {
    "input_data": DataPoolReference(id="dp-abc123", grant=my_jwt).dict(),
}
client.run(request=request)

When both a manual reference and a grant for the same key are supplied, the grant takes precedence and overwrites the manual entry.

HubPlatformClient

A thin wrapper around the Platform API used by HubServiceClient for data pool grants.

from qhub.service.client import HubPlatformClient

platform = HubPlatformClient(
    api_key="...",  # optional - resolved via DefaultCredentialsProvider
    organization_id="...",  # optional - resolved via ContextResolver
    platform_endpoint="https://api.hub.kipu-quantum.com/qc-catalog",
)

platform.organization_id
platform.data_pool_grants  # generated client for the /data-pool-grants API

The TypeScript variant does not auto-resolve credentials — apiKey must be supplied explicitly.

Advanced usage

Batch processing

Run multiple executions in parallel when you have independent inputs. Python uses a thread pool because the underlying HTTP client is synchronous; TypeScript uses Promise.all since the client is already asynchronous.

from concurrent.futures import ThreadPoolExecutor


def process_batch(batch_data):
    execution = client.run(request={"data": batch_data, "params": {"mode": "batch"}})
    execution.wait_for_final_state()
    return execution.result()


batches = [
    {"values": list(range(0, 100))},
    {"values": list(range(100, 200))},
    {"values": list(range(200, 300))},
]

with ThreadPoolExecutor(max_workers=3) as pool:
    results = list(pool.map(process_batch, batches))

print(f"Processed {len(results)} batches")

Custom polling strategy

Observe every status transition while you wait — useful for logging, progress UI, or emitting events without blocking on wait_for_final_state.

import time


def wait_with_progress(execution, timeout=None):
    start_time = time.time()
    last_status = None

    while not execution.has_finished:
        if execution.status != last_status:
            elapsed = time.time() - start_time
            print(f"[{elapsed:.1f}s] Status changed to: {execution.status}")
            last_status = execution.status

        if timeout and (time.time() - start_time) > timeout:
            raise TimeoutError("Execution timed out")

        time.sleep(5)

    print(f"Execution completed with status: {execution.status}")


execution = client.run(request={"data": data, "params": params})
wait_with_progress(execution, timeout=300)

has_finished refreshes the status from the gateway on every access, so the loop does not need an explicit refresh() call.

Reference

DTOs

ServiceExecution

Field	Type	Description
id	string	Unique identifier of the service execution.
status	ServiceExecutionStatus	Current status.
type	ServiceExecutionType?	MANAGED or WORKFLOW.
created_at / createdAt	string	ISO-8601 creation timestamp.
started_at / startedAt	string?	ISO-8601 timestamp the execution started.
ended_at / endedAt	string?	ISO-8601 timestamp the execution ended.
service_id / serviceId	string?	ID of the underlying service.
service_definition_id / serviceDefinitionId	string?	ID of the service definition.
application_id / applicationId	string?	ID of the requesting application.
tags	string[]?	Tags attached to the execution.

ServiceExecutionStatus

"UNKNOWN" | "PENDING" | "RUNNING" | "SUCCEEDED" | "CANCELLED" | "FAILED"

ResultResponse

Field	Type	Description
links / _links	{ status?: HalLink, [rel: string]: HalLink }	HAL links keyed by relation. Entries with keys status or self are navigational; everything else is a downloadable file.
embedded / _embedded	{ status?: ServiceExecution }	The embedded execution resource.

LogEntry

Field	Type	Description
message	string	Log message content.
severity	LogEntrySeverity?	One of DEBUG, NOTICE, INFO, WARNING, ERROR.
timestamp	datetime	When the entry was recorded.

DataPoolReference

Field	Type	Description
id	string	ID of the data pool.
ref	"DATAPOOL"	Discriminator, always DATAPOOL.
grant	string?	Optional short-lived JWT granting access to the pool.

DataPoolGrant

Field	Type	Description
name	string	Key under which the injected DataPoolReference is placed.
datapool_id / datapoolId	string	ID of the data pool to reference.
application_id / applicationId	string	ID of the application requesting the grant.
permission	DataPoolPermission?	VIEW or MODIFY. Defaults to MODIFY.

Reserved request keys

When building the payload for run the SDK reserves two top-level keys for metadata:

Key	Purpose
$tags	List of tags attached to the execution.
$secrets	Map of secret values resolved server-side.

If the caller already supplies either key, the corresponding keyword argument to run is ignored — caller-provided metadata wins.

Data pool grants take precedence over any pre-existing value at the same key.

Error Handling

• Bad credentials to the service gateway result in the OAuth client raising at token time. In Python, exceptions from authlib propagate out of the first API call; in TypeScript, simple-oauth2 throws on getToken.
• Missing data pool grant prerequisites: calling run with grants but no HubPlatformClient raises ValueError: api_key is required for data pool grant acquisition in Python and throws the same message in TypeScript.
• result() after failure: The Python helper retries the result fetch with exponential backoff (1s, 2s, 4s, 8s) before propagating the last exception. If the execution itself ended in FAILED, the returned result response reflects that status; consult logs() for diagnostics.
• download_result_file raises ValueError if the target path is missing or is not a directory.
• wait_for_final_state raises TimeoutError if the execution does not reach a final state before timeout elapses.

Endpoints

The SDK targets the following Kipu Quantum Hub endpoints by default. Each can be overridden via the corresponding constructor argument.

Endpoint	Default	Overridable via
Service gateway	Supplied per-subscription via service_endpoint	HubServiceClient(service_endpoint=...)
OAuth token	https://gateway.hub.kipu-quantum.com/token	HubServiceClient(token_endpoint=...)
Platform API	https://api.hub.kipu-quantum.com/qc-catalog	HubPlatformClient(platform_endpoint=...) / platformEndpoint