hyperlinken/ALT_DESIGN

1"""2client.py3Thin Python HTTP client over the IT Triage OpenEnv REST API.4Provides a clean, typed interface for use in inference scripts and notebooks.5"""6 7from __future__ import annotations8 9from typing import Any, Dict, List10 11import requests12 13from models import EnvironmentState, Observation, StepResult, TriageAction14 15 16class ITTriageClient:17    """18    Synchronous HTTP client for the IT Triage OpenEnv environment.19 20    Usage:21        client = ITTriageClient(base_url="http://localhost:7860")22        obs    = client.reset("basic_triage")23        result = client.step(action)24        st     = client.state()25    """26 27    def __init__(self, base_url: str = "http://localhost:7860", timeout: int = 30) -> None:28        self.base_url = base_url.rstrip("/")29        self.timeout  = timeout30        self._session = requests.Session()31        self._session.headers.update({"Content-Type": "application/json"})32 33    def health(self) -> Dict[str, Any]:34        """Ping the server and return health/task metadata."""35        return self._get("/")36 37    def reset(self, task_id: str = "basic_triage") -> Observation:38        """Reset the environment and return the initial observation."""39        data = self._post("/reset", {"task_id": task_id})40        return Observation(**data)41 42    def step(self, action: TriageAction) -> StepResult:43        """Submit a triage action and receive (obs, reward, done, info)."""44        data = self._post("/step", action.model_dump())45        return StepResult(**data)46 47    def state(self) -> EnvironmentState:48        """Fetch the full current environment state."""49        data = self._get("/state")50        return EnvironmentState(**data)51 52    def list_tasks(self) -> List[Dict[str, Any]]:53        """Return all registered tasks."""54        return self._get("/tasks")55 56    def _get(self, path: str) -> Any:57        resp = self._session.get(f"{self.base_url}{path}", timeout=self.timeout)58        resp.raise_for_status()59        return resp.json()60 61    def _post(self, path: str, payload: Dict) -> Any:62        resp = self._session.post(63            f"{self.base_url}{path}",64            json=payload,65            timeout=self.timeout,66        )67        resp.raise_for_status()68        return resp.json()69