Skip to main content

​
Simple Integration Guide

This guide shows you how to integrate HITL.sh into your applications to get human input when your automated systems need help making decisions.
Perfect for developers who want to add human oversight to AI systems, content moderation, quality checks, and approval workflows.

​
What You’ll Learn

1

Basic Setup

Get your API key and understand the core concepts
2

Common Patterns

Learn the most useful integration patterns with copy-paste examples
3

Handle Responses

Process human responses in your application code
4

Best Practices

Tips for reliable production integration

​
Prerequisites

  1. Sign up at app.hitl.sh
  2. Create a new API key in your dashboard
  3. Set up a loop with team members who will review requests
export HITL_API_KEY="your_api_key_here"
export HITL_LOOP_ID="your_loop_id_here"  

pip install requests

​
Integration Pattern #1: Content Moderation

Perfect for reviewing user-generated content, AI outputs, or flagged posts. 
import requests
import time

def review_content(content, priority="medium"):
    """Send content for human review and return the decision"""
    
    # Create the review request
    response = requests.post(
        f"https://api.hitl.sh/v1/api/loops/{HITL_LOOP_ID}/requests",
        headers={
            "Authorization": f"Bearer {HITL_API_KEY}",
            "Content-Type": "application/json"
        },
        json={
            "processing_type": "time-sensitive",
            "type": "markdown",
            "priority": priority,
            "request_text": f"Please review this content for community guidelines compliance:\n\n{content}",
            "timeout_seconds": 1800,  # 30 minutes
            "response_type": "single_select",
            "response_config": {
                "options": [
                    {"value": "approve", "label": "Approve - Safe to publish"},
                    {"value": "reject", "label": "Reject - Violates guidelines"},
                    {"value": "needs_review", "label": "Needs Review - Unclear content"}
                ],
                "required": true
            },
            "default_response": "reject",  # Safe default
            "platform": "api"
        }
    )
    
    if response.status_code != 201:
        return {"error": "Failed to create request"}
    
    request_data = response.json()
    request_id = request_data["data"]["request_id"]
    
    # Poll for response (in production, use webhooks instead)
    return wait_for_response(request_id)

def wait_for_response(request_id, max_wait_minutes=30):
    """Wait for human response (use webhooks in production)"""
    
    for _ in range(max_wait_minutes * 2):  # Check every 30 seconds
        response = requests.get(
            f"https://api.hitl.sh/v1/api/requests/{request_id}",
            headers={"Authorization": f"Bearer {HITL_API_KEY}"}
        )
        
        if response.status_code == 200:
            data = response.json()["data"]["request"]
            
            if data["status"] == "completed":
                return {
                    "decision": data["response_data"],
                    "reviewed_by": data.get("response_by_user", {}).get("name", "Unknown"),
                    "response_time": data.get("response_time_seconds", 0)
                }
            elif data["status"] in ["timeout", "cancelled"]:
                return {"decision": data["default_response"], "timeout": True}
        
        time.sleep(30)  # Wait 30 seconds before next check
    
    return {"error": "Timeout waiting for response"}

# Usage example
if __name__ == "__main__":
    user_post = "Check out this amazing deal on cryptocurrency!"
    
    result = review_content(user_post, priority="high")
    
    if result.get("decision") == "approve":
        print("✅ Content approved - publish it!")
    elif result.get("decision") == "reject":
        print("❌ Content rejected - don't publish")
    else:
        print(f"⚠️ Content needs review: {result}")

​
Integration Pattern #2: AI Output Quality Check

Perfect for reviewing AI-generated content, translations, or automated responses before sending them to users. 
def review_ai_output(ai_response, context="", min_quality=3):
    """Review AI output quality and get human feedback"""
    
    response = requests.post(
        f"https://api.hitl.sh/v1/api/loops/{HITL_LOOP_ID}/requests",
        headers={
            "Authorization": f"Bearer {HITL_API_KEY}",
            "Content-Type": "application/json"
        },
        json={
            "processing_type": "time-sensitive",
            "type": "markdown",
            "priority": "medium",
            "request_text": f"Rate this AI response quality (1-5 scale):\n\nContext: {context}\n\nAI Response: {ai_response}",
            "timeout_seconds": 2400,  # 40 minutes  
            "response_type": "rating",
            "response_config": {
                "min": 1,
                "max": 5
            },
            "default_response": 2,  # Conservative default
            "platform": "api"
        }
    )
    
    if response.status_code != 201:
        return {"error": "Failed to create request"}
    
    request_data = response.json()
    request_id = request_data["data"]["request_id"]
    result = wait_for_response(request_id)
    
    if isinstance(result.get("decision"), (int, float)):
        return {
            "quality_score": result["decision"],
            "approved": result["decision"] >= min_quality,
            "reviewer": result.get("reviewed_by"),
            "recommendation": "approve" if result["decision"] >= min_quality else "revise"
        }
    
    return result

