Spaces:
Sleeping
Sleeping
File size: 1,833 Bytes
c14504c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 | """
Simple HTTP client for the Code Debug OpenEnv server.
"""
from typing import Optional
import httpx
class CodeDebugClient:
def __init__(self, base_url: str = "http://localhost:7860", timeout: float = 30.0):
self.base_url = base_url.rstrip("/")
self._http = httpx.Client(base_url=self.base_url, timeout=timeout)
self.episode_id: Optional[str] = None
def reset(self, task_id: Optional[str] = None, seed: Optional[int] = None) -> dict:
payload: dict = {}
if task_id is not None:
payload["task_id"] = task_id
if seed is not None:
payload["seed"] = seed
resp = self._http.post("/reset", json=payload)
resp.raise_for_status()
data = resp.json()
self.episode_id = data["episode_id"]
return data
def step(self, code: str) -> dict:
if self.episode_id is None:
raise RuntimeError("No active episode. Call reset() first.")
resp = self._http.post(
f"/step/{self.episode_id}",
json={"action": {"code": code}},
)
resp.raise_for_status()
return resp.json()
def state(self) -> dict:
if self.episode_id is None:
raise RuntimeError("No active episode. Call reset() first.")
resp = self._http.get(f"/state/{self.episode_id}")
resp.raise_for_status()
return resp.json()
def list_tasks(self) -> list[dict]:
resp = self._http.get("/tasks")
resp.raise_for_status()
return resp.json()
def health(self) -> dict:
resp = self._http.get("/health")
resp.raise_for_status()
return resp.json()
def close(self):
self._http.close()
def __enter__(self):
return self
def __exit__(self, *args):
self.close()
|