Python Integration Guide

# ON-DEVICE — the decision is made on your machine.
# Import root is tork_governance; the entry class is Tork.
pip install tork-governance

# CLOUD / REST — Tork makes the decision server-side and writes
# a receipt to your dashboard. No Python SDK is published for this
# path; call the REST API with requests or httpx.
pip install requests  # or httpx for async

Never commit your API key. Add .env to .gitignore and use python-dotenv to load it.

import time
import requests
from requests.exceptions import RequestException, HTTPError

class TorkAPIError(Exception):
    """Custom exception for Tork API errors."""
    def __init__(self, message: str, status_code: int = None, request_id: str = None):
        self.message = message
        self.status_code = status_code
        self.request_id = request_id
        super().__init__(self.message)

def govern_with_retry(
    content: str,
    max_retries: int = 3,
    backoff_factor: float = 0.5
) -> dict:
    """Govern content with exponential backoff retry."""

    for attempt in range(max_retries):
        try:
            response = requests.post(
                "https://tork.network/api/v1/govern",
                headers={
                    "Authorization": f"Bearer {TORK_API_KEY}",
                    "Content-Type": "application/json"
                },
                json={"content": content},
                timeout=10
            )

            # Handle specific HTTP errors
            if response.status_code == 401:
                raise TorkAPIError("Invalid API key", 401)

            if response.status_code == 429:
                # Rate limited - wait and retry
                retry_after = int(response.headers.get("Retry-After", 60))
                if attempt < max_retries - 1:
                    time.sleep(retry_after)
                    continue
                raise TorkAPIError("Rate limit exceeded", 429)

            if response.status_code >= 500:
                # Server error - retry with backoff
                if attempt < max_retries - 1:
                    time.sleep(backoff_factor * (2 ** attempt))
                    continue
                raise TorkAPIError(f"Server error: {response.status_code}", response.status_code)

            response.raise_for_status()
            return response.json()

        except requests.Timeout:
            if attempt < max_retries - 1:
                time.sleep(backoff_factor * (2 ** attempt))
                continue
            raise TorkAPIError("Request timed out")

        except RequestException as e:
            raise TorkAPIError(f"Request failed: {str(e)}")

    raise TorkAPIError("Max retries exceeded")

HTTP Status Codes

from tork_governance import Tork, GovernanceAction
from typing import List, Dict, Any

# On-device governance is CPU-bound and local. There is no rate limit
# to respect and no concurrency to tune — a plain loop is the fastest
# correct answer. Async/semaphores would add overhead and buy nothing.
tork = Tork()

def govern_batch(items: List[str]) -> List[Dict[str, Any]]:
    results = []
    for content in items:
        result = tork.govern(content)
        results.append({
            "content": content[:50] + "...",
            "action": result.action.value,
            "pii_count": result.pii.count,
            "pii_types": [t.value for t in result.pii.types],
            "output": result.output,
        })
    return results

# Usage
messages = [
    "Hello, how are you?",
    "My email is john@example.com",
    "Card 4111 1111 1111 1111",
]

results = govern_batch(messages)

# Anything that wasn't simply allowed through
flagged = [r for r in results if r["action"] != GovernanceAction.ALLOW.value]
print(f"{len(flagged)} of {len(results)} messages were redacted or denied")

# For very large volumes, use multiprocessing — the work is CPU-bound,
# so processes (not threads or coroutines) are what actually scale it.

from tork_governance import Tork

# Passing an api_key turns ON metadata-only attestation reporting.
# The decision is STILL made on-device and is never delayed or
# changed by reporting. Tork never receives your input text,
# output text, or the PII values themselves.
tork = Tork(api_key="tork_live_...")

result = tork.govern("Contact me at john@example.com")

# The local decision is available immediately, as always.
print(result.action)   # GovernanceAction.REDACT
print(result.output)   # "Contact me at [EMAIL_REDACTED]"

# result.report describes the SEPARATE reporting attempt, which runs
# on a background thread. attempted and succeeded are independent —
# check succeeded, not just attempted.
if result.report:
    result.report.wait()          # optional; most callers don't need this
    print(result.report.attempted)
    print(result.report.succeeded)
    print(result.report.receipt_id)  # set only if the server persisted it
    print(result.report.reason)      # set only if it did not

# NOTE: this requires a build that is not on PyPI yet — see the
# release note below. Installing today gives you local governance
# with no reporting at all.

Supply an API key and the SDK additionally reports a metadata-only attestation of each decision — the action taken, PII type labels and counts, a risk classification, policy labels and a salted fingerprint. It never sends input text, output text or PII values. The decision itself is still made on-device and is never delayed or changed by reporting. Those attestations appear in your dashboard and are included in the daily on-chain anchor — each one recorded as a client attestation (capture_mode=edge, attested_by=client), a claim Tork recorded but did not execute and cannot independently verify. Requires tork-governance 0.24.0+ (PyPI) or 0.11.0+ (npm).

A decision reported by an on-device SDK is recorded as a client attestation (capture_mode=edge, attested_by=client): a claim Tork recorded but did not execute and cannot independently verify. A decision made by @torknetwork/sdk is recorded as capture_mode=cloud, attested_by=tork — Tork made that call itself. Both are equally immutable once anchored; they differ in what is immutable. A client attestation freezes your claim. A server-governed receipt freezes Tork's own decision.