""" FastAPI application for the WatchDog Environment. Endpoints: - GET /: Play UI (Gradio) - POST /reset: Reset the environment - POST /step: Execute an action - GET /state: Get current environment state - GET /schema: Get action/observation schemas - WS /ws: WebSocket endpoint for persistent sessions Usage: uvicorn server.app:app --reload --host 0.0.0.0 --port 8000 """ import logging logging.basicConfig(level=logging.INFO, format="%(name)s: %(message)s") from pathlib import Path # Load ../.env (openenv_hack/.env) before any imports that use env vars def _load_env() -> None: import logging import os _log = logging.getLogger("watchdog_env") _env_path = Path(__file__).resolve().parent.parent.parent / ".env" if not _env_path.is_file(): _log.info("[app] No .env at %s, skipping", _env_path) return try: from dotenv import load_dotenv load_dotenv(_env_path, override=True) # override so .env takes precedence _log.info("[app] Loaded .env from %s", _env_path) _log.info("[app] WATCHDOG_LLM_BACKEND=%s, GEMINI_API_KEY=%s", os.environ.get("WATCHDOG_LLM_BACKEND"), "set" if os.environ.get("GEMINI_API_KEY") or os.environ.get("GOOGLE_API_KEY") else "NOT SET") return except ImportError: pass # Fallback: parse .env manually when python-dotenv not installed for line in _env_path.read_text().encode("utf-8", errors="replace").decode().splitlines(): line = line.strip() if not line or line.startswith("#"): continue if "=" in line: key, _, value = line.partition("=") key, value = key.strip(), value.strip().strip("'\"") if key: os.environ[key] = value _load_env() from fastapi import FastAPI import gradio as gr from openenv.core.env_server.http_server import create_app from models import MultiTurnAction, MultiTurnObservation from .watchdog_environment import WatchDogMultiTurnEnvironment from .ui import build_ui, UI_CSS, UI_THEME # Ensure plugins are registered (Avalon, Cicero) try: import plugins # noqa: F401 except ImportError: import watchdog_env.plugins # noqa: F401 app = create_app( WatchDogMultiTurnEnvironment, MultiTurnAction, MultiTurnObservation, env_name="watchdog_env", max_concurrent_envs=4, ) # Mount Gradio play UI at root (theme/css passed to launch per Gradio 6.0) gradio_app = build_ui() app = gr.mount_gradio_app(app, gradio_app, path="/", theme=UI_THEME, css=UI_CSS) @app.get("/api") def api_info(): """API info (UI is at /).""" return { "message": "WatchDog OpenEnv Environment API", "play_ui": "/", "endpoints": { "health": "GET /health", "schema": "GET /schema", "reset": "POST /reset", "step": "POST /step", "state": "GET /state", "docs": "GET /docs", "ws": "WS /ws", }, } def main(host: str = "0.0.0.0", port: int = 8000) -> None: import uvicorn uvicorn.run(app, host=host, port=port) if __name__ == "__main__": import argparse parser = argparse.ArgumentParser() parser.add_argument("--port", type=int, default=8000) args = parser.parse_args() main(port=args.port)