openclaw-saas-platform

name	openclaw-saas-platform
description	Multi-tenant SaaS platform built on OpenClaw with auth, billing, workspace isolation, and AI agent execution gateway
triggers	["how do I set up openclaw saas","implement openclaw multi-tenant platform","create openclaw saas billing system","add user authentication to openclaw","configure openclaw workspace isolation","integrate openclaw agent gateway","deploy openclaw saas platform","test openclaw saas features"]

OpenClaw SaaS Platform

Skill by ara.so — Hermes Skills collection.

OpenClaw SaaS transforms the open-source OpenClaw AI Agent into a multi-tenant SaaS platform with account management, credit-based billing, subscription orders, workspace isolation, and real-time agent execution. Built with FastAPI backend, React 19 frontend, and dual-container Docker architecture.

Core Architecture

Dual-Container Design:

• Backend Container: FastAPI app handling auth, billing, orders, API gateway
• OpenClaw Gateway Container: Isolated AI agent execution engine
• Shared Volume: /opt/workspaces/{agent_id}/ for file isolation
• Dependencies: MySQL 8.4, Redis 7

Key Systems:

• Account system with SMS verification + JWT
• Credit-based billing with transaction ledger
• Subscription orders with state machine
• Multi-user workspace isolation
• Real-time Socket.IO communication
• Rate limiting and blacklist protection

Installation

Local Development Setup

# 1. Clone and setup dependencies
git clone https://github.com/xingzhicn/openclaw-saas.git
cd openclaw-saas

# 2. Start MySQL + Redis via Docker
docker-compose -f docker-compose.local.yml up -d

# 3. Backend setup
python3.12 -m venv venv
source venv/bin/activate
pip install -r requirements.txt

# 4. Configure environment
cat > .env <<EOF
DATABASE_URL=mysql+aiomysql://openclaw:openclaw_dev_pwd@localhost:3307/openclaw_saas
REDIS_URL=redis://localhost:6380/0
ENCRYPTION_KEY=$(openssl rand -hex 32)
JWT_SECRET_KEY=$(openssl rand -hex 32)
OPENCLAW_TOKEN=$(openssl rand -hex 16)
WORKSPACE_BASE=/opt/workspaces
ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY}
OPENAI_API_KEY=${OPENAI_API_KEY}
DEBUG=true
EOF

# 5. Initialize database
python -c "from backend.app.database import init_db; import asyncio; asyncio.run(init_db())"

# 6. Start backend
uvicorn backend.app.main:app --reload --port 8000

# 7. Start frontend (new terminal)
cd frontend
npm install
npm run dev  # http://localhost:5173

Production Deployment

# 1. Build Docker images
./scripts/pack-backend.sh
./scripts/pack-openclaw.sh

# 2. Transfer to server
scp openclaw-*.tar.gz deploy-*.sh root@your-server:/opt/

# 3. Deploy on server
ssh root@your-server
cd /opt
./deploy-all.sh

Core API Patterns

Authentication Flow

# backend/app/api/auth.py
from fastapi import APIRouter, Depends, HTTPException
from backend.app.middleware.auth import get_current_user
from backend.app.models.user import User
from backend.app.utils.sms import send_sms_code
from backend.app.utils.jwt import create_access_token

router = APIRouter(prefix="/api/v1/auth", tags=["auth"])

@router.post("/send-code")
async def send_verification_code(request: dict, db: AsyncSession = Depends(get_db)):
    """Send SMS verification code with rate limiting"""
    phone = request["phone"]
    
    # Check rate limit: 1 request per 60s
    cache_key = f"sms:ratelimit:{phone}"
    if await redis_client.exists(cache_key):
        raise HTTPException(status_code=429, detail="请等待60秒后再试")
    
    # Check daily limit: 3 requests per day
    daily_key = f"sms:daily:{phone}"
    daily_count = await redis_client.get(daily_key)
    if daily_count and int(daily_count) >= 3:
        raise HTTPException(status_code=429, detail="今日发送次数已达上限")
    
    # Generate and send code
    code = generate_sms_code()
    await send_sms_code(phone, code)
    
    
     redis_client.setex(, , code)
     redis_client.setex(cache_key, , )
     redis_client.incr(daily_key)
     redis_client.expire(daily_key, )
    
     {: }


  ():
    
    phone, code = request[], request[]
    
    
    cached_code =  redis_client.get()
      cached_code  cached_code != code:
         HTTPException(status_code=, detail=)
    
    
    result =  db.execute(select(User).where(User.phone == phone))
    user = result.scalar_one_or_none()
    
      user:
        user = User(phone=phone, credits=)  
        db.add(user)
         db.commit()
         db.refresh(user)
    
    
    token = create_access_token({: (user.)})
    
     {
        : token,
        : {
            : user.,
            : user.phone,
            : user.credits
        }
    }

