Webhook-Based Order Management
Last updated
Was this helpful?
Was this helpful?
pip install flask requests redis celery python-dotenv
## Step 1: Set Up the Clore Client
> 📦 **Using the standard Clore API client.** See [Clore API Client Reference](../reference/clore-client.md) for the full implementation and setup instructions. Save it as `clore_client.py` in your project.
```python
from clore_client import CloreClient
client = CloreClient(api_key="your-api-key")# Webhook server
WEBHOOK_SECRET = os.getenv("WEBHOOK_SECRET", "your-secret-key")
PORT = int(os.getenv("PORT", 5000))
# Redis (for task queue)
REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379/0")
# Notifications
SLACK_WEBHOOK_URL = os.getenv("SLACK_WEBHOOK_URL")
DISCORD_WEBHOOK_URL = os.getenv("DISCORD_WEBHOOK_URL")
# Order defaults
DEFAULT_IMAGE = "nvidia/cuda:12.8.0-base-ubuntu22.04"
DEFAULT_CURRENCY = "CLORE-Blockchain"
DEFAULT_PORTS = {"22": "tcp", "8888": "http"}
# Rate limiting
MAX_ORDERS_PER_HOUR = 10
MAX_REQUESTS_PER_MINUTE = 60
## Step 2: Clore.ai Client
```python
# clore_client.py
import requests
import time
import secrets
from typing import Dict, List, Optional, Any
from dataclasses import dataclass
from config import Config
@dataclass
class Server:
id: int
gpus: List[str]
spot_price: Optional[float]
ondemand_price: Optional[float]
is_available: bool
reliability: float
@dataclass
class Order:
order_id: int
server_id: int
status: str
ssh_connection: str
created_at: int
## Step 3: Task Queue (Celery Workers)
```python
# tasks.py
from celery import Celery
from clore_client import CloreClient
from notifications import notify_slack, notify_discord
from config import Config
import logging
logger = logging.getLogger(__name__)
# Initialize Celery
celery = Celery('tasks', broker=Config.REDIS_URL)
celery.conf.update(
task_serializer='json',
accept_content=['json'],
result_serializer='json',
timezone='UTC',
enable_utc=True,
task_soft_time_limit=300,
task_time_limit=600,
)
@celery.task(bind=True, max_retries=3)
def create_order_task(
self,
gpu_type: str,
max_price: float = None,
image: str = None,
use_spot: bool = True,
callback_url: str = None
):
"""
Create an order for specified GPU type.
Args:
gpu_type: GPU model to rent (e.g., "RTX 4090")
max_price: Maximum hourly price in USD
image: Docker image to use
use_spot: Use spot pricing
callback_url: URL to POST results to
"""
try:
client = CloreClient()
# Find matching servers
servers = client.find_servers(gpu_type=gpu_type, max_price=max_price)
if not servers:
result = {
"success": False,
"error": f"No {gpu_type} available under ${max_price}/hr"
}
_send_callback(callback_url, result)
return result
server = servers[0]
spot_price = server.spot_price * 1.1 if use_spot else None
# Create order
order = client.create_order(
server_id=server.id,
image=image,
use_spot=use_spot,
spot_price=spot_price
)
# Wait for ready
try:
ready_order = client.wait_for_ready(order.order_id, timeout=120)
result = {
"success": True,
"order_id": ready_order.order_id,
"server_id": ready_order.server_id,
"status": ready_order.status,
"ssh_connection": ready_order.ssh_connection,
"gpu": server.gpus,
"price_usd": server.spot_price
}
# Notify
notify_slack(f"✅ GPU Order Created\n"
f"Order: {order.order_id}\n"
f"GPU: {server.gpus}\n"
f"SSH: {ready_order.ssh_connection}")
except TimeoutError:
client.cancel_order(order.order_id)
result = {
"success": False,
"error": "Order creation timed out"
}
_send_callback(callback_url, result)
return result
except Exception as e:
logger.error(f"create_order_task failed: {e}")
if self.request.retries < self.max_retries:
raise self.retry(countdown=2 ** self.request.retries)
result = {"success": False, "error": str(e)}
_send_callback(callback_url, result)
return result
@celery.task(bind=True, max_retries=3)
def cancel_order_task(self, order_id: int, callback_url: str = None):
"""Cancel an order."""
try:
client = CloreClient()
client.cancel_order(order_id)
result = {"success": True, "order_id": order_id, "status": "cancelled"}
notify_slack(f"🛑 Order {order_id} cancelled")
_send_callback(callback_url, result)
return result
except Exception as e:
logger.error(f"cancel_order_task failed: {e}")
if self.request.retries < self.max_retries:
raise self.retry(countdown=2 ** self.request.retries)
result = {"success": False, "error": str(e)}
_send_callback(callback_url, result)
return result
@celery.task
def check_price_and_order(
gpu_type: str,
target_price: float,
image: str = None,
callback_url: str = None
):
"""Check if GPU price dropped below target and auto-order."""
client = CloreClient()
servers = client.find_servers(gpu_type=gpu_type, max_price=target_price)
if servers:
server = servers[0]
notify_slack(f"🎯 Price Alert! {gpu_type} at ${server.spot_price}/hr "
f"(target: ${target_price})")
# Auto-order
return create_order_task.delay(
gpu_type=gpu_type,
max_price=target_price,
image=image,
callback_url=callback_url
)
return {"success": False, "message": "Price not met"}
@celery.task
def monitor_orders(callback_url: str = None):
"""Monitor all active orders and report status."""
client = CloreClient()
orders = client.get_orders()
result = {
"total_orders": len(orders),
"active": len([o for o in orders if o.status == "running"]),
"creating": len([o for o in orders if o.status == "creating_order"]),
"orders": [
{
"order_id": o.order_id,
"status": o.status,
"ssh": o.ssh_connection
}
for o in orders
]
}
_send_callback(callback_url, result)
return result
def _send_callback(url: str, data: dict):
"""Send callback to specified URL."""
if not url:
return
try:
import requests
requests.post(url, json=data, timeout=10)
except Exception as e:
logger.error(f"Callback failed: {e}")# notifications.py
import requests
from config import Config
import logging
logger = logging.getLogger(__name__)
def notify_slack(message: str):
"""Send notification to Slack."""
if not Config.SLACK_WEBHOOK_URL:
return
try:
requests.post(
Config.SLACK_WEBHOOK_URL,
json={"text": message},
timeout=5
)
except Exception as e:
logger.error(f"Slack notification failed: {e}")
def notify_discord(message: str):
"""Send notification to Discord."""
if not Config.DISCORD_WEBHOOK_URL:
return
try:
requests.post(
Config.DISCORD_WEBHOOK_URL,
json={"content": message},
timeout=5
)
except Exception as e:
logger.error(f"Discord notification failed: {e}")
def notify_all(message: str):
"""Send notification to all configured channels."""
notify_slack(message)
notify_discord(message)# app.py
from flask import Flask, request, jsonify
from functools import wraps
import hmac
import hashlib
import time
from config import Config
from tasks import (
create_order_task,
cancel_order_task,
check_price_and_order,
monitor_orders
)
from clore_client import CloreClient
app = Flask(__name__)
# Rate limiting storage
request_counts = {}
def verify_signature(f):
"""Verify webhook signature."""
@wraps(f)
def decorated(*args, **kwargs):
signature = request.headers.get('X-Webhook-Signature')
if not signature:
return jsonify({"error": "Missing signature"}), 401
# Calculate expected signature
payload = request.get_data()
expected = hmac.new(
Config.WEBHOOK_SECRET.encode(),
payload,
hashlib.sha256
).hexdigest()
if not hmac.compare_digest(signature, expected):
return jsonify({"error": "Invalid signature"}), 401
return f(*args, **kwargs)
return decorated
def rate_limit(f):
"""Simple rate limiting."""
@wraps(f)
def decorated(*args, **kwargs):
ip = request.remote_addr
now = time.time()
# Clean old entries
request_counts[ip] = [t for t in request_counts.get(ip, []) if now - t < 60]
if len(request_counts.get(ip, [])) >= Config.MAX_REQUESTS_PER_MINUTE:
return jsonify({"error": "Rate limit exceeded"}), 429
request_counts.setdefault(ip, []).append(now)
return f(*args, **kwargs)
return decorated
# === Webhook Endpoints ===
@app.route('/webhook/create-order', methods=['POST'])
@rate_limit
@verify_signature
def webhook_create_order():
"""
Create a new GPU rental order.
Request body:
{
"gpu_type": "RTX 4090",
"max_price": 0.50,
"image": "nvidia/cuda:12.8.0-base-ubuntu22.04",
"use_spot": true,
"callback_url": "https://your-server.com/callback"
}
"""
data = request.json
gpu_type = data.get('gpu_type')
if not gpu_type:
return jsonify({"error": "gpu_type required"}), 400
# Queue the task
task = create_order_task.delay(
gpu_type=gpu_type,
max_price=data.get('max_price'),
image=data.get('image'),
use_spot=data.get('use_spot', True),
callback_url=data.get('callback_url')
)
return jsonify({
"status": "queued",
"task_id": task.id,
"message": f"Order creation queued for {gpu_type}"
})
@app.route('/webhook/cancel-order', methods=['POST'])
@rate_limit
@verify_signature
def webhook_cancel_order():
"""
Cancel an existing order.
Request body:
{
"order_id": 12345,
"callback_url": "https://your-server.com/callback"
}
"""
data = request.json
order_id = data.get('order_id')
if not order_id:
return jsonify({"error": "order_id required"}), 400
task = cancel_order_task.delay(
order_id=order_id,
callback_url=data.get('callback_url')
)
return jsonify({
"status": "queued",
"task_id": task.id,
"message": f"Cancellation queued for order {order_id}"
})
@app.route('/webhook/price-alert', methods=['POST'])
@rate_limit
@verify_signature
def webhook_price_alert():
"""
Set up price-based auto-ordering.
Request body:
{
"gpu_type": "RTX 4090",
"target_price": 0.35,
"image": "pytorch/pytorch:2.7.1-cuda12.8-cudnn9-runtime",
"callback_url": "https://your-server.com/callback"
}
"""
data = request.json
gpu_type = data.get('gpu_type')
target_price = data.get('target_price')
if not gpu_type or not target_price:
return jsonify({"error": "gpu_type and target_price required"}), 400
task = check_price_and_order.delay(
gpu_type=gpu_type,
target_price=target_price,
image=data.get('image'),
callback_url=data.get('callback_url')
)
return jsonify({
"status": "queued",
"task_id": task.id,
"message": f"Price alert set for {gpu_type} at ${target_price}/hr"
})
@app.route('/webhook/status', methods=['POST'])
@rate_limit
@verify_signature
def webhook_status():
"""
Get status of all orders.
Request body:
{
"callback_url": "https://your-server.com/callback"
}
"""
data = request.json or {}
task = monitor_orders.delay(
callback_url=data.get('callback_url')
)
return jsonify({
"status": "queued",
"task_id": task.id
})
@app.route('/webhook/marketplace', methods=['GET'])
@rate_limit
def webhook_marketplace():
"""Get current marketplace status (no auth required for read-only)."""
try:
client = CloreClient()
servers = client.get_marketplace()
# Summarize by GPU type
summary = {}
for s in servers:
if not s.gpus:
continue
gpu = s.gpus[0].split()[0] if s.gpus else "Unknown"
if gpu not in summary:
summary[gpu] = {"available": 0, "total": 0, "min_price": float('inf')}
summary[gpu]["total"] += 1
if s.is_available:
summary[gpu]["available"] += 1
if s.spot_price:
summary[gpu]["min_price"] = min(summary[gpu]["min_price"], s.spot_price)
# Clean up infinity
for gpu in summary:
if summary[gpu]["min_price"] == float('inf'):
summary[gpu]["min_price"] = None
return jsonify({
"status": "ok",
"summary": summary,
"total_servers": len(servers),
"total_available": len([s for s in servers if s.is_available])
})
except Exception as e:
return jsonify({"error": str(e)}), 500
# === Slack/Discord Slash Commands ===
@app.route('/slack/command', methods=['POST'])
def slack_command():
"""Handle Slack slash commands."""
# Verify Slack request (in production, verify signing secret)
text = request.form.get('text', '').strip()
parts = text.split()
if not parts:
return jsonify({
"response_type": "ephemeral",
"text": "Usage: /clore [rent|cancel|status] [args]"
})
command = parts[0].lower()
if command == 'rent':
# /clore rent RTX 4090 0.50
if len(parts) < 2:
return jsonify({
"response_type": "ephemeral",
"text": "Usage: /clore rent <gpu_type> [max_price]"
})
gpu_type = parts[1]
max_price = float(parts[2]) if len(parts) > 2 else None
create_order_task.delay(gpu_type=gpu_type, max_price=max_price)
return jsonify({
"response_type": "in_channel",
"text": f"🚀 Creating order for {gpu_type}..."
})
elif command == 'cancel':
# /clore cancel 12345
if len(parts) < 2:
return jsonify({
"response_type": "ephemeral",
"text": "Usage: /clore cancel <order_id>"
})
order_id = int(parts[1])
cancel_order_task.delay(order_id=order_id)
return jsonify({
"response_type": "in_channel",
"text": f"🛑 Cancelling order {order_id}..."
})
elif command == 'status':
monitor_orders.delay()
return jsonify({
"response_type": "in_channel",
"text": "📊 Fetching order status..."
})
else:
return jsonify({
"response_type": "ephemeral",
"text": f"Unknown command: {command}"
})
# === Health Check ===
@app.route('/health', methods=['GET'])
def health():
return jsonify({"status": "ok"})
if __name__ == '__main__':
app.run(host='0.0.0.0', port=Config.PORT, debug=False)#!/usr/bin/env python3
"""
Webhook Order Management System for Clore.ai
Usage:
# Start webhook server
python app.py
# Start Celery worker (in another terminal)
celery -A tasks worker --loglevel=info
# Test webhook
curl -X POST http://localhost:5000/webhook/create-order \
-H "Content-Type: application/json" \
-H "X-Webhook-Signature: <signature>" \
-d '{"gpu_type": "RTX 4090", "max_price": 0.50}'
"""
import os
import sys
import hmac
import hashlib
import json
import argparse
import requests
from config import Config
def generate_signature(payload: str, secret: str = None) -> str:
"""Generate webhook signature for testing."""
secret = secret or Config.WEBHOOK_SECRET
return hmac.new(
secret.encode(),
payload.encode(),
hashlib.sha256
).hexdigest()
def test_webhook(endpoint: str, data: dict, base_url: str = "http://localhost:5000"):
"""Test a webhook endpoint."""
payload = json.dumps(data)
signature = generate_signature(payload)
url = f"{base_url}/webhook/{endpoint}"
response = requests.post(
url,
data=payload,
headers={
"Content-Type": "application/json",
"X-Webhook-Signature": signature
}
)
print(f"POST {url}")
print(f"Status: {response.status_code}")
print(f"Response: {response.json()}")
return response.json()
def main():
parser = argparse.ArgumentParser(description="Webhook Order Management")
parser.add_argument("action", choices=["server", "worker", "test"], help="Action to perform")
parser.add_argument("--endpoint", help="Webhook endpoint to test")
parser.add_argument("--gpu", default="RTX 4090", help="GPU type")
parser.add_argument("--price", type=float, default=0.50, help="Max price")
args = parser.parse_args()
if args.action == "server":
from app import app
app.run(host='0.0.0.0', port=Config.PORT)
elif args.action == "worker":
os.system("celery -A tasks worker --loglevel=info")
elif args.action == "test":
if args.endpoint == "create-order":
test_webhook("create-order", {
"gpu_type": args.gpu,
"max_price": args.price
})
elif args.endpoint == "marketplace":
response = requests.get("http://localhost:5000/webhook/marketplace")
print(json.dumps(response.json(), indent=2))
else:
print("Available endpoints: create-order, cancel-order, price-alert, status, marketplace")
if __name__ == "__main__":
main()# docker-compose.yml
version: '3.8'
services:
webhook-server:
build: .
ports:
- "5000:5000"
environment:
- CLORE_API_KEY=${CLORE_API_KEY}
- WEBHOOK_SECRET=${WEBHOOK_SECRET}
- REDIS_URL=redis://redis:6379/0
- SLACK_WEBHOOK_URL=${SLACK_WEBHOOK_URL}
depends_on:
- redis
command: python app.py
worker:
build: .
environment:
- CLORE_API_KEY=${CLORE_API_KEY}
- REDIS_URL=redis://redis:6379/0
- SLACK_WEBHOOK_URL=${SLACK_WEBHOOK_URL}
depends_on:
- redis
command: celery -A tasks worker --loglevel=info
redis:
image: redis:7-alpine
volumes:
- redis-data:/data
volumes:
redis-data:curl -X POST https://your-server.com/webhook/create-order \
-H "Content-Type: application/json" \
-H "X-Webhook-Signature: <signature>" \
-d '{
"gpu_type": "RTX 4090",
"max_price": 0.50,
"image": "pytorch/pytorch:2.7.1-cuda12.8-cudnn9-runtime",
"callback_url": "https://your-server.com/callback"
}'curl -X POST https://your-server.com/webhook/price-alert \
-H "Content-Type: application/json" \
-H "X-Webhook-Signature: <signature>" \
-d '{
"gpu_type": "RTX 4090",
"target_price": 0.35,
"callback_url": "https://your-server.com/callback"
}'