Plugins

QDK/Chemistry uses a plugin system to support multiple implementations of each of the available algorithm type. This allows switching between native QDK implementations and third-party packages (e.g., PySCF, Qiskit) without modifying application code.

Plugin system

Architecture

Each algorithm in QDK/Chemistry can have multiple implementations. All implementations inherit from the same base class and conform to the same interface:

This design supports several workflows:

Benchmarking native implementations against established packages
Mixing backends (e.g., PySCF for SCF, MACIS for multi-configurational methods)
Adding custom implementations

The implementations for each algorithm type are managed by a factory class, which provides a consistent interface for creating instances and listing available implementations. We refer the reader to the factory pattern and algorithm documentation pages for more details on this design pattern.

Using plugins

To select an implementation, specify it by name:

Python API

from qdk_chemistry.algorithms import available, create
from qdk_chemistry.data import Structure

# Load H2 molecule from inline XYZ file
structure = Structure.from_xyz("""\
2
H2 molecule
H    0.000000    0.000000    0.000000
H    0.000000    0.000000    0.740848
""")

# Create a SCF solver using the factory
scf_solver = create("scf_solver", "pyscf")

# Configure it using the standard settings interface
scf_solver.settings().set("method", "hf")

# Run calculation - returns (energy, wavefunction)
energy, wavefunction = scf_solver.run(
    structure, charge=0, spin_multiplicity=1, basis_or_guess="cc-pvdz"
)
orbitals = wavefunction.get_orbitals()

print(f"SCF Energy: {energy:.10f} Hartree")

C++ API

#include <qdk/chemistry.hpp>

// Create a SCF solver that uses the QDK/Chemistry library as solver
auto scf = ScfSolverFactory::create();

// Configure it using the standard settings interface
scf->settings().set("method", "hf");

// Run calculation with the same API as native implementations
auto [energy, orbitals] =
    scf->solve(structure, charge, spin_multiplicity, basis_set_name);

To list available implementations:

Python API

# List available implementations for each algorithm type
for algorithm_name in available():
    print(f"{algorithm_name} has methods:")
    for method_name in available(algorithm_name):
        print(f"  {method_name} has settings:")
        method = create(algorithm_name, method_name)
        settings = method.settings()
        for key, value in settings.items():
            print(f"    {key}: {value}")

C++ API

#include <iostream>
#include <qdk/chemistry.hpp>

// Get a list of available SCF solver implementations
auto available_solvers = ScfSolverFactory::available();
for (const auto& solver : available_solvers) {
  std::cout << solver << std::endl;
}

// Get documentation for a specific implementation
std::cout << ScfSolverFactory::get_docstring("default") << std::endl;

Documentation pertaining to the availability and configuration of each algorithm implementation provided within QDK/Chemistry can be found on the algorithm documentation pages.

Included third-party plugins

In addition to the native implementations packaged within QDK/Chemistry, plugins are included for the following packages:

PySCF — Python-based quantum chemistry
Qiskit — Quantum algorithm primitives
OpenFermion — Quantum algorithm primitives
geomeTRIC — Molecular geometry optimization

These plugins are enabled automatically when the corresponding package is installed.

PySCF plugin details

The PySCF plugin is installed via the plugins extra:

pip install 'qdk-chemistry[plugins]'

Note

PySCF publishes no Windows wheels, so the plugins extra installs geomeTRIC but omits PySCF on native Windows. The test and all extras include plugins and likewise install without PySCF there. The jupyter extra does not include plugins.

The native QDK/Chemistry implementations are unaffected and remain available on Windows. To use the PySCF plugin on a Windows machine, work inside WSL.

Qiskit plugin details

The Qiskit plugin uses opportunistic loading to maximize compatibility across different installation configurations. When Qiskit is installed, the plugin will load and register the available algorithms. The optional ecosystem packages (Qiskit Aer and Qiskit Nature) are loaded based on their availability in your environment.

Loading behavior:

Qiskit (core): Loaded when the plugin is initialized and Qiskit is installed.
Qiskit Nature: Loaded if qiskit-nature is installed.
Qiskit Aer: Loaded if qiskit-aer is installed.

Installing optional Qiskit packages:

To install the optional Qiskit ecosystem packages, use the qiskit-extras extra when installing QDK/Chemistry:

