Python SDK

Needs Python 3.14. Fully typed. Install the etchv-sdk package, then import etchv. MIT licensed.

Install

pip install etchv-sdk

Embed and detect

Set ETCHV_API_KEY on your server. This example needs watermarks:embed and watermarks:detect, an active plan and available credits.

import os
from pathlib import Path
from etchv import Etchv

with Etchv(os.environ["ETCHV_API_KEY"]) as client:
    print(client.check_api_key())  # organization_id, key_id, scopes; no credits used
    result = client.embed_image(
        Path("photo.jpg").read_bytes(),
        {"recipient": "customer-123"},
        filename="photo.jpg",
    )
    Path(result.filename).write_bytes(result.image)
    detection = client.detect_image(result.image)
    print(detection.watermark_id, detection.confidence)

check_api_key checks your key and uses no credits. Save the output bytes exactly as returned; don't re-encode them.

Documents and video

For every media type, the watermarked file is in result.image, in its original format. Embedding also returns watermark_id, request_id, content_type, filename, asset_id, source_asset_id and storage_delivery_id.

Detection returns the watermark it found, a confidence score and results for each page or frame. It does not return the original JSON. Result fields · Credits.

Errors and recovery

Embedding and video detection retry and check for results automatically. Image and document detection, downloads and batch archives retry a dropped connection. Detection resends with the same idempotency key, and the API returns the first result at no extra charge.

It keeps that result for 24 hours; pass keep_result=False to keep nothing, and a retry then runs detection again at no charge, uploading a large file again first. The detection requests stay within one client timeout counted from the first send; the new upload is bounded like any upload. Errors from a keyed call carry its idempotency_key.

Save an idempotency key (a value you choose so a retry isn't run twice) before uploading. Then you can recover the request even if your app restarts.

from etchv import DeadlineExceededError, EtchvError

try:
    result = client.embed_image(image, {"recipient": "customer-123"},
                                idempotency_key="order-42")
except DeadlineExceededError as error:
    resume_later(error.detail["idempotency_key"], error.request_id)
except EtchvError as error:
    print(error.status_code, error.request_id)

Asset library

Scopes: assets:read for downloads, assets:write for edits, assets:delete plus owner/admin for deletion. No credits used.

page = client.list_assets(kind="watermarked", limit=25)
for item in page["items"]:
    asset = client.get_asset(item["id"])
    updated = client.update_asset(asset["id"], version=asset["version"],
                                  metadata={"campaign": "spring"})
    if updated["file_available"]:
        content = client.download_asset(updated["id"])
# Pass cursor=page["next_cursor"] with the same filters for the next page.

Traceable links

Share one file so each recipient downloads their own watermarked copy. Scopes: links:write to create, estimate and revoke, links:read to read. Each copy costs one embed of the file.

link = client.create_link(
    result.source_asset_id,  # the original of a file you watermarked
    "Board pack, October",
    recipients=["alex@example.com", "sam@example.com"],
    download_limit=3,
)
print(link["url"])  # send this to your recipients

for recipient in client.get_link(link["id"])["recipients"]:
    print(recipient["email"], recipient["opened_at"], recipient["copy"]["status"])

client.revoke_link(link["id"])

Async jobs and webhooks

To get a receipt without waiting for the result, use submit_embed / submit_detection. Media types: images, documents, videos.

Check status with get_job(request_id) for embedding and get_job(request_id, detect=True) for detection.

job = client.submit_embed("documents", pdf_bytes, {"delivery": "delivery_001"},
    filename="document.pdf", idempotency_key="delivery_001", webhook_id=webhook_id)
status = client.get_job(job["request_id"])  # queued, running, retrying, succeeded, failed
if status["status"] == "succeeded":
    result = client.get_embed_result(job["request_id"])

scan = client.submit_detection("videos", video_bytes, idempotency_key="scan_001")
detection = client.get_detection_result(scan["request_id"])

from etchv import verify_webhook, WebhookVerificationError

try:
    event = verify_webhook(raw_body, request.headers, signing_secret)
except WebhookVerificationError:
    return 400
# Deduplicate by event["id"], enqueue your work, then return 2xx.

GPU processing

On Business or Enterprise plans, pass accelerator="gpu". GPU processing uses 3× credits. If the job falls back to CPU, it uses normal credits. result.accelerator reports the accelerator used. See GPU processing.

PDF scripts and actions

A watermarked PDF keeps its scripts and actions by default. Pass active_content="remove_auto_run" to embed_document, submit_embed("documents", ...), submit_batch or submit_batch_zip to remove what a PDF viewer runs by itself. It is for PDFs only: the API refuses it for a DOCX or PPTX, and for a batch unless every file is a PDF. See scripts and actions.

Choose where results are stored

Connect your bucket, then pass its destination ID. You can also pass an object key, which is the file’s path in your bucket. Leave both out to use Etchv storage.

job = client.submit_embed("documents", pdf_bytes, {"recipient": "customer-123"},
    filename="report.pdf", storage_destination_id=destination_id,
    storage_key="reports/watermarked.pdf", idempotency_key="report-export-001")
# After the watermark job reports succeeded:
delivery = client.get_storage_delivery(job["storage_delivery_id"])
if delivery["status"] == "stored":
    data = client.download_storage_delivery(delivery["id"])