from __future__ import annotations from typing import Any from fastapi.responses import HTMLResponse from fastapi import FastAPI import uvicorn from openenv.core.env_server import create_fastapi_app from logging_utils import get_logger from .environment import CareerAction, CareerEnvironment, CareerObservation, CareerState openenv_app: FastAPI = create_fastapi_app( CareerEnvironment, action_cls=CareerAction, observation_cls=CareerObservation, ) logger = get_logger("server") app = FastAPI(title="AI Career Advisor") app.mount("/openenv", openenv_app) UI_HTML = """ AI Career Advisor

AI Career Advisor

Enter your current skills to get a career recommendation and simulation. OpenEnv routes stay available at /openenv/reset, /openenv/step, and /openenv/state.

Career recommendation
Top career matches
role current fit matched skills missing skills base salary market demand
Simulation steps
step action career salary reward feedback
Examples
Built with FastAPI HTML UI
""" @app.get("/", response_class=HTMLResponse) def root() -> HTMLResponse: logger.info("UI root endpoint hit.") return HTMLResponse(content=UI_HTML) @app.api_route("/reset", methods=["GET", "POST"]) def reset_compat(body: dict[str, Any] | None = None) -> dict[str, Any]: payload = body or {} env = CareerEnvironment() obs = env.reset( seed=payload.get("seed"), episode_id=payload.get("episode_id"), ) return { "observation": obs.model_dump(), "reward": obs.reward, "done": obs.done, } @app.api_route("/state", methods=["GET", "POST"]) def state_compat() -> dict[str, Any]: env = CareerEnvironment() return env.state.model_dump() @app.get("/schema") def schema_compat() -> dict[str, Any]: return { "action": CareerAction.model_json_schema(), "observation": CareerObservation.model_json_schema(), "state": CareerState.model_json_schema(), } @app.api_route("/step", methods=["POST"]) def step_compat(body: dict[str, Any]) -> dict[str, Any]: action_payload = body.get("action") if isinstance(action_payload, dict): query = str(action_payload.get("query", "")) else: query = str(body.get("query", "")) env = CareerEnvironment() obs = env.step(CareerAction(query=query), timeout_s=body.get("timeout_s")) return { "observation": obs.model_dump(), "reward": obs.reward, "done": obs.done, } def main() -> None: logger.info("Starting uvicorn server.app:app on 0.0.0.0:8000") uvicorn.run("server.app:app", host="0.0.0.0", port=8000) if __name__ == "__main__": main()