Client API

The qubitos.client module provides Python clients for communicating with the QubitOS HAL (Hardware Abstraction Layer) server via gRPC.

Overview

Two client classes are available:

Class	Type	Use Case
HALClient	Async	High-throughput, concurrent operations
HALClientSync	Sync	Scripts, REPL, simple use cases

Quick Start

Async Client

import asyncio
from qubitos.client import HALClient

async def main():
    async with HALClient("localhost:50051") as client:
        # Check health
        health = await client.health_check()
        print(f"Status: {health.status}")

        # Execute a pulse
        result = await client.execute_pulse(
            i_envelope=[0.1, 0.5, 0.9, 0.5, 0.1],
            q_envelope=[0.0, 0.0, 0.0, 0.0, 0.0],
            duration_ns=20,
            target_qubits=[0],
            num_shots=1024,
        )
        print(f"Counts: {result.bitstring_counts}")

asyncio.run(main())

Sync Client

from qubitos.client import HALClientSync

with HALClientSync("localhost:50051") as client:
    health = client.health_check()
    print(f"Status: {health.status}")

---

Connection Management

Using Context Manager (Recommended)

# Async
async with HALClient("localhost:50051") as client:
    # Connection is managed automatically
    result = await client.execute_pulse(...)

# Sync
with HALClientSync("localhost:50051") as client:
    result = client.execute_pulse(...)

Manual Connection

client = HALClient("localhost:50051")
await client.connect()
try:
    result = await client.execute_pulse(...)
finally:
    await client.close()

Secure Connections

import grpc

# With default TLS
client = HALClient("hal.example.com:50051", secure=True)

# With custom credentials
credentials = grpc.ssl_channel_credentials(
    root_certificates=open("ca.pem", "rb").read()
)
client = HALClient("hal.example.com:50051", credentials=credentials)

---

Executing Pulses

The execute_pulse method sends pulse envelopes to a backend for execution:

result = await client.execute_pulse(
    # Required parameters
    i_envelope=[0.1, 0.5, 0.9, 0.5, 0.1],  # In-phase envelope (MHz)
    q_envelope=[0.0, 0.0, 0.0, 0.0, 0.0],  # Quadrature envelope (MHz)
    duration_ns=20,                         # Pulse duration
    target_qubits=[0],                      # Target qubit indices

    # Optional parameters
    num_shots=1000,                         # Measurement shots
    backend_name="qutip_simulator",         # Backend to use
    measurement_basis="z",                  # Measurement basis
    return_state_vector=False,              # Return state vector
    include_noise=False,                    # Enable noise simulation
    gate_type="X",                          # Gate type hint
)

With GRAPE-Optimized Pulses

from qubitos.pulsegen import generate_pulse
from qubitos.client import HALClientSync

# Generate optimized pulse
pulse = generate_pulse("X", duration_ns=20, target_fidelity=0.999)

# Execute on backend
with HALClientSync("localhost:50051") as client:
    result = client.execute_pulse(
        i_envelope=pulse.i_envelope.tolist(),
        q_envelope=pulse.q_envelope.tolist(),
        duration_ns=20,
        target_qubits=[0],
        num_shots=1024,
    )

print(f"Measurement results: {result.bitstring_counts}")

---

Health Checks

Monitor backend status:

# Check all backends
health = await client.health_check()
print(f"Overall: {health.status}")
print(f"Message: {health.message}")

# Check specific backend
health = await client.health_check(backend_name="qutip_simulator")

# Per-backend status
for name, status in health.backends.items():
    print(f"  {name}: {status}")

---

Hardware Information

Query backend capabilities:

info = await client.get_hardware_info(backend_name="qutip_simulator")

print(f"Name: {info.name}")
print(f"Type: {info.backend_type}")
print(f"Qubits: {info.num_qubits}")
print(f"Available: {info.available_qubits}")
print(f"Gates: {info.supported_gates}")
print(f"State vector: {info.supports_state_vector}")
print(f"Noise model: {info.supports_noise_model}")

