Back to Details

modelslab-webhooks

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

ModelsLab Webhooks

ModelsLab Webhooks

Receive real-time notifications when your API requests complete processing using webhooks instead of polling.
使用Webhooks替代轮询，在API请求处理完成时接收实时通知。

When to Use This Skill

何时使用该功能

  • Handle long-running operations (video, 3D, training)
  • Build scalable applications
  • Avoid continuous polling
  • Reduce API calls
  • Get instant completion notifications
  • Process requests in the background
  • 处理长时间运行的操作（视频、3D、训练）
  • 构建可扩展的应用程序
  • 避免持续轮询
  • 减少API调用
  • 获取即时完成通知
  • 在后台处理请求

Webhook-Supported Endpoints

支持Webhook的接口

Webhooks are supported by these API categories:
Webhooks支持以下API类别：

Image Generation

图像生成

  • text2img
    , 
    img2img
    , 
    inpaint
    , 
    controlnet
  • text2img
    , 
    img2img
    , 
    inpaint
    , 
    controlnet

Video

视频生成

  • text2video
    , 
    img2video
    , 
    text2video_ultra
    , 
    img2video_ultra
  • text2video
    , 
    img2video
    , 
    text2video_ultra
    , 
    img2video_ultra

Audio

音频生成

  • text_to_speech
    , 
    voice_to_voice
    , 
    music_gen
    , 
    song_generator
  • text_to_speech
    , 
    voice_to_voice
    , 
    music_gen
    , 
    song_generator

3D

3D生成

  • text_to_3d
    , 
    image_to_3d
  • text_to_3d
    , 
    image_to_3d

Image Editing

图像编辑

  • All editing endpoints
  • 所有编辑接口

Training

模型训练

  • fine_tune
    , 
    lora_fine_tune
  • fine_tune
    , 
    lora_fine_tune

Workflows

工作流

  • run
    (workflow execution)
  • run
    （工作流执行）

How Webhooks Work

Webhook的工作原理

  1. Submit Request: Include 
    webhook
    URL in API request
  2. Get Request ID: API returns immediately with request ID
  3. Processing: ModelsLab processes your request
  4. Webhook Call: When complete, ModelsLab sends POST request to your webhook URL
  5. Handle Result: Your server receives and processes the result
  1. 提交请求：在API请求中包含
    webhook
    URL
  2. 获取请求ID：API立即返回请求ID
  3. 处理中：ModelsLab处理你的请求
  4. 调用Webhook：处理完成后，ModelsLab会向你的webhook URL发送POST请求
  5. 处理结果：你的服务器接收并处理结果

Basic Webhook Setup

基础Webhook设置

1. Create Webhook Endpoint

1. 创建Webhook接口

python
from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/webhook/modelslab', methods=['POST'])
def handle_modelslab_webhook():
    """Handle webhook callbacks from ModelsLab."""
    try:
        # Get the webhook data
        data = request.json

        # Extract important fields
        status = data.get('status')
        output = data.get('output', [])
        track_id = data.get('track_id')  # Your custom identifier
        request_id = data.get('id')

        if status == 'success':
            # Process successful result
            result_url = output[0] if output else None
            print(f"Generation complete! Track ID: {track_id}")
            print(f"Result: {result_url}")

            # Store in database, notify user, etc.
            save_result(track_id, result_url)

        elif status == 'failed':
            # Handle failure
            error = data.get('message', 'Unknown error')
            print(f"Generation failed: {error}")
            handle_failure(track_id, error)

        return jsonify({"received": True}), 200

    except Exception as e:
        print(f"Webhook error: {e}")
        return jsonify({"error": str(e)}), 500

if __name__ == '__main__':
    app.run(port=8080)
