Add downstream SDK integration examples

CHANGELOG.md

@@ -8,6 +8,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 
 ### Added
 
+- Sync and async downstream-SDK integration examples for `ClientExtension`, event hooks, and error mapping.
 - `QctrlQaoaJobCreationPayload` and `QctrlQaoaJobInput` for submitting Q-CTRL QAOA maxcut combinatorial-optimization jobs via `create_job`. The `create_job` body union now also accepts `QctrlQaoaJobCreationPayload`.
 - `cost_model` optional field on `BaseJob`, `GetCircuitJobResponse`, and `GetJobResponse`, typed as `CostModel` (`"quantum_compute_time"` or `"execution_time"`).
 

README.md

@@ -63,6 +63,10 @@ print(probs.additional_properties)
 
 Each generated endpoint module exposes four callables: `sync`, `sync_detailed`, `asyncio`, and `asyncio_detailed`. The `sync` and `asyncio` variants return the parsed body; the `_detailed` variants return a `Response[T]` with the status code, headers, and parsed body.
 
+## Examples
+
+See [examples/](examples/README.md) for sync and async downstream-SDK integration examples using the extension API.
+
 For options (`api_key`, `base_url`, `max_retries`, `timeout`, `extension`), error classes, retry behavior, pagination, polling, sessions, and downstream-SDK extension hooks, see the [API reference](https://ionq.github.io/ionq-core-python/).
 
 ## Versioning

examples/README.md

@@ -0,0 +1,41 @@
+# IonQ Core Examples
+
+This directory contains runnable examples for building downstream SDK integrations on top of `ionq-core`.
+The examples submit a Bell-state circuit to the free `simulator` backend, wait for completion, and print
+the result probabilities.
+
+## Setup
+
+Install the package:
+
+```sh
+pip install ionq-core
+```
+
+Create an IonQ Cloud account at <https://identity.ionq.com/create-account>, then export your API key:
+
+```sh
+export IONQ_API_KEY="your-api-key"
+```
+
+## Downstream SDK Integration
+
+Run the sync example:
+
+```sh
+python examples/downstream_integration.py
+```
+
+Run the async example:
+
+```sh
+python examples/downstream_integration_async.py
+```
+
+Both examples demonstrate how a downstream SDK can pass a `ClientExtension` to `IonQClient` to customize
+client behavior without modifying `ionq-core`:
+
+- `user_agent_token` identifies the downstream SDK in the `User-Agent` header.
+- `default_headers` adds SDK-specific request headers.
+- `EventHook` and `AsyncEventHook` observe request and response activity.
+- `error_mapper` wraps `ionq-core` exceptions in SDK-defined exception types.

examples/downstream_integration.py

@@ -0,0 +1,107 @@
+# SPDX-FileCopyrightText: 2026 IonQ, Inc.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Downstream SDK integration example for the sync ionq-core API."""
+
+from __future__ import annotations
+
+import os
+from typing import cast
+
+import httpx
+
+from ionq_core import (
+    APIError,
+    AuthenticatedClient,
+    ClientExtension,
+    EventHook,
+    IonQClient,
+    RateLimitError,
+    wait_for_job,
+)
+from ionq_core.api.default import create_job, get_job_probabilities
+from ionq_core.models.circuit_job_creation_payload import CircuitJobCreationPayload
+from ionq_core.models.get_results_response import GetResultsResponse
+from ionq_core.models.job_creation_response import JobCreationResponse
+
+
+class DownstreamSDKError(RuntimeError):
+    """Base exception raised by this example downstream SDK."""
+
+
+class DownstreamRateLimitError(DownstreamSDKError):
+    """Raised when IonQ returns a rate-limit response."""
+
+
+class LoggingHook(EventHook):
+    """Minimal request/response logger for a downstream SDK wrapper."""
+
+    def on_request(self, request: httpx.Request) -> None:
+        print(f"sdk -> {request.method} {request.url.path}")
+
+    def on_response(self, request: httpx.Request, response: httpx.Response) -> None:
+        print(f"sdk <- {response.status_code} {request.url.path}")
+
+
+def map_ionq_error(exc: Exception) -> Exception:
+    """Convert ionq-core transport exceptions into downstream SDK errors."""
+    if isinstance(exc, RateLimitError):
+        retry_after = f" Retry after {exc.retry_after}s." if exc.retry_after is not None else ""
+        return DownstreamRateLimitError(f"IonQ API rate limit exceeded.{retry_after}")
+    if isinstance(exc, APIError):
+        return DownstreamSDKError(f"IonQ API request failed: {exc.message}")
+    return exc
+
+
+def bell_state_payload() -> CircuitJobCreationPayload:
+    """Build a Bell-state job payload for the simulator backend."""
+    return CircuitJobCreationPayload.from_dict(
+        {
+            "type": "ionq.circuit.v1",
+            "backend": "simulator",
+            "shots": 100,
+            "input": {
+                "gateset": "qis",
+                "qubits": 2,
+                "circuit": [
+                    {"gate": "h", "target": 0},
+                    {"gate": "cnot", "control": 0, "target": 1},
+                ],
+            },
+        }
+    )
+
+
+def build_client() -> AuthenticatedClient:
+    """Create an IonQ client configured as a downstream SDK would configure it."""
+    extension = ClientExtension(
+        user_agent_token="example-downstream-sdk/0.1",
+        default_headers={"X-Downstream-SDK": "example-sync"},
+        event_hooks=(LoggingHook(),),
+        error_mapper=map_ionq_error,
+    )
+    return IonQClient(extension=extension)
+
+
+def main() -> None:
+    if "IONQ_API_KEY" not in os.environ:
+        raise SystemExit("Set IONQ_API_KEY before running this example.")
+
+    client = build_client()
+    job_response = create_job.sync(client=client, body=bell_state_payload())
+    if job_response is None:
+        raise DownstreamSDKError("IonQ API did not return a job creation response.")
+
+    job = cast(JobCreationResponse, job_response)
+    completed_job = wait_for_job(client, job.id)
+
+    probabilities_response = get_job_probabilities.sync(uuid=completed_job.id, client=client)
+    if probabilities_response is None:
+        raise DownstreamSDKError(f"IonQ API did not return probabilities for job {completed_job.id}.")
+
+    probabilities = cast(GetResultsResponse, probabilities_response)
+    print(probabilities.additional_properties)
+
+
+if __name__ == "__main__":
+    main()

examples/downstream_integration_async.py

@@ -0,0 +1,108 @@
+# SPDX-FileCopyrightText: 2026 IonQ, Inc.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Downstream SDK integration example for the async ionq-core API."""
+
+from __future__ import annotations
+
+import asyncio
+import os
+from typing import cast
+
+import httpx
+
+from ionq_core import (
+    APIError,
+    AsyncEventHook,
+    AuthenticatedClient,
+    ClientExtension,
+    IonQClient,
+    RateLimitError,
+    async_wait_for_job,
+)
+from ionq_core.api.default import create_job, get_job_probabilities
+from ionq_core.models.circuit_job_creation_payload import CircuitJobCreationPayload
+from ionq_core.models.get_results_response import GetResultsResponse
+from ionq_core.models.job_creation_response import JobCreationResponse
+
+
+class DownstreamSDKError(RuntimeError):
+    """Base exception raised by this example downstream SDK."""
+
+
+class DownstreamRateLimitError(DownstreamSDKError):
+    """Raised when IonQ returns a rate-limit response."""
+
+
+class AsyncLoggingHook(AsyncEventHook):
+    """Minimal async request/response logger for a downstream SDK wrapper."""
+
+    async def on_request(self, request: httpx.Request) -> None:
+        print(f"sdk -> {request.method} {request.url.path}")
+
+    async def on_response(self, request: httpx.Request, response: httpx.Response) -> None:
+        print(f"sdk <- {response.status_code} {request.url.path}")
+
+
+def map_ionq_error(exc: Exception) -> Exception:
+    """Convert ionq-core transport exceptions into downstream SDK errors."""
+    if isinstance(exc, RateLimitError):
+        retry_after = f" Retry after {exc.retry_after}s." if exc.retry_after is not None else ""
+        return DownstreamRateLimitError(f"IonQ API rate limit exceeded.{retry_after}")
+    if isinstance(exc, APIError):
+        return DownstreamSDKError(f"IonQ API request failed: {exc.message}")
+    return exc
+
+
+def bell_state_payload() -> CircuitJobCreationPayload:
+    """Build a Bell-state job payload for the simulator backend."""
+    return CircuitJobCreationPayload.from_dict(
+        {
+            "type": "ionq.circuit.v1",
+            "backend": "simulator",
+            "shots": 100,
+            "input": {
+                "gateset": "qis",
+                "qubits": 2,
+                "circuit": [
+                    {"gate": "h", "target": 0},
+                    {"gate": "cnot", "control": 0, "target": 1},
+                ],
+            },
+        }
+    )
+
+
+def build_client() -> AuthenticatedClient:
+    """Create an IonQ client configured as a downstream SDK would configure it."""
+    extension = ClientExtension(
+        user_agent_token="example-downstream-sdk/0.1",
+        default_headers={"X-Downstream-SDK": "example-async"},
+        async_event_hooks=(AsyncLoggingHook(),),
+        error_mapper=map_ionq_error,
+    )
+    return IonQClient(extension=extension)
+
+
+async def run() -> None:
+    if "IONQ_API_KEY" not in os.environ:
+        raise SystemExit("Set IONQ_API_KEY before running this example.")
+
+    async with build_client() as client:
+        job_response = await create_job.asyncio(client=client, body=bell_state_payload())
+        if job_response is None:
+            raise DownstreamSDKError("IonQ API did not return a job creation response.")
+
+        job = cast(JobCreationResponse, job_response)
+        completed_job = await async_wait_for_job(client, job.id)
+
+        probabilities_response = await get_job_probabilities.asyncio(uuid=completed_job.id, client=client)
+        if probabilities_response is None:
+            raise DownstreamSDKError(f"IonQ API did not return probabilities for job {completed_job.id}.")
+
+        probabilities = cast(GetResultsResponse, probabilities_response)
+        print(probabilities.additional_properties)
+
+
+if __name__ == "__main__":
+    asyncio.run(run())