openquantum-sdk 0.3.1

Python SDK for Open Quantum

Open Quantum Python SDK

A lightweight, modern Python SDK for the Open Quantum Platform.

Features:

• Management API — list organizations, backend classes, and providers
• Scheduler API — job categories, upload, prepare, submit, poll, and one-call submit_job()
• Smart defaults — auto-select execution plan & queue priority from quote
• Thread-safe auth — Keycloak client-credentials with auto-refresh
• Automatic output download — download_job_output() returns parsed JSON

Installation

pip install -U openquantum-sdk
# or local dev:
pip install -e .

Python 3.9+ required.

Authentication

Option 1: SDK Key File (Recommended)

Download from Open Quantum SDK Keys:

{
  "client_id": "s_abc123...",
  "client_secret": "e460c8..."
}

export OPENQUANTUM_SDK_KEY=/path/to/sdk-key.json

Option 2: Direct in Code

from openquantum_sdk.auth import ClientCredentials, ClientCredentialsAuth

auth = ClientCredentialsAuth(
    creds=ClientCredentials(client_id="s_...", client_secret="e460c8...")
)

Quickstart: One-Call Submission (Recommended)

from openquantum_sdk.clients import SchedulerClient, ManagementClient, JobSubmissionConfig
from openquantum_sdk.enums import ExecutionPlanType, QueuePriorityType
import json

# 1. Instantiate clients (auto-loads auth from env)
scheduler = SchedulerClient()
management = ManagementClient()

try:
    # 2. Discover your Organization ID
    print("Discovering organization...")
    orgs = management.list_user_organizations()
    org_id = orgs.organizations[0].id
    print(f" > Using Org: {orgs.organizations[0].name}")

    # 2.5 Discover available backend providers
    print("\nAvailable providers:")
    providers = management.list_providers()
    for p in providers.providers:
        print(f" > {p.name} ({p.short_code or 'no short code'}) — {p.description}")

    # 2.6 Discover backend classes
    print("\nAvailable backend classes:")
    classes = management.list_backend_classes()
    for bc in classes.backend_classes:
        print(f" > {bc.name} ({bc.short_code or 'no short code'}) — {bc.type}")
        print(f"   ├ Status: {bc.status or 'Unknown'} | Queue depth: {bc.queue_depth or 'N/A'}")
        print(f"   └ Use: '{bc.short_code or bc.id}' in backend_class_id")

    # 3. Define config
    config = JobSubmissionConfig(
        organization_id=org_id,
        backend_class_id="rigetti:ankaa-3",
        name="Bell State SDK Quickstart",
        job_subcategory_id="fin:port", # Finance Portfolio Optimization
        shots=100,
        execution_plan="auto",
        queue_priority="auto",
        auto_approve_quote=True,
        verbose=True
    )

    # 4. Submit + wait + get result
    dummy_qasm = b"OPENQASM 2.0; include 'qelib1.inc'; qreg q[2]; creg c[2]; h q[0]; cx q[0],q[1]; measure q -> c;"
    job = scheduler.submit_job(config, file_content=dummy_qasm)

    # 5. Download and print parsed JSON output
    print("\n--- JOB SUCCEEDED ---")
    print(f"Job ID: {job.id}")
    print(f"Status: {job.status}")

    if job.output_data_url:
        output_json = scheduler.download_job_output(job)
        print("\n--- JOB OUTPUT (JSON) ---")
        print(json.dumps(output_json, indent=2))
    else:
        print("No output available.")

except Exception as e:
    print(f"\n--- JOB FAILED ---")
    print(f"Error: {e}")

finally:
    scheduler.close()
    management.close()

Handles everything in one call — including automatic JSON download.

Quickstart: Low-Level (Advanced)

from openquantum_sdk.clients import SchedulerClient, ManagementClient
from openquantum_sdk.models import JobPreparationCreate, JobCreate
from openquantum_sdk.utils import poll_for_status
import json

scheduler = SchedulerClient()
management = ManagementClient()

