SDKs
Python SDK
Python watermarking SDK for Etchv. Watermark images, PDFs and videos, detect watermarks, and manage assets with a fully typed client. Install with pip.
Needs Python 3.14. Fully typed. Install the etchv-sdk package, then import etchv. MIT licensed.
Install
pip install etchv-sdkEmbed 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.
Client configuration
Etchv(key, base_url=..., timeout=..., transport=...)
| Option | Default / behavior |
|---|---|
base_url | https: |
timeout | 120 seconds |
transport | Optional HTTPX transport |
Media filename, idempotency_key | File name and stable retry key |
base_url- Default / behavior
https:/ /api. etchv. com
timeout- Default / behavior
120seconds
transport- Default / behavior
- Optional HTTPX transport
- Media
filename,idempotency_key - Default / behavior
- File name and stable retry key
Use a with block (a context manager) so the client closes when you are done. Run the later examples inside the same with block, or open another client.
Errors and recovery
Embedding and video detection retry and check for results automatically. Image and PDF detection do not. 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.
| Situation | Action |
|---|---|
| Client timeout | The accepted job keeps running. Resume it, or resend the same inputs with the same key |
| Resume a job | get_embed_result( / get_detection_result( |
| Changed inputs | Use a new key. Reusing the old key returns 409 |
| Collect a saved result | Within 24 hours; see retention |
- Client timeout
- Action
- The accepted job keeps running. Resume it, or resend the same inputs with the same key
- Resume a job
- Action
get_embed_result(/request_id) get_detection_result(request_id)
- Changed inputs
- Action
- Use a new key. Reusing the old key returns
409
- Collect a saved result
- Action
- Within 24 hours; see retention
Error types and handling
EtchvError exposes status_code, detail and request_id. DeadlineExceededError uses status 0 and includes the identifiers you need to recover the request.
Typed errors: AuthenticationError (401), PaymentRequiredError (402), PermissionDeniedError (403), NotFoundError (404), ConflictError (409), GoneError (410) and RateLimitError (429). Other statuses raise EtchvError.
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)See HTTP errors and retry behavior. Save the request ID when reporting failures.
Asset library
Scopes: assets:read for downloads, assets:write for edits, assets:delete plus owner/admin for deletion. No credits used.
List, edit and download assets
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.An edit replaces all metadata and must send the asset’s current version. A 409 means the asset changed since you read it; reload and resolve the conflict. delete_asset deletes one asset. delete_assets deletes 1–50 at once: either all are deleted or none are. Asset fields, filters and retention.
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.
Submit and collect an async job
Load the media bytes before submitting.
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"])The webhook is optional (Java/Rust: pass null / None). Save the receipt. Job handling.
Manage endpoints and verify webhooks
Methods: create_webhook, list_webhooks, update_webhook, delete_webhook, list_webhook_deliveries, redeliver_webhook. Use webhooks:read to inspect or webhooks:write plus owner/admin to manage. Save the signing secret; it is shown only once.
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.verify_webhook verifies the signature, timestamp and event ID, or raises WebhookVerificationError. Verify the raw request bytes, with a 300-second timestamp tolerance. Skip events you have already handled, queue your work, then return 2xx. Verification rules.
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.
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.
Deliver to your bucket
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"])After the job succeeds, check the delivery with storage:read. Checking earlier returns 404. When the status is stored, download the file with download_storage_delivery.
With storage:write and owner/admin access, retry_storage_delivery retries a failed or cancelled delivery at no cost. Manage destinations and deliveries with: create_storage_destination, verify_storage_destination, update_storage_destination, delete_storage_destination, create_storage_delivery. See delivery recovery.