pip install 'qdk-chemistry[qiskit-extras]'

Alternatively, you can install them directly:

pip install qiskit-aer qiskit-nature

Note

On Python 3.14, qiskit-aer is omitted from qiskit-extras on Linux ARM64 (aarch64), because Qiskit does not yet publish a Python 3.14 wheel for that platform. All other platforms install the full set.

Checking what is loaded:

To determine which Qiskit components are available in your environment, you can check the following module-level variables:

from qdk_chemistry.plugins.qiskit import (
    QDK_CHEMISTRY_HAS_QISKIT,
    QDK_CHEMISTRY_HAS_QISKIT_NATURE,
    QDK_CHEMISTRY_HAS_QISKIT_AER,
)

print(f"Qiskit core available: {QDK_CHEMISTRY_HAS_QISKIT}")
print(f"Qiskit Nature available: {QDK_CHEMISTRY_HAS_QISKIT_NATURE}")
print(f"Qiskit Aer available: {QDK_CHEMISTRY_HAS_QISKIT_AER}")

These boolean variables are set at module load time and reflect the actual availability of each package in your Python environment.

Warning

If you attempt to use an algorithm that requires an optional Qiskit package that is not installed, the algorithm will not be available in the factory. Use the listing-implementations pattern to see which implementations are currently available.

OpenFermion plugin details

The OpenFermion plugin integrates QDK/Chemistry with OpenFermion. Like the Qiskit plugin, it uses opportunistic loading: the plugin loads when OpenFermion is installed.

Loading behavior:

OpenFermion: Loaded when the plugin is initialized and OpenFermion is installed.

Installing OpenFermion packages:

To install OpenFermion, use the openfermion-extras extra when installing QDK/Chemistry:

pip install 'qdk-chemistry[openfermion-extras]'

Checking what is loaded:

To determine which OpenFermion components are available in your environment, you can check the following module-level variables:

from qdk_chemistry.plugins.openfermion import (
    QDK_CHEMISTRY_HAS_OPENFERMION,
)

print(f"OpenFermion available: {QDK_CHEMISTRY_HAS_OPENFERMION}")

This boolean variable is set at module load time and reflects the actual availability of each package in your Python environment.

Warning

If you attempt to use an algorithm that requires OpenFermion but the package is not installed, the algorithm will not be available in the factory. Use the listing-implementations pattern to see which implementations are currently available.

Community-developed plugins are also welcome. See Creating plugins for guidance on creating new plugins.

Creating plugins

An installed Python package can contribute any combination of:

Implementations of existing algorithm types
New algorithm types and their implementations
DataClass types used in algorithm inputs or outputs
Remote execution backends
Cache backends

The following sections provide complete remote backend and algorithm examples. The same plugin object can register multiple capabilities through PluginRegistrar.

Registration names must be unique within their registry. Registering a second algorithm, algorithm type, data class, remote backend, or cache backend under an existing name raises DuplicateRegistrationError, a ValueError subclass. The rejected registration does not replace the existing implementation.

Automatic discovery

The plugin contract is a QdkChemistryPlugin subclass exposed through the qdk_chemistry.plugins entry-point group. A plugin package declares this in its pyproject.toml:

[project.entry-points."qdk_chemistry.plugins"]
custom = "custom_package.plugin:CustomScfPlugin"

Importing qdk_chemistry discovers the installed package and calls its register method; users do not need to import the plugin module themselves. A plugin registers its capabilities through the supplied registrar:

from qdk_chemistry.plugins import PluginRegistrar, QdkChemistryPlugin

class CustomPlugin(QdkChemistryPlugin):
    def register(self, registrar: PluginRegistrar) -> None:
        registrar.register_algorithm(lambda: CustomAlgorithm())
        registrar.register_dataclass(CustomResult)
        registrar.register_remote_backend("custom", CustomRemoteBackend)
        registrar.register_cache_backend("custom", CustomCacheBackend)

Each plugin-defined DataClass used in algorithm inputs or outputs must declare a non-empty wire-format identifier in its own class body:

from qdk_chemistry.data import DataClass

class CustomResult(DataClass):
    @staticmethod
    def data_type_name() -> str:
        return "custom_result"
    ...

