Skip to main content

SDK Quickstart

The QCOS Python SDK provides a Pythonic interface for executing quantum circuits on GPU-accelerated infrastructure.

Installation​

pip install qcos-sdk

Basic Usage​

from qcos import QCOSClient

# Initialize client (uses QCOS_API_KEY environment variable)
client = QCOSClient()

# Execute a circuit
result = client.execute(
    qasm="""
    OPENQASM 2.0;
    include "qelib1.inc";
    qreg q[2];
    creg c[2];
    h q[0];
    cx q[0],q[1];
    measure q -> c;
    """,
    shots=1024
)

print(result.counts)
# {'00': 512, '11': 512}

Client Configuration​

from qcos import QCOSClient

# Full configuration
client = QCOSClient(
    api_key="your-api-key",          # Required (or use env var)
    api_url="https://api.softquantus.com",  # Optional
    timeout=60,                       # Request timeout (seconds)
    max_retries=3,                    # Automatic retries on failure
    retry_delay=1.0,                  # Initial retry delay
    log_level="INFO",                 # Logging level
)

# Check connection
health = client.health()
print(health)
# {'status': 'healthy', 'version': '1.0.0', 'queue_connected': True}

Execution Methods​

Synchronous Execution​

# Execute and wait for result
result = client.execute(
    qasm=circuit,
    shots=1024,
    wait=True,          # Wait for completion (default)
    timeout=120         # Max wait time
)

Submit Without Waiting​

# Submit job and get job ID
job = client.submit(qasm=circuit, shots=1024)
print(f"Submitted: {job.job_id}")

# Check status later
status = client.get_status(job.job_id)
print(f"Status: {status.status}")

# Get results when ready
result = client.get_result(job.job_id)

Async Execution​

import asyncio
from qcos import AsyncQCOSClient

async def run_circuits():
    async with AsyncQCOSClient() as client:
        # Submit multiple circuits concurrently
        tasks = [
            client.submit(qasm=circuit1, shots=1024),
            client.submit(qasm=circuit2, shots=1024),
            client.submit(qasm=circuit3, shots=1024),
        ]
        jobs = await asyncio.gather(*tasks)
        
        # Wait for all results
        results = await asyncio.gather(*[job.wait() for job in jobs])
        
        for i, result in enumerate(results):
            print(f"Circuit {i+1}: {result.counts}")

asyncio.run(run_circuits())

Circuit Input Formats​

OpenQASM String​

qasm = """
OPENQASM 2.0;
include "qelib1.inc";
qreg q[2];
creg c[2];
h q[0];
cx q[0],q[1];
measure q -> c;
"""
result = client.execute(qasm=qasm, shots=1024)

From File​

result = client.execute_file("circuit.qasm", shots=1024)

Qiskit QuantumCircuit​

from qiskit import QuantumCircuit

qc = QuantumCircuit(2, 2)
qc.h(0)
qc.cx(0, 1)
qc.measure([0, 1], [0, 1])

result = client.execute_qiskit(qc, shots=1024)

Cirq Circuit​

import cirq

q0, q1 = cirq.LineQubit.range(2)
circuit = cirq.Circuit([
    cirq.H(q0),
    cirq.CNOT(q0, q1),
    cirq.measure(q0, q1, key='result')
])

result = client.execute_cirq(circuit, shots=1024)

Result Object​

result = client.execute(qasm=circuit, shots=1024)

# === Basic Properties ===
result.job_id           # str: Job identifier
result.status           # str: 'completed', 'failed', etc.
result.counts           # dict: {'00': 512, '11': 512}

# === Execution Details ===
result.num_qubits       # int: Number of qubits
result.shots            # int: Shots executed
result.execution_time   # float: Time in seconds
result.backend          # str: 'cuStateVec', 'AerSimulator'
result.device           # str: 'NVIDIA A100', 'CPU'

# === Timestamps ===
result.created_at       # datetime: Job submission time
result.completed_at     # datetime: Completion time

# === Methods ===
result.probabilities()  # dict: {'00': 0.5, '11': 0.5}
result.most_likely()    # str: Most probable state
result.to_dict()        # dict: Full result as dictionary
result.to_json()        # str: JSON representation

Batch Processing​

Execute multiple circuits efficiently:

circuits = [
    {"qasm": circuit1, "shots": 1024},
    {"qasm": circuit2, "shots": 2048},
    {"qasm": circuit3, "shots": 512},
]

# Submit all at once
batch = client.batch_submit(circuits)
print(f"Batch ID: {batch.batch_id}")
print(f"Jobs: {len(batch.jobs)}")

# Wait for all results
results = batch.wait_all()

# Or process as they complete
for result in batch.as_completed():
    print(f"Job {result.job_id}: {result.counts}")

Error Handling​

from qcos import (
    QCOSClient,
    QCOSError,
    AuthenticationError,
    RateLimitError,
    QuotaExceededError,
    CircuitError,
    TimeoutError,
)

client = QCOSClient()

try:
    result = client.execute(qasm=circuit, shots=1024)
    
except AuthenticationError as e:
    print(f"Authentication failed: {e.message}")
    
except RateLimitError as e:
    print(f"Rate limit exceeded. Reset at: {e.reset_time}")
    
except QuotaExceededError as e:
    print(f"Quota exceeded: {e.limit} qubits max, you requested {e.requested}")
    
except CircuitError as e:
    print(f"Invalid circuit: {e.message}")
    print(f"Line {e.line}: {e.details}")
    
except TimeoutError as e:
    print(f"Job timed out after {e.timeout}s")
    
except QCOSError as e:
    print(f"QCOS error: {e.message}")

Logging​

import logging
from qcos import QCOSClient

# Configure logging
logging.basicConfig(level=logging.DEBUG)

# Or set on client
client = QCOSClient(log_level="DEBUG")

# Log messages:
# DEBUG: Connecting to https://api.softquantus.com
# DEBUG: Submitting circuit (2 qubits, 1024 shots)
# DEBUG: Job abc123 queued
# DEBUG: Polling status...
# INFO: Job abc123 completed in 1.2s

Context Manager​

from qcos import QCOSClient

# Automatic cleanup
with QCOSClient() as client:
    result = client.execute(qasm=circuit, shots=1024)
    print(result.counts)
# Connection closed automatically

Type Hints​

The SDK is fully typed for IDE support:

from qcos import QCOSClient, ExecutionResult

def run_circuit(qasm: str, shots: int = 1024) -> dict[str, int]:
    client: QCOSClient = QCOSClient()
    result: ExecutionResult = client.execute(qasm=qasm, shots=shots)
    return result.counts

Next Steps​