""" FastAPI application for the Shopify Store Audit & Remediation Environment. Endpoints: POST /reset - Reset the environment (accepts task selection) POST /step - Execute an action GET /state - Get current environment state GET /schema - Get action/observation schemas WS /ws - WebSocket for persistent sessions """ try: from openenv.core.env_server.http_server import create_app except Exception as e: raise ImportError( "openenv is required. Install with: pip install openenv-core[core]" ) from e try: from ..models import ShopifyStoreAuditAction, ShopifyStoreAuditObservation from .shopify_store_audit_environment import ShopifyStoreAuditEnvironment except ImportError: from models import ShopifyStoreAuditAction, ShopifyStoreAuditObservation from server.shopify_store_audit_environment import ShopifyStoreAuditEnvironment from fastapi.responses import HTMLResponse, JSONResponse try: from .tasks import ALL_TASKS except ImportError: from server.tasks import ALL_TASKS try: from .graders import grade_product_listing_qa, grade_seo_collection_optimization, grade_full_store_audit except ImportError: from server.graders import grade_product_listing_qa, grade_seo_collection_optimization, grade_full_store_audit GRADER_MAP = { "product_listing_qa": grade_product_listing_qa, "seo_collection_optimization": grade_seo_collection_optimization, "full_store_audit": grade_full_store_audit, } app = create_app( ShopifyStoreAuditEnvironment, ShopifyStoreAuditAction, ShopifyStoreAuditObservation, env_name="shopify_store_audit", max_concurrent_envs=2, ) LANDING_HTML = """ Shopify Store Audit - OpenEnv

🛒 Shopify Store Audit

RL environment with real Shopify product data — 45 products, 184 discoverable issues, shaped rewards

Easy · Full hints

Product Listing QA

8 issues · 25 steps
Issues listed with suggested commands. Agent fills in params.

Medium · Descriptions only

SEO & Collections

12 issues · 35 steps
Issue descriptions shown. Agent picks commands & params.

Hard · Explore

Full Store Audit

20 issues · 50 steps
Only category counts. Agent must discover issues itself.

API Endpoints

GET/health — health check

GET/tasks — enumerate tasks & graders

POST/reset — reset (body: {})

POST/step — action (body: {"action":{"command":"...","params":{}}})

GET/state — current state

GET/schema — action/observation schemas

POST/grade — run grader on sample

WS/ws — persistent WebSocket session

""" @app.get("/tasks") async def list_tasks(): """Enumerate all tasks with their grader info. Used by competition validators.""" tasks = [] for task_id, cfg in ALL_TASKS.items(): tasks.append({ "id": cfg.task_id, "title": cfg.title, "description": cfg.description, "difficulty": cfg.difficulty, "max_steps": cfg.max_steps, "num_issues": cfg.num_issues, "hint_level": cfg.hint_level, "has_grader": task_id in GRADER_MAP, "grader": f"server.graders.grade_{task_id}", }) return {"tasks": tasks, "count": len(tasks)} @app.post("/grade") async def grade_task(body: dict): """Run a grader on the provided observation/sample. Used by competition validators.""" task_id = body.get("task_id", body.get("task", "")) sample = body.get("sample", body.get("observation", body)) grader = GRADER_MAP.get(task_id) if grader is None: return JSONResponse( status_code=400, content={"error": f"Unknown task '{task_id}'. Available: {list(GRADER_MAP.keys())}"}, ) try: score = grader(sample) return {"task_id": task_id, "score": score, "valid": 0.0 <= score <= 1.0} except Exception as e: return JSONResponse(status_code=500, content={"error": str(e)}) @app.get("/", response_class=HTMLResponse) async def root(): return LANDING_HTML def main(host: str = "0.0.0.0", port: int = 8000): import uvicorn uvicorn.run(app, host=host, port=port) if __name__ == "__main__": main()