The value returned by data_type_name() identifies the serialized format during remote and cache deserialization. A canonical loader must declare this static method directly and return a non-empty string. A subclass must declare a unique identifier and register as its own loader. Registration raises TypeError when the declaration is missing or empty, and DuplicateRegistrationError when another loader already owns the identifier.

Register these classes with register_dataclass() or pass them through the data_classes argument of register_algorithm(). Python return annotations are not used for discovery.

Remote backend MCP configuration

Remote backends expose no constructor options to MCP clients by default. To allow a client-controlled option, declare mcp_safe_config_options directly on the concrete backend class. Registration validates that it is a frozenset of non-empty constructor parameter names.

class CustomRemoteBackend(RemoteBackend):
   mcp_safe_config_options = frozenset({"poll_interval", "timeout"})

   def __init__(self, *, endpoint, poll_interval=5.0, timeout=3600.0):
      ...

Do not declare executable paths, credentials, endpoint selection, storage locations, or other options that can redirect execution or access. These remain backend- or server-owned.

Naming and call order

An algorithm implementation name must be unique within its algorithm type; remote backend and cache backend names must be unique within their respective registries. Third-party plugins should use package- or organization-prefixed names to avoid collisions with built-in implementations and other plugins.

Registration is first come, first served and does not override an existing name. Core built-ins are registered before external registrations reach each registry. Unified plugin entry points are called in the order returned by Python’s entry-point discovery, followed by bundled optional integrations. If a plugin reuses an existing name, registration raises DuplicateRegistrationError and keeps the earlier implementation unchanged. During entry-point discovery, QDK/Chemistry catches that exception, emits a UserWarning identifying the plugin that failed to register, and continues loading other plugins. Because entry-point order can vary between environments, plugins must not rely on discovery order to override another implementation.

Implementing a remote backend

A remote backend implements the transport and job lifecycle required to execute a serialized QDK/Chemistry request outside the calling process. The following example uses the system ssh and scp commands to transfer files and launch a background process.

Note

This example targets a directly SSH-accessible machine with python3 and QDK/Chemistry available in its default environment. It does not submit through a queue scheduler such as SLURM or PBS. Queue-managed systems should provide a backend designed for their scheduler and site policy.

The example retains each remote job directory, including its inputs, outputs, PID file, and logs, after the job reaches a terminal state. Applications are responsible for removing these artifacts through job cleanup. That cleanup does not remove caller-owned local job records or result directories.

Backend implementation

Implement RemoteBackend for the target transport and execution environment:

from __future__ import annotations

import shlex
import subprocess
import uuid
from pathlib import Path, PurePosixPath

from qdk_chemistry.remote.backends import (
    DEFAULT_POLL_INTERVAL,
    DEFAULT_TIMEOUT,
    JobState,
    JobStatus,
    RemoteBackend,
)


def _parse_remote_pid(output: str) -> str | None:
    """Return a normalized positive PID from remote command output.

    Args:
        output: Standard output containing the remote process identifier.

    """
    pid = output.strip()
    if not pid.isascii() or not pid.isdecimal():
        return None
    return pid.lstrip("0") or None


