Kanopy Python SDK

kanopy-ai is the Python client for Kanopy's infrastructure inspection API. It covers the core integration workflow: create a project, upload footage, monitor processing, query network inventory, and download results.

License and service boundary

The client library in this repository is licensed under the Apache License, Version 2.0. Copyright 2026 Kanopy AI, Inc.

The license applies to the SDK source code only. It does not grant access to the Kanopy API or rights in Kanopy's hosted service, backend implementation, models, processing systems, customer data, trade names, or trademarks. API access requires credentials issued by Kanopy and remains subject to the applicable customer agreement and service terms.

Applications that merely use the SDK through its interfaces remain separable works under Apache-2.0. Utilities retain their independently developed applications and data; Kanopy retains its SDK, platform, and pre-existing intellectual property.

Install

pip install kanopy-ai

Quick start

from kanopy import Kanopy

with Kanopy(api_key="kpy_live_...") as kanopy:
    project = kanopy.create_project(
        name="North corridor",
        description="Q3 inspection",
    )
    upload = kanopy.upload(
        "flight.mp4",
        metadata="flight.csv",
        capture_device="drone",
        project_id=project["id"],
        title="North corridor flight 01",
        upload_request_id="north-corridor-flight-01",
    )

    # Reconstruction is queued automatically when the upload completes.
    job = kanopy.wait_for_job(upload["job_id"], timeout=60 * 60)
    trees = kanopy.list_project_trees(project["id"], job_id=job["id"])

The default base URL is https://app.kanopy-ai.com/api/v1. For staging or local development, pass base_url= when constructing Kanopy.

Large video uploads

Use upload_large for production inspection footage. It creates an idempotent multipart session, uploads parts directly to object storage, retries each failed part with a fresh presigned URL, and queues reconstruction when all parts are complete:

def report_progress(sent: int, total: int) -> None:
    print(f"{sent / total:.0%}")

upload = kanopy.upload_large(
    "large-flight.mp4",
    metadata="flight.txt",
    project_id=project["id"],
    title="North corridor flight 02",
    upload_request_id="north-corridor-flight-02",
    capture_device="drone",
    line_clearance=True,
    progress=report_progress,
)

The default uses 64 MiB parts, four parallel workers, and three attempts per part. part_size, max_workers, and part_retries are configurable. Memory use is approximately part_size * max_workers while transfers are active.

upload_large sends the original source bytes without client-side compression. For declared drone and action_cam uploads it requests background server preparation by default: Kanopy validates and localizes the original, normalizes GPS, and extracts reconstruction frames at the configured processing rate. It does not create a compressed replacement for the source. Pass server_side_upload_prep=False only when an integration must retain the legacy synchronous completion path.

Video and GPS inputs

Declare capture_device so the SDK can validate the upload before transferring a large video:

Capture device	Video	GPS/metadata contract
drone	Original inspection video	One or more .csv or .txt flight logs are required. Encrypted DJI .txt logs are converted server-side.
action_cam	Original GoPro/action-camera MP4	A separate flight log is optional. Kanopy can extract embedded GoPro GPMF GPS; do not transcode the source before upload.
phone	Original phone video	GPS is optional; provide canonical gps_track.json when georeferencing is required.

For multiple drone logs, pass metadata_files=[...]. A canonical GPS track can be supplied with gps_track="gps_track.json". .srt subtitle files are not a supported flight-log input.

Omitting capture_device remains supported for compatibility with older SDK integrations, but it skips client-side combination validation and does not opt the upload into background server preparation. New integrations should always declare it.

# GoPro GPS is embedded in the original MP4.
upload = kanopy.upload_large(
    "GX010123.MP4",
    capture_device="action_cam",
    project_id=project["id"],
    title="North corridor action-camera run",
)

Downloads and exports

