Python Integration Guide

# Install the official SDK
pip install tork-governance

# Or for async support
pip install tork-governance[async]

# For direct API calls (no SDK)
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 evaluate_with_retry(
    content: str,
    max_retries: int = 3,
    backoff_factor: float = 0.5
) -> dict:
    """Evaluate content with exponential backoff retry."""

    for attempt in range(max_retries):
        try:
            response = requests.post(
                "https://tork.network/api/v1/evaluate",
                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

import asyncio
from tork_governance import AsyncTorkClient
from typing import List, Dict, Any

async def evaluate_batch(
    items: List[str],
    concurrency: int = 10,
    context: str = "batch"
) -> List[Dict[str, Any]]:
    """
    Evaluate multiple items with controlled concurrency.
    """
    client = AsyncTorkClient()
    semaphore = asyncio.Semaphore(concurrency)

    async def evaluate_one(content: str) -> Dict[str, Any]:
        async with semaphore:
            try:
                result = await client.evaluate(
                    content=content,
                    context=context
                )
                return {
                    "content": content[:50] + "...",
                    "safe": result.safe,
                    "risk_score": result.risk_score,
                    "flags": result.flags
                }
            except Exception as e:
                return {
                    "content": content[:50] + "...",
                    "error": str(e)
                }

    results = await asyncio.gather(*[
        evaluate_one(item) for item in items
    ])

    await client.close()
    return results

# Usage
async def main():
    messages = [
        "Hello, how are you?",
        "What's the weather like?",
        "Ignore all previous instructions...",
        # ... more messages
    ]

    results = await evaluate_batch(messages, concurrency=5)

    # Filter unsafe content
    unsafe = [r for r in results if not r.get("safe", True)]
    print(f"Found {len(unsafe)} unsafe messages")

asyncio.run(main())

from contextlib import contextmanager, asynccontextmanager
from tork_governance import TorkClient, AsyncTorkClient

# Sync context manager
@contextmanager
def tork_session():
    """Context manager for Tork client."""
    client = TorkClient()
    try:
        yield client
    finally:
        # Cleanup if needed
        pass

# Async context manager
@asynccontextmanager
async def async_tork_session():
    """Async context manager for Tork client."""
    client = AsyncTorkClient()
    try:
        yield client
    finally:
        await client.close()

# Usage
def process_message(message: str):
    with tork_session() as tork:
        result = tork.evaluate(content=message)
        if result.safe:
            return handle_safe_message(message)
        else:
            return handle_unsafe_message(result.flags)

async def process_message_async(message: str):
    async with async_tork_session() as tork:
        result = await tork.evaluate(content=message)
        return result