---

Listing Backends

Get available backends:

backends = await client.list_backends()
print(f"Available backends: {backends}")

---

Error Handling

from qubitos.client import HALClient, HALClientError

try:
    async with HALClient("localhost:50051") as client:
        result = await client.execute_pulse(...)
except HALClientError as e:
    print(f"Error: {e}")
    print(f"Code: {e.code}")  # gRPC error code

Common error codes:

Code	Meaning
UNAVAILABLE	Server not reachable
DEADLINE_EXCEEDED	Request timed out
INVALID_ARGUMENT	Bad request parameters
NOT_FOUND	Backend not found

---

API Reference

qubitos.client.hal.HALClient

HALClient(
    address: str = "localhost:50051",
    timeout: float = 30.0,
    secure: bool = False,
    credentials: ChannelCredentials | None = None,
)

Async gRPC client for the HAL server.

The client provides methods for: - Executing pulse sequences - Checking backend health - Getting hardware information

Usage

async with HALClient("localhost:50051") as client: ... result = await client.execute_pulse(...)

Or without context manager

client = HALClient("localhost:50051") await client.connect() try: ... result = await client.execute_pulse(...) ... finally: ... await client.close()

Initialize the HAL client.

Parameters:

Name	Type	Description	Default
address	str	HAL server address (host:port)	'localhost:50051'
timeout	float	Default timeout for RPC calls in seconds	30.0
secure	bool	Whether to use TLS	False
credentials	ChannelCredentials | None	gRPC credentials for secure connections	None

Source code in src/qubitos/client/hal.py

def __init__(
    self,
    address: str = "localhost:50051",
    timeout: float = 30.0,
    secure: bool = False,
    credentials: grpc.ChannelCredentials | None = None,
):
    """Initialize the HAL client.

    Args:
        address: HAL server address (host:port)
        timeout: Default timeout for RPC calls in seconds
        secure: Whether to use TLS
        credentials: gRPC credentials for secure connections
    """
    self.address = address
    self.timeout = timeout
    self.secure = secure
    self.credentials = credentials

    self._channel: grpc.aio.Channel | None = None
    self._stub: QuantumBackendServiceStub | None = None
    self._connected = False

connect async

connect() -> None

Connect to the HAL server.

Source code in src/qubitos/client/hal.py

async def connect(self) -> None:
    """Connect to the HAL server."""
    if self._connected:
        return

    logger.info(f"Connecting to HAL server at {self.address}")

    try:
        if self.secure:
            if self.credentials is None:
                self.credentials = grpc.ssl_channel_credentials()
            self._channel = grpc.aio.secure_channel(self.address, self.credentials)
        else:
            self._channel = grpc.aio.insecure_channel(self.address)

        # Create the gRPC stub using generated proto classes
        self._stub = QuantumBackendServiceStub(self._channel)
        self._connected = True

        logger.info("Connected to HAL server")

    except Exception as e:
        logger.error(f"Failed to connect to HAL server: {e}")
        raise HALClientError(f"Connection failed: {e}", code="CONNECTION_ERROR") from e

close async

close() -> None

Close the connection to the HAL server.

Source code in src/qubitos/client/hal.py

async def close(self) -> None:
    """Close the connection to the HAL server."""
    if self._channel is not None:
        await self._channel.close()
        self._channel = None
        self._stub = None
        self._connected = False
        logger.info("Disconnected from HAL server")

health_check async

health_check(
    backend_name: str | None = None,
) -> HealthCheckResult

Check the health of the HAL server and backends.

Parameters:

Name	Type	Description	Default
backend_name	str | None	Specific backend to check (or None for all)	None

Returns:

Type	Description
HealthCheckResult	HealthCheckResult with status information

Source code in src/qubitos/client/hal.py

