Skip to main content
Build your first HITL (Human-in-the-Loop) application in under 10 minutes. This tutorial walks you through creating a content moderation system where human reviewers evaluate user-generated content.
What You’ll Build: A content moderation system that automatically sends user comments to human reviewers for approval or rejection, with real-time webhook notifications.

​
Prerequisites

HITL.sh Account

Sign up for free and get your API key from the dashboard.

Development Environment

Python 3.7+, Node.js 16+, or your preferred language with HTTP client support.

​
Step 1: Get Your API Key

1

Sign Up

Create your free account at hitl.sh/signup
2

Generate API Key

Go to your dashboard and create a new API key. Copy and save it securely.
3

Test Connection

Verify your API key works using the dedicated test endpoint:
curl -X GET https://api.hitl.sh/v1/test \
  -H "Authorization: Bearer your_api_key_here"
You should get a JSON response confirming your API key is valid and showing your account information.

​
Step 2: Create Your First Loop

A loop is a group of human reviewers who will evaluate your requests. Let’s create a content moderation team:
  • Python
  • Node.js
  • cURL
import requests

# Your API credentials
API_KEY = "your_api_key_here"
BASE_URL = "https://api.hitl.sh/v1"

headers = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json"
}

# Create a content moderation loop
loop_data = {
    "name": "Content Moderation Team",
    "description": "Reviews user-generated content for community guidelines compliance",
    "icon": "🛡️"
}

response = requests.post(f"{BASE_URL}/api/loops", headers=headers, json=loop_data)
loop_response = response.json()

print(f"✅ Created loop: {loop_response['data']['loop']['name']}")
print(f"   Loop ID: {loop_response['data']['loop']['id']}")
print(f"   Invite Code: {loop_response['data']['invite_code']}")
print(f"   QR Code URL: {loop_response['data']['qr_code_url']}")
print(f"   Join URL: {loop_response['data']['join_url']}")

