Python SDK Reference

Package: marqov (v0.2.0-dev)

Install: pip install marqov

Optional extras: pip install marqov[qiskit], marqov[cirq], marqov[pennylane], marqov[openqasm]

marqov.circuits

Circuit

Backend-agnostic quantum circuit with a fluent API. All gate methods return self for chaining.

from marqov import Circuit
 
circuit = Circuit().h(0).cnot(0, 1)

Constructor:

Circuit() -> Circuit

Creates an empty circuit.

Properties:

Single-qubit gates:

Rotation gates (angles in radians):

Two-qubit gates:

Backend conversion:

Import methods (class methods):

Serialization:

Simulation:

bell_state

def bell_state() -> Circuit

Create a Bell state circuit (|00> + |11>)/sqrt(2). Equivalent to Circuit().h(0).cnot(0, 1).

ghz_state

def ghz_state(num_qubits: int) -> Circuit

Create a GHZ state circuit for num_qubits qubits. Applies H to qubit 0, then CNOT chain.

marqov.device

MarqovDevice

Unified interface for quantum backend execution. Accepts any supported circuit type and automatically converts to the target backend’s native format.

from marqov import MarqovDevice, get_device
 
device = get_device({"backend": "sv1", "device_arn": "...", "s3_bucket": "..."})
counts = device.run(circuit, shots=1000)

Constructor:

MarqovDevice(backend: str, params: dict) -> MarqovDevice

Properties:

Methods:

run

def run(self, circuit, shots: int = 1000) -> dict[str, int]

Execute a circuit and return measurement counts.

Returns: Dictionary mapping bitstring outcomes to counts (e.g., {"00": 512, "11": 488}).

Raises: TypeError if circuit type is not supported. ValueError if S3 configuration is missing for AWS devices.

Supported circuit types for auto-conversion:

• marqov.Circuit (native)
• str (OpenQASM)
• braket.circuits.Circuit
• qiskit.QuantumCircuit
• cirq.Circuit
• pennylane.tape.QuantumScript

get_device

def get_device(params: dict) -> MarqovDevice

Factory function to create a MarqovDevice from execution parameters.

Required params keys by provider:

Raises: ValueError if backend is missing.

marqov.executors

BaseExecutor (abstract)

Abstract base class for all quantum executors.

class BaseExecutor(ABC):
    async def execute(self, circuit: Circuit, shots: int = 1000, **kwargs) -> ExecutionResult: ...
    async def cancel(self, job_id: str) -> bool: ...
    @property
    def name(self) -> str: ...

ExecutionResult

Dataclass holding results from circuit execution.

@dataclass
class ExecutionResult:
    counts: dict[str, int]
    backend: str
    execution_time_ms: float
    shots: int = 0
    raw_result: Any = None
    metadata: dict[str, Any] = field(default_factory=dict)

Properties:

LocalExecutor

Execute circuits on the QuantumFlow local state vector simulator. No cloud credentials required.

from marqov import LocalExecutor
 
executor = LocalExecutor()
result = await executor.execute(circuit, shots=1000)

Constructor:

LocalExecutor(config: LocalExecutorConfig | None = None)

LocalExecutorConfig fields:

BraketExecutor

Execute circuits on AWS Braket devices (simulators and QPUs).

from marqov.executors import BraketExecutor, BraketExecutorConfig
 
config = BraketExecutorConfig(
    device_arn="arn:aws:braket:::device/quantum-simulator/amazon/sv1",
    s3_bucket="amazon-braket-my-bucket",
)
executor = BraketExecutor(config)
result = await executor.execute(circuit, shots=1000)

BraketExecutorConfig fields:

Additional methods:

AzureQuantumExecutor

Execute circuits on Azure Quantum backends (Quantinuum, PASQAL, IonQ, Rigetti).

AzureQuantumExecutorConfig fields:

ExecutorFactory

Factory for creating executors based on provider configuration.

from marqov.executors import ExecutorFactory
 
executor = ExecutorFactory.create_executor("sv1", {
    "provider": "AWS Braket",
    "device_arn": "arn:aws:braket:::device/quantum-simulator/amazon/sv1",
    "s3_bucket": "my-bucket",
})

Class methods:

Supported providers: AWS Braket, Azure Quantum, Quantum Brilliance, Local. Coming soon: IBM Quantum, IonQ Direct.

marqov.workflows

task

Decorator to mark a function as a task (unit of work in a workflow).

from marqov import task
 
@task
def add(x, y):
    return x + y
 
# With options:
@task(executor="braket", retries=3, timeout=600)
async def measure(circuit):
    return await executor.run(circuit)

Parameters (when used with options):

Behavior:

• Outside a @workflow: executes normally and returns the result.
• Inside a @workflow: returns a TaskProxy, registering the call in the transport graph for deferred execution.

workflow

Decorator to mark a function as a workflow (composition of tasks).

from marqov import task, workflow
 
@workflow
def vqe_step(theta):
    circuit = build(theta)
    z0 = measure(circuit, "ZI")  # runs in parallel
    z1 = measure(circuit, "IZ")  # runs in parallel
    return compute(z0, z1)
 
dispatch = vqe_step(0.5)
result = await dispatch.run(client)

Parameters:

Returns: WorkflowDispatch object.

WorkflowDispatch

Handle for dispatching a workflow to Temporal.

dispatch = my_workflow(args)

Methods:

Attributes:

TemporalConfig

Configuration for Temporal connection.

from marqov import TemporalConfig
 
config = TemporalConfig(
    host="localhost",
    port=7233,
    namespace="default",
    task_queue="marqov-workflows",
)
print(config.address)  # "localhost:7233"

Properties:

create_worker

async def create_worker(client, task_queue: str = "marqov-workflows") -> Worker

Create a Temporal worker that processes Marqov workflows. The worker registers the JobWorkflow and execute_task activity.

TaskConfig

Configuration for an individual task.

Deprecated Aliases

The following are deprecated and will be removed in v0.3.0: