Zequent Client SDK (Python) — Asyncio Patterns

The Java Client SDK integrates tightly with Quarkus / CDI / Mutiny. The Python Client SDK is plain asyncio + grpc.aio. This document covers the lifecycle, streaming, cancellation, and error patterns that make the Python SDK pleasant to use in real applications.

This is the Python counterpart of SPRING_BOOT.md.

Lifecycle: one client per process

ZequentClient is stateful — it owns three grpc.aio channels. Create exactly one per process and reuse it:

async with ZequentClient.from_env() as client:
    await do_work(client)

Under the hood:

__aenter__() creates the three channels and lazily-initialised stubs.
__aexit__() closes them in reverse order.

For a script, the async with form is enough. For a long-running server (FastAPI, aiohttp, custom asyncio loop), wire it into a lifespan so it stays alive between requests:

from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    client = ZequentClient.from_env()
    await client.__aenter__()
    app.state.zequent = client
    try:
        yield
    finally:
        await client.__aexit__(None, None, None)

app = FastAPI(lifespan=lifespan)

For a Click / Typer CLI:

@click.command()
def takeoff(sn: str):
    asyncio.run(_takeoff(sn))

async def _takeoff(sn: str):
    async with ZequentClient.from_env() as client:
        await client.remote_control.takeoff(TakeoffRequest(sn=sn))

Concurrency

grpc.aio channels are HTTP/2-multiplexed and re-entrant. You can issue many concurrent calls on a single ZequentClient:

async with ZequentClient.from_env() as client:
    drones = ["DOCK-1", "DOCK-2", "DOCK-3"]
    results = await asyncio.gather(*[
        client.remote_control.takeoff(TakeoffRequest(sn=sn)) for sn in drones
    ])

There's no need for connection pools or thread pools. Do not wrap calls in asyncio.to_thread — they're already non-blocking.

Cancellation

Cancelling the calling task cancels the underlying gRPC call. Use it for timeouts or for early termination of streaming reads:

try:
    async with asyncio.timeout(2.0):
        await client.mission_autonomy.get_skill_execution("execution-id")
except TimeoutError:
    log.warning("get_skill_execution timed out")

Streams run in a background task and call your callback for every frame. stream_telemetry returns a StreamHandle: stop it, or let an async with block exit, to cancel the underlying gRPC stream:

from client_sdk import StreamTelemetryRequest

async def on_frame(frame):
    print(frame)

async with client.live_data.stream_telemetry(StreamTelemetryRequest(sn="DOCK-1"), on_frame):
    await asyncio.sleep(60)   # leaving the block stops the stream

Streaming

Streams reconnect on their own after transient gRPC errors. For a long-lived stream that you start and stop from different coroutines, keep the StreamHandle:

handle = client.live_data.stream_telemetry(StreamTelemetryRequest(sn="DOCK-1"), on_frame, on_error)
# ... later, from anywhere:
await handle.stop()

For manual control, which streams your inputs to the platform, use the start_manual_control_input async context:

from client_sdk import ManualControlInput

async with client.remote_control.start_manual_control_input("DOCK-1") as session:
    await session.send_input(ManualControlInput(throttle=0.5, yaw=0.1))
    response = await session.complete()

Error handling

All client errors derive from ZequentClientError:

Exception	When it's raised
`ZequentClientError`	Generic base — catch this if you don't care about specifics
`ZequentRetryExhaustedError`	Retries gave up; the underlying gRPC call kept failing
`CircuitBreakerOpen`	The breaker for this method is open; refusing to send the call
`grpc.aio.AioRpcError`	Low-level gRPC errors that don't get wrapped (rare)

Recommended pattern:

from client_sdk import ZequentClientError, ZequentRetryExhaustedError

try:
    await client.remote_control.takeoff(req)
except ZequentRetryExhaustedError as e:
    # Likely: backend down. Surface as 503.
    raise HTTPException(503, str(e)) from e
except ZequentClientError as e:
    # Anything else from the SDK
    raise HTTPException(502, str(e)) from e

Don't catch Exception broadly — let programmer errors (validation, type errors) surface.

Resilience tuning

Override the resilience policy when you have requirements that differ from the defaults:

from client_sdk.config.resilience import ResilienceConfig

resilience = ResilienceConfig(
    max_attempts=3,
    initial_backoff_ms=100,
    max_backoff_ms=2_000,
    backoff_multiplier=2.0,
    breaker_failure_threshold=5,
    breaker_reset_seconds=15,
)

async with ZequentClient(config) as client:
    ...

Retries apply only to unary RPCs and only on retryable gRPC status codes. See CONFIGURATION_PYTHON.md for the full list.

Testing

The Python SDK is built around MagicMock-friendly stubs. The recommended pattern in tests:

from unittest.mock import AsyncMock, MagicMock
from client_sdk.remote_control.client import RemoteControlClient


async def test_takeoff_sends_correct_request():
    stub = MagicMock()
    stub.Takeoff = AsyncMock(return_value=...)

    client = RemoteControlClient(stub=stub)
    await client.takeoff(TakeoffRequest(sn="DOCK-1"))

    stub.Takeoff.assert_awaited_once()
    sent = stub.Takeoff.await_args.args[0]
    assert sent.sn == "DOCK-1"

For integration tests, spin up the platform services (compose) and use a real ZequentClient.from_env().

Comparison to the Java SDK

Concern	Java	Python
Concurrency	`CompletableFuture<T>`, Mutiny	`async def` / `await` + `grpc.aio`
DI	CDI (`@Inject ZequentClient`)	Pass the client manually (FastAPI / DI of choice)
Configuration	`application.properties` + env	Env (`from_env()`) or explicit `ServiceConfig`
Lifecycle	CDI `@PostConstruct` / `@PreDestroy`	`async with` / lifespan hook
Streaming	`Multi<T>` (Mutiny)	Callback + `StreamHandle`
Retries	SmallRye fault tolerance	Built-in `GrpcResilience`
Error type	`ErrorInfo` on the response; Application/SkillExecution calls complete exceptionally (`MissionAutonomyClientException`)	`ZequentClientError` and its subclasses (e.g. `MissionAutonomyError`)

The Python SDK is intentionally lean — there is no DI, no annotations, and no codegen step: the generated protocol code ships with the zqnt-utils package. If something feels missing, check whether it can be expressed with stdlib asyncio + the patterns in this document.