Credit Billing System

# backend/app/api/credits.py
from backend.app.models.credit_transaction import CreditTransaction, TransactionType
from backend.app.utils.credits import freeze_credits, consume_credits, refund_credits

@router.post("/freeze")
async def freeze_user_credits(
    request: dict,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db)
):
    """Freeze credits before agent execution"""
    amount = request["amount"]
    request_id = request["request_id"]
    
    # Check sufficient credits
    if current_user.credits < amount:
        raise HTTPException(status_code=400, detail="积分不足")
    
    # Freeze in Redis
    await redis_client.hset(
        f"freeze:{request_id}",
        mapping={"user_id": current_user.id, "amount": amount}
    )
    await redis_client.expire(f"freeze:{request_id}", 3600)
    
    # Deduct from user balance
    current_user.credits -= amount
    await db.commit()
    
    # Create freeze transaction
    transaction = CreditTransaction(
        user_id=current_user.id,
        amount=amount,
        type=TransactionType.FREEZE,
        description=f"冻结积分: {request_id}",
        request_id=request_id
    )
    db.add(transaction)
    await db.commit()
    
     {: amount, : current_user.credits}


  ():
    
    request_id = request[]
    actual_amount = request[]
    
    
    freeze_data =  redis_client.hgetall()
      freeze_data:
         HTTPException(status_code=, detail=)
    
    frozen_amount = (freeze_data[])
    user_id = (freeze_data[])
    
    
    transaction = CreditTransaction(
        user_id=user_id,
        amount=actual_amount,
        =TransactionType.CONSUME,
        description=,
        request_id=request_id
    )
    db.add(transaction)
    
    
     actual_amount < frozen_amount:
        refund_amount = frozen_amount - actual_amount
        result =  db.execute(select(User).where(User. == user_id))
        user = result.scalar_one()
        user.credits += refund_amount
        
        refund_tx = CreditTransaction(
            user_id=user_id,
            amount=refund_amount,
            =TransactionType.REFUND,
            description=,
            request_id=request_id
        )
        db.add(refund_tx)
    
     redis_client.delete()
     db.commit()
    
     {: actual_amount, : frozen_amount - actual_amount}

Workspace Isolation

# backend/app/utils/workspace.py
import os
import shutil
from pathlib import Path

class WorkspaceManager:
    def __init__(self, base_path: str = "/opt/workspaces"):
        self.base_path = Path(base_path)
    
    def get_workspace_path(self, agent_id: str) -> Path:
        """Get isolated workspace path for agent"""
        workspace = self.base_path / str(agent_id)
        workspace.mkdir(parents=True, exist_ok=True)
        return workspace
    
    def create_workspace(self, agent_id: str) -> Path:
        """Create new workspace with initial structure"""
        workspace = self.get_workspace_path(agent_id)
        
        # Create subdirectories
        (workspace / "input").mkdir(exist_ok=True)
        (workspace / "output").mkdir(exist_ok=True)
        (workspace / "logs").mkdir(exist_ok=True)
        
        return workspace
    
    def cleanup_workspace(self, agent_id: str):
        """Remove workspace files (keep for 7 days in production)"""
        workspace = self.get_workspace_path(agent_id)
         workspace.exists():
            shutil.rmtree(workspace)
    
     ():
        
        workspace = .get_workspace_path(agent_id)
        input_path = workspace /  / filename
        input_path.write_text(content)
         input_path
    
     () -> :
        
        workspace = .get_workspace_path(agent_id)
        output_path = workspace /  / filename
          output_path.exists():
             FileNotFoundError()
         output_path.read_text()



  ():
    agent_id = (uuid.uuid4())
    workspace_manager = WorkspaceManager()
    
    
    workspace = workspace_manager.create_workspace(agent_id)
    
    
    workspace_manager.write_input_file(agent_id, , request[])
    
    
    
    result =  execute_openclaw_agent(agent_id, workspace)
    
    
    output = workspace_manager.read_output_file(agent_id, )
    
     {: agent_id, : output}

