openclaw-saas-platform

# OpenClaw SaaS Platform > Skill by [ara.so](https://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 ```bash # 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 ```bash # 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 ```python # 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) # Cache code for 5 minutes await redis_client.setex(f"sms:code:{phone}", 300, code) await redis_client.setex(cache_key, 60, "1") await redis_client.incr(daily_key) await redis_client.expire(daily_key, 86400) return {"message": "验证码已发送"} @router.post("/login") async def login(request: dict, db: AsyncSession = Depends(get_db)): """Login with phone and SMS code""" phone, code = request["phone"], request["code"] # Verify code cached_code = await redis_client.get(f"sms:code:{phone}") if not cached_code or cached_code != code: raise HTTPException(status_code=400, detail="验证码错误或已过期") # Get or create user result = await db.execute(select(User).where(User.phone == phone)) user = result.scalar_one_or_none() if not user: user = User(phone=phone, credits=1000) # 1000 initial credits db.add(user) await db.commit() await db.refresh(user) # Generate JWT token = create_access_token({"sub": str(user.id)}) return { "access_token": token, "user": { "id": user.id, "phone": user.phone, "credits": user.credits } } ``` ### Credit Billing System ```python # 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() return {"frozen_amount": amount, "remaining_credits": current_user.credits} @router.post("/consume") async def consume_frozen_credits( request: dict, db: AsyncSession = Depends(get_db) ): """Consume frozen credits after agent execution""" request_id = request["request_id"] actual_amount = request["actual_amount"] # Get freeze info freeze_data = await redis_client.hgetall(f"freeze:{request_id}") if not freeze_data: raise HTTPException(status_code=404, detail="冻结记录不存在") frozen_amount = int(freeze_data["amount"]) user_id = int(freeze_data["user_id"]) # Consume transaction transaction = CreditTransaction( user_id=user_id, amount=actual_amount, type=TransactionType.CONSUME, description=f"消耗积分: {request_id}", request_id=request_id ) db.add(transaction) # Refund excess if any if actual_amount < frozen_amount: refund_amount = frozen_amount - actual_amount result = await db.execute(select(User).where(User.id == user_id)) user = result.scalar_one() user.credits += refund_amount refund_tx = CreditTransaction( user_id=user_id, amount=refund_amount, type=TransactionType.REFUND, description=f"退回冻结余额: {request_id}", request_id=request_id ) db.add(refund_tx) await redis_client.delete(f"freeze:{request_id}") await db.commit() return {"consumed": actual_amount, "refunded": frozen_amount - actual_amount} ``` ### Workspace Isolation ```python # 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) if workspace.exists(): shutil.rmtree(workspace) def write_input_file(self, agent_id: str, filename: str, content: str): """Write file to workspace input directory""" workspace = self.get_workspace_path(agent_id) input_path = workspace / "input" / filename input_path.write_text(content) return input_path def read_output_file(self, agent_id: str, filename: str) -> str: """Read file from workspace output directory""" workspace = self.get_workspace_path(agent_id) output_path = workspace / "output" / filename if not output_path.exists(): raise FileNotFoundError(f"{filename} not found in output") return output_path.read_text() # Usage in API @router.post("/agents/execute") async def execute_agent( request: dict, current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db) ): agent_id = str(uuid.uuid4()) workspace_manager = WorkspaceManager() # Create isolated workspace workspace = workspace_manager.create_workspace(agent_id) # Write input files workspace_manager.write_input_file(agent_id, "task.txt", request["task"]) # Execute agent in OpenClaw Gateway container # (shares /opt/workspaces via Docker volume) result = await execute_openclaw_agent(agent_id, workspace) # Read output output = workspace_manager.read_output_file(agent_id, "result.txt") return {"agent_id": agent_id, "output": output} ``` ### Agent Execution Gateway ```python # 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: str, user_id: int): """Stream agent execution status to frontend""" await sio.emit('agent_status', { 'agent_id': agent_id, 'status': 'running' }, room=f"user_{user_id}") # ... agent execution ... await sio.emit('agent_status', { 'agent_id': agent_id, 'status': 'completed', 'result': result }, room=f"user_{user_id}") ``` ## Testing Patterns OpenClaw SaaS uses "real" testing with database/Redis operations and four-dimensional verification: ```python # 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 )