""" FastAPI application for the AETHER-TaskFlow Environment. Endpoints: GET / - Interactive Mission Control Dashboard POST /reset - Reset the environment, return initial observation POST /step - Execute an action, return next observation GET /state - Return current internal state GET /schema - Return action/observation/state JSON schemas GET /health - Health check WS /ws - WebSocket for persistent sessions GET /docs - Swagger UI """ from __future__ import annotations import os import sys from pathlib import Path from typing import Any, Dict from fastapi import Body, HTTPException, status from fastapi.responses import HTMLResponse from openenv.core.env_server.http_server import create_app from openenv.core.env_server.types import ResetRequest, ResetResponse, SchemaResponse, StepResponse from openenv.core.env_server.web_interface import WebInterfaceManager from pydantic import ValidationError _REPO_ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(_REPO_ROOT)) try: from models import AetherTaskFlowAction, AetherTaskFlowObservation, AetherTaskFlowState from env.aether_env import AetherTaskFlowEnvironment except ModuleNotFoundError: sys.path.insert(0, str(_REPO_ROOT)) from models import AetherTaskFlowAction, AetherTaskFlowObservation, AetherTaskFlowState from env.aether_env import AetherTaskFlowEnvironment _DIFFICULTY = os.getenv("AETHER_DIFFICULTY", "easy") def _make_env() -> AetherTaskFlowEnvironment: difficulty = os.getenv("AETHER_DIFFICULTY", _DIFFICULTY) return AetherTaskFlowEnvironment(difficulty=difficulty) def _create_persistent_manager() -> WebInterfaceManager: temp_env = _make_env() metadata: dict = {} try: if hasattr(temp_env, "get_metadata"): metadata = temp_env.get_metadata() except Exception: pass finally: if hasattr(temp_env, "close"): try: temp_env.close() except Exception: pass return WebInterfaceManager(_make_env, AetherTaskFlowAction, AetherTaskFlowObservation, metadata=metadata) app = create_app( _make_env, AetherTaskFlowAction, AetherTaskFlowObservation, env_name="aether_taskflow", max_concurrent_envs=4, ) _persistent_manager = _create_persistent_manager() def _remove_route(path: str, method: str) -> None: app.router.routes = [ route for route in app.router.routes if not (getattr(route, "path", None) == path and method in (getattr(route, "methods", set()) or set())) ] for _path, _method in (("/reset", "POST"), ("/step", "POST"), ("/state", "GET"), ("/schema", "GET")): _remove_route(_path, _method) def _action_to_payload(action: Any) -> Dict[str, Any]: if hasattr(action, "model_dump"): return action.model_dump(exclude={"metadata"}) if isinstance(action, dict): return action raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Invalid action") def _extract_action_payload(payload: Dict[str, Any]) -> Dict[str, Any]: env = _persistent_manager.env if "message" in payload and isinstance(payload["message"], str): if hasattr(env, "message_to_action"): return _action_to_payload(env.message_to_action(payload["message"])) return {"message": payload["message"]} action_payload = payload.get("action", payload) if isinstance(action_payload, str): if hasattr(env, "message_to_action"): return _action_to_payload(env.message_to_action(action_payload)) return {"message": action_payload} if isinstance(action_payload, dict): if "message" in action_payload and isinstance(action_payload["message"], str): if hasattr(env, "message_to_action"): return _action_to_payload(env.message_to_action(action_payload["message"])) return {"message": action_payload["message"]} return action_payload raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Invalid payload") @app.get("/", response_class=HTMLResponse, tags=["UI"], summary="Mission Control Dashboard") async def root() -> HTMLResponse: difficulty = os.getenv("AETHER_DIFFICULTY", "easy") html = f""" AETHER-TaskFlow · Mission Control
LIVE
Swagger Schema State Health
Difficulty STANDBY
📋 TASK QUEUE 0 tasks
Reset the environment
to load tasks
⏱ Time Remaining
—
⚡ Energy
—
💰 Budget
—
🏆 Cumulative Value
—
SELECT ACTION
⚡
EXECUTE
Full reward
🔧
OPTIMIZE
Reduce risk
📤
DELEGATE
35% reward
⏳
DEFER
Small penalty
› Waiting for first action...
REWARD HISTORY — per step
No data yet
—
Episode Score
0
Steps
0.00
Last Reward
0
Completed
0
Failed
SYSTEM HEALTH —
ACTION LOG
› System online. Select difficulty and press RESET.
API ONLINE
NO EPISODE
Meta PyTorch OpenEnv Hackathon 2025
OpenAPI ↗
""" return HTMLResponse(content=html) # --------------------------------------------------------------------------- # API Routes # --------------------------------------------------------------------------- @app.post("/reset", response_model=ResetResponse, tags=["Environment Control"], summary="Reset the environment") async def reset(request: ResetRequest = Body(default_factory=ResetRequest)) -> ResetResponse: response = await _persistent_manager.reset_environment(request.model_dump(exclude_unset=True)) return ResetResponse(**response) @app.post("/step", response_model=StepResponse, tags=["Environment Control"], summary="Execute an action") async def step(payload: Dict[str, Any] = Body(default_factory=dict)) -> StepResponse: try: action_payload = _extract_action_payload(payload) response = await _persistent_manager.step_environment(action_payload) except ValidationError as exc: raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=exc.errors()) from exc return StepResponse(**response) @app.get("/state", response_model=AetherTaskFlowState, tags=["State Management"], summary="Get current state") async def get_state() -> AetherTaskFlowState: return _persistent_manager.env.state @app.get("/schema", response_model=SchemaResponse, tags=["Schema"], summary="Get JSON schemas") async def get_schemas() -> SchemaResponse: return SchemaResponse( action=AetherTaskFlowAction.model_json_schema(), observation=AetherTaskFlowObservation.model_json_schema(), state=AetherTaskFlowState.model_json_schema(), ) def _run_server(host: str, port: int, difficulty: str) -> None: if difficulty != _DIFFICULTY: os.environ["AETHER_DIFFICULTY"] = difficulty import uvicorn uvicorn.run(app, host=host, port=port) def main() -> None: import argparse parser = argparse.ArgumentParser(description="AETHER-TaskFlow server") parser.add_argument("--host", default="0.0.0.0") parser.add_argument("--port", type=int, default=7860) parser.add_argument("--difficulty", choices=["easy", "medium", "hard"], default=_DIFFICULTY) args = parser.parse_args() _run_server(host=args.host, port=args.port, difficulty=args.difficulty) if __name__ == "__main__": main()