try:
    # 1. Get org
    org = management.list_user_organizations().organizations[0]

    # 1.5 List providers & backend classes
    print("Providers:")
    for p in management.list_providers().providers:
        print(f" - {p.name} ({p.short_code})")

    print("\nBackend classes:")
    for bc in management.list_backend_classes().backend_classes:
        print(f" - {bc.name} [{bc.type}] — {bc.short_code or bc.id}")

    # 2. Choose IDs
    MY_BACKEND = "rigetti:ankaa-3"
    MY_SUBCATEGORY = "fin:port"

    # 3. Upload
    upload_id = scheduler.upload_job_input(file_path="circuit.qasm")

    # 4. Prepare
    prep = JobPreparationCreate(
        organization_id=org.id,
        backend_class_id=MY_BACKEND,
        name="Low-Level Test",
        upload_endpoint_id=upload_id,
        job_subcategory_id=MY_SUBCATEGORY,
        shots=100,
        configuration_data={}
    )
    prep_resp = scheduler.prepare_job(prep)

    # 5. Poll preparation
    def get_prep_status(prep_id: str):
        result = scheduler.get_preparation_result(prep_id)
        done = result.status in ("Completed", "Failed")
        return done, result

    result = poll_for_status(
        get_status_fn=get_prep_status,
        resource_id=prep_resp.id,
        interval=2.0,
        timeout=300
    )

    if result.status == "Failed":
        raise RuntimeError(f"Preparation failed: {result.message or 'Unknown error'}")

    # Quote inspection
    print("\nQuote received:")
    for plan in result.quote:
        print(f" • {plan.name} — Base: {plan.price} credit{'s' if plan.price != 1 else ''}")
        for qp in plan.queue_priorities:
            total = plan.price + qp.price_increase
            print(f"    ├ {qp.name}: +{qp.price_increase} → Total: {total}")
        print(f"    └ Plan ID: {plan.execution_plan_id}")

    # 6. Select cheapest
    cheapest_plan = min(result.quote, key=lambda p: p.price)
    cheapest_prio = min(cheapest_plan.queue_priorities, key=lambda q: q.price_increase)

    # 7. Create job
    job = JobCreate(
        organization_id=org.id,
        job_preparation_id=prep_resp.id,
        execution_plan_id=cheapest_plan.execution_plan_id,
        queue_priority_id=cheapest_prio.queue_priority_id,
    )

    job_resp = scheduler.create_job(job)

    # 8. Poll completion
    def get_job_status(job_id: str):
        job_obj = scheduler.get_job(job_id)
        done = job_obj.status in ("Completed", "Failed", "Canceled")
        return done, job_obj

    final_job = poll_for_status(
        get_status_fn=get_job_status,
        resource_id=job_resp.id,
        interval=5.0,
        timeout=86_400,
    )

    print(f"\nJob finished: {final_job.status}")

    # 9. Download and print JSON output
    if final_job.output_data_url:
        output_json = scheduler.download_job_output(final_job)
        print("\n--- JOB OUTPUT (JSON) ---")
        print(json.dumps(output_json, indent=2))
    else:
        print("No output available.")

finally:
    scheduler.close()
    management.close()

High-Level API: submit_job()

Parameter	Type	Default	Description
config	JobSubmissionConfig	—	All job metadata
file_path	str	—	Input file
file_content	bytes	—	Or pass bytes

JobSubmissionConfig

Field	Type	Default	Notes
organization_id	str	—	Required
backend_class_id	str	—	Required (short code or UUID)
job_subcategory_id	str	—	Required
name	str	—	Required
shots	int	—	Required
configuration_data	Dict[str, Any]	None
execution_plan	ExecutionPlanType | "auto"	"auto"
queue_priority	QueuePriorityType | "auto"	"auto"
auto_approve_quote	bool	True	Set False in notebooks
job_timeout_seconds	int	86400	1 day

API Reference

SchedulerClient

Method	Returns	Description
__init__(auth[ClientCredentialsAuth])
get_job_categories(limit=20, cursor=...)	PaginatedJobCategories	List top-level categories
get_job_subcategories(category_id, limit=20, cursor=...)	PaginatedJobCategories	List subcategories
upload_job_input(file_path=... or file_content=...)	str	Returns upload ID
prepare_job(JobPreparationCreate)	JobPreparationResponse	Starts preparation
get_preparation_result(preparation_id)	JobPreparationResultResponse	Includes .quote
create_job(JobCreate)	JobRead	Submits job
get_job(job_id)	JobRead	Fetch job status
cancel_job(job_id)	None	Cancel running job
list_jobs(organization_id, limit=20, cursor=..., status=...)	PaginatedJobs	List your jobs
download_job_output(job)	Any	Downloads + parses JSON output
submit_job(config, file_path=... or file_content=...)	JobRead	One-call submission + polling

ManagementClient

Method	Returns	Description
__init__(auth[ClientCredentialsAuth])
list_user_organizations(limit=20, cursor=...)	PaginatedOrganizations	Your orgs
list_providers(limit=20, cursor=...)	PaginatedProviders	All backend providers
list_backend_classes(provider_id=None, limit=20, cursor=...)	PaginatedBackendClasses	Available backends

Data Models

