Troubleshooting & FAQ

# Test your API key
curl -X GET https://api.hitl.sh/v1/api/loops \
  -H "Authorization: Bearer your_api_key_here" \
  -v

# Check response for detailed error information

headers = {
    "Authorization": "Bearer your_api_key_here",  # Correct
    "Content-Type": "application/json"
}

# Wrong formats:
# "Authorization": "Bearer your_api_key_here"             # Wrong header name
# "authorization": "Bearer your_api_key_here"  # Wrong case
# "Authorization": "your_api_key_here"         # Missing Bearer prefix

import time
import random
from requests.exceptions import HTTPError

def api_call_with_backoff(func, max_retries=3):
    """Make API call with exponential backoff for rate limiting"""
    for attempt in range(max_retries):
        try:
            return func()
        except HTTPError as e:
            if e.response.status_code == 429:
                # Extract retry-after header if available
                retry_after = e.response.headers.get('Retry-After', '60')
                wait_time = int(retry_after) + random.uniform(1, 5)
                
                print(f"Rate limited. Waiting {wait_time:.1f} seconds...")
                time.sleep(wait_time)
                continue
            raise
    raise Exception("Max retries exceeded")

# Usage
def make_request():
    return requests.post(url, headers=headers, json=data)

result = api_call_with_backoff(make_request)

# Minimum required fields
request_data = {
    "loop_id": "65f1234567890abcdef12345",  # Required
    "request_text": "Please review this content...",  # Required
    "response_type": "single_select",  # Required
    "response_config": {  # Required for most response types
        "options": ["Approve", "Reject"]
    }
}

# Single Select
{
    "response_type": "single_select",
    "response_config": {
        "options": ["Option 1", "Option 2", "Option 3"]  # 2-10 options
    }
}

# Rating
{
    "response_type": "rating", 
    "response_config": {
        "scale_max": 5  # Required
        # scale_min defaults to 1
        # scale_step defaults to 1
    }
}

# Text
{
    "response_type": "text",
    "response_config": {
        "max_length": 1000,  # Optional, default 1000
        "placeholder": "Enter feedback..."  # Optional
    }
}

# Verify loop exists and get correct ID
response = requests.get(
    "https://api.hitl.sh/v1/api/loops",
    headers={"Authorization": "Bearer your_api_key_here"}
)

loops = response.json()["data"]["loops"]
for loop in loops:
    print(f"Loop: {loop['name']} - ID: {loop['id']}")

def create_request_with_length_check(request_text, **kwargs):
    MAX_LENGTH = 10000
    
    if len(request_text) > MAX_LENGTH:
        # Truncate and add notice
        truncated_text = request_text[:MAX_LENGTH-100]  # Leave room for notice
        request_text = f"{truncated_text}... [Content truncated due to length]"
    
    return create_request(request_text=request_text, **kwargs)

# Good loop names
"Content Moderation Team"
"AI Safety Review - Internal"
"Customer Support Escalation Q3"

# Invalid names
""  # Too short
"A" * 101  # Too long
"Loop-Name-With-Too@Many#Special$Characters%"  # Invalid characters

response = requests.get(f"{BASE_URL}/loops/{loop_id}/members", headers=headers)
members = response.json()["data"]
print(f"Active members: {members['active_count']}")
print(f"Pending invitations: {members['pending_count']}")

response = requests.get(f"{BASE_URL}/loops/{loop_id}", headers=headers)
loop = response.json()["data"]["loop"]
print(f"Invite code: {loop['invite_code']}")
print(f"QR code: {loop['qr_code_url']}")

# Test if your webhook URL is accessible
curl -X POST https://your-domain.com/webhook/hitl \
  -H "Content-Type: application/json" \
  -d '{"test": "webhook accessibility"}'

# Should return 200 OK or your custom response

# Use ngrok for local development
npm install -g ngrok
ngrok http 3000

# Use the https URL provided by ngrok
# https://abc123.ngrok.io/webhook/hitl

import hmac
import hashlib

def verify_webhook_signature(payload, signature, secret):
    """Verify HITL webhook signature"""
    
    # Remove sha256= prefix if present
    if signature.startswith('sha256='):
        signature = signature[7:]
    
    # Calculate expected signature using raw payload
    expected_signature = hmac.new(
        secret.encode('utf-8'),
        payload.encode('utf-8') if isinstance(payload, str) else payload,
        hashlib.sha256
    ).hexdigest()
    
    # Use secure comparison
    return hmac.compare_digest(expected_signature, signature)

