Python SDK

The certivu Python package provides sync and async clients for signing and verifying AI-generated content. It is designed for AI/ML pipelines where Python is the primary language.

Install

pip install certivu

Requires Python 3.9+.

Initialize

from certivu import CertivuClient

client = CertivuClient(
    api_key="ctv_key_abc123",
    generator_id="your-generator-uuid",   # required for signing
)

Config can also come from environment variables:

export CERTIVU_API_KEY=ctv_key_abc123
export CERTIVU_GENERATOR_ID=your-generator-uuid

---

Sign content

from pathlib import Path

content_bytes = Path("output.jpg").read_bytes()  # or .mp3, .mp4, .pdf, etc.

result = client.sign(
    content=content_bytes,
    model="stable-diffusion-xl",
    # format="image"  # optional — auto-detected from magic bytes
)

print(result.token)               # ctv_7f3kx9mq2...
print(result.record_id)           # rec-uuid
print(len(result.watermarked_content))  # signed content bytes — use this one

The API handles watermarking, hashing, and ML-DSA signing server-side. result.watermarked_content is the signed content with the ctv_ token injected into a format-native container (XMP, ID3v2, HTML meta, cerv atom, etc.) and the watermark embedded.

---

Verify content

Verification is always free — no API key required.

result = client.verify(content=content_bytes)

if result.authentic and result.confidence == "high":
    print(f"Verified — {result.provenance.org} · {result.provenance.signed_at}")
elif result.tampered:
    print("Content has been modified since signing")
else:
    print(f"Not verified: {result.reason}")

Pass a token explicitly to skip automatic extraction:

result = client.verify(content=content_bytes, token="ctv_7f3kx9mq2...")

Confidence levels

Level	Signals	Meaning
high	Watermark ✓ + Record ✓ + Signature ✓	Full chain intact
medium	Record ✓ + Signature ✓	Re-uploaded without watermark
low	Partial	Something is off
none	Nothing found	Not signed by Certivu

---

Token status

Lightweight lookup without re-uploading content — CDN-cacheable:

status = client.get_token_status("ctv_7f3kx9mq2...")
print(status.generator_status)  # "active" or "revoked"
print(status.signed_at)

# As of v2.4.0 the response also includes the org's display name and branding
print(status.org_name)  # "Acme AI"
print(status.branding)  # dict | None — display_name, logo_url, primary_color, support_url

The branding fields power the public certificate page at https://certivu.ai/certificate?t=<ctv_token> (configured by Growth+ orgs via PATCH /v1/account/branding).

---

Batch verify

Up to 50 items per call:

results = client.verify_batch([
    {"content": content1_bytes},
    {"content": content2_bytes, "token": "ctv_..."},
])

for r in results:
    print(r.authentic, r.confidence)

---

Batch sign

Up to 50 records per call (HTTP 207 multi-status):

results = client.sign_batch([
    {"content": content1_bytes, "model": "sdxl"},
    {"content": content2_bytes, "model": "sdxl"},
])

for r in results:
    if "error" in r:
        print("Failed:", r["error"])
    else:
        print(r["token"])

---

Audit log

page = client.get_audit_log(page=1, limit=50)

for event in page.events:
    print(event.type, event.timestamp)

print(f"Total: {page.total}")

---

Analytics

# Overview — plan-gated (Free=7d, Starter=30d, Growth+=90d)
overview = client.get_analytics_overview(days=30)
print(overview.total_verifications)
print(overview.tamper_count)
print(overview.daily_trend)   # list of AnalyticsTrendPoint
print(overview.top_records)   # list of AnalyticsTopRecord

# Per-record drill-down (Growth+ required)
record = client.get_record_analytics("rec-uuid")
print(record.confidence_breakdown.high)
print(record.recent_events)

---

Webhooks

# List endpoints
result = client.list_webhooks()
print(result.webhooks_enabled)   # False on Free/Starter plans
for ep in result.endpoints:
    print(ep.webhook_id, ep.url, ep.status)

# Register an endpoint (Growth+ required)
endpoint = client.create_webhook(
    url="https://your-app.example.com/certivu",
    events=["record.created", "verify.tamper_detected", "quota.warning"],
)
print(endpoint.webhook_id)
# Store endpoint.secret — shown once, cannot be retrieved again

# Delete an endpoint
client.delete_webhook(endpoint.webhook_id)

---

Compliance attestation

Generate a factual attestation of your org’s provenance and verification activity, mapped to EU AI Act Article 50. Enterprise plan only.

# Fetch the attestation as a dict
attestation = client.get_attestation(from_="2026-01-01", to="2026-03-31")
print(attestation.summary.records_signed)
print(attestation.eu_ai_act_article_50.mapping)

# Export a downloadable JSON or HTML document
html = client.export_attestation(format="html", from_="2026-01-01", to="2026-03-31")
# html is the printable attestation document — write it to disk or attach to records

An attestation is a factual record of activity, not a certification, legal opinion, or guarantee of compliance, and Certivu makes no claim to detect all AI-generated content. See the Attestation API reference.

---

Post-quantum C2PA

Embed a real, ML-DSA-signed C2PA manifest into a signed JPEG/PNG, bound to a record. Pass the signed asset bytes. Requires a Growth plan or above; assets up to 50 MB.

embedded = client.embed_c2pa(record_id, signed_image_bytes)
# embedded: bytes — the asset with the C2PA manifest embedded
with open("output.c2pa.jpg", "wb") as f:
    f.write(embedded)

The manifest is interoperable with the Content Credentials ecosystem but signed with ML-DSA-65 — standard verifiers report the algorithm as unrecognized (the quantum-resistant trade-off). AsyncCertivuClient exposes the same method as await client.embed_c2pa(...).

---

Async client

All methods are available on AsyncCertivuClient:

from certivu import AsyncCertivuClient

async def main():
    async with AsyncCertivuClient(api_key="ctv_key_abc123") as client:
        result = await client.verify(content=content_bytes)
        print(result.authentic, result.confidence)

        results = await client.verify_batch([
            {"content": item} for item in content_list
        ])

asyncio.run(main())

---

Error handling

from certivu import AuthError, QuotaError, NotFoundError, CertivuError

try:
    result = client.sign(content=content_bytes, model="sdxl")
except QuotaError as e:
    print("Quota exceeded. Upgrade at:", e.upgrade_url)
except AuthError:
    print("Invalid API key")
except CertivuError as e:
    print(f"Error {e.status_code}: {e}")

---

Source

The Python SDK lives at packages/sdk-python/ in the Certivu repository.