async def health_check(
    self,
    backend_name: str | None = None,
) -> HealthCheckResult:
    """Check the health of the HAL server and backends.

    Args:
        backend_name: Specific backend to check (or None for all)

    Returns:
        HealthCheckResult with status information
    """
    self._ensure_connected()
    assert self._stub is not None

    try:
        request = HealthRequest(backend_name=backend_name or "")
        response = await self._stub.Health(
            request,
            timeout=self.timeout,
        )

        # Parse response
        status = _parse_health_status(response.status)
        backends = {}
        for backend_status in response.backend_statuses:
            backends[backend_status.name] = _parse_health_status(backend_status.status)

        return HealthCheckResult(
            status=status,
            message=response.message if hasattr(response, "message") else "",
            backends=backends,
        )

    except grpc.RpcError as e:
        raise HALClientError(
            f"Health check failed: {e.details()}",
            code=e.code().name,
        ) from e

get_hardware_info async

get_hardware_info(
    backend_name: str | None = None,
) -> HardwareInfo

Get hardware information for a backend.

Parameters:

Name	Type	Description	Default
backend_name	str | None	Backend name (or None for default)	None

Returns:

Type	Description
HardwareInfo	HardwareInfo with backend details

Source code in src/qubitos/client/hal.py

async def get_hardware_info(
    self,
    backend_name: str | None = None,
) -> HardwareInfo:
    """Get hardware information for a backend.

    Args:
        backend_name: Backend name (or None for default)

    Returns:
        HardwareInfo with backend details
    """
    self._ensure_connected()
    assert self._stub is not None

    try:
        request = GetHardwareInfoRequest(backend_name=backend_name or "")
        response = await self._stub.GetHardwareInfo(
            request,
            timeout=self.timeout,
        )

        info = response.info
        return HardwareInfo(
            name=info.name,
            backend_type=BackendType.SIMULATOR
            if info.backend_type == 0
            else BackendType.HARDWARE,
            tier=info.tier,
            num_qubits=info.num_qubits,
            available_qubits=list(info.available_qubits),
            supported_gates=list(info.supported_gates),
            supports_state_vector=info.supports_state_vector,
            supports_noise_model=info.supports_noise_model,
            software_version=info.software_version,
        )

    except grpc.RpcError as e:
        raise HALClientError(
            f"Get hardware info failed: {e.details()}",
            code=e.code().name,
        ) from e

execute_pulse async

execute_pulse(
    i_envelope: Sequence[float],
    q_envelope: Sequence[float],
    duration_ns: int,
    target_qubits: Sequence[int],
    num_shots: int = 1000,
    pulse_id: str | None = None,
    backend_name: str | None = None,
    measurement_basis: str = "z",
    return_state_vector: bool = False,
    include_noise: bool = False,
    gate_type: str = "CUSTOM",
) -> MeasurementResult

Execute a pulse sequence on a backend.

Parameters:

Name	Type	Description	Default
i_envelope	Sequence[float]	I (in-phase) pulse envelope (MHz)	required
q_envelope	Sequence[float]	Q (quadrature) pulse envelope (MHz)	required
duration_ns	int	Pulse duration in nanoseconds	required
target_qubits	Sequence[int]	Target qubit indices	required
num_shots	int	Number of measurement shots	1000
pulse_id	str | None	Optional pulse identifier	None
backend_name	str | None	Backend to use (or None for default)	None
measurement_basis	str	Measurement basis ("x", "y", or "z")	'z'
return_state_vector	bool	Whether to return the state vector	False
include_noise	bool	Whether to include noise simulation	False
gate_type	str	Gate type name (e.g., "X", "H", "CZ", "CUSTOM")	'CUSTOM'

Returns:

Type	Description
MeasurementResult	MeasurementResult with bitstring counts and metadata

Source code in src/qubitos/client/hal.py

