Zequent Client SDK (Python) — Customer Example

A complete, runnable example of a small FastAPI service that uses the Python Client SDK to expose a REST API for drone operations. This is the Python counterpart of CUSTOMER_EXAMPLE.md.

Project layout

my-drone-gateway/
├── pyproject.toml
├── .env
└── app/
    ├── __init__.py
    ├── main.py
    └── models.py

`pyproject.toml`

[project]
name = "my-drone-gateway"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = [
    "zqnt-client-sdk>=2.0.0",
    "fastapi>=0.115.0",
    "uvicorn[standard]>=0.30.0",
    "python-dotenv>=1.0.0",
]

[tool.uv.sources]
zqnt-client-sdk = { git = "https://github.com/zequent/zqnt-client-sdk-python", tag = "v2.0.0" }
zqnt-utils = { git = "https://github.com/zequent/zqnt-utils-python", tag = "v2.0.0" }

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

The SDK is not on PyPI yet, so [tool.uv.sources] installs it, and zqnt-utils, from their Git release tags — see Quickstart. Install with:

uv sync

`.env`

ZQNT_CLIENT_TOKEN=<your client credential>
REMOTE_CONTROL_SERVICE_HOST=localhost
REMOTE_CONTROL_SERVICE_PORT=8002
MISSION_AUTONOMY_SERVICE_HOST=localhost
MISSION_AUTONOMY_SERVICE_PORT=8004
LIVE_DATA_SERVICE_HOST=localhost
LIVE_DATA_SERVICE_PORT=8003

ZQNT_CLIENT_TOKEN is the service's client credential, issued in the Admin Console under Manage → Access & Integrations → Credentials. Without one the platform refuses every call.

`app/models.py`

Lightweight request models — keep your wire format separate from the SDK's:

from pydantic import BaseModel, Field


class TakeoffBody(BaseModel):
    latitude: float = Field(..., ge=-90, le=90)
    longitude: float = Field(..., ge=-180, le=180)
    altitude: float = Field(..., ge=0, le=500)


class GoToBody(TakeoffBody):
    pass


class SkillRunBody(BaseModel):
    application_id: str
    skill_id: str
    parameters: dict | None = None

`app/main.py`

import asyncio
import logging
from contextlib import asynccontextmanager

from dotenv import load_dotenv
from fastapi import Depends, FastAPI, HTTPException, Request
from google.protobuf.json_format import MessageToDict

from client_sdk import (
    GoToRequest,
    ReturnToHomeRequest,
    StreamTelemetryRequest,
    TakeoffRequest,
    ZequentClient,
    ZequentClientError,
    ZequentRetryExhaustedError,
)

from .models import GoToBody, SkillRunBody, TakeoffBody

load_dotenv()
logging.basicConfig(level=logging.INFO)
log = logging.getLogger("drone-gateway")


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


app = FastAPI(title="Drone Gateway", lifespan=lifespan)


def get_client(request: Request) -> ZequentClient:
    return request.app.state.zequent


# ----------------------------------------------------------------------
# Remote Control
# ----------------------------------------------------------------------

@app.post("/drones/{sn}/takeoff")
async def takeoff(
    sn: str, body: TakeoffBody,
    client: ZequentClient = Depends(get_client),
):
    try:
        return await client.remote_control.takeoff(
            TakeoffRequest(sn=sn, **body.model_dump())
        )
    except ZequentRetryExhaustedError as e:
        raise HTTPException(503, f"Service unavailable: {e}") from e
    except ZequentClientError as e:
        raise HTTPException(502, str(e)) from e


@app.post("/drones/{sn}/goto")
async def goto(
    sn: str, body: GoToBody,
    client: ZequentClient = Depends(get_client),
):
    return await client.remote_control.go_to(GoToRequest(sn=sn, **body.model_dump()))


@app.post("/drones/{sn}/return-home")
async def return_home(sn: str, client: ZequentClient = Depends(get_client)):
    return await client.remote_control.return_to_home(ReturnToHomeRequest(sn=sn))


# ----------------------------------------------------------------------
# Mission Autonomy — Skill executions
# See: ../concepts/applications-and-skills.md
# ----------------------------------------------------------------------

@app.post("/drones/{sn}/skills/run")
async def run_skill(
    sn: str, body: SkillRunBody,
    client: ZequentClient = Depends(get_client),
):
    """Run a Skill of a deployed Application (Production version, or the newest)."""
    try:
        execution = await client.mission_autonomy.execute_application(
            sn, body.application_id, body.skill_id, parameters=body.parameters
        )
    except ZequentClientError as e:
        raise HTTPException(502, str(e)) from e
    return MessageToDict(execution)


@app.get("/executions/{execution_id}")
async def get_execution(execution_id: str, client: ZequentClient = Depends(get_client)):
    return MessageToDict(await client.mission_autonomy.get_skill_execution(execution_id))


@app.post("/executions/{execution_id}/cancel")
async def cancel_execution(execution_id: str, client: ZequentClient = Depends(get_client)):
    return MessageToDict(await client.mission_autonomy.cancel_skill_execution(execution_id))


# ----------------------------------------------------------------------
# Live Data (server-streaming)
# ----------------------------------------------------------------------

@app.get("/drones/{sn}/telemetry")
async def telemetry_window(
    sn: str, frames: int = 5, client: ZequentClient = Depends(get_client),
):
    """Collect the next N telemetry frames and return them as a list."""
    out = []
    enough = asyncio.Event()

    def on_frame(frame):
        out.append(str(frame))
        if len(out) >= frames:
            enough.set()

    async with client.live_data.stream_telemetry(StreamTelemetryRequest(sn=sn), on_frame):
        await asyncio.wait_for(enough.wait(), timeout=30)
    return out

Skill executions come back as protobuf messages, so the handlers convert them with MessageToDict before FastAPI returns them. Failed Application and Skill execution calls raise MissionAutonomyError, a ZequentClientError.

Run it

uv run uvicorn app.main:app --reload

Test:

curl -X POST http://localhost:8000/drones/DOCK-1/takeoff \
  -H 'content-type: application/json' \
  -d '{"latitude":47.3769,"longitude":8.5417,"altitude":100}'

Notes for production

Use a long-lived ZequentClient per process (the lifespan does this) — never instantiate one per request.
Wrap each handler in proper error mapping; the example covers the SDK's two main exception types.
For high-fanout streaming endpoints, push frames to the caller as they arrive (for example over a WebSocket) instead of buffering them.
Add structured request logging that captures sn + the execution id so you can correlate with platform logs.
Don't forget --workers 1 if you rely on a single ZequentClient in app.state — for multi-worker setups, each worker creates its own client instance, which is fine.
Configure TLS via custom channels (see CONFIGURATION_PYTHON.md) when deploying outside a private network.