python
from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/webhook/modelslab', methods=['POST'])
def handle_modelslab_webhook():
    """Handle webhook callbacks from ModelsLab."""
    try:
        # Get the webhook data
        data = request.json

        # Extract important fields
        status = data.get('status')
        output = data.get('output', [])
        track_id = data.get('track_id')  # Your custom identifier
        request_id = data.get('id')

        if status == 'success':
            # Process successful result
            result_url = output[0] if output else None
            print(f"Generation complete! Track ID: {track_id}")
            print(f"Result: {result_url}")

            # Store in database, notify user, etc.
            save_result(track_id, result_url)

        elif status == 'failed':
            # Handle failure
            error = data.get('message', 'Unknown error')
            print(f"Generation failed: {error}")
            handle_failure(track_id, error)

        return jsonify({"received": True}), 200

    except Exception as e:
        print(f"Webhook error: {e}")
        return jsonify({"error": str(e)}), 500

if __name__ == '__main__':
    app.run(port=8080)

2. Make Your Endpoint Publicly Accessible

2. 让接口可公开访问

bash
undefined
bash
undefined

Use ngrok for local testing

Use ngrok for local testing

ngrok http 8080
ngrok http 8080

Your webhook URL will be:

Your webhook URL will be:

https://abc123.ngrok.io/webhook/modelslab

https://abc123.ngrok.io/webhook/modelslab

undefined
undefined

Using Webhooks with APIs

在API中使用Webhook

Image Generation with Webhook

结合Webhook的图像生成

python
import requests

def generate_image_with_webhook(prompt, api_key, webhook_url, track_id):
    """Generate image and receive result via webhook."""
    response = requests.post(
        "https://modelslab.com/api/v6/images/text2img",
        json={
            "key": api_key,
            "model_id": "midjourney",
            "prompt": prompt,
            "negative_prompt": "low quality",
            "width": 768,
            "height": 768,
            "webhook": webhook_url,  # Your webhook endpoint
            "track_id": track_id     # Your custom identifier
        }
    )

    data = response.json()

    if data["status"] == "error":
        raise Exception(f"Error: {data['message']}")

    # Return request ID for tracking
    return {
        "request_id": data["id"],
        "track_id": track_id,
        "status": data["status"]
    }
python
import requests

def generate_image_with_webhook(prompt, api_key, webhook_url, track_id):
    """Generate image and receive result via webhook."""
    response = requests.post(
        "https://modelslab.com/api/v6/images/text2img",
        json={
            "key": api_key,
            "model_id": "midjourney",
            "prompt": prompt,
            "negative_prompt": "low quality",
            "width": 768,
            "height": 768,
            "webhook": webhook_url,  # Your webhook endpoint
            "track_id": track_id     # Your custom identifier
        }
    )

    data = response.json()

    if data["status"] == "error":
        raise Exception(f"Error: {data['message']}")

    # Return request ID for tracking
    return {
        "request_id": data["id"],
        "track_id": track_id,
        "status": data["status"]
    }

Usage

Usage

result = generate_image_with_webhook( "A futuristic cityscape at sunset", "your_api_key", "https://yourserver.com/webhook/modelslab", "image_001" ) print(f"Request submitted: {result['request_id']}")
result = generate_image_with_webhook( "A futuristic cityscape at sunset", "your_api_key", "https://yourserver.com/webhook/modelslab", "image_001" ) print(f"Request submitted: {result['request_id']}")

Result will be sent to your webhook when ready

Result will be sent to your webhook when ready

undefined
undefined

Video Generation with Webhook

结合Webhook的视频生成

python
def generate_video_with_webhook(prompt, api_key, webhook_url, track_id):
    """Generate video and receive result via webhook."""
    response = requests.post(
        "https://modelslab.com/api/v6/video/text2video",
        json={
            "key": api_key,
            "model_id": "cogvideox",
            "prompt": prompt,
            "num_frames": 25,
            "webhook": webhook_url,
            "track_id": track_id
        }
    )

    data = response.json()
    return {
        "request_id": data["id"],
        "track_id": track_id
    }
python
def generate_video_with_webhook(prompt, api_key, webhook_url, track_id):
    """Generate video and receive result via webhook."""
    response = requests.post(
        "https://modelslab.com/api/v6/video/text2video",
        json={
            "key": api_key,
            "model_id": "cogvideox",
            "prompt": prompt,
            "num_frames": 25,
            "webhook": webhook_url,
            "track_id": track_id
        }
    )

    data = response.json()
    return {
        "request_id": data["id"],
        "track_id": track_id
    }

