vettly 0.1.8

Content moderation SDK for App Store Guideline 1.2 compliance. Filtering, reporting, blocking, and audit trails.

vettly

Decision infrastructure for Python. Every decision recorded, versioned, and auditable—with full async support.

The Problem

Your platform makes thousands of automated decisions daily. But when a user appeals—or a regulator asks—can you prove:

• What policy was applied?
• Why that action was taken?
• What evidence supports the decision?

Classification APIs return True or False. That's not enough for DSA Article 17 compliance. That's not enough when your Trust & Safety team needs to explain decisions months later.

The Solution

Vettly is decision infrastructure. Every API call returns an id that links to:

• The exact policy version applied
• Category scores and thresholds that triggered the action
• Content fingerprint for tamper-evident verification

from vettly import ModerationClient

client = ModerationClient("sk_live_...")

result = client.check(
    content=user_content,
    policy_id="community-guidelines-v2"
)

# Store decision ID with your content
db.posts.create(
    content=user_content,
    moderation_decision_id=result.id,  # Link to audit record
    action=result.action.value
)

# Later: retrieve for compliance review
decision = client.get_decision(result.id)
print(decision.policy.version)  # Exact policy version

Installation

pip install vettly

Quick Start

from vettly import ModerationClient

client = ModerationClient("sk_live_...")

result = client.check(
    content="User-generated text",
    policy_id="community-safe"
)

if result.action == "block":
    print(f"Blocked: {result.id}")
else:
    print(f"Allowed: {result.action}")

Get Your API Key

1. Sign up at vettly.dev
2. Go to Dashboard > API Keys
3. Create and copy your key

---

Core Features

Text Decisions

result = client.check(
    content="User-generated text",
    policy_id="community-safe",
    content_type="text",
    metadata={"user_id": "user_123"},
    user_id="user_123",
    request_id="unique-request-id"  # For idempotency
)

print(result.id)          # Decision ID for audit trail
print(result.action)      # Action.ALLOW | Action.WARN | Action.FLAG | Action.BLOCK
print(result.safe)        # bool
print(result.flagged)     # bool
print(result.categories)  # List[CategoryResult]
print(result.latency_ms)  # Response time

Image Decisions

# From URL
result = client.check_image(
    image_url="https://example.com/image.jpg",
    policy_id="strict"
)

# From base64 data URI
result = client.check_image(
    image_url="data:image/jpeg;base64,/9j/4AAQ...",
    policy_id="strict"
)

Batch Operations

Process multiple items efficiently:

from vettly import BatchItem

# Synchronous batch
batch_result = client.batch_check(
    policy_id="community-safe",
    items=[
        BatchItem(id="post-1", content="First post"),
        BatchItem(id="post-2", content="Second post"),
        BatchItem(id="post-3", content="Third post"),
    ]
)

for result in batch_result.results:
    print(f"{result.id}: {result.action}")

# Async batch with webhook delivery
async_batch = client.batch_check_async(
    policy_id="community-safe",
    items=[...],
    webhook_url="https://your-app.com/webhooks/batch-complete"
)

print(f"Batch ID: {async_batch.batch_id}")

Policy Replay

Re-evaluate historical decisions with different policies:

# User appeals a decision - test with updated policy
original_decision = client.get_decision("original-decision-id")
replay = client.replay_decision(
    decision_id="original-decision-id",
    policy_id="community-guidelines-v3"  # New policy version
)

# Compare outcomes
print(f"Original: {original_decision.action}")
print(f"With new policy: {replay.action}")

Dry Run

Test policies without making provider calls:

dry_run = client.dry_run(
    policy_id="new-policy-id",
    mock_scores={
        "hate_speech": 0.7,
        "harassment": 0.3,
        "violence": 0.1
    }
)

print(dry_run["evaluation"]["action"])  # What action would be taken

---

Async Support

Full async/await support with AsyncModerationClient:

from vettly import AsyncModerationClient

async with AsyncModerationClient("sk_live_...") as client:
    result = await client.check(
        content="User content",
        policy_id="community-safe"
    )

    if result.action == "block":
        print(f"Blocked: {result.id}")

With FastAPI

from fastapi import FastAPI, HTTPException
from vettly import AsyncModerationClient

app = FastAPI()
client = AsyncModerationClient("sk_live_...")

@app.post("/comments")
async def create_comment(content: str):
    result = await client.check(
        content=content,
        policy_id="community-safe"
    )

    if result.action == "block":
        raise HTTPException(
            status_code=403,
            detail={
                "error": "Content blocked",
                "decision_id": result.id
            }
        )

    # Save with audit trail
    await db.comments.create(
        content=content,
        moderation_decision_id=result.id
    )

    return {"status": "ok", "decision_id": result.id}

@app.on_event("shutdown")
async def shutdown():
    await client.close()

With Django (async views)

from django.http import JsonResponse
from vettly import AsyncModerationClient

client = AsyncModerationClient("sk_live_...")

async def create_post(request):
    content = request.POST.get("content")

    result = await client.check(
        content=content,
        policy_id="community-safe"
    )

    if result.action == "block":
        return JsonResponse(
            {"error": "Content blocked", "decision_id": result.id},
            status=403
        )

    # Save post
    post = await Post.objects.acreate(
        content=content,
        moderation_decision_id=result.id
    )

    return JsonResponse({"id": post.id})

---

Decision Retrieval & Audit

Get Decision Details

decision = client.get_decision("decision-uuid")

print(decision.id)
print(decision.action)
print(decision.categories)
print(decision.policy.id)
print(decision.policy.version)
print(decision.created_at)

List Recent Decisions

decisions = client.list_decisions(limit=50, offset=0)