# Flask example
@app.route('/webhook/hitl', methods=['POST'])
def handle_webhook():
    # Get raw payload - don't parse JSON first!
    payload = request.get_data(as_text=True)
    signature = request.headers.get('X-HITL-Signature-256')
    
    if not verify_webhook_signature(payload, signature, WEBHOOK_SECRET):
        return {'error': 'Invalid signature'}, 401
    
    # Now parse the JSON
    data = json.loads(payload)
    # ... process webhook

# Return 200 immediately, process asynchronously
from flask import Flask, request
import threading

app = Flask(__name__)

@app.route('/webhook/hitl', methods=['POST'])
def handle_webhook_quickly():
    # Verify signature first
    if not verify_signature(request.get_data(), request.headers.get('X-HITL-Signature-256')):
        return {'error': 'Invalid signature'}, 401
    
    # Return success immediately
    webhook_data = request.json
    
    # Process in background thread
    threading.Thread(target=process_webhook_async, args=(webhook_data,)).start()
    
    return {'status': 'received'}, 200

def process_webhook_async(data):
    # Long-running processing here
    # Database updates, external API calls, etc.
    pass

# Verify webhook is configured for all needed events
webhook_config = {
    "url": "https://your-domain.com/webhook/hitl",
    "events": [
        "request.completed",  # When requests are completed
        "request.claimed",    # When requests are claimed  
        "request.cancelled",  # When requests are cancelled
        "request.timeout",    # When requests time out
        "loop.member_joined", # When new members join
        # Add all events you need
    ],
    "secret": "your_webhook_secret"
}

{
  "event": "request.completed",
  "timestamp": "2024-03-15T14:30:00Z",
  "data": {
    "request": {
      "id": "65f1234567890abcdef12348",
      "loop_id": "65f1234567890abcdef12345", 
      "status": "completed",
      "response_data": "Approve",
      "response_by": "65f1234567890abcdef12350",
      "response_at": "2024-03-15T14:30:00Z"
      // ... other request fields
    },
    "loop": {
      "id": "65f1234567890abcdef12345",
      "name": "Content Moderation Team"
    },
    "reviewer": {
      "user_id": "65f1234567890abcdef12350",
      "email": "reviewer@example.com"
    }
  }
}

def process_webhook(event_data):
    event_type = event_data['event']
    data = event_data['data']
    
    if event_type == 'request.completed':
        request = data['request']
        reviewer = data['reviewer']
        # Process completed request
        handle_completed_request(request, reviewer)
        
    elif event_type == 'request.claimed':
        request = data['request'] 
        reviewer = data['reviewer']
        # Handle request claimed
        handle_claimed_request(request, reviewer)
        
    elif event_type == 'loop.member_joined':
        loop = data['loop']
        member = data['member']
        # Handle new member
        handle_new_member(loop, member)

# Get invite code for manual entry
response = requests.get(f"{BASE_URL}/loops/{loop_id}", headers=headers)
loop = response.json()["data"]["loop"]
print(f"Manual invite code: {loop['invite_code']}")

# Check if loop still exists and get fresh invite code
def verify_loop_invite(loop_id):
    try:
        response = requests.get(f"{BASE_URL}/loops/{loop_id}", headers=headers)
        if response.status_code == 200:
            loop = response.json()["data"]["loop"]
            return {
                "exists": True,
                "invite_code": loop["invite_code"], 
                "member_count": loop["member_count"],
                "name": loop["name"]
            }
    except:
        return {"exists": False}

# Use connection pooling
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

session = requests.Session()

# Configure retries and connection pooling
retry_strategy = Retry(
    total=3,
    status_forcelist=[429, 500, 502, 503, 504],
    backoff_factor=1
)

adapter = HTTPAdapter(
    pool_connections=10,
    pool_maxsize=20,
    max_retries=retry_strategy
)

session.mount("http://", adapter)
session.mount("https://", adapter)

# Use session for all requests
response = session.get(url, headers=headers)

# Cache loop IDs to avoid repeated lookups
import functools
import time

@functools.lru_cache(maxsize=100)
def get_loop_id_by_name(loop_name):
    """Cache loop ID lookups"""
    response = requests.get(f"{BASE_URL}/loops", headers=headers)
    loops = response.json()["data"]["loops"]
    
    for loop in loops:
        if loop["name"] == loop_name:
            return loop["id"]
    return None

# Cache expires after 1 hour
@functools.lru_cache(maxsize=100)
def get_cached_loop_details(loop_id, cache_time=3600):
    """Cache loop details with TTL"""
    current_time = int(time.time())
    cache_key = f"{loop_id}_{current_time // cache_time}"  # TTL buckets
    
    response = requests.get(f"{BASE_URL}/loops/{loop_id}", headers=headers)
    return response.json()["data"]["loop"]