# Usage
ai_content = "The weather today is quite pleasant with sunny skies."
context = "Customer asked about today's weather"

quality_check = review_ai_output(ai_content, context, min_quality=3)

if quality_check.get("approved"):
    print(f"✅ AI response approved (score: {quality_check['quality_score']}/5)")
    # Send AI response to customer
else:
    print(f"❌ AI response needs improvement (score: {quality_check['quality_score']}/5)")
    # Generate new AI response or escalate to human agent

​
Integration Pattern #3: Document Approval Workflow

Great for reviewing contracts, proposals, marketing materials, or any documents that need human approval. 
def approve_document(document_title, document_content, urgency="medium"):
    """Send document for approval with detailed feedback"""
    
    response = requests.post(
        f"https://api.hitl.sh/v1/api/loops/{HITL_LOOP_ID}/requests",
        headers={
            "Authorization": f"Bearer {HITL_API_KEY}",
            "Content-Type": "application/json"
        },
        json={
            "processing_type": "time-sensitive" if urgency == "high" else "deferred",
            "type": "markdown",
            "priority": urgency,
            "request_text": f"Please review this document for approval:\n\n**Title:** {document_title}\n\n**Content:**\n{document_content}",
            "timeout_seconds": 3600 if urgency == "high" else 86400,  # 1 hour vs 24 hours
            "response_type": "multi_select",
            "response_config": {
                "options": [
                    {"value": "approve_as_is", "label": "Approve as-is"},
                    {"value": "approve_minor_changes", "label": "Approve with minor changes"},
                    {"value": "needs_major_revisions", "label": "Needs major revisions"},
                    {"value": "legal_review_required", "label": "Legal review required"},
                    {"value": "reject_start_over", "label": "Reject - start over"}
                ],
                "min_selections": 1,
                "max_selections": 5,
                "required": true
            },
            "default_response": "needs_major_revisions",  # Conservative default
            "platform": "api"
        }
    )
    
    if response.status_code != 201:
        return {"error": "Failed to create request"}
    
    request_data = response.json()
    request_id = request_data["data"]["request_id"] 
    
    return wait_for_response(request_id)

# Usage
doc_title = "Q4 Marketing Proposal"
doc_content = """
## Objective
Increase brand awareness by 25% through targeted social media campaigns.

## Budget
$50,000 for Q4 campaigns

## Timeline  
October 1 - December 31, 2024
"""

approval = approve_document(doc_title, doc_content, urgency="high")

if isinstance(approval.get("decision"), list):
    decisions = approval["decision"]
    
    if "approve_as_is" in decisions:
        print("✅ Document fully approved - proceed with implementation")
    elif "approve_minor_changes" in decisions:
        print("⚠️ Document approved with minor changes needed")
    elif "legal_review_required" in decisions:
        print("⚖️ Document needs legal review before approval")  
    else:
        print(f"❌ Document needs work: {', '.join(decisions)}")
Instead of polling for responses, set up webhooks to get notified instantly when reviews complete: 
from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/hitl-webhook', methods=['POST'])
def handle_hitl_response():
    """Handle webhook notifications from HITL.sh"""
    
    webhook_data = request.json
    
    if webhook_data.get('event') == 'request.completed':
        request_id = webhook_data['data']['request_id']
        decision = webhook_data['data']['response_data']
        reviewer = webhook_data['data']['response_by_user']['name']
        
        # Process the decision in your application
        process_human_decision(request_id, decision, reviewer)
        
        return jsonify({"status": "success"})
    
    return jsonify({"status": "ignored"})

def process_human_decision(request_id, decision, reviewer):
    """Process the human decision in your application"""
    print(f"Request {request_id} completed by {reviewer}: {decision}")
    
    # Add your business logic here:
    # - Update database
    # - Send notifications  
    # - Trigger next steps in workflow
    # - Log the decision

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=3000)

​
Production Best Practices

  • Store your API key securely (environment variables, secret managers)
  • Use HTTPS for all webhook endpoints
  • Validate webhook signatures if available
  • Implement rate limiting on your webhook endpoints