async def execute_pulse(
    self,
    i_envelope: Sequence[float],
    q_envelope: Sequence[float],
    duration_ns: int,
    target_qubits: Sequence[int],
    num_shots: int = 1000,
    pulse_id: str | None = None,
    backend_name: str | None = None,
    measurement_basis: str = "z",
    return_state_vector: bool = False,
    include_noise: bool = False,
    gate_type: str = "CUSTOM",
) -> MeasurementResult:
    """Execute a pulse sequence on a backend.

    Args:
        i_envelope: I (in-phase) pulse envelope (MHz)
        q_envelope: Q (quadrature) pulse envelope (MHz)
        duration_ns: Pulse duration in nanoseconds
        target_qubits: Target qubit indices
        num_shots: Number of measurement shots
        pulse_id: Optional pulse identifier
        backend_name: Backend to use (or None for default)
        measurement_basis: Measurement basis ("x", "y", or "z")
        return_state_vector: Whether to return the state vector
        include_noise: Whether to include noise simulation
        gate_type: Gate type name (e.g., "X", "H", "CZ", "CUSTOM")

    Returns:
        MeasurementResult with bitstring counts and metadata
    """
    self._ensure_connected()
    assert self._stub is not None

    # Generate pulse_id if not provided
    if pulse_id is None:
        pulse_id = str(uuid.uuid4())

    # Map gate type string to enum
    gate_type_enum = _parse_gate_type(gate_type)

    try:
        # Build PulseShape
        pulse = PulseShape(
            pulse_id=pulse_id,
            algorithm="grape",
            gate_type=gate_type_enum,
            target_qubit_indices=list(target_qubits),
            duration_ns=duration_ns,
            num_time_steps=len(i_envelope),
            time_step_ns=duration_ns / len(i_envelope) if i_envelope else 0,
            i_envelope=list(i_envelope),
            q_envelope=list(q_envelope),
        )

        # Build request
        request = ExecutePulseRequest(
            backend_name=backend_name or "qutip_simulator",
            pulse=pulse,
            num_shots=num_shots,
            measurement_basis=measurement_basis.lower(),
            measurement_qubits=list(target_qubits),
            return_state_vector=return_state_vector,
            include_noise=include_noise,
        )

        response = await self._stub.ExecutePulse(
            request,
            timeout=self.timeout,
        )

        # Check for errors
        if not response.success and response.error:
            raise HALClientError(
                response.error.message,
                code=str(response.error.code),
            )

        # Parse result
        result = response.result
        bitstring_counts = dict(result.bitstring_counts)

        state_vector = None
        if result.state_vector and result.state_vector.amplitudes:
            # Amplitudes are stored as [re_0, im_0, re_1, im_1, ...]
            amps = result.state_vector.amplitudes
            state_vector = [(amps[i], amps[i + 1]) for i in range(0, len(amps), 2)]

        return MeasurementResult(
            request_id=str(uuid.uuid4()),  # Generate locally since response may not have it
            pulse_id=pulse_id,
            bitstring_counts={k: int(v) for k, v in bitstring_counts.items()},
            total_shots=result.total_shots,
            successful_shots=result.successful_shots,
            fidelity_estimate=result.fidelity_estimate if result.fidelity_estimate else None,
            state_vector=state_vector,
        )

    except grpc.RpcError as e:
        raise HALClientError(
            f"Execute pulse failed: {e.details()}",
            code=e.code().name,
        ) from e

list_backends async

list_backends() -> list[str]

List available backends.

Returns:

Type	Description
list[str]	List of backend names

Source code in src/qubitos/client/hal.py

async def list_backends(self) -> list[str]:
    """List available backends.

    Returns:
        List of backend names
    """
    self._ensure_connected()
    assert self._stub is not None

    try:
        request = ListBackendsRequest(include_details=False)
        response = await self._stub.ListBackends(
            request,
            timeout=self.timeout,
        )
        return list(response.backend_names)

    except grpc.RpcError as e:
        raise HALClientError(
            f"List backends failed: {e.details()}",
            code=e.code().name,
        ) from e

qubitos.client.hal.HALClientSync