Submit video generation

Submit video generation

video_request = generate_video_with_webhook( "A spaceship flying through an asteroid field", "your_api_key", "https://yourserver.com/webhook/video", "video_123" )
undefined
video_request = generate_video_with_webhook( "A spaceship flying through an asteroid field", "your_api_key", "https://yourserver.com/webhook/video", "video_123" )
undefined

Audio Generation with Webhook

结合Webhook的音频生成

python
def generate_audio_with_webhook(text, api_key, webhook_url, track_id):
    """Generate audio and receive result via webhook."""
    response = requests.post(
        "https://modelslab.com/api/v6/audio/text_to_speech",
        json={
            "key": api_key,
            "text": text,
            "voice_id": "alloy",
            "language": "en",
            "webhook": webhook_url,
            "track_id": track_id
        }
    )

    data = response.json()
    return data["id"]
python
def generate_audio_with_webhook(text, api_key, webhook_url, track_id):
    """Generate audio and receive result via webhook."""
    response = requests.post(
        "https://modelslab.com/api/v6/audio/text_to_speech",
        json={
            "key": api_key,
            "text": text,
            "voice_id": "alloy",
            "language": "en",
            "webhook": webhook_url,
            "track_id": track_id
        }
    )

    data = response.json()
    return data["id"]

Track ID System

Track ID系统

The 
track_id
parameter lets you identify which webhook callback corresponds to which request:
python
import uuid

def generate_unique_track_id(user_id, request_type):
    """Generate a unique tracking ID."""
    unique_id = str(uuid.uuid4())[:8]
    return f"{user_id}_{request_type}_{unique_id}"
track_id
参数可让你识别哪个Webhook回调对应哪个请求：
python
import uuid

def generate_unique_track_id(user_id, request_type):
    """Generate a unique tracking ID."""
    unique_id = str(uuid.uuid4())[:8]
    return f"{user_id}_{request_type}_{unique_id}"

Usage

Usage

track_id = generate_unique_track_id("user123", "image")
track_id = generate_unique_track_id("user123", "image")

Returns: "user123_image_a1b2c3d4"

Returns: "user123_image_a1b2c3d4"

Use in API request

Use in API request

generate_image_with_webhook( prompt, api_key, webhook_url, track_id=track_id )
undefined
generate_image_with_webhook( prompt, api_key, webhook_url, track_id=track_id )
undefined

Advanced Webhook Handler

高级Webhook处理器

python
from flask import Flask, request, jsonify
import logging
from datetime import datetime

app = Flask(__name__)
logging.basicConfig(level=logging.INFO)

class WebhookHandler:
    """Handle ModelsLab webhook callbacks."""

    def __init__(self):
        self.handlers = {
            'image': self.handle_image_result,
            'video': self.handle_video_result,
            'audio': self.handle_audio_result,
            '3d': self.handle_3d_result
        }

    def process_webhook(self, data):
        """Process incoming webhook data."""
        track_id = data.get('track_id', '')
        status = data.get('status')
        request_id = data.get('id')

        logging.info(f"Webhook received: {track_id} - {status}")

        # Extract request type from track_id
        request_type = self.get_request_type(track_id)

        if status == 'success':
            output = data.get('output', [])
            result_url = output[0] if output else None

            # Route to appropriate handler
            if request_type in self.handlers:
                self.handlers[request_type](track_id, result_url, data)
            else:
                self.handle_generic_result(track_id, result_url, data)

        elif status == 'failed':
            error = data.get('message', 'Unknown error')
            self.handle_failure(track_id, error)

        return True

    def get_request_type(self, track_id):
        """Extract request type from track_id."""
        # Assuming track_id format: "user_type_id"
        parts = track_id.split('_')
        return parts[1] if len(parts) > 1 else 'unknown'

    def handle_image_result(self, track_id, result_url, data):
        """Handle image generation result."""
        logging.info(f"Image ready: {track_id}")
        # Store in database
        # Notify user
        # Post-process image
        pass

    def handle_video_result(self, track_id, result_url, data):
        """Handle video generation result."""
        logging.info(f"Video ready: {track_id}")
        # Store video
        # Trigger post-processing
        pass

    def handle_audio_result(self, track_id, result_url, data):
        """Handle audio generation result."""
        logging.info(f"Audio ready: {track_id}")
        pass

    def handle_3d_result(self, track_id, result_url, data):
        """Handle 3D generation result."""
        logging.info(f"3D model ready: {track_id}")
        pass

    def handle_generic_result(self, track_id, result_url, data):
        """Handle generic result."""
        logging.info(f"Result ready: {track_id} - {result_url}")

    def handle_failure(self, track_id, error):
        """Handle failed generation."""
        logging.error(f"Generation failed: {track_id} - {error}")
        # Notify user
        # Log error
        # Retry if appropriate