import os
from cryptography.fernet import Fernet

# Secure API key storage
HITL_API_KEY = os.environ.get('HITL_API_KEY')
if not HITL_API_KEY:
    raise ValueError("HITL_API_KEY environment variable is required")
  • Use webhooks instead of polling in production
  • Implement retry logic with exponential backoff
  • Set appropriate timeouts for different request types
  • Cache frequently used loop IDs and configurations
import time
import random

def create_request_with_retry(request_data, max_retries=3):
    """Create request with retry logic"""
    
    for attempt in range(max_retries):
        try:
            response = requests.post(
                f"https://api.hitl.sh/v1/api/loops/{HITL_LOOP_ID}/requests",
                headers={"Authorization": f"Bearer {HITL_API_KEY}"},
                json=request_data,
                timeout=30
            )
            
            if response.status_code == 201:
                return response.json()
                
        except requests.exceptions.RequestException as e:
            if attempt == max_retries - 1:
                raise e
            
            # Exponential backoff with jitter
            wait_time = (2 ** attempt) + random.uniform(0, 1)
            time.sleep(wait_time)
    
    raise Exception("Max retries exceeded")
  • Log all HITL requests with unique identifiers
  • Monitor response times and success rates
  • Set up alerts for timeout rates above acceptable thresholds
  • Track reviewer performance and availability
import logging
from datetime import datetime

# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def log_hitl_request(request_type, content_summary, request_id=None):
    """Log HITL request for monitoring"""
    
    log_data = {
        "timestamp": datetime.utcnow().isoformat(),
        "request_type": request_type,
        "content_summary": content_summary[:100] + "..." if len(content_summary) > 100 else content_summary,
        "request_id": request_id,
        "status": "created" if request_id else "failed"
    }
    
    logger.info(f"HITL Request: {log_data}")
  • Provide clear, specific instructions to reviewers
  • Use appropriate default responses for timeout scenarios
  • Test different response types to find what works best
  • Regularly review and update your request templates
def create_clear_request(content, request_type="approval"):
    """Create clear, actionable requests for better response quality"""
    
    templates = {
        "approval": {
            "instructions": "Please review this content and decide if it's safe to publish:",
            "options": ["✅ Approve - Safe to publish", "❌ Reject - Do not publish", "⚠️ Needs changes - Specify in comments"]
        },
        "quality": {
            "instructions": "Rate the quality of this content (1=Poor, 5=Excellent):",
            "context": "Consider accuracy, clarity, usefulness, and engagement"
        }
    }
    
    template = templates.get(request_type, templates["approval"])
    
    return {
        "request_text": f"{template['instructions']}\n\n{content}",
        "response_config": template.get("options", {}),
        "context": {"template_used": request_type}
    }

​
Next Steps

Set Up Webhooks

Configure real-time notifications for production use

API Reference

Explore all available options and configurations

Response Types Guide

Learn about advanced response configurations

Best Practices

Advanced tips for optimizing your integration

​
Common Issues & Solutions

Common causes:
  • Invalid API key or loop ID
  • Missing required fields
  • Invalid response configuration
Solution:
# Validate configuration before sending
def validate_request_config(config):
    required_fields = ["processing_type", "type", "priority", "request_text", "response_type", "platform"]
    
    for field in required_fields:
        if field not in config:
            raise ValueError(f"Missing required field: {field}")
    
    if config["response_type"] == "single_select" and not config.get("response_config", {}).get("options"):
        raise ValueError("Single select requires options array")
    
    return True
Common causes:
  • Timeout too short for request complexity
  • Reviewers not available or not notified
  • Unclear instructions leading to hesitation
Solution:
# Adjust timeouts based on request complexity
def calculate_timeout(request_type, content_length):
    base_timeout = {
        "simple_approval": 1800,    # 30 minutes
        "detailed_review": 3600,    # 1 hour  
        "complex_analysis": 7200    # 2 hours
    }
    
    timeout = base_timeout.get(request_type, 1800)
    
    # Add extra time for longer content
    if content_length > 1000:
        timeout += 1800  # Add 30 minutes
    
    return min(timeout, 86400)  # Max 24 hours
Common causes:
  • Loop members haven’t installed the mobile app
  • Notification permissions disabled
  • Device tokens expired
Solution:
  • Ensure all reviewers have installed the HITL mobile app
  • Test notifications in your loop settings
  • Verify reviewers have enabled push notifications
  • Consider SMS or email fallback notifications
⌘I