Agent Execution Gateway

# backend/app/tasks/agent_executor.py
import httpx
from backend.app.config import settings

async def execute_openclaw_agent(agent_id: str, workspace: Path) -> dict:
    """Execute agent in isolated OpenClaw Gateway container"""
    
    # OpenClaw Gateway runs in separate container
    gateway_url = settings.OPENCLAW_GATEWAY_URL
    
    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{gateway_url}/execute",
            json={
                "agent_id": agent_id,
                "workspace": str(workspace),
                "model": "claude-3-5-sonnet-20241022",
                "api_key": settings.ANTHROPIC_API_KEY
            },
            headers={"Authorization": f"Bearer {settings.OPENCLAW_TOKEN}"},
            timeout=300.0
        )
        
        if response.status_code != 200:
            raise Exception(f"Agent execution failed: {response.text}")
        
        return response.json()

# Real-time status updates via Socket.IO
from socketio import AsyncServer

sio = AsyncServer(async_mode='asgi', cors_allowed_origins='*')

async def stream_agent_status(agent_id: , user_id: ):
    
    
     sio.emit(, {
        : agent_id,
        : 
    }, room=)
    
    
    
     sio.emit(, {
        : agent_id,
        : ,
        : result
    }, room=)

Testing Patterns

OpenClaw SaaS uses "real" testing with database/Redis operations and four-dimensional verification:

# tests/test_credits.py
import pytest
from tests.helpers.verifier import verify_db, verify_cache, verify_workspace, verify_response

def test_freeze_consume_workflow(client, auth_headers, test_user, db, redis_client):
    """Test complete freeze -> consume -> refund workflow"""
    request_id = "req_freeze_001"
    
    # 1. Freeze credits
    response = client.post(
        "/api/v1/credits/freeze",
        json={"amount": 500, "request_id": request_id},
        headers=auth_headers
    )
    
    verify_response(response, 200, {"frozen_amount": 500})
    verify_db(db, "credit_transactions", {
        "user_id": test_user.id,
        "request_id": request_id,
        "type": "freeze"
    }, {"amount": 500})
    verify_cache(redis_client, f"freeze:{request_id}", {
        "user_id": str(test_user.id),
        "amount": "500"
    })
    
    # 2. Consume credits (less than frozen)
    response = client.post(
        "/api/v1/credits/consume",
        json={"request_id": request_id, "actual_amount": 300}
    )
    
    verify_response(response, 200, {"consumed": 300, "refunded": 200})
    verify_db(db, "credit_transactions", {
        : request_id,
        : 
    }, {: })
    verify_db(db, , {
        : request_id,
        : 
    }, {: })
    
    
      redis_client.exists()

 ():
    
    agent_id_1 = 
    agent_id_2 = 
    
    
    workspace_manager.write_input_file(agent_id_1, , )
    verify_workspace(agent_id_1, expected_files=[])
    
    
    workspace_manager.write_input_file(agent_id_2, , )
    verify_workspace(agent_id_2, expected_files=[])
    
    
    content_1 = workspace_manager.read_output_file(agent_id_1, )
     content_1 == 
    
    content_2 = workspace_manager.read_output_file(agent_id_2, )
     content_2 == 

Run tests:

# All tests
pytest tests/ -v

# Specific module
pytest tests/test_credits.py -v

# With coverage
pytest tests/ --cov=backend/app --cov-report=html

Frontend Integration

// frontend/src/services/api.ts
import axios from 'axios';

const api = axios.create({
  baseURL: import.meta.env.VITE_API_URL || 'http://localhost:8000',
  headers: { 'Content-Type': 'application/json' }
});