webhook_handler = WebhookHandler()

@app.route('/webhook/modelslab', methods=['POST'])
def webhook_endpoint():
    """Webhook endpoint for all ModelsLab callbacks."""
    try:
        data = request.json
        webhook_handler.process_webhook(data)
        return jsonify({"received": True}), 200
    except Exception as e:
        logging.error(f"Webhook error: {e}")
        return jsonify({"error": str(e)}), 500

if __name__ == '__main__':
    app.run(port=8080)
python
from flask import Flask, request, jsonify
import logging
from datetime import datetime

app = Flask(__name__)
logging.basicConfig(level=logging.INFO)

class WebhookHandler:
    """Handle ModelsLab webhook callbacks."""

    def __init__(self):
        self.handlers = {
            'image': self.handle_image_result,
            'video': self.handle_video_result,
            'audio': self.handle_audio_result,
            '3d': self.handle_3d_result
        }

    def process_webhook(self, data):
        """Process incoming webhook data."""
        track_id = data.get('track_id', '')
        status = data.get('status')
        request_id = data.get('id')

        logging.info(f"Webhook received: {track_id} - {status}")

        # Extract request type from track_id
        request_type = self.get_request_type(track_id)

        if status == 'success':
            output = data.get('output', [])
            result_url = output[0] if output else None

            # Route to appropriate handler
            if request_type in self.handlers:
                self.handlers[request_type](track_id, result_url, data)
            else:
                self.handle_generic_result(track_id, result_url, data)

        elif status == 'failed':
            error = data.get('message', 'Unknown error')
            self.handle_failure(track_id, error)

        return True

    def get_request_type(self, track_id):
        """Extract request type from track_id."""
        # Assuming track_id format: "user_type_id"
        parts = track_id.split('_')
        return parts[1] if len(parts) > 1 else 'unknown'

    def handle_image_result(self, track_id, result_url, data):
        """Handle image generation result."""
        logging.info(f"Image ready: {track_id}")
        # Store in database
        # Notify user
        # Post-process image
        pass

    def handle_video_result(self, track_id, result_url, data):
        """Handle video generation result."""
        logging.info(f"Video ready: {track_id}")
        # Store video
        # Trigger post-processing
        pass

    def handle_audio_result(self, track_id, result_url, data):
        """Handle audio generation result."""
        logging.info(f"Audio ready: {track_id}")
        pass

    def handle_3d_result(self, track_id, result_url, data):
        """Handle 3D generation result."""
        logging.info(f"3D model ready: {track_id}")
        pass

    def handle_generic_result(self, track_id, result_url, data):
        """Handle generic result."""
        logging.info(f"Result ready: {track_id} - {result_url}")

    def handle_failure(self, track_id, error):
        """Handle failed generation."""
        logging.error(f"Generation failed: {track_id} - {error}")
        # Notify user
        # Log error
        # Retry if appropriate

webhook_handler = WebhookHandler()

@app.route('/webhook/modelslab', methods=['POST'])
def webhook_endpoint():
    """Webhook endpoint for all ModelsLab callbacks."""
    try:
        data = request.json
        webhook_handler.process_webhook(data)
        return jsonify({"received": True}), 200
    except Exception as e:
        logging.error(f"Webhook error: {e}")
        return jsonify({"error": str(e)}), 500