# Save the loop ID for the next steps
LOOP_ID = loop_response['data']['loop']['id']
Loop Creation Response: The API returns the created loop along with invitation details:
  • loop - The loop object with id, name, description, icon, etc.
  • invite_code - 6-character code for reviewers to join (e.g., “ABC123”)
  • qr_code_base64 - Base64-encoded QR code image
  • qr_code_url - Public URL to the QR code image
  • join_url - Direct link to join the loop (e.g., “https://hitl.sh/join/ABC123”)
Save the loop.id for creating requests!

​
Step 3: Invite Reviewers

Your loop needs human reviewers to evaluate requests. You have several options:
For Testing: You can proceed without reviewers for now. Later, you can test by downloading the HITL mobile app yourself and joining your loop.

​
Step 4: Create Your First Request

Now let’s create a content moderation request. This represents user-generated content that needs human review:
  • Python
  • Node.js
  • cURL
# Sample user comment to moderate
user_comment = """
Hey everyone! Just wanted to share my experience with this amazing product. 
It's been life-changing and I think you should all try it too! 
Check out my link in bio for a special discount. 
BTW, anyone who disagrees with me is totally wrong!
"""

# Create moderation request
request_data = {
    "loop_id": LOOP_ID,
    "request_text": f"Please review this user comment for community guidelines compliance:\n\n\"{user_comment}\"",
    "response_type": "single_select",
    "response_config": {
        "options": [
            "✅ Approve - Follows guidelines",
            "⚠️ Approve with Warning - Minor issues",
            "❌ Reject - Violates guidelines", 
            "🚨 Reject and Flag - Serious violation"
        ]
    },
    "priority": "medium",
    "processing_type": "time-sensitive",
    "default_response": "❌ Reject - Review timeout",
    "timeout_seconds": 3600  # 1 hour
}

response = requests.post(f"{BASE_URL}/api/loops/{LOOP_ID}/requests", headers=headers, json=request_data)
request = response.json()

if response.status_code == 201:
    print("✅ Content moderation request created!")
    print(f"   Request ID: {request['data']['request']['id']}")
    print(f"   Status: {request['data']['request']['status']}")
    print("   📱 Reviewers will receive this on their mobile app")
    
    REQUEST_ID = request['data']['request']['id']
else:
    print("❌ Error creating request:", request)

​
Step 5: Monitor Request Status

You can check the status of your request and see when it gets completed:
  • Python
  • Node.js
import time

def check_request_status(request_id):
    """Check and display request status"""
    response = requests.get(f"{BASE_URL}/api/requests/{request_id}", headers=headers)
    
    if response.status_code == 200:
        request_data = response.json()['data']['request']
        status = request_data['status']
        
        print(f"📊 Request Status: {status}")
        
        if status == 'completed':
            print(f"✅ Review completed!")
            print(f"   Decision: {request_data['response_data']}")
            print(f"   Reviewed by: {request_data.get('response_by', 'Unknown')}")
            print(f"   Response time: {request_data.get('response_time_seconds', 0):.1f} seconds")
            return True
        elif status == 'claimed':
            print(f"👤 Request claimed by reviewer (in progress)")
        elif status == 'pending':
            print(f"⏳ Waiting for reviewer to claim")
        elif status == 'cancelled':
            print(f"❌ Request was cancelled")
            return True
        elif status == 'timeout':
            print(f"⏰ Request timed out")
            print(f"   Default response used: {request_data.get('default_response')}")
            return True
    else:
        print(f"❌ Error checking status: {response.status_code}")
    
    return False

# Monitor the request
print("🔍 Monitoring request status...")
for i in range(12):  # Check for up to 2 minutes
    if check_request_status(REQUEST_ID):
        break
    time.sleep(10)  # Wait 10 seconds between checks
else:
    print("⏰ Still waiting for review after 2 minutes")
    print("   Request will continue processing in the background")

​
Step 6: Set Up Real-time Webhooks (Optional)

Instead of polling for updates, you can receive instant notifications when requests are completed:
  • Python (Flask)
  • Node.js (Express)
from flask import Flask, request, jsonify
import hmac
import hashlib
import json

app = Flask(__name__)

# Your webhook secret (set this in your HITL dashboard)
WEBHOOK_SECRET = "your_webhook_secret_here"

def verify_webhook_signature(payload, signature, secret):
    """Verify webhook signature for security"""
    expected_signature = hmac.new(
        secret.encode('utf-8'),
        payload.encode('utf-8'),
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(f"sha256={expected_signature}", signature)

@app.route('/webhook/hitl', methods=['POST'])
def handle_webhook():
    # Get the raw payload and signature
    payload = request.get_data(as_text=True)
    signature = request.headers.get('X-HITL-Signature-256')
    
    # Verify the webhook signature
    if not verify_webhook_signature(payload, signature, WEBHOOK_SECRET):
        return jsonify({'error': 'Invalid signature'}), 401
    
    # Parse the webhook data
    webhook_data = json.loads(payload)
    event_type = webhook_data['event']
    data = webhook_data['data']
    
    if event_type == 'request.completed':
        handle_request_completed(data)
    elif event_type == 'request.claimed':
        handle_request_claimed(data)
    
    return jsonify({'status': 'success'})

def handle_request_completed(data):
    """Process completed moderation request"""
    request_data = data['request']
    reviewer = data['reviewer']
    
    print("🎉 Content moderation completed!")
    print(f"   Request ID: {request_data['id']}")
    print(f"   Decision: {request_data['response_data']}")
    print(f"   Reviewer: {reviewer['email']}")
    
    # Apply the moderation decision
    decision = request_data['response_data']
    if "Approve" in decision:
        print("✅ Content approved - taking no action")
    elif "Reject" in decision:
        print("❌ Content rejected - hiding from users")
        # Your logic to hide/remove content
    elif "Flag" in decision:
        print("🚨 Content flagged - escalating for review")
        # Your logic to escalate serious violations

def handle_request_claimed(data):
    """Process when request is claimed by reviewer"""
    request_data = data['request']
    reviewer = data['reviewer']
    
    print(f"👤 Request {request_data['id']} claimed by {reviewer['email']}")

if __name__ == '__main__':
    print("🔗 Webhook server starting...")
    print("   Configure webhook URL: http://your-domain.com/webhook/hitl")
    app.run(debug=True, port=5000)

​
Step 7: Complete Integration Example

Here’s a complete content moderation system that ties everything together:
  • Python
import requests
import time

class ContentModerationSystem:
    def __init__(self, api_key, loop_id):
        self.api_key = api_key
        self.loop_id = loop_id
        self.base_url = "https://api.hitl.sh/v1"
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }
    
    def moderate_content(self, user_id, content, content_type="comment"):
        """Submit content for human moderation"""
        
        print(f"🔍 Submitting {content_type} from user {user_id} for moderation...")
        
        request_data = {
            "loop_id": self.loop_id,
            "request_text": f"Review this {content_type} from user {user_id}:\n\n\"{content}\"\n\nDoes this content follow community guidelines?",
            "response_type": "single_select",
            "response_config": {
                "options": [
                    "✅ Approve - Follows guidelines",
                    "⚠️ Approve with Warning", 
                    "❌ Reject - Violates guidelines",
                    "🚨 Reject and Flag User"
                ]
            },
            "priority": "medium",
            "processing_type": "time-sensitive",
            "metadata": {
                "user_id": user_id,
                "content_type": content_type,
                "original_content": content[:100]  # First 100 chars for reference
            }
        }
        
        response = requests.post(f"{self.base_url}/api/loops/{self.loop_id}/requests", headers=self.headers, json=request_data)
        
        if response.status_code == 201:
            request = response.json()['data']['request']
            print(f"✅ Moderation request created: {request['id']}")
            return request['id']
        else:
            print(f"❌ Error creating request: {response.text}")
            return None
    
    def get_moderation_stats(self):
        """Get overall moderation statistics"""
        response = requests.get(f"{self.base_url}/api/requests?limit=100", headers=self.headers)
        
        if response.status_code == 200:
            requests_data = response.json()['data']['requests']
            
            total = len(requests_data)
            completed = len([r for r in requests_data if r['status'] == 'completed'])
            pending = len([r for r in requests_data if r['status'] == 'pending'])
            
            if completed > 0:
                approved = len([r for r in requests_data if r['status'] == 'completed' and 'Approve' in r.get('response_data', '')])
                approval_rate = (approved / completed) * 100
            else:
                approval_rate = 0
            
            return {
                'total_requests': total,
                'completed': completed,
                'pending': pending,
                'approval_rate': approval_rate
            }
        return None

# Example usage
API_KEY = "your_api_key_here"
LOOP_ID = "your_loop_id_here"

moderation_system = ContentModerationSystem(API_KEY, LOOP_ID)

# Test with sample content
test_content = [
    ("user123", "This is a great product! Really helped me out.", "comment"),
    ("user456", "Buy my crypto course! Link in bio! Make money fast!", "comment"),
    ("user789", "I disagree with your opinion but respect your perspective.", "comment")
]

request_ids = []
for user_id, content, content_type in test_content:
    request_id = moderation_system.moderate_content(user_id, content, content_type)
    if request_id:
        request_ids.append(request_id)
    time.sleep(1)  # Avoid rate limiting

print(f"\n📊 Created {len(request_ids)} moderation requests")
print("🔍 Monitor these requests in your HITL dashboard or via webhooks")

# Get stats
stats = moderation_system.get_moderation_stats()
if stats:
    print(f"\n📈 Moderation Statistics:")
    print(f"   Total requests: {stats['total_requests']}")
    print(f"   Completed: {stats['completed']}")
    print(f"   Pending: {stats['pending']}")
    print(f"   Approval rate: {stats['approval_rate']:.1f}%")

​
What’s Next?

Congratulations! 🎉 You’ve built a complete human-in-the-loop content moderation system. Here’s what you can explore next:

Response Types

Learn about different response types: text, multiple choice, ratings, and more.

Webhooks Integration

Set up real-time webhooks for instant notifications when requests are completed.

Mobile App Guide

Learn how reviewers use the mobile app to respond to your requests.

Best Practices

Optimize your loops for better performance and reviewer experience.

SDKs

Use our official SDKs for easier integration in Python, Node.js, Go, and PHP.

API Playground

Test API endpoints interactively in your browser.

​
Real-World Use Cases

Now that you understand the basics, consider these practical applications:
Challenge: Moderate thousands of user posts per day Solution: Create multiple specialized loops (text content, images, reported content) with different priorities and response types
# Different loops for different content types
loops = {
    'text_moderation': create_loop("Text Content Review", "Review text posts and comments"),
    'image_moderation': create_loop("Image Content Review", "Review uploaded images"),
    'reported_content': create_loop("Reported Content", "Priority review of user-reported content")
}
Challenge: Verify accuracy of user-submitted business information Solution: Human reviewers check addresses, phone numbers, and business details
business_verification = {
    "request_text": f"Verify this business information:\nName: {business_name}\nAddress: {address}\nPhone: {phone}",
    "response_type": "multi_select",
    "response_config": {
        "options": ["Address verified", "Phone verified", "Hours verified", "Website verified"]
    }
}
Challenge: Review AI-generated content before publication Solution: Human editors review AI-written articles, code, or creative content
ai_review = {
    "request_text": f"Review this AI-generated article:\n\n{ai_content}",
    "response_type": "rating",
    "response_config": {
        "scale_max": 5
        # scale_min defaults to 1
    }
}

​
Support

Need help? We’re here for you:

Documentation

Browse our complete API documentation and guides

Community

Join our Discord community for developer discussions and support

Email Support

Contact [email protected] for technical questions and account issues

Status Page

Check api-status.hitl.sh for current system status and maintenance updates
You’re all set! 🚀 You now have a working human-in-the-loop system that can scale to handle real production workloads.