modelslab-webhooks

ModelsLab Webhooks

使用 Webhook 代替轮询，在 API 请求处理完成时接收实时通知。

何时使用此技能

• 处理长时间运行的操作（视频、3D、训练）
• 构建可扩展的应用程序
• 避免持续轮询
• 减少 API 调用
• 即时接收完成通知
• 在后台处理请求

支持 Webhook 的端点

以下 API 类别支持 Webhook：

图像生成

• text2img、img2img、inpaint、controlnet

视频

• text2video、img2video、text2video_ultra、img2video_ultra

音频

• text_to_speech、voice_to_voice、music_gen、song_generator

3D

• text_to_3d、image_to_3d

图像编辑

• 所有编辑端点

训练

• fine_tune、lora_fine_tune

工作流

• run（工作流执行）

Webhook 的工作方式

1. 提交请求：在 API 请求中加入 webhook URL
2. 获取请求 ID：API 立即返回请求 ID
3. 处理中：ModelsLab 处理你的请求
4. 调用 Webhook：处理完成后，ModelsLab 会向你的 Webhook URL 发送 POST 请求
5. 处理结果：你的服务器接收并处理结果

Webhook 基本设置

1. 创建 Webhook 端点

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. 让端点可公开访问

# Use ngrok for local testing
ngrok http 8080

# Your webhook URL will be:
# https://abc123.ngrok.io/webhook/modelslab

在 API 中使用 Webhook

使用 Webhook 生成图像

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
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

使用 Webhook 生成视频

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
video_request = generate_video_with_webhook(
    "A spaceship flying through an asteroid field",
    "your_api_key",
    "https://yourserver.com/webhook/video",
    "video_123"
)

使用 Webhook 生成音频

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"]

跟踪 ID 系统

track_id 参数可帮助你识别每个 Webhook 回调对应的请求：

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
track_id = generate_unique_track_id("user123", "image")
# Returns: "user123_image_a1b2c3d4"

# Use in API request
generate_image_with_webhook(
    prompt,
    api_key,
    webhook_url,
    track_id=track_id
)

高级 Webhook 处理程序

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)

最佳实践

1. 始终返回 200 OK

@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. 异步处理

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. 验证 Webhook 数据

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. 实现重试逻辑

# ModelsLab will retry failed webhooks automatically
# 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...

5. 使用 HTTPS

# Always use HTTPS for webhook URLs
webhook_url = "https://yourserver.com/webhook"  # ✓
webhook_url = "http://yourserver.com/webhook"   # ✗

Webhook 负载结构

{
  "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
  }
}

常见用例

批量处理

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
batch_generate_images(
    ["prompt 1", "prompt 2", ...],  # 100 prompts
    api_key,
    "https://yourserver.com/webhook/batch"
)

用户通知系统

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

在本地测试 Webhook

# 1. Install ngrok
brew install ngrok

# 2. Start your webhook server
python webhook_server.py

# 3. Expose with ngrok
ngrok http 8080

# 4. Use the ngrok URL in your API requests
# https://abc123.ngrok.io/webhook/modelslab

错误处理

@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

资源

• Webhook 文档：https://docs.modelslab.com/webhooks
• 支持的端点：查看各 API 文档
• 获取 API 密钥：https://modelslab.com/dashboard

相关技能

• 所有 ModelsLab 生成技能均支持 Webhook
• modelslab-sdk-usage - SDK 内置了 Webhook 支持