Everything operational that can be downloaded from the Kanopy platform is available through the API and this SDK. A complete job package contains the reconstruction and segmented point clouds, camera poses, trees/poles/spans analytics, and summary. Merged PLY point clouds contain the shipped scalar measurements as standard vertex properties.

To discover what a job produced rather than assuming a ZIP layout, list its outputs. Each entry has a stable id, a kind, a format, and a version token that only changes when the bytes change, so a synchronizing client can skip work it has already done:

for output in kanopy.list_job_outputs(job_id):
    if output["kind"] != "merged_point_cloud":
        continue
    if output["version"] == seen.get(output["id"]):
        continue
    kanopy.download_job_output(job_id, output["id"], f"{output['id']}.ply")
    seen[output["id"]] = output["version"]

A job that has not produced anything yet returns an empty list rather than raising, so this is safe to call as soon as a job.completed webhook arrives. Outputs stored in object storage are served through a short-lived presigned URL, which the SDK follows without ever sending your API key to the storage host.

# Complete job, or only the export-ready point clouds in a chosen metre CRS.
kanopy.download_job_folder(job_id, "job-results.zip")
kanopy.download_job_folder(
    job_id,
    "point-clouds.zip",
    include="point_cloud",
    point_cloud_epsg=26916,
)

# One job analytics table.
kanopy.download_job_table(job_id, "trees", "trees.csv")

# Project analytics in platform-compatible portable formats.
kanopy.download_project_table(project_id, "trees", "trees.csv")
kanopy.download_project_table(project_id, "trees", "trees.json", format="json")
kanopy.download_project_table(project_id, "trees", "trees.kml", format="kml")
kanopy.download_project_table(
    project_id, "trees", "trees.geojson", format="geojson"
)
kanopy.download_project_table(project_id, "poles", "poles.csv")
kanopy.download_project_table(project_id, "spans", "spans.json", format="json")

# Portable detail reports, including available inspection imagery.
kanopy.download_tree_report(project_id, tree_id, "tree-analysis.pdf")
kanopy.download_pole_report(project_id, pole_id, "pole-analysis.pdf")

For a small project, download_project_folder streams a complete project ZIP directly. For large projects, the async helper creates or reuses a background export, polls it, and downloads the resulting presigned archive without sending the API credential to object storage:

kanopy.download_project_export(
    project_id,
    "project-results.zip",
    timeout=60 * 60,
)

Organization administrators can also export the activity log or the staff-access-only report:

kanopy.download_audit_events("audit-events.csv")
kanopy.download_audit_events("staff-access.csv", staff_only=True)

Personal privacy archives are deliberately excluded from API-key access. They remain available only through an interactive platform session with re-authentication.

Pagination

List methods return a Page. Offset pagination is used by default. Pass cursor="" to start keyset pagination, then use page.next_cursor:

page = kanopy.list_jobs(cursor="", limit=100)
while True:
    for job in page.items:
        print(job["id"], job["status"])
    if not page.next_cursor:
        break
    page = kanopy.list_jobs(cursor=page.next_cursor, limit=100)

Errors and request IDs

Non-successful API responses raise KanopyError. The exception exposes the HTTP status, Kanopy error code, detail payload, and support request ID:

from kanopy import KanopyError

try:
    kanopy.get_job("missing-id")
except KanopyError as exc:
    print(exc.status_code, exc.code, exc.request_id)

Isolated local API smoke test

The smoke harness builds the current backend and runs the installed SDK against a disposable Docker stack with Postgres, Redis, and MinIO. It uses real bearer authentication, enables API_KEY_CONTRACT_MODE=block, performs a two-part presigned upload, and removes all containers and volumes when it finishes.

./scripts/run_local_smoke.sh

The stack uses localhost:18000 for the API and localhost:19100 for MinIO so it can run alongside the normal development Compose project.

The SDK keeps a reviewed copy of Kanopy's public OpenAPI schema under tests/fixtures/. From the Kanopy development repository root, refresh that copy after an intentional public API change with:

./scripts/sync_public_openapi.sh