SYNAPSE-X

""" 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=["*"], ) @app.get("/", response_class=HTMLResponse, include_in_schema=False) def root(): return HTMLResponse( content=""" SYNAPSE-X

Redirecting to SYNAPSE-X API docs...

""" ) _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 {} @app.get("/health") def health(): return {"status": "healthy", "environment": "SYNAPSE-X", "version": APP_VERSION} @app.get("/metadata") 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, } @app.get("/schema") 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(), } @app.get("/mcp") def mcp_manifest(): return { "status": "healthy", "mode": "simulation", "environment": "SYNAPSE-X", "metadata_url": "/metadata", "schema_url": "/schema", } @app.post("/mcp") 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", }, } @app.get("/reset") 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()) @app.post("/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()) @app.post("/step") 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 {}, } @app.get("/state") def state(): return get_env().state() @app.get("/tasks") def list_tasks(): return {name: cfg for name, cfg in TASK_REGISTRY.items()} @app.get("/validate") 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", }, } @app.post("/grade") 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)