Error Handling

{
  "error": {
    "code": "policy_violation",
    "message": "Action blocked by policy: block-pii-sharing",
    "status": 403,
    "details": {
      "policy_name": "block-pii-sharing",
      "agent_id": "agent-123",
      "violation_type": "pii_detected"
    },
    "request_id": "req_abc123def456",
    "documentation_url": "https://docs.tork.network/errors/policy_violation"
  }
}

import os
import time
import requests

# There is no Python cloud SDK — use the REST API.
TORK_API_KEY = os.environ["TORK_API_KEY"]

response = requests.post(
    "https://tork.network/api/v1/govern",
    headers={
        "Authorization": f"Bearer {TORK_API_KEY}",
        "Content-Type": "application/json",
    },
    json={"content": "..."},
)

if response.ok:
    result = response.json()
else:
    error = response.json().get("error", {})
    code = error.get("code")

    if code == "policy_violation":
        print(f"Policy blocked: {error.get('message')}")
        # Log for audit, notify user

    elif code == "budget_exceeded":
        print("Budget exceeded")
        # Queue for later or request a budget increase

    elif code == "approval_required":
        print(f"Approval needed: {error.get('details', {}).get('approval_id')}")
        # Wait for approval or notify approvers

    elif code == "rate_limit_exceeded":
        # Honour Retry-After when the server sends it
        wait = int(response.headers.get("Retry-After", 1))
        print(f"Rate limited, retry after: {wait}s")
        time.sleep(wait)

    else:
        print(f"API error {response.status_code}: {code} - {error.get('message')}")

    # Always log the request_id — support needs it to trace the call
    print(f"Request ID: {error.get('request_id')}")

import time
import requests

def with_retry(send, max_retries=3):
    """Send a Tork API request, retrying 429s and 5xx with backoff."""
    for attempt in range(max_retries):
        response = send()

        if response.status_code == 429:
            if attempt == max_retries - 1:
                return response
            wait = int(response.headers.get("Retry-After", 2 ** attempt))
            time.sleep(wait)
            continue

        # Only retry server-side failures; 4xx will not succeed on replay
        if response.status_code >= 500:
            if attempt == max_retries - 1:
                return response
            time.sleep(2 ** attempt)
            continue

        return response

    return response

# Usage
response = with_retry(lambda: requests.post(
    "https://tork.network/api/v1/govern",
    headers={"Authorization": f"Bearer {TORK_API_KEY}"},
    json={"content": "..."},
))