Spaces:
Runtime error
Runtime error
File size: 2,385 Bytes
dbf5e96 b2b578f dbf5e96 b2b578f 74c9d03 b2b578f 74c9d03 dbf5e96 b2b578f dbf5e96 caf52d5 b2b578f dbf5e96 b2b578f | 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 | """
CodeReviewEnv Client — typed async/sync client for MCP tool-calling interface.
Usage:
async with CodeReviewEnv(base_url="https://your-space.hf.space") as env:
result = await env.reset(seed=42)
# List available tools
result = await env.step(CodeReviewAction(action_type="ListToolsAction"))
# Call a tool
result = await env.step(CodeReviewAction(
action_type="ToolCallAction",
tool_name="get_code",
arguments={},
))
"""
from typing import Any, Dict
from openenv.core.env_client import EnvClient, StepResult
try:
from .models import CodeReviewAction, CodeReviewObservation, CodeReviewState
except ImportError:
from models import CodeReviewAction, CodeReviewObservation, CodeReviewState
class CodeReviewEnv(EnvClient[CodeReviewAction, CodeReviewObservation, CodeReviewState]):
"""OpenEnv client for CodeReviewEnv.
Supports MCP tool-calling: ListToolsAction and ToolCallAction.
"""
def _step_payload(self, action: CodeReviewAction) -> Dict[str, Any]:
return action.model_dump(exclude_none=True)
def _parse_result(self, payload: Dict[str, Any]) -> StepResult[CodeReviewObservation]:
obs_data = payload.get("observation", payload.get("data", payload))
observation = CodeReviewObservation(**obs_data)
reward = payload.get("reward", getattr(observation, "reward", 0.0))
done = payload.get("done", getattr(observation, "done", False))
return StepResult(observation=observation, reward=reward, done=done)
def _parse_state(self, payload: Dict[str, Any]) -> CodeReviewState:
state_data = payload.get("data", payload)
return CodeReviewState(**state_data)
# ─── Convenience methods ─────────────────────────────────────────
async def list_tools(self) -> StepResult[CodeReviewObservation]:
"""List available tools."""
return await self.step(CodeReviewAction(action_type="ListToolsAction"))
async def call_tool(self, tool_name: str, **kwargs) -> StepResult[CodeReviewObservation]:
"""Call a tool by name."""
return await self.step(CodeReviewAction(
action_type="ToolCallAction",
tool_name=tool_name,
arguments=kwargs,
))
|