Spaces:
Sleeping
Sleeping
| """ | |
| SYNAPSE-X API server - FastAPI application for HF Spaces deployment. | |
| Endpoints required by OpenEnv spec and hackathon pre-submission checklist: | |
| GET /health - liveness check | |
| POST /reset - start new episode | |
| GET /reset - same, query-param form | |
| POST /step - apply one action | |
| GET /state - inspect current env state | |
| GET /tasks - enumerate task presets | |
| GET /validate - submission readiness check | |
| POST /grade - score an action trace | |
| """ | |
| import json | |
| import os | |
| import sys | |
| from pathlib import Path | |
| from typing import Any, Optional | |
| from urllib.parse import parse_qs | |
| from fastapi import FastAPI, HTTPException, Request | |
| from fastapi.middleware.cors import CORSMiddleware | |
| from fastapi.responses import HTMLResponse | |
| from pydantic import BaseModel, ValidationError | |
| PROJECT_ROOT = Path(__file__).resolve().parents[1] | |
| if str(PROJECT_ROOT) not in sys.path: | |
| sys.path.insert(0, str(PROJECT_ROOT)) | |
| from env.environment import SynapseXEnvironment | |
| from env.grader import TASK_REGISTRY, TASK_SEEDS, grade as run_grade | |
| from env.models import Action, GradeResult, Observation, StepResult | |
| app = FastAPI( | |
| title="SYNAPSE-X", | |
| description="Predictive decision intelligence environment - OpenEnv Hackathon submission", | |
| version="1.0.0", | |
| ) | |
| LOCAL_IMAGE_NAME = os.environ.get("LOCAL_IMAGE_NAME", "synapse-x") | |
| APP_VERSION = "1.0.0" | |
| APP_DESCRIPTION = "Predictive decision intelligence environment - OpenEnv Hackathon submission" | |
| API_ENDPOINTS = [ | |
| "/health", | |
| "/metadata", | |
| "/schema", | |
| "/mcp", | |
| "/reset", | |
| "/step", | |
| "/state", | |
| "/tasks", | |
| "/grade", | |
| "/validate", | |
| ] | |
| app.add_middleware( | |
| CORSMiddleware, | |
| allow_origins=["*"], | |
| allow_methods=["*"], | |
| allow_headers=["*"], | |
| ) | |
| def root(): | |
| return HTMLResponse( | |
| content=""" | |
| <!DOCTYPE html> | |
| <html> | |
| <head> | |
| <meta charset="utf-8"> | |
| <title>SYNAPSE-X</title> | |
| <meta http-equiv="refresh" content="0; url=/docs"> | |
| <style> | |
| body { font-family: sans-serif; background: #0f172a; color: #e2e8f0; | |
| display: flex; align-items: center; justify-content: center; | |
| height: 100vh; margin: 0; } | |
| a { color: #818cf8; } | |
| </style> | |
| </head> | |
| <body> | |
| <p>Redirecting to <a href="/docs">SYNAPSE-X API docs</a>...</p> | |
| </body> | |
| </html> | |
| """ | |
| ) | |
| _env: Optional[SynapseXEnvironment] = None | |
| def get_env() -> SynapseXEnvironment: | |
| global _env | |
| if _env is None: | |
| _env = SynapseXEnvironment(task_config=TASK_REGISTRY["easy"], seed=42) | |
| _env.reset() | |
| return _env | |
| def to_dict(model: Any) -> dict[str, Any]: | |
| if hasattr(model, "model_dump"): | |
| return model.model_dump() | |
| return model.dict() | |
| class ResetRequest(BaseModel): | |
| task_name: Optional[str] = "easy" | |
| seed: int = 42 | |
| class GradeRequest(BaseModel): | |
| task_name: str | |
| actions: list | |
| def _normalize_task_name(task_name: Optional[str]) -> str: | |
| candidate = (task_name or "easy").strip() | |
| return candidate if candidate in TASK_REGISTRY else "easy" | |
| def _resolve_reset_request( | |
| body_payload: Optional[dict[str, Any]] = None, | |
| task_name: Optional[str] = None, | |
| seed: Optional[str] = None, | |
| ) -> ResetRequest: | |
| payload: dict[str, Any] = {} | |
| if task_name is not None: | |
| payload["task_name"] = task_name | |
| if seed is not None: | |
| payload["seed"] = seed | |
| if body_payload: | |
| payload.update(body_payload) | |
| try: | |
| return ResetRequest(**payload) | |
| except ValidationError: | |
| return ResetRequest() | |
| async def _extract_reset_payload(request: Request) -> dict[str, Any]: | |
| try: | |
| raw_body = await request.body() | |
| except Exception: | |
| return {} | |
| if not raw_body: | |
| return {} | |
| body_text = raw_body.decode("utf-8", errors="ignore").strip() | |
| if not body_text: | |
| return {} | |
| content_type = request.headers.get("content-type", "").lower() | |
| if "application/x-www-form-urlencoded" in content_type: | |
| parsed = parse_qs(body_text, keep_blank_values=False) | |
| return {key: values[-1] for key, values in parsed.items() if values} | |
| try: | |
| decoded = json.loads(body_text) | |
| except json.JSONDecodeError: | |
| return {} | |
| return decoded if isinstance(decoded, dict) else {} | |
| def health(): | |
| return {"status": "healthy", "environment": "SYNAPSE-X", "version": APP_VERSION} | |
| def metadata(): | |
| return { | |
| "name": "synapse-x", | |
| "environment": "SYNAPSE-X", | |
| "version": APP_VERSION, | |
| "description": APP_DESCRIPTION, | |
| "mode": "simulation", | |
| "entrypoints": { | |
| "api_app": "api.app:app", | |
| "server_script": "server.app:main", | |
| "environment": "env.environment:SynapseXEnvironment", | |
| "inference": "inference.py", | |
| }, | |
| "task_names": list(TASK_REGISTRY.keys()), | |
| "task_seeds": TASK_SEEDS, | |
| "endpoints": API_ENDPOINTS, | |
| } | |
| def schema(): | |
| return { | |
| "environment": { | |
| "name": "SynapseXEnvironment", | |
| "module": "env.environment", | |
| }, | |
| "observation": Observation.model_json_schema(), | |
| "action": Action.model_json_schema(), | |
| "state": { | |
| "type": "object", | |
| "properties": { | |
| "time": {"type": "integer"}, | |
| "resources": {"type": "number"}, | |
| "total_reward": {"type": "number"}, | |
| "done": {"type": "boolean"}, | |
| "tasks": {"type": "array"}, | |
| "history": {"type": "array"}, | |
| "seed": {"type": "integer"}, | |
| }, | |
| "required": ["time", "resources", "done", "tasks"], | |
| }, | |
| "step_result": StepResult.model_json_schema(), | |
| "grade_result": GradeResult.model_json_schema(), | |
| } | |
| def mcp_manifest(): | |
| return { | |
| "status": "healthy", | |
| "mode": "simulation", | |
| "environment": "SYNAPSE-X", | |
| "metadata_url": "/metadata", | |
| "schema_url": "/schema", | |
| } | |
| async def mcp_rpc(request: Request): | |
| try: | |
| payload = await request.json() | |
| except Exception: | |
| payload = {} | |
| return { | |
| "jsonrpc": "2.0", | |
| "id": payload.get("id"), | |
| "result": { | |
| "status": "healthy", | |
| "mode": "simulation", | |
| "environment": "SYNAPSE-X", | |
| "metadata_url": "/metadata", | |
| "schema_url": "/schema", | |
| }, | |
| } | |
| def reset_get(task_name: Optional[str] = None, seed: Optional[str] = None): | |
| global _env | |
| req = _resolve_reset_request(task_name=task_name, seed=seed) | |
| task = _normalize_task_name(req.task_name) | |
| _env = SynapseXEnvironment(task_config=TASK_REGISTRY[task], seed=req.seed) | |
| return to_dict(_env.reset()) | |
| async def reset(request: Request, task_name: Optional[str] = None, seed: Optional[str] = None): | |
| global _env | |
| body_payload = await _extract_reset_payload(request) | |
| req = _resolve_reset_request(body_payload=body_payload, task_name=task_name, seed=seed) | |
| task = _normalize_task_name(req.task_name) | |
| _env = SynapseXEnvironment(task_config=TASK_REGISTRY[task], seed=req.seed) | |
| return to_dict(_env.reset()) | |
| def step(action_dict: dict): | |
| env = get_env() | |
| try: | |
| action = Action(**action_dict) | |
| except ValidationError as exc: | |
| raise HTTPException(status_code=400, detail=str(exc)) from exc | |
| try: | |
| result = env.step(action) | |
| except RuntimeError as exc: | |
| raise HTTPException(400, str(exc)) from exc | |
| return { | |
| "observation": to_dict(result.observation), | |
| "reward": float(result.reward), | |
| "done": bool(result.done), | |
| "info": result.info or {}, | |
| } | |
| def state(): | |
| return get_env().state() | |
| def list_tasks(): | |
| return {name: cfg for name, cfg in TASK_REGISTRY.items()} | |
| def validate(): | |
| return { | |
| "status": "ok", | |
| "environment": "SYNAPSE-X", | |
| "version": APP_VERSION, | |
| "mode": "simulation", | |
| "task_names": list(TASK_REGISTRY.keys()), | |
| "task_seeds": TASK_SEEDS, | |
| "endpoints": API_ENDPOINTS, | |
| "docker": { | |
| "image_name": LOCAL_IMAGE_NAME, | |
| "app_import": "api.app:app", | |
| }, | |
| "multi_mode": { | |
| "server_script": "server.app:main", | |
| "lockfile": "uv.lock", | |
| }, | |
| "openenv_spec": { | |
| "reset": "POST /reset", | |
| "step": "POST /step", | |
| "state": "GET /state", | |
| }, | |
| } | |
| def grade_endpoint(req: GradeRequest): | |
| if req.task_name not in TASK_REGISTRY: | |
| raise HTTPException(400, f"Unknown task_name '{req.task_name}'. Choose from: {list(TASK_REGISTRY.keys())}") | |
| result: GradeResult = run_grade(req.task_name, req.actions) | |
| return to_dict(result) | |
| if __name__ == "__main__": | |
| import uvicorn | |
| port = int(os.environ.get("PORT", 7860)) | |
| uvicorn.run("api.app:app", host="0.0.0.0", port=port, reload=False) | |