Model	Key Fields
BackendClassRead	id, name, description, type, provider_id, short_code, queue_depth, accepting_jobs, status
ErrorResponse	status_code, message: List[str], error_code
JobCategoryRead	id, name, short_code
JobCreate	organization_id, job_preparation_id, execution_plan_id, queue_priority_id
JobList	id, backend_class_id, name, status, backend_name, output_data_url, credits_used, submitted_at
JobPreparationCreate	organization_id, backend_class_id, name, upload_endpoint_id, job_subcategory_id, shots, configuration_data
JobPreparationResponse	id
JobPreparationResultResponse	organization_id, backend_class_id, status, name, input_data_url, job_category_id, job_subcategory_id, shots, quote: List[QuotePlan], message, configuration_data
JobPreparationUploadResponse	id, url
JobRead	id, status, input_data_url, job_preparation_id, execution_plan_id, queue_priority_id, message, output_data_url, transaction_id, submitted_at
OrganizationRead	id, name
PaginatedBackendClasses	backend_classes: List[BackendClassRead], pagination: PaginationInfo
PaginatedJobCategories	categories: List[JobCategoryRead], pagination: PaginationInfo
PaginatedJobs	jobs: List[JobList], pagination: PaginationInfo
PaginatedOrganizations	organizations: List[OrganizationRead], pagination: PaginationInfo
PaginatedProviders	providers: List[ProviderRead], pagination: PaginationInfo
PaginationInfo	next_cursor
ProviderRead	id, name, description, short_code
QueuePriority	name, description, price_increase, queue_priority_id
QuotePlan	name, price, description, execution_plan_id, queue_priorities: List[QueuePriority]

Utils: poll_for_status()

from openquantum_sdk.utils import poll_for_status

Parameter	Type	Default	Description
get_status_fn	Callable[[str], Tuple[bool, Any]]		Function to be polled
resource_id	str		id to be passed to status function: get_status_fn(resource_id)
interval	int	5	Number of seconds to wait before next call of status function
timeout	int	300	Number of seconds before timeout

Enums

from openquantum_sdk.enums import ExecutionPlanType, QueuePriorityType

Enum	Values
ExecutionPlanType	PUBLIC, PRIVATE
QueuePriorityType	STANDARD, PRIORITY, INSTANT

CLI (Experimental)

python -m openquantum_sdk --sdk-key ./key.json \
                          --input bell.qasm \
                          --backend "ionq:aria-1" \
                          --name "CLI Circuit" \
                          --subcategory "mathematics:linear-systems" \
                          --shots 100 \
                          --auto-approve

Automatically downloads and pretty-prints JSON output.

Short Codes

Backends

Provider	Backend	Short Code
Ion-Q	Aria-1	ionq:aria-1
IQM	Emerald	iqm:emerald
IQM	Garnet	iqm:garnet
Rigetti	Ankaa-3	rigetti:ankaa-3

Note: This list may be modified as backends are added or removed from the platform. For the most up-to-date list, run the following code:

backends = management.list_backend_classes().backend_classes

print('| Name | Short Code | Online |')
for b in backends:
    print(f"| {b.name} | {b.short_code} | {b.accepting_jobs} |")

Subcategories

Category	Subcategory	Short Code
Chemistry	Molecular Ground State Estimation	chem:mgse
Chemistry	Other (Specify)	chem:oth
Chemistry	Quantum System Simulation	chem:qss
Chemistry	Variational Quantum Eigensolver (VQE)	chem:vqe
Cryptography	Discrete Logarithms	crypto:dlog
Cryptography	Hash Function Attacks	crypto:hash
Cryptography	Other (Specify)	crypto:oth
Cryptography	Shor's Factoring (RSA Cryptanalysis)	crypto:shor
Finance	Option Pricing	fin:opt
Finance	Other (Specify)	fin:oth
Finance	Portfolio Optimization	fin:port
Finance	Risk Analysis	fin:risk
Graph and Network Analysis	Graph Connectivity	gna:con
Graph and Network Analysis	Other (Specify)	gna:oth
Graph and Network Analysis	Pattern Matching	gna:pat
Graph and Network Analysis	Quantum Walks on Graphs	gna:qwg
Machine Learning	Other (Specify)	ml:oth
Machine Learning	Quantum Classification	ml:qcl
Machine Learning	Quantum Clustering	ml:qcu
Machine Learning	Quantum Recommendation Systems	ml:qrs
Mathematics	Differential Equations	math:deq
Mathematics	Hidden Subgroup Problems	math:hsp
Mathematics	Linear Systems (HHL Algorithm)	math:hhl
Mathematics	Other (Specify)	math:oth
Optimization	Adiabatic Optimization	opt:adi
Optimization	Constraint Satisfaction	opt:csp
Optimization	Other (Specify)	opt:oth
Optimization	Quantum Approximate Optimization Algorithm (QAOA)	opt:qaoa
Other	Other (Specify)	oth:oth
Physics	Hamiltonian Dynamics Simulation	phys:hds
Physics	Other (Specify)	phys:oth
Physics	Partition Functions	phys:pfn
Physics	Topological Invariants (e.g., Jones Polynomials)	phys:tin

Note: This list may be modified as categories and subcategories are added or removed. For the most up-to-date list, run the following code:

cats = scheduler.get_job_categories().categories

print('| Category | Subcategory | Short Code |')
for cat in cats:
    subcats = scheduler.get_job_subcategories(cat.short_code).categories
    for subcat in subcats:
        print(f'| {cat.name} | {subcat.name} | {subcat.short_code} |')

FAQ

Q: How do I get results as JSON?
A: Use scheduler.download_job_output(job) — returns parsed object.

Q: Why does auto mode fail on some QPUs?
A: auto only allows PUBLIC + STANDARD. Use explicit enums for private/instant access.