class SSHBackend(RemoteBackend):
    """Run QDK/Chemistry jobs on a directly SSH-accessible machine.

    Only ``connect``, ``disconnect``, ``upload``, and ``download`` are abstract
    in ``RemoteBackend``. The asynchronous job hooks ``_submit``, ``check``,
    ``cancel``, and ``fetch`` have default implementations that raise
    ``NotImplementedError``, so a backend can leave unsupported hooks unchanged.
    This example overrides all five, including ``cleanup_job``, to support the
    complete asynchronous job lifecycle.

    Job directories under ``remote_workdir`` are retained after they reach a
    terminal state. Job cleanup removes only these remote directories; callers
    remain responsible for local job records and result directories.
    """

    # MCP clients may tune polling behavior but cannot select the host,
    # credentials, remote workspace, SSH arguments, or worker executable.
    mcp_safe_config_options = frozenset({"poll_interval", "timeout"})

    def __init__(
        self,
        *,
        host: str,
        poll_interval: float = DEFAULT_POLL_INTERVAL,
        timeout: float = DEFAULT_TIMEOUT,
        remote_workdir: str = "/tmp/qdk_remote",
        identity_file: str | Path | None = None,
        ssh_options: list[str] | None = None,
        python_path: str = "python3",
    ) -> None:
        """Initialize the backend with a required SSH host.

        Args:
            host: SSH destination, such as ``"user@hostname"``.
            poll_interval: Seconds between job status checks.
            timeout: Maximum duration for one SSH or SCP operation.
            remote_workdir: Remote directory containing job artifacts.
            identity_file: Optional private-key path passed to SSH and SCP.
            ssh_options: Additional SSH and SCP command-line options.
            python_path: Remote Python executable used to start the worker.

        """
        if not host:
            raise ValueError("SSHBackend requires a host (e.g., 'user@hostname')")
        super().__init__(
            host=host,
            poll_interval=poll_interval,
            timeout=timeout,
            remote_workdir=remote_workdir,
            identity_file=str(identity_file) if identity_file else None,
            ssh_options=list(ssh_options or []),
            python_path=python_path,
        )
        self.host: str = host
        self.timeout = timeout
        self.remote_workdir = remote_workdir
        self.identity_file = identity_file
        self.ssh_options = list(ssh_options or [])
        self.python_path = python_path

    def connect(self) -> None:
        """Implement the abstract connection hook by testing SSH and creating a workdir."""
        result = subprocess.run(
            self._ssh_cmd(["echo", "connected"]),
            check=False,
            capture_output=True,
            text=True,
            timeout=30,
        )
        if result.returncode != 0:
            raise ConnectionError(f"SSH connection failed: {result.stderr}")

        subprocess.run(
            self._ssh_cmd(["mkdir", "-p", self.remote_workdir]),
            check=True,
            timeout=30,
        )

    def disconnect(self) -> None:
        """Implement the abstract disconnection hook as a no-op for one-shot commands."""

    def upload(self, local_path: str | Path, remote_path: str) -> None:
        """Implement the abstract upload hook with SCP.

        Args:
            local_path: Source file on the local machine.
            remote_path: Destination file path on the remote machine.

        """
        local_path = Path(local_path)
        command = [
            "scp",
            *self._ssh_options(),
            str(local_path),
            f"{self._ssh_target()}:{remote_path}",
        ]
        result = subprocess.run(
            command,
            check=False,
            capture_output=True,
            text=True,
            timeout=self.timeout,
        )
        if result.returncode != 0:
            raise RuntimeError(f"SCP upload failed: {result.stderr}")

    def download(self, remote_path: str, local_path: str | Path) -> None:
        """Implement the abstract download hook with SCP.

        Args:
            remote_path: Source file path on the remote machine.
            local_path: Destination file on the local machine.

        """
        local_path = Path(local_path)
        local_path.parent.mkdir(parents=True, exist_ok=True)

        command = [
            "scp",
            *self._ssh_options(),
            f"{self._ssh_target()}:{remote_path}",
            str(local_path),
        ]
        result = subprocess.run(
            command,
            check=False,
            capture_output=True,
            text=True,
            timeout=self.timeout,
        )
        if result.returncode != 0:
            raise RuntimeError(f"SCP download failed: {result.stderr}")

    def _ssh_target(self) -> str:
        """Return the configured SSH target."""
        return self.host

    def _ssh_options(self) -> list[str]:
        """Build the common SSH and SCP options."""
        options = []
        if self.identity_file is not None:
            options.extend(["-i", str(Path(self.identity_file).expanduser())])
        options.extend(self.ssh_options)
        return options

    def _ssh_cmd(self, remote_command: list[str]) -> list[str]:
        """Build an SSH command.

        Args:
            remote_command: Command and arguments to execute on the remote machine.

        """
        command = ["ssh", *self._ssh_options(), self._ssh_target()]
        command.append(shlex.join(remote_command))
        return command

    def _run_remote(
        self, command: str, *, timeout: int | None = None
    ) -> subprocess.CompletedProcess:
        """Run a shell command on the remote machine.

        Args:
            command: Shell command to execute remotely.
            timeout: Optional command timeout in seconds.

        """
        ssh_command = ["ssh", *self._ssh_options(), self._ssh_target(), command]
        return subprocess.run(
            ssh_command,
            check=False,
            capture_output=True,
            text=True,
            timeout=timeout or self.timeout,
        )

    def _submit(self, payload: dict) -> tuple[str, dict]:
        """Override the optional async submission hook to launch an SSH worker.

        Args:
            payload: Serialized algorithm execution request.

        """
        import shutil
        import tempfile

        from qdk_chemistry.remote.serialization import serialize_inputs

        job_id = uuid.uuid4().hex[:12]
        remote_job_dir = f"{self.remote_workdir}/job_{job_id}"
        remote_input_dir = f"{remote_job_dir}/input"
        remote_output_dir = f"{remote_job_dir}/output"

        self._run_remote(
            f"mkdir -p {shlex.quote(remote_input_dir)} {shlex.quote(remote_output_dir)}",
            timeout=30,
        )

        local_input_dir = Path(tempfile.mkdtemp(prefix="qdk_ssh_input_"))
        try:
            input_files = serialize_inputs(
                local_input_dir,
                args=payload["args"],
                kwargs=payload["kwargs"],
                algorithm_type=payload["algorithm_type"],
                algorithm_name=payload["algorithm_name"],
                settings=payload["settings"],
                run_hash=payload.get("run_hash"),
                input_hashes=payload.get("input_hashes"),
                force_rerun=payload.get("force_rerun", False),
                remote_cache=payload.get("remote_cache"),
                remote_cache_backend=payload.get("remote_cache_backend"),
            )

            for local_file in input_files:
                self.upload(local_file, f"{remote_input_dir}/{local_file.name}")
        finally:
            shutil.rmtree(local_input_dir, ignore_errors=True)

        # A detached worker and PID file are choices made by this direct-SSH
        # transport. A scheduler backend would submit through its scheduler and
        # record the resulting scheduler job ID instead.
        background_command = (
            f"cd {shlex.quote(remote_job_dir)} && "
            f"nohup {shlex.quote(self.python_path)} -m qdk_chemistry.remote.worker "
            f"--input-dir {shlex.quote(remote_input_dir)} --output-dir {shlex.quote(remote_output_dir)} "
            f"> {shlex.quote(f'{remote_job_dir}/stdout.log')} "
            f"2> {shlex.quote(f'{remote_job_dir}/stderr.log')} & "
            f"echo $! > {shlex.quote(f'{remote_job_dir}/pid')}"
        )
        result = self._run_remote(background_command, timeout=30)
        if result.returncode != 0:
            raise RuntimeError(f"Failed to launch remote job: {result.stderr}")

        # Job persists this opaque state in JSON and passes it back to check,
        # cancel, and fetch, so every value must be JSON-serializable.
        backend_state = {
            "job_id": job_id,
            "remote_job_dir": remote_job_dir,
            "remote_output_dir": remote_output_dir,
        }
        return job_id, backend_state

    def check(self, backend_state: dict) -> JobStatus:
        """Override the optional async status hook to inspect the SSH worker.

        ``JobState`` defines the canonical case-insensitive lifecycle states.
        Backend-specific status strings are also supported and remain
        nonterminal until mapped to a terminal state.

        Args:
            backend_state: Persisted state for the submitted remote job.

        """
        remote_job_dir = backend_state["remote_job_dir"]

        pid_path = shlex.quote(f"{remote_job_dir}/pid")
        pid_result = self._run_remote(f"cat {pid_path}", timeout=10)
        if pid_result.returncode != 0:
            return JobStatus(
                job_id=backend_state["job_id"],
                status=JobState.FAILED,
                error="Could not read PID file",
            )

        pid = _parse_remote_pid(pid_result.stdout)
        if pid is None:
            return JobStatus(
                job_id=backend_state["job_id"],
                status=JobState.FAILED,
                error="Invalid PID file",
            )

        # ``kill -0`` is this transport's process-liveness probe. A scheduler
        # backend would query scheduler state using its persisted job ID.
        alive = self._run_remote(
            f"kill -0 {pid} 2>/dev/null && echo alive || echo done", timeout=10
        )
        if "alive" in alive.stdout:
            status = JobState.RUNNING
        else:
            manifest_path = shlex.quote(f"{remote_job_dir}/output/manifest.json")
            manifest_check = self._run_remote(
                f"test -f {manifest_path} && echo ok || echo missing",
                timeout=10,
            )
            status = (
                JobState.SUCCEEDED if "ok" in manifest_check.stdout else JobState.FAILED
            )

        stderr_path = shlex.quote(f"{remote_job_dir}/stderr.log")
        logs_result = self._run_remote(
            f"tail -50 {stderr_path} 2>/dev/null", timeout=10
        )
        logs = logs_result.stdout if logs_result.returncode == 0 else ""

        return JobStatus(
            job_id=backend_state["job_id"],
            status=status,
            logs=logs,
            metadata={"pid": pid, "remote_job_dir": remote_job_dir},
        )

    def cancel(self, backend_state: dict) -> None:
        """Override the optional async cancellation hook by signaling the worker PID.

        Args:
            backend_state: Persisted state for the submitted remote job.

        """
        remote_job_dir = backend_state["remote_job_dir"]
        pid_path = shlex.quote(f"{remote_job_dir}/pid")
        pid_result = self._run_remote(f"cat {pid_path}", timeout=10)
        if pid_result.returncode == 0:
            pid = _parse_remote_pid(pid_result.stdout)
            if pid is not None:
                self._run_remote(f"kill {pid} 2>/dev/null", timeout=10)

    def fetch(
        self,
        backend_state: dict,
        local_dir: str | Path | None = None,
    ) -> dict:
        """Override the optional async fetch hook to download and deserialize results.

        Args:
            backend_state: Persisted state for the completed remote job.
            local_dir: Optional directory for downloaded result files.

        """
        import json
        import shutil
        import tempfile

        from qdk_chemistry.remote.serialization import (
            deserialize_outputs,
            get_serialized_file_names,
        )

        remote_output_dir = backend_state["remote_output_dir"]

        own_temporary_directory = local_dir is None
        if own_temporary_directory:
            resolved_dir = Path(tempfile.mkdtemp(prefix="qdk_ssh_fetch_"))
        else:
            assert local_dir is not None
            resolved_dir = Path(local_dir)
            resolved_dir.mkdir(parents=True, exist_ok=True)
        local_dir = resolved_dir

        try:
            manifest_local = local_dir / "manifest.json"
            self.download(f"{remote_output_dir}/manifest.json", manifest_local)

            with open(manifest_local) as manifest_file:
                manifest = json.load(manifest_file)

            for entry in manifest.get("results", []):
                for filename in get_serialized_file_names(entry):
                    self.download(
                        f"{remote_output_dir}/{filename}", local_dir / filename
                    )

            return deserialize_outputs(local_dir)
        finally:
            if own_temporary_directory:
                shutil.rmtree(local_dir, ignore_errors=True)

    def cleanup_job(self, backend_state: dict) -> None:
        """Remove artifacts owned by one completed SSH job.

        Args:
            backend_state: Persisted state for the terminal remote job.

        """
        remote_workdir = PurePosixPath(self.remote_workdir)
        remote_job_dir = PurePosixPath(backend_state["remote_job_dir"])
        remote_output_dir = PurePosixPath(backend_state["remote_output_dir"])
        if (
            remote_job_dir.parent != remote_workdir
            or remote_output_dir.parent != remote_job_dir
        ):
            raise ValueError(
                "Remote job paths are inconsistent with the configured work directory"
            )

        result = self._run_remote(
            f"rm -rf -- {shlex.quote(str(remote_job_dir))}", timeout=30
        )
        if result.returncode != 0:
            raise RuntimeError(f"Failed to clean up remote job: {result.stderr}")