if __name__ == '__main__':
    app.run(port=8080)

Best Practices

最佳实践

1. Always Return 200 OK

1. 始终返回200 OK

python
@app.route('/webhook', methods=['POST'])
def webhook():
    try:
        data = request.json
        # Process data
        process_webhook(data)
    except Exception as e:
        logging.error(f"Error: {e}")
        # Still return 200 to acknowledge receipt
    return jsonify({"received": True}), 200
python
@app.route('/webhook', methods=['POST'])
def webhook():
    try:
        data = request.json
        # Process data
        process_webhook(data)
    except Exception as e:
        logging.error(f"Error: {e}")
        # Still return 200 to acknowledge receipt
    return jsonify({"received": True}), 200

2. Process Asynchronously

2. 异步处理

python
from threading import Thread

@app.route('/webhook', methods=['POST'])
def webhook():
    data = request.json
    # Process in background thread
    Thread(target=process_webhook, args=(data,)).start()
    return jsonify({"received": True}), 200
python
from threading import Thread

@app.route('/webhook', methods=['POST'])
def webhook():
    data = request.json
    # Process in background thread
    Thread(target=process_webhook, args=(data,)).start()
    return jsonify({"received": True}), 200

3. Validate Webhook Data

3. 验证Webhook数据

python
def validate_webhook(data):
    """Validate webhook payload."""
    required_fields = ['status', 'id', 'track_id']
    return all(field in data for field in required_fields)

@app.route('/webhook', methods=['POST'])
def webhook():
    data = request.json
    if not validate_webhook(data):
        return jsonify({"error": "Invalid payload"}), 400
    # Process...
python
def validate_webhook(data):
    """Validate webhook payload."""
    required_fields = ['status', 'id', 'track_id']
    return all(field in data for field in required_fields)

@app.route('/webhook', methods=['POST'])
def webhook():
    data = request.json
    if not validate_webhook(data):
        return jsonify({"error": "Invalid payload"}), 400
    # Process...

4. Implement Retry Logic

4. 实现重试逻辑

python
undefined
python
undefined

ModelsLab will retry failed webhooks automatically

ModelsLab will retry failed webhooks automatically

But handle idempotency in your endpoint

But handle idempotency in your endpoint

seen_requests = set()
def process_webhook(data): request_id = data.get('id')
# Check if already processed
if request_id in seen_requests:
    logging.info(f"Duplicate webhook: {request_id}")
    return

# Mark as seen
seen_requests.add(request_id)

# Process...
undefined
seen_requests = set()
def process_webhook(data): request_id = data.get('id')
# Check if already processed
if request_id in seen_requests:
    logging.info(f"Duplicate webhook: {request_id}")
    return

# Mark as seen
seen_requests.add(request_id)

# Process...
undefined

5. Use HTTPS

5. 使用HTTPS

python
undefined
python
undefined

Always use HTTPS for webhook URLs

Always use HTTPS for webhook URLs

webhook_url = "https://yourserver.com/webhook" # ✓ webhook_url = "http://yourserver.com/webhook" # ✗
undefined
webhook_url = "https://yourserver.com/webhook" # ✓ webhook_url = "http://yourserver.com/webhook" # ✗
undefined

Webhook Payload Structure

Webhook负载结构

json
{
  "status": "success",
  "id": "request_id_here",
  "track_id": "your_custom_track_id",
  "output": [
    "https://modelslab.com/path/to/result.jpg"
  ],
  "meta": {
    "prompt": "The original prompt",
    "model_id": "midjourney",
    "seed": 12345
  }
}
json
{
  "status": "success",
  "id": "request_id_here",
  "track_id": "your_custom_track_id",
  "output": [
    "https://modelslab.com/path/to/result.jpg"
  ],
  "meta": {
    "prompt": "The original prompt",
    "model_id": "midjourney",
    "seed": 12345
  }
}

Common Use Cases

常见用例

Batch Processing

批量处理