# Use appropriate timeouts
import requests

# Configure reasonable timeouts
response = requests.post(
    url,
    json=data,
    headers=headers,
    timeout=(5, 30)  # (connection_timeout, read_timeout)
)

# Implement request compression
headers = {
    "Authorization": "Bearer your_api_key",
    "Content-Type": "application/json",
    "Accept-Encoding": "gzip, deflate"
}

# Use async requests for parallel operations
import aiohttp
import asyncio

async def create_multiple_requests(request_data_list):
    async with aiohttp.ClientSession() as session:
        tasks = []
        for data in request_data_list:
            task = session.post(
                f"{BASE_URL}/requests",
                json=data,
                headers=headers
            )
            tasks.append(task)
        
        responses = await asyncio.gather(*tasks)
        return [await resp.json() for resp in responses]

def diagnose_pending_request(request_id):
    # Get request details
    req_response = requests.get(f"{BASE_URL}/requests/{request_id}", headers=headers)
    request = req_response.json()["data"]["request"]
    
    # Get loop member status  
    loop_response = requests.get(f"{BASE_URL}/loops/{request['loop_id']}/members", headers=headers)
    members = loop_response.json()["data"]
    
    diagnosis = {
        "request_age_minutes": calculate_age(request["created_at"]),
        "priority": request["priority"],
        "active_reviewers": members["active_count"],
        "timeout_minutes": request.get("timeout_minutes", 60)
    }
    
    # Provide recommendations
    if members["active_count"] == 0:
        diagnosis["issue"] = "No active reviewers in loop"
        diagnosis["solution"] = "Invite reviewers or check existing members' status"
    elif diagnosis["request_age_minutes"] > diagnosis["timeout_minutes"]:
        diagnosis["issue"] = "Request has timed out"
        diagnosis["solution"] = "Check for timeout response or create new request"
    
    return diagnosis

def handle_response_data_safely(request_data):
    """Safely handle different response data formats"""
    response_type = request_data["response_type"]
    response_data = request_data["response_data"]
    
    if response_type == "single_select":
        # Should be string
        assert isinstance(response_data, str)
        return response_data
    
    elif response_type == "multi_select":  
        # Should be array of strings
        assert isinstance(response_data, list)
        return [str(item) for item in response_data]
    
    elif response_type == "rating":
        # Should be number
        assert isinstance(response_data, (int, float))
        return float(response_data)
    
    elif response_type == "text":
        # Should be string
        assert isinstance(response_data, str)
        return response_data.strip()
    
    elif response_type == "boolean":
        # Should be boolean
        assert isinstance(response_data, bool)
        return response_data
        
    elif response_type == "number":
        # Should be number
        assert isinstance(response_data, (int, float))
        return response_data
    
    else:
        raise ValueError(f"Unknown response type: {response_type}")

# Install the SDK
pip install hitl-python

# Or if using conda
conda install -c conda-forge hitl-python

# Verify installation
python -c "import hitl; print(hitl.__version__)"

# Make sure you're in the right virtual environment
which python
which pip

# Install in correct environment  
pip install hitl-python

# Or create new environment
python -m venv hitl_env
source hitl_env/bin/activate  # On Windows: hitl_env\Scripts\activate
pip install hitl-python

# For development only - don't use in production
import requests
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry
import ssl

# Disable SSL warnings (development only)
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

# Create custom session with SSL settings
session = requests.Session()
session.verify = True  # Keep True for production

# Or specify certificate bundle location
session.verify = '/path/to/certificate/bundle.pem'

# Use with HITL client
from hitl import HITLClient
client = HITLClient(api_key="your_key", session=session)

# Install SDK
npm install hitl-js

# Or with yarn
yarn add hitl-js

# Check installation
npm list hitl-js

# Clear npm cache if needed
npm cache clean --force
rm -rf node_modules package-lock.json
npm install

// ES6 modules (newer Node.js or with "type": "module" in package.json)
import { HITLClient } from 'hitl-js';

// CommonJS (older Node.js)
const { HITLClient } = require('hitl-js');

// TypeScript
import { HITLClient, Loop, Request } from 'hitl-js';

// Make sure function is async
async function main() {
  const client = new HITLClient({ apiKey: 'your_key' });
  
  try {
    const loops = await client.loops.list();
    console.log(loops);
  } catch (error) {
    console.error('Error:', error);
  }
}

// Or use promises
function mainPromises() {
  const client = new HITLClient({ apiKey: 'your_key' });
  
  client.loops.list()
    .then(loops => console.log(loops))
    .catch(error => console.error('Error:', error));
}

// Top-level await (Node.js 14+)
const loops = await client.loops.list();