Registration and discovery

Register the backend through PluginRegistrar from a QdkChemistryPlugin:

from qdk_chemistry.plugins import PluginRegistrar, QdkChemistryPlugin


class SSHRemoteBackendPlugin(QdkChemistryPlugin):
    """Register the illustrative SSH remote backend."""

    def register(self, registrar: PluginRegistrar) -> None:
        """Register the backend with QDK/Chemistry.

        Args:
            registrar: Plugin registrar receiving the SSH backend.

        """
        registrar.register_remote_backend("ssh", SSHBackend)

Expose that plugin class through the unified entry-point group in the plugin package’s pyproject.toml:

[project.entry-points."qdk_chemistry.plugins"]
ssh = "custom_package.ssh_backend:SSHRemoteBackendPlugin"

After the package is installed, importing qdk_chemistry discovers the entry point and registers the backend. Application code does not import the plugin module explicitly.

Usage

The discovered backend is available through the standard remote backend registry:

def create_ssh_backend():
    """Create the automatically discovered SSH backend."""
    from qdk_chemistry.remote import available_backends, create_remote

    assert "ssh" in available_backends()
    return create_remote("ssh", host="user@compute.example.com")

Algorithms created through qdk_chemistry.algorithms.create() accept remote and cache keyword arguments on run. create_remote() returns a connected backend, which the caller disconnects when it is no longer needed:

from qdk_chemistry.algorithms import create
from qdk_chemistry.remote import create_remote

scf = create("scf_solver")
remote = create_remote("ssh", host="user@compute.example.com")
try:
    energy, wavefunction = scf.run(
        structure,
        0,
        1,
        "cc-pvdz",
        remote=remote,
        cache="./cache",
    )
finally:
    remote.disconnect()

Remote argument and result values support QDK Chemistry data classes, NumPy arrays with non-object and non-structured data types, None, booleans, integers, floats, strings, NumPy scalar equivalents, and lists or tuples recursively containing supported values. AlgorithmRef values are also supported in arguments and settings, including nested algorithm-reference settings. Generic dictionaries are not supported as argument or result values. Keyword arguments and algorithm settings remain mappings because the protocol serializes their entries separately; use a QDK Chemistry data class for other structured values.

Disconnecting closes connection-scoped resources but does not remove artifacts belonging to submitted jobs. For an asynchronous Job, pass cleanup=True to fetch() to remove backend artifacts after the result is successfully retrieved and persisted. Call cleanup() to remove artifacts separately for any terminal job. Cleanup is idempotent; failed retrieval leaves artifacts available for inspection or retry.

Passing a path as cache creates a local FolderCache. On a completed cache hit, run reconstructs and returns the result without submitting another remote job. If the cache contains an in-flight job for the same algorithm, settings, and inputs, polling resumes instead of creating a duplicate. Pass force_rerun=True to bypass the lookup and execute again.

