openenv/echo_env

1# SPDX-License-Identifier: BSD-3-Clause2 3"""4Generic environment client that works with raw dictionaries.5 6This module provides a GenericEnvClient that doesn't require installing7environment-specific packages. It's useful for connecting to remote servers8without running any untrusted code locally.9"""10 11from typing import Any, Dict12 13from .client_types import StepResult14from .env_client import EnvClient15 16 17class GenericEnvClient(EnvClient[Dict[str, Any], Dict[str, Any], Dict[str, Any]]):18    """19    Environment client that works with raw dictionaries instead of typed classes.20 21    This client doesn't require installing environment-specific packages, making it22    ideal for:23    - Connecting to remote servers without installing their packages24    - Quick prototyping and testing25    - Environments where type safety isn't needed26    - Security-conscious scenarios where you don't want to run remote code27 28    The trade-off is that you lose type safety and IDE autocomplete for actions29    and observations. Instead of typed objects, you work with plain dictionaries.30 31    Examples:32 33        ```python34        # Direct connection to a running server (no installation needed)35        with GenericEnvClient(base_url="http://localhost:8000") as env:36            result = env.reset()37            result = env.step({"code": "print('hello')"})38            print(result.observation)  # Dict[str, Any]39            print(result.observation.get("output"))40 41        # From local Docker image (chain .sync() for synchronous use)42        env = GenericEnvClient.from_docker_image("coding-env:latest").sync()43        result = env.reset()44        result = env.step({"code": "x = 1 + 2"})45        env.close()46 47        # From HuggingFace Hub (pulls Docker image, no pip install)48        env = GenericEnvClient.from_env("user/my-env", use_docker=True).sync()49        result = env.reset()50        env.close()51        ```52 53    `GenericEnvClient` inherits `from_docker_image()` and `from_env()` from54    `EnvClient`, so you can use it with Docker containers and HuggingFace55    Spaces without any package installation.56    """57 58    def _step_payload(self, action: Dict[str, Any]) -> Dict[str, Any]:59        """60        Convert action to payload for the server.61 62        For GenericEnvClient, this handles both raw dictionaries and63        typed Action objects (Pydantic models). If a Pydantic model is64        passed, it will be converted to a dictionary using model_dump().65 66        Args:67            action (`dict` or `BaseModel`):68                Action as a dictionary or Pydantic `BaseModel`.69 70        Returns:71            The action as a dictionary for the server72        """73        # If it's already a dict, return as-is74        if isinstance(action, dict):75            return action76 77        # If it's a Pydantic model (Action subclass), convert to dict78        if hasattr(action, "model_dump"):79            return action.model_dump()80 81        # Fallback for other objects with __dict__82        if hasattr(action, "__dict__"):83            return vars(action)84 85        # Last resort: try to convert to dict86        return dict(action)87 88    def _parse_result(self, payload: Dict[str, Any]) -> StepResult[Dict[str, Any]]:89        """90        Parse server response into a StepResult.91 92        Extracts the observation, reward, and done fields from the93        server response.94 95        Args:96            payload (`dict`):97                Response payload from the server.98 99        Returns:100            StepResult with observation as a dictionary101        """102        return StepResult(103            observation=payload.get("observation", {}),104            reward=payload.get("reward"),105            done=payload.get("done", False),106            metadata=payload.get("metadata"),107        )108 109    def _parse_state(self, payload: Dict[str, Any]) -> Dict[str, Any]:110        """111        Parse state response from the server.112 113        For GenericEnvClient, this returns the payload as-is since114        we're working with dictionaries.115 116        Args:117            payload (`dict`):118                State payload from the server.119 120        Returns:121            The state as a dictionary122        """123        return payload124 125 126class GenericAction(Dict[str, Any]):127    """128    A dictionary subclass for creating actions when using GenericEnvClient.129 130    This provides a semantic wrapper around dictionaries to make code more131    readable when working with GenericEnvClient. It behaves exactly like a132    dict but signals intent that this is an action for an environment.133 134    Examples:135 136        ```python137        # Without GenericAction (works fine)138        env.step({"code": "print('hello')"})139 140        # With GenericAction (more explicit)141        action = GenericAction(code="print('hello')")142        env.step(action)143 144        # With multiple fields145        action = GenericAction(code="x = 1", timeout=30, metadata={"tag": "test"})146        env.step(action)147        ```148 149    `GenericAction` is just a dict with a constructor that accepts keyword150    arguments. It's provided for symmetry with typed Action classes and151    to make code more readable.152    """153 154    def __init__(self, **kwargs: Any) -> None:155        """156        Create a GenericAction from keyword arguments.157 158        Args:159            **kwargs:160                Action fields as keyword arguments.161 162        Examples:163 164            ```python165            action = GenericAction(code="print(1)", timeout=30)166            action["code"]  # 'print(1)'167            ```168        """169        super().__init__(kwargs)170 171    def __repr__(self) -> str:172        """Return a readable representation."""173        items = ", ".join(f"{k}={v!r}" for k, v in self.items())174        return f"GenericAction({items})"175