HALClientSync(*args, **kwargs)

Synchronous wrapper for HALClient.

Useful for scripts and REPL usage where async isn't convenient.

Example

with HALClientSync("localhost:50051") as client: ... health = client.health_check()

Source code in src/qubitos/client/hal.py

def __init__(self, *args, **kwargs):
    self._client = HALClient(*args, **kwargs)
    self._loop: asyncio.AbstractEventLoop | None = None

health_check

health_check(
    backend_name: str | None = None,
) -> HealthCheckResult

Source code in src/qubitos/client/hal.py

def health_check(self, backend_name: str | None = None) -> HealthCheckResult:
    return self._get_loop().run_until_complete(self._client.health_check(backend_name))

get_hardware_info

get_hardware_info(
    backend_name: str | None = None,
) -> HardwareInfo

Source code in src/qubitos/client/hal.py

def get_hardware_info(self, backend_name: str | None = None) -> HardwareInfo:
    return self._get_loop().run_until_complete(self._client.get_hardware_info(backend_name))

execute_pulse

execute_pulse(**kwargs) -> MeasurementResult

Source code in src/qubitos/client/hal.py

def execute_pulse(self, **kwargs) -> MeasurementResult:
    return self._get_loop().run_until_complete(self._client.execute_pulse(**kwargs))

list_backends

list_backends() -> list[str]

Source code in src/qubitos/client/hal.py

def list_backends(self) -> list[str]:
    return self._get_loop().run_until_complete(self._client.list_backends())

Data Types

qubitos.client.hal.HealthStatus

Bases: Enum

Backend health status.

qubitos.client.hal.BackendType

Bases: Enum

Backend type.

qubitos.client.hal.HardwareInfo dataclass

HardwareInfo(
    name: str,
    backend_type: BackendType,
    tier: str,
    num_qubits: int,
    available_qubits: list[int],
    supported_gates: list[str],
    supports_state_vector: bool,
    supports_noise_model: bool,
    software_version: str,
)

Information about a quantum backend.

Attributes:

Name	Type	Description
name	str	Backend name
backend_type	BackendType	Type of backend (simulator or hardware)
tier	str	Backend tier (local, cloud, etc.)
num_qubits	int	Number of qubits
available_qubits	list[int]	List of available qubit indices
supported_gates	list[str]	List of supported gate names
supports_state_vector	bool	Whether state vector output is supported
supports_noise_model	bool	Whether noise modeling is supported
software_version	str	Backend software version

qubitos.client.hal.MeasurementResult dataclass

MeasurementResult(
    request_id: str,
    pulse_id: str,
    bitstring_counts: dict[str, int],
    total_shots: int,
    successful_shots: int,
    fidelity_estimate: float | None = None,
    state_vector: list[tuple[float, float]] | None = None,
)

Result of a pulse execution.

Attributes:

Name	Type	Description
request_id	str	Unique request identifier
pulse_id	str	Pulse identifier
bitstring_counts	dict[str, int]	Dictionary mapping bitstrings to counts
total_shots	int	Total number of shots requested
successful_shots	int	Number of successful shots
fidelity_estimate	float | None	Estimated gate fidelity (if computed)
state_vector	list[tuple[float, float]] | None	State vector as list of (real, imag) tuples (if requested)

qubitos.client.hal.HealthCheckResult dataclass

HealthCheckResult(
    status: HealthStatus,
    message: str = "",
    backends: dict[str, HealthStatus] = dict(),
)

Result of a health check.

Attributes:

Name	Type	Description
status	HealthStatus	Overall health status
message	str	Optional status message
backends	dict[str, HealthStatus]	Per-backend health status

qubitos.client.hal.HALClientError

HALClientError(message: str, code: str | None = None)

Bases: Exception

Error from HAL client operations.

Source code in src/qubitos/client/hal.py

def __init__(self, message: str, code: str | None = None):
    self.code = code
    super().__init__(message)