python
def batch_generate_images(prompts, api_key, webhook_url):
    """Generate multiple images in batch."""
    request_ids = []

    for i, prompt in enumerate(prompts):
        result = generate_image_with_webhook(
            prompt,
            api_key,
            webhook_url,
            track_id=f"batch_{i}"
        )
        request_ids.append(result['request_id'])
        print(f"Submitted {i+1}/{len(prompts)}")

    return request_ids
python
def batch_generate_images(prompts, api_key, webhook_url):
    """Generate multiple images in batch."""
    request_ids = []

    for i, prompt in enumerate(prompts):
        result = generate_image_with_webhook(
            prompt,
            api_key,
            webhook_url,
            track_id=f"batch_{i}"
        )
        request_ids.append(result['request_id'])
        print(f"Submitted {i+1}/{len(prompts)}")

    return request_ids

Generate 100 images

Generate 100 images

batch_generate_images( ["prompt 1", "prompt 2", ...], # 100 prompts api_key, "https://yourserver.com/webhook/batch" )
undefined
batch_generate_images( ["prompt 1", "prompt 2", ...], # 100 prompts api_key, "https://yourserver.com/webhook/batch" )
undefined

User Notification System

用户通知系统

python
from flask import Flask, request, jsonify
import smtplib

@app.route('/webhook/notify', methods=['POST'])
def webhook_with_notification():
    """Webhook that notifies users when complete."""
    data = request.json
    track_id = data.get('track_id')

    # Extract user email from track_id or database
    user_email = get_user_email(track_id)

    if data['status'] == 'success':
        result_url = data['output'][0]
        send_email(
            user_email,
            "Your content is ready!",
            f"Download here: {result_url}"
        )

    return jsonify({"received": True}), 200
python
from flask import Flask, request, jsonify
import smtplib

@app.route('/webhook/notify', methods=['POST'])
def webhook_with_notification():
    """Webhook that notifies users when complete."""
    data = request.json
    track_id = data.get('track_id')

    # Extract user email from track_id or database
    user_email = get_user_email(track_id)

    if data['status'] == 'success':
        result_url = data['output'][0]
        send_email(
            user_email,
            "Your content is ready!",
            f"Download here: {result_url}"
        )

    return jsonify({"received": True}), 200

Testing Webhooks Locally

本地测试Webhook

bash
undefined
bash
undefined

1. Install ngrok

1. Install ngrok

brew install ngrok
brew install ngrok

2. Start your webhook server

2. Start your webhook server

python webhook_server.py
python webhook_server.py

3. Expose with ngrok

3. Expose with ngrok

ngrok http 8080
ngrok http 8080

4. Use the ngrok URL in your API requests

4. Use the ngrok URL in your API requests

https://abc123.ngrok.io/webhook/modelslab

https://abc123.ngrok.io/webhook/modelslab

undefined
undefined

Error Handling

错误处理

python
@app.route('/webhook', methods=['POST'])
def webhook():
    try:
        data = request.json

        if not data:
            logging.error("Empty webhook payload")
            return jsonify({"error": "Empty payload"}), 400

        if data['status'] == 'failed':
            handle_failed_generation(data)
        else:
            handle_successful_generation(data)

        return jsonify({"received": True}), 200

    except Exception as e:
        logging.error(f"Webhook processing error: {e}")
        # Log but still return 200
        return jsonify({"received": True, "error": str(e)}), 200
python
@app.route('/webhook', methods=['POST'])
def webhook():
    try:
        data = request.json

        if not data:
            logging.error("Empty webhook payload")
            return jsonify({"error": "Empty payload"}), 400

        if data['status'] == 'failed':
            handle_failed_generation(data)
        else:
            handle_successful_generation(data)

        return jsonify({"received": True}), 200

    except Exception as e:
        logging.error(f"Webhook processing error: {e}")
        # Log but still return 200
        return jsonify({"received": True, "error": str(e)}), 200

Resources

资源

Related Skills

相关功能

  • All ModelsLab generation skills support webhooks
  • modelslab-sdk-usage
    - SDKs have built-in webhook support
  • 所有ModelsLab生成类功能均支持Webhooks
  • modelslab-sdk-usage
    - SDK内置Webhook支持