// Auto-attach JWT token
api.interceptors.request.use(config => {
  const token = localStorage.getItem('access_token');
  if (token) {
    config.headers.Authorization = `Bearer ${token}`;
  }
  return config;
});

// Auth API
export const authAPI = {
  sendCode: (phone: string) => api.post('/api/v1/auth/send-code', { phone }),
  login: (phone: string, code: string) => api.post('/api/v1/auth/login', { phone, code }),
  getCurrentUser: () => api.()
};


  creditsAPI = {
  :  api.(),
  :  
    api.(, { page, size }),
  : 
    api.(, { amount, : requestId })
};


  agentAPI = {
  :  api.(, { task }),
  :  api.()
};

// frontend/src/hooks/useAgentExecution.ts
import { io } from 'socket.io-client';
import { useEffect, useState } from 'react';

export function useAgentExecution(userId: number) {
  const [status, setStatus] = useState<string>('idle');
  const [result, setResult] = useState<any>(null);
  
  useEffect(() => {
    const socket = io(import.meta.env.VITE_API_URL);
    
    socket.emit('join', `user_${userId}`);
    
    socket.on('agent_status', (data) => {
      setStatus(data.status);
      if (data.status === 'completed') {
        setResult(data.result);
      }
    });
    
    return () => { socket.disconnect(); };
  }, [userId]);
  
  return { status, result };
}

Configuration

Key environment variables:

# Database
DATABASE_URL=mysql+aiomysql://user:pass@host:3306/dbname

# Redis
REDIS_URL=redis://host:6379/0

# Security
ENCRYPTION_KEY=<64-char-hex>  # openssl rand -hex 32
JWT_SECRET_KEY=<64-char-hex>
JWT_ALGORITHM=HS256
JWT_EXPIRATION_HOURS=720

# OpenClaw Gateway
OPENCLAW_GATEWAY_URL=http://openclaw-gateway:8001
OPENCLAW_TOKEN=<32-char-hex>  # openssl rand -hex 16
WORKSPACE_BASE=/opt/workspaces

# AI Models
ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY}
OPENAI_API_KEY=${OPENAI_API_KEY}

# Features
DEBUG=false
ENABLE_SMS=true
SMS_PROVIDER=aliyun

Common Troubleshooting

SMS code not received (DEBUG mode):

# In DEBUG mode, code is always "123456"
if settings.DEBUG:
    code = "123456"

Workspace permission denied:

# Ensure shared volume has correct permissions
chown -R 1000:1000 /opt/workspaces
chmod -R 755 /opt/workspaces

Database connection pool exhausted:

# backend/app/database.py
engine = create_async_engine(
    DATABASE_URL,
    pool_size=20,          # Increase from default 5
    max_overflow=40,       # Increase from default 10
    pool_pre_ping=True
)

Redis connection timeout:

# backend/app/database.py
redis_client = redis.from_url(
    REDIS_URL,
    decode_responses=True,
    socket_timeout=5.0,
    socket_connect_timeout=5.0
)

JWT token expired:

// frontend/src/services/api.ts
api.interceptors.response.use(
  response => response,
  error => {
    if (error.response?.status === 401) {
      localStorage.removeItem('access_token');
      window.location.href = '/login';
    }
    return Promise.reject(error);
  }
);

Production Deployment Checklist

1. Security:

   • Set DEBUG=false
   • Rotate ENCRYPTION_KEY and JWT_SECRET_KEY
   • Enable HTTPS with valid SSL certificate
   • Configure firewall rules (only 80/443 public)

2. Database:

   • Enable MySQL slow query log
   • Configure automated backups (daily)
   • Set proper connection pool limits

3. Monitoring:

   • Setup Sentry for error tracking
   • Configure log aggregation (ELK/Loki)
   • Enable Prometheus metrics

4. Scaling:

   • Use Redis Sentinel for HA
   • Deploy multiple backend replicas behind load balancer
   • Setup auto-scaling for OpenClaw Gateway containers

5. Workspace Cleanup:

   • Schedule cron job to delete workspaces older than 7 days

   0 2 * * * find /opt/workspaces -mtime +7 -exec rm -rf {} \;