for d in decisions:
    print(f"{d.id}: {d.action} ({d.created_at})")

Get cURL for Debugging

curl_command = client.get_curl_command("decision-uuid")
print(curl_command)
# curl -X POST https://api.vettly.dev/v1/check ...

---

Policy Management

Create or Update Policy

policy = client.create_policy(
    policy_id="my-policy",
    yaml_content="""
name: My Community Policy
version: "1.0"
rules:
  - category: hate_speech
    threshold: 0.7
    provider: openai
    action: block
  - category: harassment
    threshold: 0.8
    provider: openai
    action: flag
fallback:
  provider: mock
  on_timeout: true
  timeout_ms: 5000
"""
)

print(policy.version)  # Immutable version hash

List Policies

policies = client.list_policies()

for p in policies:
    print(f"{p.id}: v{p.version}")

---

Webhooks

Register a Webhook

webhook = client.register_webhook(
    url="https://your-app.com/webhooks/vettly",
    events=["decision.blocked", "decision.flagged"],
    description="Production webhook for blocked content"
)

print(webhook.id)
print(webhook.secret)  # Use for signature verification

Webhook Signature Verification

from flask import Flask, request
from vettly import verify_webhook_signature, construct_webhook_event

app = Flask(__name__)
WEBHOOK_SECRET = "whsec_..."

@app.route("/webhooks/vettly", methods=["POST"])
def handle_webhook():
    payload = request.get_data(as_text=True)
    signature = request.headers.get("X-Vettly-Signature")

    if not verify_webhook_signature(payload, signature, WEBHOOK_SECRET):
        return "Invalid signature", 401

    event = construct_webhook_event(payload)

    if event["type"] == "decision.blocked":
        # Handle blocked content
        decision_id = event["data"]["decision_id"]
        notify_moderator(decision_id)
    elif event["type"] == "decision.flagged":
        # Queue for human review
        add_to_review_queue(event["data"])

    return "OK", 200

Manage Webhooks

# List all webhooks
webhooks = client.list_webhooks()

# Update a webhook
client.update_webhook(
    webhook_id="webhook-id",
    events=["decision.blocked"],
    enabled=True
)

# Test a webhook
test_result = client.test_webhook("webhook-id", "decision.blocked")
print(f"Test {'passed' if test_result['success'] else 'failed'}")

# View delivery logs
deliveries = client.get_webhook_deliveries("webhook-id", limit=10)

# Delete a webhook
client.delete_webhook("webhook-id")

---

Idempotency

Prevent duplicate processing with request IDs:

result = client.check(
    content="Hello",
    policy_id="default",
    request_id="unique-request-id-123"
)

# Same request_id returns cached result
duplicate = client.check(
    content="Hello",
    policy_id="default",
    request_id="unique-request-id-123"
)

assert result.id == duplicate.id  # Same decision

---

Error Handling

The SDK provides typed exceptions for precise handling:

from vettly import (
    VettlyAPIError,
    VettlyAuthError,
    VettlyRateLimitError,
    VettlyQuotaError,
    VettlyValidationError,
    VettlyServerError,
)

try:
    result = client.check(content="test", policy_id="default")
except VettlyAuthError:
    # Invalid or expired API key
    print("Authentication failed - check your API key")
except VettlyRateLimitError as e:
    # Rate limited - SDK retries automatically, this means retries exhausted
    print(f"Rate limited. Retry after {e.retry_after}s")
except VettlyQuotaError as e:
    # Monthly quota exceeded
    print(f"Quota exceeded: {e.quota}")
except VettlyValidationError as e:
    # Invalid request parameters
    print(f"Validation error: {e.errors}")
except VettlyServerError as e:
    # Server error (5xx)
    print(f"Server error: {e.status_code}")
except VettlyAPIError as e:
    # Other API errors
    print(f"{e.code}: {e.message}")

---

Configuration

from vettly import ModerationClient

client = ModerationClient(
    api_key="sk_live_...",
    api_url="https://api.vettly.dev",  # Optional: custom API URL
    timeout=30.0,                       # Request timeout in seconds
    max_retries=3,                      # Max retries for failures
    retry_delay=1.0,                    # Base delay for exponential backoff
)

---

Context Manager

Use as a context manager for automatic cleanup:

# Sync client
with ModerationClient("sk_live_...") as client:
    result = client.check(content="Hello", policy_id="default")

# Async client
async with AsyncModerationClient("sk_live_...") as client:
    result = await client.check(content="Hello", policy_id="default")

---

Response Models

CheckResponse

from vettly import Action, CheckResponse, CategoryResult

result: CheckResponse = client.check(...)

result.id           # str - Decision ID (UUID)
result.safe         # bool
result.flagged      # bool
result.action       # Action (ALLOW, WARN, FLAG, BLOCK)
result.categories   # List[CategoryResult]
result.latency_ms   # int
result.policy_id    # str

CategoryResult

for category in result.categories:
    category.category   # str - Category name
    category.score      # float - 0.0 to 1.0
    category.threshold  # float - Configured threshold
    category.triggered  # bool - Whether threshold was exceeded

Action Enum

from vettly import Action

Action.ALLOW   # Content passes all checks
Action.WARN    # Minor concern, user should be notified
Action.FLAG    # Needs human review
Action.BLOCK   # Violates policy

---

Who Uses Vettly

Trust & Safety Teams

• Audit trail for every decision
• Policy version history
• Evidence preservation for appeals

Legal & Compliance

• DSA Article 17 compliance
• Decision records for legal discovery
• Policy approval workflows

Engineering Teams

• Type-safe Pydantic models
• Automatic retries with exponential backoff
• Full async support

---

Links

• vettly.dev - Sign up
• docs.vettly.dev - Documentation
• PyPI - Package