By default, a cache is local to the calling machine. Set is_shared=True only when the same backing store is reachable from both the calling machine and remote compute node, such as a network-mounted directory:

from qdk_chemistry.remote.cache import FolderCache

cache = FolderCache("/mnt/shared/qdk-cache", is_shared=True)
result = algorithm.run(*args, remote=remote, cache=cache)

A shared cache lets the remote worker reuse content-addressed inputs already present there and publish results without transferring those files through the backend. Do not mark a caller-local directory as shared; the remote worker must be able to recreate and access the configured cache.

Implementing a new algorithm backend

This section demonstrates how to integrate an external SCF solver as a QDK/Chemistry plugin, enabling access through the standard API.

Interface requirements

Each algorithm type in QDK/Chemistry defines an abstract base class specifying the interface that all implementations must satisfy:

A name() method that returns a unique identifier for the implementation
A _run_impl() method containing the computational logic
A settings() object for runtime configuration

Defining custom settings

When an implementation requires configuration options beyond those provided by the base settings class, a derived settings class can be defined:

Python API

from qdk_chemistry.data import ElectronicStructureSettings


class CustomScfSettings(ElectronicStructureSettings):
    """Settings for the custom SCF solver."""

    def __init__(self):
        super().__init__()
        # Define additional settings beyond the inherited defaults
        self._set_default(
            "custom_option",
            "string",
            "default_value",
            "Description of the custom option",
        )

C++ API

class CustomScfSettings
    : public qdk::chemistry::algorithms::ElectronicStructureSettings {
 public:
  CustomScfSettings() : ElectronicStructureSettings() {
    // Define additional settings beyond the inherited defaults
    set_default("custom_option", "default_value");
  }
};

Implementation structure

The implementation class inherits from the algorithm base class and overrides the required methods. The _run_impl() method is responsible for:

Converting QDK/Chemistry data structures to the external package’s format
Invoking the external computation
Converting results back to QDK/Chemistry data structures

Python API

from qdk_chemistry.algorithms import ScfSolver
from qdk_chemistry.data import (
    BasisSet,
    Orbitals,
    Structure,
    Wavefunction,
)


