Spaces:
Sleeping
Sleeping
File size: 3,562 Bytes
72de9a9 d73bfc0 72de9a9 d73bfc0 72de9a9 d73bfc0 72de9a9 | 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 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 | """
Client SDK for the API Debug Environment.
Implements the three required abstract methods from EnvClient:
- _step_payload: converts APIDebugAction to JSON dict
- _parse_result: converts server response to StepResult
- _parse_state: converts server state to State object
Usage:
async with APIDebugEnv(base_url="http://localhost:8000") as env:
result = await env.reset(task="easy")
result = await env.step(APIDebugAction(error_type="missing_required_field"))
"""
from typing import Any, Dict
from openenv.core import EnvClient
from openenv.core.client_types import StepResult
from openenv.core.env_server.types import State
from models import APIDebugAction, APIDebugObservation
class APIDebugEnv(EnvClient[APIDebugAction, APIDebugObservation, State]):
def _step_payload(self, action: APIDebugAction) -> Dict[str, Any]:
"""Convert action to JSON dict, including only non-None fields."""
payload = {}
if action.error_type is not None:
payload["error_type"] = action.error_type
if action.error_types is not None:
payload["error_types"] = action.error_types
if action.affected_fields is not None:
payload["affected_fields"] = action.affected_fields
if action.fixed_request is not None:
payload["fixed_request"] = action.fixed_request
if action.fixed_headers is not None:
payload["fixed_headers"] = action.fixed_headers
if action.explanation is not None:
payload["explanation"] = action.explanation
if action.response_issues is not None:
payload["response_issues"] = action.response_issues
if action.expected_status_code is not None:
payload["expected_status_code"] = action.expected_status_code
return payload
def _parse_result(self, payload: Dict[str, Any]) -> StepResult[APIDebugObservation]:
"""Convert server JSON response to StepResult.
The server sends:
{
"observation": { ...fields except reward/done/metadata... },
"reward": float,
"done": bool,
}
"""
obs_data = payload.get("observation", {})
observation = APIDebugObservation(
task=obs_data.get("task", "easy"),
api_name=obs_data.get("api_name", ""),
http_method=obs_data.get("http_method", "POST"),
endpoint=obs_data.get("endpoint", ""),
broken_request=obs_data.get("broken_request", ""),
broken_headers=obs_data.get("broken_headers", {}),
api_spec=obs_data.get("api_spec", ""),
error_count=obs_data.get("error_count", 1),
step_number=obs_data.get("step_number", 0),
max_steps=obs_data.get("max_steps", 3),
response_body=obs_data.get("response_body", ""),
response_status_code=obs_data.get("response_status_code", 0),
feedback=obs_data.get("feedback", ""),
message=obs_data.get("message", ""),
done=payload.get("done", False),
reward=payload.get("reward", 0.0),
)
return StepResult(
observation=observation,
reward=payload.get("reward", 0.0),
done=payload.get("done", False),
)
def _parse_state(self, payload: Dict[str, Any]) -> State:
"""Convert server state JSON to State object."""
return State(
episode_id=payload.get("episode_id", ""),
step_count=payload.get("step_count", 0),
)
|