class CustomScfSolver(ScfSolver):
    """Custom SCF solver wrapping an external chemistry package."""

    def __init__(self):
        super().__init__()
        self._settings = CustomScfSettings()

    def name(self) -> str:
        return "custom"

    def _run_impl(
        self,
        structure: Structure,
        charge: int,
        spin_multiplicity: int,
        basis_or_guess: Orbitals | BasisSet | str | None = None,
    ) -> tuple[float, Wavefunction]:
        """Perform a self-consistent field (SCF) calculation using a custom backend.

        This method should convert the input structure to the external format, run the SCF calculation
        using the specified method and basis set, and return the electronic energy and wavefunction
        in QDK/Chemistry format.

        Args:
            structure: The molecular structure to be calculated.
            charge: The total charge of the molecular system.
            spin_multiplicity: The spin multiplicity (2S+1) of the system.
            basis_or_guess: Basis set information or initial guess, which can be:
                - An Orbitals object (used as initial guess)
                - A BasisSet object
                - A string specifying the basis set name
                - None (use default from settings)

        Returns:
            Tuple of (energy, wavefunction)
        """
        # Convert to external format

        # Execute external calculation

        # Convert results to QDK format

        # energy = 0.0
        # wavefunction = Wavefunction(...)
        # return energy, wavefunction
        return 0.0, None

C++ API

#include <qdk/chemistry/algorithms/scf.hpp>

#include "external_chemistry_package.hpp"

class CustomScfSolver : public qdk::chemistry::algorithms::ScfSolver {
 public:
  CustomScfSolver() { _settings = std::make_unique<CustomScfSettings>(); }

  std::string name() const override { return "custom"; }

 protected:
  std::pair<double, std::shared_ptr<qdk::chemistry::data::Wavefunction>>
  _run_impl(std::shared_ptr<qdk::chemistry::data::Structure> structure,
            int charge, int spin_multiplicity,
            std::optional<std::shared_ptr<qdk::chemistry::data::Orbitals>>
                initial_guess) override {
    // Convert to external format
    auto external_mol = convert_to_external_format(structure);

    // Execute external calculation
    auto basis = _settings->get<std::string>("basis_set");
    auto [energy, external_orbitals] =
        external_package::run_scf(external_mol, basis);

    // Convert results to QDK format
    auto wavefunction = convert_to_qdk_wavefunction(external_orbitals);

    return {energy, wavefunction};
  }
};

Registration

Implementations are registered with the algorithm factory to enable discovery and instantiation by name. The plugin registrar delegates to that existing factory registry:

Python API

from qdk_chemistry.plugins import PluginRegistrar, QdkChemistryPlugin


class CustomScfPlugin(QdkChemistryPlugin):
    """Register the capabilities provided by the custom SCF package."""

    def register(self, registrar: PluginRegistrar) -> None:
        registrar.register_algorithm(lambda: CustomScfSolver())


# Installed plugins are registered automatically through their entry point.
# This call only makes the standalone documentation example executable.
CustomScfPlugin().register(PluginRegistrar())

C++ API

#include <qdk/chemistry/algorithms/scf.hpp>

// Static registration during library initialization
static auto registration =
    qdk::chemistry::algorithms::ScfSolver::register_implementation(
        []() { return std::make_unique<CustomScfSolver>(); });

Following registration, the implementation is accessible through the standard API:

from qdk_chemistry.algorithms import available, create
from qdk_chemistry.data import Structure

# Define a molecular structure (e.g., H2 molecule)
coords = [[0.0, 0.0, 0.0], [0.0, 0.0, 1.4]]
molecule = Structure(coords, symbols=["H", "H"])

# Instantiate the custom solver
solver = create("scf_solver", "custom")
# energy, wavefunction = solver.run(
#     molecule, charge=0, spin_multiplicity=1, basis_or_guess="sto-3g"
# )

# Verify registration
print(available("scf_solver"))  # [..., 'custom']

Plugins

Plugin system

Architecture

Using plugins

Included third-party plugins

PySCF plugin details

Qiskit plugin details

OpenFermion plugin details

Creating plugins

Automatic discovery

Remote backend MCP configuration

Implementing a remote backend

Implementing a new algorithm backend

Defining a new algorithm type

Further reading