+
+
+
+
+
+
+ """.replace('{_generate_action_form_fields(action_fields)}', _generate_action_form_fields(action_fields))
+
+
+def _generate_instructions_section(metadata: Optional[EnvironmentMetadata]) -> str:
+ """Generate the instructions section with environment documentation."""
+ if not metadata or not metadata.readme_content:
+ return ''
+
+ # Convert markdown to HTML (basic conversion)
+ import re
+ html_content = _markdown_to_html(metadata.readme_content)
+
+ return f'''
+
+
+
+
+
+
+
+ HumanAgent Interface
+
+
+
+ {_generate_instructions_section(metadata)}
+
+
+ {_generate_action_interface(action_fields, is_chat_env)}
+
+
+
+
+
+
+
+
+
+
+
+ Current State
+
+
+
+ Status:
+ Not initialized
+
+
+ Episode ID:
+ -
+
+
+ Step Count:
+ 0
+
+
+
+
+ State Observer
+
+
+
+
+
+
+
+
+ Current Observation
+
+ No observation yet
+
+
+
+ Action History
+
+ No actions taken yet
+
+
+
+ '''
+
+
+def _extract_action_fields(action_cls: Type[Action]) -> List[Dict[str, Any]]:
+ """Extract enhanced field metadata from Action class for form generation."""
+ import typing
+ from typing import get_origin, get_args
+
+ action_fields = []
+ if not hasattr(action_cls, '__dataclass_fields__'):
+ return action_fields
+
+ for field_name, field_info in action_cls.__dataclass_fields__.items():
+ if field_name == 'metadata':
+ continue
+
+ field_type = field_info.type
+ field_metadata = _extract_field_metadata(field_name, field_info)
+
+ # Determine input type based on field type
+ input_type = _determine_input_type(field_type)
+
+ # Check if field is required
+ is_required = field_info.default is field_info.default_factory
+
+ action_fields.append({
+ 'name': field_name,
+ 'type': input_type,
+ 'required': is_required,
+ 'description': field_metadata.get('description', ''),
+ 'default_value': field_metadata.get('default_value'),
+ 'choices': field_metadata.get('choices', []),
+ 'min_value': field_metadata.get('min_value'),
+ 'max_value': field_metadata.get('max_value'),
+ 'placeholder': field_metadata.get('placeholder', ''),
+ 'help_text': field_metadata.get('help_text', ''),
+ })
+
+ return action_fields
+
+
+def _extract_field_metadata(field_name: str, field_info) -> Dict[str, Any]:
+ """Extract metadata from dataclass field including docstring and type hints."""
+ import typing
+ from typing import get_origin, get_args, Literal, Union, Optional
+
+ metadata = {}
+
+ # Extract description from field docstring or annotation
+ if hasattr(field_info, 'metadata') and field_info.metadata:
+ # Check for custom metadata
+ for meta in field_info.metadata:
+ if isinstance(meta, dict):
+ metadata.update(meta)
+
+ # Extract type information
+ field_type = field_info.type
+ origin = get_origin(field_type)
+
+ # Handle Literal types for dropdown choices
+ if origin is Literal:
+ args = get_args(field_type)
+ metadata['choices'] = list(args)
+
+ # Handle Optional types
+ if origin is Union:
+ args = get_args(field_type)
+ if len(args) == 2 and type(None) in args:
+ # This is Optional[SomeType]
+ non_none_type = args[0] if args[1] is type(None) else args[1]
+ metadata['optional'] = True
+ # Recursively check the non-None type for choices
+ if get_origin(non_none_type) is Literal:
+ metadata['choices'] = list(get_args(non_none_type))
+ else:
+ # Regular Union type
+ metadata['choices'] = [str(arg) for arg in args if arg is not type(None)]
+
+ # Handle numeric constraints
+ if field_type in (int, float):
+ # Check for common constraint patterns in field name
+ if 'count' in field_name.lower() or 'num' in field_name.lower():
+ metadata['min_value'] = 0
+ if 'id' in field_name.lower():
+ metadata['min_value'] = 0
+
+ # Generate placeholder text
+ if 'message' in field_name.lower():
+ metadata['placeholder'] = f'Enter {field_name.replace("_", " ")}...'
+ elif 'code' in field_name.lower():
+ metadata['placeholder'] = 'Enter Python code here...'
+ elif 'tokens' in field_name.lower():
+ metadata['placeholder'] = 'Enter comma-separated token IDs (e.g., 1,2,3,4,5)'
+ else:
+ metadata['placeholder'] = f'Enter {field_name.replace("_", " ")}...'
+
+ # Generate help text based on field name and type
+ if 'action_id' in field_name.lower():
+ metadata['help_text'] = 'The action ID to execute in the environment'
+ elif 'game_name' in field_name.lower():
+ metadata['help_text'] = 'Name of the game or environment'
+ elif 'tokens' in field_name.lower():
+ metadata['help_text'] = 'Token IDs as a comma-separated list of integers'
+ elif 'code' in field_name.lower():
+ metadata['help_text'] = 'Python code to execute in the environment'
+ elif 'message' in field_name.lower():
+ metadata['help_text'] = 'Text message to send'
+
+ return metadata
+
+
+def _determine_input_type(field_type) -> str:
+ """Determine the appropriate HTML input type for a field type."""
+ import typing
+ from typing import get_origin, get_args, Literal, Union
+
+ # Handle direct types
+ if field_type == str:
+ return "text"
+ elif field_type == int:
+ return "number"
+ elif field_type == float:
+ return "number"
+ elif field_type == bool:
+ return "checkbox"
+
+ # Handle complex types
+ origin = get_origin(field_type)
+
+ if origin is Literal:
+ return "select"
+ elif origin is Union:
+ args = get_args(field_type)
+ if len(args) == 2 and type(None) in args:
+ # Optional type - use the non-None type
+ non_none_type = args[0] if args[1] is type(None) else args[1]
+ return _determine_input_type(non_none_type)
+ elif all(isinstance(arg, str) for arg in args if arg is not type(None)):
+ return "select"
+ else:
+ return "text"
+ elif hasattr(field_type, '__name__') and 'Tensor' in field_type.__name__:
+ return "tensor"
+ else:
+ return "text"
+
+
+def _markdown_to_html(markdown: str) -> str:
+ """Convert basic markdown to HTML for README display."""
+ import html
+ import re
+
+ # Escape HTML first
+ html_content = html.escape(markdown)
+
+ # Convert headers
+ html_content = re.sub(r'^# (.*?)$', r'
+
+ {metadata.name}
+ +
+
+
+ {html_content}
+
+ \1
', html_content, flags=re.MULTILINE) + html_content = re.sub(r'^## (.*?)$', r'\1
', html_content, flags=re.MULTILINE) + html_content = re.sub(r'^### (.*?)$', r'\1
', html_content, flags=re.MULTILINE) + + # Convert code blocks + html_content = re.sub(r'```(.*?)\n(.*?)\n```', r'\2\1', html_content)
+
+ # Convert bold and italic
+ html_content = re.sub(r'\*\*(.*?)\*\*', r'\1', html_content)
+ html_content = re.sub(r'\*(.*?)\*', r'\1', html_content)
+
+ # Convert lists
+ html_content = re.sub(r'^- (.*?)$', r'- \1
') + + return html_content + + +def _generate_action_interface(action_fields: List[Dict[str, Any]], is_chat_env: bool) -> str: + """Generate either a chat interface or action form based on environment type.""" + if is_chat_env: + return _generate_chat_interface() + else: + return _generate_action_form(action_fields) + +def _generate_chat_interface() -> str: + """Generate a chat-style interface for chat environments.""" + return ''' + +
+
+ '''
+
+def _generate_action_form(action_fields: List[Dict[str, Any]]) -> str:
+ """Generate a traditional action form for non-chat environments."""
+ return f'''
+
+ Chat Interface
+ +
+
+
+
+
+
+
+
+
+ '''
+
+def _generate_action_form_fields(action_fields: List[Dict[str, Any]]) -> str:
+ """Generate HTML form fields for action input with enhanced metadata."""
+ if not action_fields:
+ return 'Take Action
+ +No action fields available
' + + fields_html = [] + for field in action_fields: + field_html = _generate_single_field(field) + fields_html.append(field_html) + + return '\n'.join(fields_html) + + +def _generate_single_field(field: Dict[str, Any]) -> str: + """Generate HTML for a single form field with enhanced metadata.""" + field_name = field['name'] + field_type = field['type'] + required = field['required'] + placeholder = field.get('placeholder', '') + help_text = field.get('help_text', '') + choices = field.get('choices', []) + min_value = field.get('min_value') + max_value = field.get('max_value') + default_value = field.get('default_value') + + # Build label with required indicator + label_text = field_name.replace('_', ' ').title() + if required: + label_text += ' *' + + # Build input attributes + input_attrs = [] + if required: + input_attrs.append('required') + if placeholder: + input_attrs.append(f'placeholder="{placeholder}"') + if min_value is not None: + input_attrs.append(f'min="{min_value}"') + if max_value is not None: + input_attrs.append(f'max="{max_value}"') + if default_value is not None: + input_attrs.append(f'value="{default_value}"') + + attrs_str = ' '.join(input_attrs) + + if field_type == 'checkbox': + return f''' +
+
+ {f'{help_text}' if help_text else ''}
+
+ '''
+
+ elif field_type == 'select':
+ options_html = []
+ if not required:
+ options_html.append(f'')
+
+ for choice in choices:
+ selected = 'selected' if str(choice) == str(default_value) else ''
+ options_html.append(f'')
+
+ return f'''
+
+
+
+ {f'{help_text}' if help_text else ''}
+
+ '''
+
+ elif field_type == 'tensor':
+ return f'''
+
+
+
+ {help_text or 'Enter token IDs as comma-separated integers (e.g., 1,2,3,4,5)'}
+
+ '''
+
+ elif field_type == 'text' and ('message' in field_name.lower() or 'code' in field_name.lower()):
+ return f'''
+
+
+
+ {f'{help_text}' if help_text else ''}
+
+ '''
+
+ else:
+ return f'''
+
+
+
+ {f'{help_text}' if help_text else ''}
+
+ '''
diff --git a/src/core/core/http_env_client.py b/src/core/core/http_env_client.py
new file mode 100644
index 0000000000000000000000000000000000000000..7f21a31fcbdf65054ebc70dc1cc6785353f9fa6b
--- /dev/null
+++ b/src/core/core/http_env_client.py
@@ -0,0 +1,246 @@
+"""
+core/runner_env.py
+Minimal HTTP-based environment client.
+- Talks to a single env worker exposing: POST /reset, POST /step
+
+Future hooks (commented below) for:
+- episode_id, seed on reset
+- request_id on step
+- custom headers (auth/trace)
+"""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from typing import Any, Dict, Generic, Optional, Type, TYPE_CHECKING, TypeVar
+
+import requests
+
+from .client_types import StepResult
+from .containers.runtime import LocalDockerProvider
+from .containers.runtime.uv_provider import UVProvider
+
+if TYPE_CHECKING:
+ from .containers.runtime import ContainerProvider
+
+ActT = TypeVar("ActT")
+ObsT = TypeVar("ObsT")
+EnvClientT = TypeVar("EnvClientT", bound="HTTPEnvClient")
+
+
+class HTTPEnvClient(ABC, Generic[ActT, ObsT]):
+ def __init__(
+ self,
+ base_url: str,
+ request_timeout_s: float = 15.0,
+ default_headers: Optional[Dict[str, str]] = None,
+ provider: Optional["ContainerProvider"] = None,
+ ):
+ self._base = base_url.rstrip("/")
+ self._timeout = float(request_timeout_s)
+ self._http = requests.Session()
+ self._headers = default_headers or {}
+ self._provider = provider
+
+ @classmethod
+ def from_docker_image(
+ cls: Type[EnvClientT],
+ image: str,
+ provider: Optional["ContainerProvider"] = None,
+ **kwargs: Any,
+ ) -> EnvClientT:
+ """
+ Create an environment client by spinning up a Docker container locally.
+
+ This is a development utility that:
+ 1. Starts a Docker container from the specified image
+ 2. Waits for the server to be ready
+ 3. Creates and returns a client instance connected to the container
+
+ Note:
+ The caller or a higher-level orchestrator manages the container
+ lifecycle. The container continues running until it is stopped.
+
+ Args:
+ image: Docker image name to run (e.g., "echo-env:latest")
+ provider: Container provider to use (defaults to
+ ``LocalDockerProvider``)
+ **kwargs: Additional arguments passed to
+ ``provider.start_container()`` (e.g., env_vars, port)
+
+ Returns:
+ An instance of the client class connected to the running container
+
+ Example:
+ >>> from envs.coding_env.client import CodingEnv
+ >>> from envs.coding_env.models import CodeAction
+ >>>
+ >>> # Create environment from image
+ >>> env = CodingEnv.from_docker_image("coding-env:latest")
+ >>>
+ >>> # Create environment with custom env vars
+ >>> env = CodingEnv.from_docker_image(
+ ... "coding-env:latest",
+ ... env_vars={"MY_VAR": "value"}
+ ... )
+ >>>
+ >>> # Use the environment
+ >>> result = env.reset()
+ >>> print(result.observation)
+ >>>
+ >>> step_result = env.step(CodeAction(code="print('hello')"))
+ >>> print(step_result.observation.stdout)
+ >>>
+ >>> # Cleanup (optional)
+ >>> env.close()
+ """
+
+ # Use default provider if none provided
+ if provider is None:
+ provider = LocalDockerProvider()
+
+ # 1. Start container with optional kwargs (e.g., env_vars, port)
+ base_url = provider.start_container(image, **kwargs)
+
+ # 2. Wait for server to be ready
+ provider.wait_for_ready(base_url)
+
+ # 3. Create and return client instance with provider reference
+ return cls(base_url=base_url, provider=provider)
+
+ @classmethod
+ def from_hub(
+ cls: Type[EnvClientT],
+ space_id: str,
+ *,
+ use_docker: bool = True,
+ provider: Optional["ContainerProvider"] = None,
+ host: str = "0.0.0.0",
+ port: Optional[int] = None,
+ reload: bool = False,
+ timeout_s: float = 60.0,
+ runner: Optional[UVProvider] = None,
+ project_url: Optional[str] = None,
+ connect_host: Optional[str] = None,
+ extra_env: Optional[Dict[str, str]] = None,
+ **provider_kwargs: Any,
+ ) -> EnvClientT:
+ """Create a client from a Hugging Face Space.
+
+ Set ``use_docker=True`` to launch the registry image with a container
+ provider. The default ``use_docker=False`` runs the Space locally using
+ ``uv run`` through :class:`UVProvider`.
+ """
+
+ if use_docker:
+ if provider is None:
+ provider = LocalDockerProvider()
+
+ tag = provider_kwargs.pop("tag", "latest")
+ image = provider_kwargs.pop(
+ "image",
+ f"registry.hf.space/{space_id.replace('/', '-')}:" f"{tag}",
+ )
+
+ base_url = provider.start_container(image, **provider_kwargs)
+ provider.wait_for_ready(base_url)
+ return cls(base_url=base_url, provider=provider)
+
+ uv_runner = runner or UVProvider(
+ space_id=space_id,
+ host=host,
+ port=port,
+ reload=reload,
+ project_url=project_url,
+ connect_host=connect_host,
+ extra_env=extra_env,
+ )
+
+ base_url = uv_runner.start()
+
+ try:
+ uv_runner.wait_for_ready(timeout_s=timeout_s)
+ except Exception:
+ uv_runner.stop()
+ raise
+
+ return cls(base_url=base_url, provider=uv_runner)
+
+ @abstractmethod
+ def _step_payload(self, action: ActT) -> dict:
+ """Convert an Action object to the JSON body expected by the env server."""
+ raise NotImplementedError
+
+ @abstractmethod
+ def _parse_result(self, payload: dict) -> StepResult[ObsT]:
+ """Convert a JSON response from the env server to StepResult[ObsT]."""
+ raise NotImplementedError
+
+ @abstractmethod
+ def _parse_state(self, payload: dict) -> Any:
+ """Convert a JSON response from the state endpoint to a State object."""
+ raise NotImplementedError
+
+ # ---------- Environment Server Interface Methods ----------
+ def reset(self) -> StepResult[ObsT]:
+ body: Dict[str, Any] = {}
+ # TODO: later:
+ # body["seed"] = seed
+ # body["episode_id"] = episode_id
+ r = self._http.post(
+ f"{self._base}/reset",
+ json=body,
+ headers=self._headers,
+ timeout=self._timeout,
+ )
+ r.raise_for_status()
+ return self._parse_result(r.json())
+
+ def step(self, action: ActT) -> StepResult[ObsT]:
+ body: Dict[str, Any] = {
+ "action": self._step_payload(action),
+ "timeout_s": int(self._timeout),
+ }
+ # TODO: later:
+ # body["request_id"] = str(uuid.uuid4())
+ # body["episode_id"] = current_episode_id
+ r = self._http.post(
+ f"{self._base}/step",
+ json=body,
+ headers=self._headers,
+ timeout=self._timeout,
+ )
+ r.raise_for_status()
+ return self._parse_result(r.json())
+
+ def state(self) -> Any:
+ """
+ Get the current environment state from the server.
+
+ Returns:
+ State object with environment state information (e.g., episode_id, step_count)
+
+ Example:
+ >>> client = EchoEnv.from_docker_image("echo-env:latest")
+ >>> result = client.reset()
+ >>> state = client.state()
+ >>> print(state.episode_id)
+ >>> print(state.step_count)
+ """
+ r = self._http.get(
+ f"{self._base}/state",
+ headers=self._headers,
+ timeout=self._timeout,
+ )
+ r.raise_for_status()
+ return self._parse_state(r.json())
+
+ def close(self) -> None:
+ """
+ Close the environment and clean up resources.
+
+ If this client was created via from_docker_image(), this will stop
+ and remove the associated container.
+ """
+ if self._provider is not None:
+ self._provider.stop_container()
diff --git a/src/core/core/pyproject.toml b/src/core/core/pyproject.toml
new file mode 100644
index 0000000000000000000000000000000000000000..32602f58cb6f066e6874f977f1838483a2ea931b
--- /dev/null
+++ b/src/core/core/pyproject.toml
@@ -0,0 +1,46 @@
+[build-system]
+requires = ["setuptools>=45", "wheel"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "openenv-core"
+version = "0.1.0"
+description = "Core components for OpenEnv - HTTP-based agentic environments"
+readme = "README.md"
+requires-python = ">=3.8"
+license = {text = "BSD-3-Clause"}
+authors = [
+ {name = "Meta Platforms, Inc.", email = "opensource@meta.com"}
+]
+keywords = ["environment", "agent", "http", "docker", "fastapi"]
+
+dependencies = [
+ "requests>=2.25.0",
+ "fastapi>=0.104.0",
+ "uvicorn>=0.24.0",
+]
+
+[project.optional-dependencies]
+dev = [
+ "pytest>=7.0.0",
+ "black>=23.0.0",
+ "ruff>=0.1.0",
+ "mypy>=1.0.0",
+]
+
+[project.urls]
+Homepage = "https://github.com/facebookresearch/OpenEnv"
+Repository = "https://github.com/facebookresearch/OpenEnv"
+Documentation = "https://github.com/facebookresearch/OpenEnv/blob/main/README.md"
+"Bug Tracker" = "https://github.com/facebookresearch/OpenEnv/issues"
+
+[tool.setuptools]
+py-modules = ["openenv_core.__init__", "openenv_core.http_env_client", "openenv_core.client_types"]
+packages = [
+ "openenv_core",
+ "openenv_core.containers",
+ "openenv_core.containers.runtime",
+ "openenv_core.env_server",
+ "openenv_core.tools"
+]
+package-dir = {"openenv_core" = "."}
diff --git a/src/core/core/tools/__init__.py b/src/core/core/tools/__init__.py
new file mode 100644
index 0000000000000000000000000000000000000000..034e7f0686b81bcdda4f30cf1e32b86cd4fa2bf5
--- /dev/null
+++ b/src/core/core/tools/__init__.py
@@ -0,0 +1,16 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+# All rights reserved.
+#
+# This source code is licensed under the BSD-style license found in the
+# LICENSE file in the root directory of this source tree.
+
+"""Core tools for code execution and other utilities."""
+
+from .git_server_client import GitServerClient, RepoInfo
+from .local_python_executor import PyExecutor
+
+__all__ = [
+ "PyExecutor",
+ "GitServerClient",
+ "RepoInfo",
+]
\ No newline at end of file
diff --git a/src/core/core/tools/git_server_client.py b/src/core/core/tools/git_server_client.py
new file mode 100644
index 0000000000000000000000000000000000000000..143bc363b3db278a5cfd09a1b47a736fda8d3bfa
--- /dev/null
+++ b/src/core/core/tools/git_server_client.py
@@ -0,0 +1,362 @@
+#!/usr/bin/env python3
+"""
+Git Server Client for connecting to external Gitea instance.
+
+This module provides a lightweight client for interacting with a shared
+Gitea service, optimized for task-based isolation where multiple environment
+instances share the same Gitea server but have isolated workspaces.
+"""
+
+import json
+import os
+import shutil
+import subprocess
+import time
+from dataclasses import dataclass
+from pathlib import Path
+from urllib.parse import urlparse
+
+
+@dataclass
+class RepoInfo:
+ """Information about a repository."""
+
+ name: str
+ url: str
+ commit: str
+ clone_url: str
+
+
+class GitServerClient:
+ """
+ Client for connecting to an external Gitea server.
+
+ This client is optimized for task-based isolation where:
+ - Multiple tasks share the same Gitea instance
+ - Each task has its own isolated workspace
+ - Fast reset() via git operations (no server restart)
+ - Repos are pre-migrated to Gitea once
+
+ Args:
+ gitea_url: URL of the Gitea server (e.g., "http://gitea:3000")
+ username: Gitea username for authentication
+ password: Gitea password for authentication
+ workspace_dir: Local workspace directory for cloning repos
+
+ Example:
+ >>> # Connect to shared Gitea (credentials from environment)
+ >>> import os
+ >>> client = GitServerClient(
+ ... gitea_url=os.getenv("GITEA_URL"),
+ ... username=os.getenv("GITEA_USERNAME"),
+ ... password=os.getenv("GITEA_PASSWORD")
+ ... )
+ >>> client.wait_for_ready()
+ >>> # Clone repo to workspace
+ >>> path = client.clone_to_workspace("my-repo", commit="abc123")
+ >>> # Fast reset to base state
+ >>> client.reset_workspace("my-repo", commit="abc123")
+ """
+
+ def __init__(
+ self,
+ gitea_url: str,
+ username: str,
+ password: str,
+ workspace_dir: str = "/workspace",
+ ):
+ """Initialize Git Server Client."""
+ self.gitea_url = gitea_url.rstrip("/")
+ self.username = username
+ self.password = password
+ self.workspace_dir = Path(workspace_dir)
+ self.is_ready = False
+
+ # Parse Gitea URL
+ parsed = urlparse(self.gitea_url)
+ self.domain = parsed.hostname or "localhost"
+ self.port = parsed.port or 3000
+
+ # Ensure workspace exists
+ os.makedirs(self.workspace_dir, exist_ok=True)
+
+ # Configure git credentials
+ self._configure_git()
+
+ def _configure_git(self):
+ """Configure git credentials for automatic authentication."""
+ home_dir = Path.home()
+
+ # Git config
+ git_config = f"""[user]
+ name = {self.username}
+ email = {self.username}@local.env
+[init]
+ defaultBranch = main
+[credential]
+ helper = store
+"""
+ gitconfig_path = home_dir / ".gitconfig"
+ gitconfig_path.write_text(git_config)
+
+ # Git credentials
+ git_credentials = f"http://{self.username}:{self.password}@{self.domain}:{self.port}\n"
+ gitcreds_path = home_dir / ".git-credentials"
+ gitcreds_path.write_text(git_credentials)
+ gitcreds_path.chmod(0o600)
+
+ def wait_for_ready(self, timeout: int = 30) -> bool:
+ """
+ Wait for Gitea server to be ready.
+
+ Args:
+ timeout: Maximum seconds to wait
+
+ Returns:
+ True if server is ready, False otherwise
+ """
+ start_time = time.time()
+ while time.time() - start_time < timeout:
+ try:
+ result = subprocess.run(
+ ["curl", "-sf", f"{self.gitea_url}/"],
+ capture_output=True,
+ timeout=5,
+ )
+ if result.returncode == 0:
+ self.is_ready = True
+ return True
+ except subprocess.TimeoutExpired:
+ pass
+ except Exception:
+ pass
+
+ time.sleep(1)
+
+ return False
+
+ def list_repositories(self) -> list[dict[str, str]]:
+ """
+ List all repositories in Gitea.
+
+ Returns:
+ List of repository information dictionaries
+ """
+ if not self.is_ready:
+ raise RuntimeError("Gitea server is not ready")
+
+ result = subprocess.run(
+ [
+ "curl",
+ "-s",
+ f"{self.gitea_url}/api/v1/user/repos",
+ "-u",
+ f"{self.username}:{self.password}",
+ ],
+ capture_output=True,
+ text=True,
+ )
+
+ if result.returncode != 0:
+ return []
+
+ try:
+ repos = json.loads(result.stdout)
+ return [
+ {
+ "name": repo["name"],
+ "full_name": repo["full_name"],
+ "clone_url": repo["clone_url"],
+ "description": repo.get("description", ""),
+ }
+ for repo in repos
+ ]
+ except (json.JSONDecodeError, KeyError):
+ return []
+
+ def clone_to_workspace(
+ self, repo_name: str, target_dir: str | None = None, commit: str = "main"
+ ) -> str:
+ """
+ Clone a repository to the workspace at a specific commit.
+
+ This creates a fresh clone optimized for task isolation.
+
+ Args:
+ repo_name: Name of repository to clone
+ target_dir: Target directory name (defaults to repo_name)
+ commit: Commit hash or branch to check out
+
+ Returns:
+ Path to cloned repository
+
+ Raises:
+ RuntimeError: If clone fails
+ """
+ if not self.is_ready:
+ raise RuntimeError("Gitea server is not ready")
+
+ target_dir = target_dir or repo_name
+ target_path = self.workspace_dir / target_dir
+
+ # Remove existing directory if present
+ if target_path.exists():
+ shutil.rmtree(target_path)
+
+ clone_url = f"{self.gitea_url}/{self.username}/{repo_name}.git"
+
+ # Clone repository
+ result = subprocess.run(
+ ["git", "clone", clone_url, str(target_path)],
+ capture_output=True,
+ text=True,
+ )
+
+ if result.returncode != 0:
+ raise RuntimeError(f"Clone failed: {result.stderr}")
+
+ # Checkout specific commit
+ if commit != "main":
+ result = subprocess.run(
+ ["git", "checkout", commit],
+ cwd=str(target_path),
+ capture_output=True,
+ text=True,
+ )
+
+ if result.returncode != 0:
+ raise RuntimeError(f"Checkout failed: {result.stderr}")
+
+ return str(target_path)
+
+ def reset_workspace(self, repo_name: str, commit: str = "main") -> bool:
+ """
+ Fast reset of workspace to base state (optimized for task resets).
+
+ This is much faster than re-cloning. It:
+ 1. Checks out the target commit
+ 2. Resets to that commit (hard)
+ 3. Cleans untracked files
+
+ Args:
+ repo_name: Name of repository (directory in workspace)
+ commit: Commit hash or branch to reset to
+
+ Returns:
+ True if reset successful
+
+ Raises:
+ RuntimeError: If reset fails
+ """
+ repo_path = self.workspace_dir / repo_name
+
+ if not repo_path.exists():
+ raise RuntimeError(f"Repository not found in workspace: {repo_name}")
+
+ # Fetch latest (in case commit is new)
+ subprocess.run(
+ ["git", "fetch", "--all"],
+ cwd=str(repo_path),
+ capture_output=True,
+ )
+
+ # Checkout and hard reset to commit
+ result = subprocess.run(
+ ["git", "checkout", commit],
+ cwd=str(repo_path),
+ capture_output=True,
+ text=True,
+ )
+
+ if result.returncode != 0:
+ raise RuntimeError(f"Checkout failed: {result.stderr}")
+
+ result = subprocess.run(
+ ["git", "reset", "--hard", f"origin/{commit}" if commit != "main" else commit],
+ cwd=str(repo_path),
+ capture_output=True,
+ text=True,
+ )
+
+ if result.returncode != 0:
+ # Try without origin/ prefix
+ result = subprocess.run(
+ ["git", "reset", "--hard", commit],
+ cwd=str(repo_path),
+ capture_output=True,
+ text=True,
+ )
+ if result.returncode != 0:
+ raise RuntimeError(f"Reset failed: {result.stderr}")
+
+ # Clean untracked files and directories
+ subprocess.run(
+ ["git", "clean", "-fdx"],
+ cwd=str(repo_path),
+ capture_output=True,
+ )
+
+ return True
+
+ def execute_git_command(
+ self, command: str, working_dir: str = ""
+ ) -> tuple[int, str, str]:
+ """
+ Execute a git command in the workspace.
+
+ Args:
+ command: Git command to execute (without 'git' prefix)
+ working_dir: Working directory relative to workspace
+
+ Returns:
+ Tuple of (exit_code, stdout, stderr)
+ """
+ work_path = (
+ self.workspace_dir / working_dir if working_dir else self.workspace_dir
+ )
+
+ if not work_path.exists():
+ return (1, "", f"Working directory does not exist: {work_path}")
+
+ # Split command safely
+ cmd_parts = ["git"] + command.split()
+
+ result = subprocess.run(
+ cmd_parts,
+ cwd=str(work_path),
+ capture_output=True,
+ text=True,
+ )
+
+ return (result.returncode, result.stdout, result.stderr)
+
+ def get_current_commit(self, repo_name: str) -> str:
+ """
+ Get current commit hash of a workspace repository.
+
+ Args:
+ repo_name: Name of repository in workspace
+
+ Returns:
+ Commit hash
+ """
+ repo_path = self.workspace_dir / repo_name
+
+ if not repo_path.exists():
+ raise RuntimeError(f"Repository not found: {repo_name}")
+
+ result = subprocess.run(
+ ["git", "rev-parse", "HEAD"],
+ cwd=str(repo_path),
+ capture_output=True,
+ text=True,
+ )
+
+ if result.returncode != 0:
+ raise RuntimeError(f"Failed to get commit: {result.stderr}")
+
+ return result.stdout.strip()
+
+ def workspace_exists(self, repo_name: str) -> bool:
+ """Check if a repository exists in workspace."""
+ return (self.workspace_dir / repo_name).exists()
diff --git a/src/core/core/tools/local_python_executor.py b/src/core/core/tools/local_python_executor.py
new file mode 100644
index 0000000000000000000000000000000000000000..ba4477d52f3f0f295ec568dc21d807a711ea2cc5
--- /dev/null
+++ b/src/core/core/tools/local_python_executor.py
@@ -0,0 +1,105 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+# All rights reserved.
+#
+# This source code is licensed under the BSD-style license found in the
+# LICENSE file in the root directory of this source tree.
+
+"""
+Local Python Executor.
+
+This module provides functionality for executing Python code locally by wrapping
+the smolagents LocalPythonExecutor.
+"""
+
+from smolagents import LocalPythonExecutor
+
+from core.env_server.types import CodeExecResult
+
+
+class PyExecutor:
+ """
+ Wrapper around smolagents LocalPythonExecutor for executing Python code.
+
+ This class provides a simple interface to execute Python code in a subprocess
+ and capture the results including stdout, stderr, and exit code.
+
+ Args:
+ additional_imports: List of additional module imports to authorize.
+ For example: ["numpy", "pandas", "matplotlib"]
+ These will be added to the base authorized imports.
+
+ Example:
+ >>> # Basic usage with default imports
+ >>> executor = PyExecutor()
+ >>> result = executor.run("print('Hello, World!')")
+ >>> print(result.stdout) # "Hello, World!\n"
+ >>> print(result.exit_code) # 0
+ >>>
+ >>> # Usage with additional imports
+ >>> executor = PyExecutor(additional_imports=["numpy", "pandas"])
+ >>> result = executor.run("import numpy as np\\nprint(np.array([1, 2, 3]))")
+ >>> print(result.stdout) # "[1 2 3]\n"
+ """
+
+ def __init__(self, additional_imports: list[str] | None = None):
+ """
+ Initialize the PyExecutor with a LocalPythonExecutor instance.
+
+ Args:
+ additional_imports: List of additional module names to authorize for import.
+ Defaults to an empty list if not provided.
+ """
+ if additional_imports is None:
+ additional_imports = []
+ self._executor = LocalPythonExecutor(
+ additional_authorized_imports=additional_imports
+ )
+ # Initialize tools to make BASE_PYTHON_TOOLS available (including print)
+ self._executor.send_tools({})
+
+ def run(self, code: str) -> CodeExecResult:
+ """
+ Execute Python code and return the result.
+
+ Args:
+ code: Python code string to execute
+
+ Returns:
+ CodeExecResult containing stdout, stderr, and exit_code
+
+ Example:
+ >>> executor = PyExecutor()
+ >>> result = executor.run("x = 5 + 3\\nprint(x)")
+ >>> print(result.stdout) # "8\n"
+ >>> print(result.exit_code) # 0
+ >>>
+ >>> # Error handling
+ >>> result = executor.run("1 / 0")
+ >>> print(result.exit_code) # 1
+ >>> print(result.stderr) # Contains error message
+ """
+ try:
+ # Execute the code using LocalPythonExecutor
+ # LocalPythonExecutor returns a CodeOutput object with output, logs, is_final_answer
+ exec_result = self._executor(code)
+
+ # Extract the logs (which contain print outputs) as stdout
+ # The output field contains the return value of the code
+ stdout = exec_result.logs
+ stderr = ""
+ exit_code = 0 # Success
+
+ return CodeExecResult(
+ stdout=stdout,
+ stderr=stderr,
+ exit_code=exit_code,
+ )
+
+ except Exception as e:
+ # LocalPythonExecutor raises InterpreterError for various issues
+ # (syntax errors, forbidden operations, runtime errors, etc.)
+ return CodeExecResult(
+ stdout="",
+ stderr=str(e),
+ exit_code=1, # Non-zero indicates error
+ )
diff --git a/src/core/env_server/__init__.py b/src/core/env_server/__init__.py
new file mode 100644
index 0000000000000000000000000000000000000000..79e66535f0e74ae18d84181a234365fca8f3ffc1
--- /dev/null
+++ b/src/core/env_server/__init__.py
@@ -0,0 +1,35 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+# All rights reserved.
+#
+# This source code is licensed under the BSD-style license found in the
+# LICENSE file in the root directory of this source tree.
+
+"""Core environment interfaces and types."""
+
+from .base_transforms import CompositeTransform, NullTransform
+from .http_server import HTTPEnvServer, create_app, create_fastapi_app
+from .interfaces import Environment, Message, ModelTokenizer, Transform
+from .types import Action, Observation, State
+from .web_interface import create_web_interface_app, WebInterfaceManager
+
+__all__ = [
+ # Core interfaces
+ "Environment",
+ "Transform",
+ "Message",
+ "ModelTokenizer",
+ # Types
+ "Action",
+ "Observation",
+ "State",
+ # Base transforms
+ "CompositeTransform",
+ "NullTransform",
+ # HTTP Server
+ "HTTPEnvServer",
+ "create_app",
+ "create_fastapi_app",
+ # Web Interface
+ "create_web_interface_app",
+ "WebInterfaceManager",
+]
diff --git a/src/core/env_server/__pycache__/__init__.cpython-313.pyc b/src/core/env_server/__pycache__/__init__.cpython-313.pyc
new file mode 100644
index 0000000000000000000000000000000000000000..9aa78f3c757381d9be765a5b246d7b9b1fc4497f
Binary files /dev/null and b/src/core/env_server/__pycache__/__init__.cpython-313.pyc differ
diff --git a/src/core/env_server/__pycache__/base_transforms.cpython-313.pyc b/src/core/env_server/__pycache__/base_transforms.cpython-313.pyc
new file mode 100644
index 0000000000000000000000000000000000000000..61982cdddd803e60ffdfa66896d9032c3fa8e570
Binary files /dev/null and b/src/core/env_server/__pycache__/base_transforms.cpython-313.pyc differ
diff --git a/src/core/env_server/__pycache__/http_server.cpython-313.pyc b/src/core/env_server/__pycache__/http_server.cpython-313.pyc
new file mode 100644
index 0000000000000000000000000000000000000000..458e51a9b6ffbd9518a4734b27bff139e182a063
Binary files /dev/null and b/src/core/env_server/__pycache__/http_server.cpython-313.pyc differ
diff --git a/src/core/env_server/__pycache__/interfaces.cpython-313.pyc b/src/core/env_server/__pycache__/interfaces.cpython-313.pyc
new file mode 100644
index 0000000000000000000000000000000000000000..cdbe3d01c42ac5ab35a18ea39dfa56289a5958f3
Binary files /dev/null and b/src/core/env_server/__pycache__/interfaces.cpython-313.pyc differ
diff --git a/src/core/env_server/__pycache__/types.cpython-313.pyc b/src/core/env_server/__pycache__/types.cpython-313.pyc
new file mode 100644
index 0000000000000000000000000000000000000000..50dce0dc82cbdd39e1f9b6633e1acd82450af39a
Binary files /dev/null and b/src/core/env_server/__pycache__/types.cpython-313.pyc differ
diff --git a/src/core/env_server/__pycache__/web_interface.cpython-313.pyc b/src/core/env_server/__pycache__/web_interface.cpython-313.pyc
new file mode 100644
index 0000000000000000000000000000000000000000..b269e6d4dfe878dbb1f6a13e3dd9307ebff59c24
Binary files /dev/null and b/src/core/env_server/__pycache__/web_interface.cpython-313.pyc differ
diff --git a/src/core/env_server/base_transforms.py b/src/core/env_server/base_transforms.py
new file mode 100644
index 0000000000000000000000000000000000000000..d8165e3d77ba23bbd4c765e46cc38bc6c475ad4c
--- /dev/null
+++ b/src/core/env_server/base_transforms.py
@@ -0,0 +1,29 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+# All rights reserved.
+#
+# This source code is licensed under the BSD-style license found in the
+# LICENSE file in the root directory of this source tree.
+
+"""Base transform implementations for composing environment-specific transforms."""
+
+from .interfaces import Transform
+from .types import Observation
+
+
+class CompositeTransform(Transform):
+ """Combines multiple transforms into a single transform."""
+
+ def __init__(self, transforms: list[Transform]):
+ self.transforms = transforms
+
+ def __call__(self, observation: Observation) -> Observation:
+ for transform in self.transforms:
+ observation = transform(observation)
+ return observation
+
+
+class NullTransform(Transform):
+ """Default transform that passes through unchanged."""
+
+ def __call__(self, observation: Observation) -> Observation:
+ return observation
\ No newline at end of file
diff --git a/src/core/env_server/http_server.py b/src/core/env_server/http_server.py
new file mode 100644
index 0000000000000000000000000000000000000000..d18873f0126d8c9198ac24f7308c6a358ccd9435
--- /dev/null
+++ b/src/core/env_server/http_server.py
@@ -0,0 +1,233 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+# All rights reserved.
+#
+# This source code is licensed under the BSD-style license found in the
+# LICENSE file in the root directory of this source tree.
+
+"""
+HTTP server wrapper for Environment instances.
+
+This module provides utilities to wrap any Environment subclass and expose it
+over HTTP endpoints that HTTPEnvClient can consume.
+"""
+
+from __future__ import annotations
+
+import os
+from dataclasses import asdict
+from typing import Any, Dict, Type
+
+from .interfaces import Environment
+from .types import Action, Observation
+from fastapi import Body, FastAPI
+
+class HTTPEnvServer:
+ """
+ HTTP server wrapper for Environment instances.
+
+ This class wraps an Environment and exposes its reset(), step(), and state
+ methods as HTTP endpoints compatible with HTTPEnvClient.
+
+ The server expects:
+ - Action deserialization: Converts JSON dict to Action subclass
+ - Observation serialization: Converts Observation subclass to JSON dict
+
+ Example:
+ >>> from core.env_server import HTTPEnvServer
+ >>> from envs.coding_env.server import CodeExecutionEnvironment
+ >>>
+ >>> env = CodeExecutionEnvironment()
+ >>> server = HTTPEnvServer(env)
+ >>>
+ >>> # Register routes with FastAPI
+ >>> from fastapi import FastAPI
+ >>> app = FastAPI()
+ >>> server.register_routes(app)
+ """
+
+ def __init__(
+ self,
+ env: Environment,
+ action_cls: Type[Action],
+ observation_cls: Type[Observation],
+ ):
+ """
+ Initialize HTTP server wrapper.
+
+ Args:
+ env: The Environment instance to wrap
+ action_cls: The Action subclass this environment expects
+ observation_cls: The Observation subclass this environment returns
+ """
+ self.env = env
+ self.action_cls = action_cls
+ self.observation_cls = observation_cls
+
+ def register_routes(self, app: Any) -> None:
+ """
+ Register HTTP routes on a FastAPI application.
+
+ Args:
+ app: FastAPI application instance
+ """
+
+ if not isinstance(app, FastAPI):
+ raise TypeError("app must be a FastAPI instance")
+
+ @app.post("/reset")
+ async def reset(request: Dict[str, Any] = Body(default={})) -> Dict[str, Any]:
+ """Reset endpoint - returns initial observation."""
+ # TODO: Handle seed, episode_id from request if provided
+ observation = self.env.reset()
+ return self._serialize_observation(observation)
+
+ @app.post("/step")
+ async def step(request: Dict[str, Any]) -> Dict[str, Any]:
+ """Step endpoint - executes action and returns observation."""
+ action_data = request.get("action", {})
+ # TODO: Handle timeout_s, request_id, episode_id from request if provided
+
+ # Deserialize action
+ action = self._deserialize_action(action_data)
+
+ # Execute step
+ observation = self.env.step(action)
+
+ # Return serialized observation
+ return self._serialize_observation(observation)
+
+ @app.get("/state")
+ async def get_state() -> Dict[str, Any]:
+ """State endpoint - returns current environment state."""
+ state = self.env.state
+ return asdict(state)
+
+ @app.get("/health")
+ async def health() -> Dict[str, str]:
+ """Health check endpoint."""
+ return {"status": "healthy"}
+
+
+ def _deserialize_action(self, action_data: Dict[str, Any]) -> Action:
+ """
+ Convert JSON dict to Action instance.
+
+ Args:
+ action_data: Dictionary containing action data
+
+ Returns:
+ Action instance
+
+ Note:
+ This is a simple implementation. Subclasses may need to override
+ for more complex deserialization logic.
+ """
+ # Remove metadata if present (it will be set via kw_only field)
+ metadata = action_data.pop("metadata", {})
+ action = self.action_cls(**action_data)
+ action.metadata = metadata
+ return action
+
+ def _serialize_observation(self, observation: Observation) -> Dict[str, Any]:
+ """
+ Convert Observation instance to JSON-compatible dict.
+
+ Args:
+ observation: Observation instance
+
+ Returns:
+ Dictionary compatible with HTTPEnvClient._parse_result()
+
+ The format matches what HTTPEnvClient expects:
+ {
+ "observation": {...}, # Observation fields
+ "reward": float | None,
+ "done": bool,
+ }
+ """
+ obs_dict = asdict(observation)
+
+ # Extract reward and done (these are part of StepResult on client side)
+ reward = obs_dict.pop("reward", None)
+ done = obs_dict.pop("done", False)
+ obs_dict.pop("metadata", None) # Remove metadata from observation
+
+ # Return in HTTPEnvClient expected format
+ return {
+ "observation": obs_dict,
+ "reward": reward,
+ "done": done,
+ }
+
+def create_app(
+ env: Environment,
+ action_cls: Type[Action],
+ observation_cls: Type[Observation],
+ env_name: Optional[str] = None,
+) -> Any:
+ """
+ Create a FastAPI application with or without web interface.
+
+ This function creates a FastAPI app with the web interface enabled by default,
+ including README integration for better user experience.
+
+ Args:
+ env: The Environment instance to serve
+ action_cls: The Action subclass this environment expects
+ observation_cls: The Observation subclass this environment returns
+ env_name: Optional environment name for README loading
+
+ Returns:
+ FastAPI application instance with or without web interface and README integration
+ """
+ # Check if web interface should be enabled
+ # This can be controlled via environment variable or build argument
+ enable_web = (
+ os.getenv("ENABLE_WEB_INTERFACE", "false").lower() in ("true", "1", "yes")
+ )
+
+ if enable_web:
+ # Import web interface only when needed
+ from .web_interface import create_web_interface_app
+ return create_web_interface_app(env, action_cls, observation_cls, env_name)
+ else:
+ # Use standard FastAPI app without web interface
+ return create_fastapi_app(env, action_cls, observation_cls)
+
+
+def create_fastapi_app(
+ env: Environment,
+ action_cls: Type[Action],
+ observation_cls: Type[Observation],
+) -> Any:
+ """
+ Create a FastAPI application with routes for the given environment.
+
+ Args:
+ env: The Environment instance to serve
+ action_cls: The Action subclass this environment expects
+ observation_cls: The Observation subclass this environment returns
+
+ Returns:
+ FastAPI application instance with routes registered
+
+ Example:
+ >>> from envs.coding_env.server import CodeExecutionEnvironment
+ >>> from envs.coding_env.models import CodeAction, CodeObservation
+ >>>
+ >>> env = CodeExecutionEnvironment()
+ >>> app = create_fastapi_app(env, CodeAction, CodeObservation)
+ >>>
+ >>> # Run with: uvicorn module:app --host 0.0.0.0 --port 8000
+ """
+ try:
+ from fastapi import FastAPI
+ except ImportError:
+ raise ImportError(
+ "FastAPI is required. Install with: pip install fastapi uvicorn"
+ )
+
+ app = FastAPI(title="Environment HTTP Server")
+ server = HTTPEnvServer(env, action_cls, observation_cls)
+ server.register_routes(app)
+ return app
diff --git a/src/core/env_server/interfaces.py b/src/core/env_server/interfaces.py
new file mode 100644
index 0000000000000000000000000000000000000000..caa2d76db1b079f0c15277d1dac2db44e4173ac9
--- /dev/null
+++ b/src/core/env_server/interfaces.py
@@ -0,0 +1,118 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+# All rights reserved.
+#
+# This source code is licensed under the BSD-style license found in the
+# LICENSE file in the root directory of this source tree.
+
+from abc import ABC, abstractmethod
+from typing import Any, Protocol, TypedDict
+
+from .types import Action, Observation, State
+
+
+class Message(TypedDict):
+ """A message in a conversation.
+
+ Compatible with Huggingface chat template format.
+ """
+
+ role: str
+ content: str
+
+
+class ModelTokenizer(Protocol):
+ """Protocol for tokenizers that support chat templates.
+
+ This protocol defines the interface that tokenizers must implement
+ to work with chat-based environments. It's compatible with
+ Huggingface transformers tokenizers.
+ """
+
+ def apply_chat_template(
+ self,
+ conversation: list[Message],
+ tokenize: bool = True,
+ return_tensors: str | None = None,
+ **kwargs: Any,
+ ) -> Any:
+ """Apply a chat template to format and optionally tokenize a conversation.
+
+ Args:
+ conversation: List of message dictionaries with 'role' and 'content'
+ tokenize: Whether to tokenize the output
+ return_tensors: Format for returned tensors ('pt' for PyTorch)
+ **kwargs: Additional arguments
+
+ Returns:
+ Formatted and optionally tokenized conversation
+ """
+ ...
+
+ def decode(
+ self, token_ids: Any, skip_special_tokens: bool = False, **kwargs: Any
+ ) -> str:
+ """Decode token IDs back to text.
+
+ Args:
+ token_ids: Token IDs to decode
+ skip_special_tokens: Whether to skip special tokens in output
+ **kwargs: Additional arguments
+
+ Returns:
+ Decoded text string
+ """
+ ...
+
+
+class Transform(ABC):
+ """Transform observations to add rewards, metrics, or other modifications.
+
+ Transforms follow the TorchRL pattern where they take an observation
+ and return a (potentially modified) observation. This allows for
+ flexible reward computation and observation augmentation.
+ """
+
+ @abstractmethod
+ def __call__(self, observation: Observation) -> Observation:
+ """Transform an observation.
+
+ Args:
+ observation: The input observation
+
+ Returns:
+ The transformed observation
+ """
+ pass
+
+
+class Environment(ABC):
+ """Base class for all environment servers following Gym/Gymnasium API.
+
+ Args:
+ transform: Optional transform to apply to observations
+ """
+
+ def __init__(self, transform: Transform | None = None):
+ self.transform = transform
+
+ @abstractmethod
+ def reset(self) -> Observation:
+ """Reset the environment and return initial observation."""
+ pass
+
+ @abstractmethod
+ def step(self, action: Action) -> Observation:
+ """Take a step in the environment."""
+ pass
+
+ @property
+ @abstractmethod
+ def state(self) -> State:
+ """Get the current environment state."""
+ pass
+
+ def _apply_transform(self, observation: Observation) -> Observation:
+ """Apply transform if one is provided."""
+ if self.transform is not None:
+ return self.transform(observation)
+ return observation
diff --git a/src/core/env_server/types.py b/src/core/env_server/types.py
new file mode 100644
index 0000000000000000000000000000000000000000..70da9f3ca2257ba6c27fb95a58db0a6ec37ccf3e
--- /dev/null
+++ b/src/core/env_server/types.py
@@ -0,0 +1,57 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+# All rights reserved.
+#
+# This source code is licensed under the BSD-style license found in the
+# LICENSE file in the root directory of this source tree.
+
+from dataclasses import dataclass, field
+from typing import Any, Dict, List, Optional, Union
+
+
+# Type aliases
+Scalar = Union[int, float, bool]
+
+
+@dataclass(kw_only=True)
+class Action:
+ """Base class for all environment actions."""
+
+ metadata: Dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass(kw_only=True)
+class Observation:
+ """Base class for all environment observations."""
+
+ done: bool = False
+ reward: Union[bool, int, float, None] = None
+ metadata: Dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass
+class State:
+ """Base class for environment state."""
+
+ episode_id: Optional[str] = None
+ step_count: int = 0
+
+
+@dataclass
+class CodeExecResult:
+ """Result of code execution containing stdout, stderr, and exit code."""
+
+ stdout: str
+ stderr: str
+ exit_code: int
+
+
+@dataclass
+class EnvironmentMetadata:
+ """Metadata about an environment for documentation and UI purposes."""
+
+ name: str
+ description: str
+ readme_content: Optional[str] = None
+ version: Optional[str] = None
+ author: Optional[str] = None
+ documentation_url: Optional[str] = None
diff --git a/src/core/env_server/web_interface.py b/src/core/env_server/web_interface.py
new file mode 100644
index 0000000000000000000000000000000000000000..3c36aa1de336f86161429f8a382054ec89db1912
--- /dev/null
+++ b/src/core/env_server/web_interface.py
@@ -0,0 +1,1613 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+# All rights reserved.
+#
+# This source code is licensed under the BSD-style license found in the
+# LICENSE file in the root directory of this source tree.
+
+"""
+Web interface for OpenEnv environments.
+
+This module provides a web-based interface for interacting with OpenEnv environments,
+including a two-pane layout for HumanAgent interaction and state observation.
+"""
+
+from __future__ import annotations
+
+import json
+import time
+from dataclasses import asdict, dataclass
+from typing import Any, Dict, List, Optional, Type
+from datetime import datetime
+
+from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Request
+from fastapi.responses import HTMLResponse, FileResponse
+from fastapi.staticfiles import StaticFiles
+from pydantic import BaseModel
+
+from .interfaces import Environment
+from .types import Action, Observation, State, EnvironmentMetadata
+
+
+def load_environment_metadata(env: Environment, env_name: Optional[str] = None) -> EnvironmentMetadata:
+ """
+ Load environment metadata including README content.
+
+ Args:
+ env: The environment instance
+ env_name: Optional environment name for README file lookup
+
+ Returns:
+ EnvironmentMetadata with loaded information
+ """
+ # Try to get metadata from environment if it has a method for it
+ if hasattr(env, 'get_metadata'):
+ return env.get_metadata()
+
+ # Default metadata
+ metadata = EnvironmentMetadata(
+ name=env_name or env.__class__.__name__,
+ description=f"{env.__class__.__name__} environment",
+ version="1.0.0"
+ )
+
+ # Try to load README from file system
+ readme_content = _load_readme_from_filesystem(env_name)
+ if readme_content:
+ metadata.readme_content = readme_content
+
+ return metadata
+
+
+def _load_readme_from_filesystem(env_name: Optional[str]) -> Optional[str]:
+ """
+ Load README content from the filesystem.
+
+ Tries multiple locations:
+ 1. Container filesystem: /app/README.md
+ 2. Local development: src/envs/{env_name}/README.md
+ 3. Environment variable: ENV_README_PATH
+ """
+ import os
+ from pathlib import Path
+
+ # Try container filesystem first
+ container_readme = Path("/app/README.md")
+ if container_readme.exists():
+ try:
+ return container_readme.read_text(encoding='utf-8')
+ except Exception:
+ pass
+
+ # Try environment variable path
+ custom_path = os.environ.get("ENV_README_PATH")
+ if custom_path and Path(custom_path).exists():
+ try:
+ return Path(custom_path).read_text(encoding='utf-8')
+ except Exception:
+ pass
+
+ # Try local development path
+ if env_name:
+ local_readme = Path(f"src/envs/{env_name}/README.md")
+ if local_readme.exists():
+ try:
+ return local_readme.read_text(encoding='utf-8')
+ except Exception:
+ pass
+
+ return None
+
+
+@dataclass
+class ActionLog:
+ """Log entry for an action taken."""
+ timestamp: str
+ action: Dict[str, Any]
+ observation: Dict[str, Any]
+ reward: Optional[float]
+ done: bool
+ step_count: int
+
+
+@dataclass
+class EpisodeState:
+ """Current episode state for the web interface."""
+ episode_id: Optional[str]
+ step_count: int
+ current_observation: Optional[Dict[str, Any]]
+ action_logs: List[ActionLog]
+ is_reset: bool = True
+
+
+class WebInterfaceManager:
+ """Manages the web interface for an environment."""
+
+ def __init__(
+ self,
+ env: Environment,
+ action_cls: Type[Action],
+ observation_cls: Type[Observation],
+ metadata: Optional[EnvironmentMetadata] = None,
+ ):
+ self.env = env
+ self.action_cls = action_cls
+ self.observation_cls = observation_cls
+ self.metadata = metadata or EnvironmentMetadata(
+ name=env.__class__.__name__,
+ description=f"{env.__class__.__name__} environment"
+ )
+ self.episode_state = EpisodeState(
+ episode_id=None,
+ step_count=0,
+ current_observation=None,
+ action_logs=[]
+ )
+ self.connected_clients: List[WebSocket] = []
+
+ async def connect_websocket(self, websocket: WebSocket):
+ """Connect a new WebSocket client."""
+ await websocket.accept()
+ self.connected_clients.append(websocket)
+
+ # Send current state to the new client
+ await self._send_state_update()
+
+ async def disconnect_websocket(self, websocket: WebSocket):
+ """Disconnect a WebSocket client."""
+ if websocket in self.connected_clients:
+ self.connected_clients.remove(websocket)
+
+ async def _send_state_update(self):
+ """Send current state to all connected clients."""
+ if not self.connected_clients:
+ return
+
+ state_data = {
+ "type": "state_update",
+ "episode_state": asdict(self.episode_state)
+ }
+
+ # Send to all connected clients
+ disconnected_clients = []
+ for client in self.connected_clients:
+ try:
+ await client.send_text(json.dumps(state_data))
+ except:
+ disconnected_clients.append(client)
+
+ # Remove disconnected clients
+ for client in disconnected_clients:
+ self.connected_clients.remove(client)
+
+ async def reset_environment(self) -> Dict[str, Any]:
+ """Reset the environment and update state."""
+ observation = self.env.reset()
+ state = self.env.state
+
+ # Update episode state
+ self.episode_state.episode_id = state.episode_id
+ self.episode_state.step_count = 0
+ self.episode_state.current_observation = asdict(observation)
+ self.episode_state.action_logs = []
+ self.episode_state.is_reset = True
+
+ # Send state update
+ await self._send_state_update()
+
+ return {
+ "observation": asdict(observation),
+ "reward": observation.reward,
+ "done": observation.done,
+ }
+
+ async def step_environment(self, action_data: Dict[str, Any]) -> Dict[str, Any]:
+ """Execute a step in the environment and update state."""
+ # Deserialize action
+ action = self._deserialize_action(action_data)
+
+ # Execute step
+ observation = self.env.step(action)
+ state = self.env.state
+
+ # Create action log
+ action_log = ActionLog(
+ timestamp=datetime.now().isoformat(),
+ action=asdict(action),
+ observation=asdict(observation),
+ reward=observation.reward,
+ done=observation.done,
+ step_count=state.step_count
+ )
+
+ # Update episode state
+ self.episode_state.episode_id = state.episode_id
+ self.episode_state.step_count = state.step_count
+ self.episode_state.current_observation = asdict(observation)
+ self.episode_state.action_logs.append(action_log)
+ self.episode_state.is_reset = False
+
+ # Send state update
+ await self._send_state_update()
+
+ return {
+ "observation": asdict(observation),
+ "reward": observation.reward,
+ "done": observation.done,
+ }
+
+ def get_state(self) -> Dict[str, Any]:
+ """Get current environment state."""
+ state = self.env.state
+ return asdict(state)
+
+ def _deserialize_action(self, action_data: Dict[str, Any]) -> Action:
+ """Convert JSON dict to Action instance."""
+ metadata = action_data.pop("metadata", {})
+
+ # Handle tensor fields that come from JSON as lists
+ processed_data = {}
+ for key, value in action_data.items():
+ if key == "tokens" and isinstance(value, (list, str)):
+ # Convert list or string to tensor
+ if isinstance(value, str):
+ # If it's a string, try to parse it as a list of numbers
+ try:
+ import json
+ value = json.loads(value)
+ except:
+ # If parsing fails, treat as empty list
+ value = []
+ if isinstance(value, list):
+ import torch
+ processed_data[key] = torch.tensor(value, dtype=torch.long)
+ else:
+ processed_data[key] = value
+ elif key == "action_id" and isinstance(value, str):
+ # Convert action_id from string to int
+ try:
+ processed_data[key] = int(value)
+ except ValueError:
+ # If conversion fails, keep original value
+ processed_data[key] = value
+ else:
+ processed_data[key] = value
+
+ action = self.action_cls(**processed_data)
+ action.metadata = metadata
+ return action
+
+
+def create_web_interface_app(
+ env: Environment,
+ action_cls: Type[Action],
+ observation_cls: Type[Observation],
+ env_name: Optional[str] = None,
+) -> FastAPI:
+ """
+ Create a FastAPI application with web interface for the given environment.
+
+ Args:
+ env: The Environment instance to serve
+ action_cls: The Action subclass this environment expects
+ observation_cls: The Observation subclass this environment returns
+ env_name: Optional environment name for README loading
+
+ Returns:
+ FastAPI application instance with web interface
+ """
+ from .http_server import create_fastapi_app
+
+ # Create the base environment app
+ app = create_fastapi_app(env, action_cls, observation_cls)
+
+ # Load environment metadata
+ metadata = load_environment_metadata(env, env_name)
+
+ # Create web interface manager
+ web_manager = WebInterfaceManager(env, action_cls, observation_cls, metadata)
+
+ # Add web interface routes
+ @app.get("/web", response_class=HTMLResponse)
+ async def web_interface():
+ """Serve the web interface."""
+ return get_web_interface_html(action_cls, web_manager.metadata)
+
+ @app.get("/web/metadata")
+ async def web_metadata():
+ """Get environment metadata."""
+ return asdict(web_manager.metadata)
+
+ @app.websocket("/ws")
+ async def websocket_endpoint(websocket: WebSocket):
+ """WebSocket endpoint for real-time updates."""
+ await web_manager.connect_websocket(websocket)
+ try:
+ while True:
+ # Keep connection alive
+ await websocket.receive_text()
+ except WebSocketDisconnect:
+ await web_manager.disconnect_websocket(websocket)
+
+ @app.post("/web/reset")
+ async def web_reset():
+ """Reset endpoint for web interface."""
+ return await web_manager.reset_environment()
+
+ @app.post("/web/step")
+ async def web_step(request: Dict[str, Any]):
+ """Step endpoint for web interface."""
+ # Check if this is a message-based request (chat environment)
+ if "message" in request:
+ message = request["message"]
+ # Convert message to action using the environment's message_to_action method
+ action = web_manager.env.message_to_action(message)
+ action_data = {"tokens": action.tokens.tolist()}
+ else:
+ action_data = request.get("action", {})
+
+ return await web_manager.step_environment(action_data)
+
+ @app.get("/web/state")
+ async def web_state():
+ """State endpoint for web interface."""
+ return web_manager.get_state()
+
+ return app
+
+
+def get_web_interface_html(action_cls: Type[Action], metadata: Optional[EnvironmentMetadata] = None) -> str:
+ """Generate the HTML for the web interface."""
+
+ # Check if this is a chat environment by looking for tokens field
+ is_chat_env = False
+ if hasattr(action_cls, '__dataclass_fields__'):
+ for field_name, field_info in action_cls.__dataclass_fields__.items():
+ if field_name == 'tokens' and hasattr(field_info.type, '__name__') and 'Tensor' in field_info.type.__name__:
+ is_chat_env = True
+ break
+
+ # Get action fields for dynamic form generation with enhanced metadata
+ action_fields = _extract_action_fields(action_cls)
+
+ return f"""
+
+
+
+
+
+
+
+
+
+
+
+
+ """.replace('{_generate_action_form_fields(action_fields)}', _generate_action_form_fields(action_fields))
+
+
+def _generate_instructions_section(metadata: Optional[EnvironmentMetadata]) -> str:
+ """Generate the instructions section with environment documentation."""
+ if not metadata or not metadata.readme_content:
+ return ''
+
+ # Convert markdown to HTML (basic conversion)
+ import re
+ html_content = _markdown_to_html(metadata.readme_content)
+
+ return f'''
+
+
+
+
+
+
+
+ HumanAgent Interface
+
+
+
+ {_generate_instructions_section(metadata)}
+
+
+ {_generate_action_interface(action_fields, is_chat_env)}
+
+
+
+
+
+
+
+
+
+
+
+ Current State
+
+
+
+ Status:
+ Not initialized
+
+
+ Episode ID:
+ -
+
+
+ Step Count:
+ 0
+
+
+
+
+ State Observer
+
+
+
+
+
+
+
+
+ Current Observation
+
+ No observation yet
+
+
+
+ Action History
+
+ No actions taken yet
+
+
+
+ '''
+
+
+def _extract_action_fields(action_cls: Type[Action]) -> List[Dict[str, Any]]:
+ """Extract enhanced field metadata from Action class for form generation."""
+ import typing
+ from typing import get_origin, get_args
+
+ action_fields = []
+ if not hasattr(action_cls, '__dataclass_fields__'):
+ return action_fields
+
+ for field_name, field_info in action_cls.__dataclass_fields__.items():
+ if field_name == 'metadata':
+ continue
+
+ field_type = field_info.type
+ field_metadata = _extract_field_metadata(field_name, field_info)
+
+ # Determine input type based on field type
+ input_type = _determine_input_type(field_type)
+
+ # Check if field is required
+ is_required = field_info.default is field_info.default_factory
+
+ action_fields.append({
+ 'name': field_name,
+ 'type': input_type,
+ 'required': is_required,
+ 'description': field_metadata.get('description', ''),
+ 'default_value': field_metadata.get('default_value'),
+ 'choices': field_metadata.get('choices', []),
+ 'min_value': field_metadata.get('min_value'),
+ 'max_value': field_metadata.get('max_value'),
+ 'placeholder': field_metadata.get('placeholder', ''),
+ 'help_text': field_metadata.get('help_text', ''),
+ })
+
+ return action_fields
+
+
+def _extract_field_metadata(field_name: str, field_info) -> Dict[str, Any]:
+ """Extract metadata from dataclass field including docstring and type hints."""
+ import typing
+ from typing import get_origin, get_args, Literal, Union, Optional
+
+ metadata = {}
+
+ # Extract description from field docstring or annotation
+ if hasattr(field_info, 'metadata') and field_info.metadata:
+ # Check for custom metadata
+ for meta in field_info.metadata:
+ if isinstance(meta, dict):
+ metadata.update(meta)
+
+ # Extract type information
+ field_type = field_info.type
+ origin = get_origin(field_type)
+
+ # Handle Literal types for dropdown choices
+ if origin is Literal:
+ args = get_args(field_type)
+ metadata['choices'] = list(args)
+
+ # Handle Optional types
+ if origin is Union:
+ args = get_args(field_type)
+ if len(args) == 2 and type(None) in args:
+ # This is Optional[SomeType]
+ non_none_type = args[0] if args[1] is type(None) else args[1]
+ metadata['optional'] = True
+ # Recursively check the non-None type for choices
+ if get_origin(non_none_type) is Literal:
+ metadata['choices'] = list(get_args(non_none_type))
+ else:
+ # Regular Union type
+ metadata['choices'] = [str(arg) for arg in args if arg is not type(None)]
+
+ # Handle numeric constraints
+ if field_type in (int, float):
+ # Check for common constraint patterns in field name
+ if 'count' in field_name.lower() or 'num' in field_name.lower():
+ metadata['min_value'] = 0
+ if 'id' in field_name.lower():
+ metadata['min_value'] = 0
+
+ # Generate placeholder text
+ if 'message' in field_name.lower():
+ metadata['placeholder'] = f'Enter {field_name.replace("_", " ")}...'
+ elif 'code' in field_name.lower():
+ metadata['placeholder'] = 'Enter Python code here...'
+ elif 'tokens' in field_name.lower():
+ metadata['placeholder'] = 'Enter comma-separated token IDs (e.g., 1,2,3,4,5)'
+ else:
+ metadata['placeholder'] = f'Enter {field_name.replace("_", " ")}...'
+
+ # Generate help text based on field name and type
+ if 'action_id' in field_name.lower():
+ metadata['help_text'] = 'The action ID to execute in the environment'
+ elif 'game_name' in field_name.lower():
+ metadata['help_text'] = 'Name of the game or environment'
+ elif 'tokens' in field_name.lower():
+ metadata['help_text'] = 'Token IDs as a comma-separated list of integers'
+ elif 'code' in field_name.lower():
+ metadata['help_text'] = 'Python code to execute in the environment'
+ elif 'message' in field_name.lower():
+ metadata['help_text'] = 'Text message to send'
+
+ return metadata
+
+
+def _determine_input_type(field_type) -> str:
+ """Determine the appropriate HTML input type for a field type."""
+ import typing
+ from typing import get_origin, get_args, Literal, Union
+
+ # Handle direct types
+ if field_type == str:
+ return "text"
+ elif field_type == int:
+ return "number"
+ elif field_type == float:
+ return "number"
+ elif field_type == bool:
+ return "checkbox"
+
+ # Handle complex types
+ origin = get_origin(field_type)
+
+ if origin is Literal:
+ return "select"
+ elif origin is Union:
+ args = get_args(field_type)
+ if len(args) == 2 and type(None) in args:
+ # Optional type - use the non-None type
+ non_none_type = args[0] if args[1] is type(None) else args[1]
+ return _determine_input_type(non_none_type)
+ elif all(isinstance(arg, str) for arg in args if arg is not type(None)):
+ return "select"
+ else:
+ return "text"
+ elif hasattr(field_type, '__name__') and 'Tensor' in field_type.__name__:
+ return "tensor"
+ else:
+ return "text"
+
+
+def _markdown_to_html(markdown: str) -> str:
+ """Convert basic markdown to HTML for README display."""
+ import html
+ import re
+
+ # Escape HTML first
+ html_content = html.escape(markdown)
+
+ # Convert headers
+ html_content = re.sub(r'^# (.*?)$', r'
+
+ {metadata.name}
+ +
+
+
+ {html_content}
+
+ \1
', html_content, flags=re.MULTILINE) + html_content = re.sub(r'^## (.*?)$', r'\1
', html_content, flags=re.MULTILINE) + html_content = re.sub(r'^### (.*?)$', r'\1
', html_content, flags=re.MULTILINE) + + # Convert code blocks + html_content = re.sub(r'```(.*?)\n(.*?)\n```', r'\2\1', html_content)
+
+ # Convert bold and italic
+ html_content = re.sub(r'\*\*(.*?)\*\*', r'\1', html_content)
+ html_content = re.sub(r'\*(.*?)\*', r'\1', html_content)
+
+ # Convert lists
+ html_content = re.sub(r'^- (.*?)$', r'- \1
') + + return html_content + + +def _generate_action_interface(action_fields: List[Dict[str, Any]], is_chat_env: bool) -> str: + """Generate either a chat interface or action form based on environment type.""" + if is_chat_env: + return _generate_chat_interface() + else: + return _generate_action_form(action_fields) + +def _generate_chat_interface() -> str: + """Generate a chat-style interface for chat environments.""" + return ''' + +
+
+ '''
+
+def _generate_action_form(action_fields: List[Dict[str, Any]]) -> str:
+ """Generate a traditional action form for non-chat environments."""
+ return f'''
+
+ Chat Interface
+ +
+
+
+
+
+
+
+
+
+ '''
+
+def _generate_action_form_fields(action_fields: List[Dict[str, Any]]) -> str:
+ """Generate HTML form fields for action input with enhanced metadata."""
+ if not action_fields:
+ return 'Take Action
+ +No action fields available
' + + fields_html = [] + for field in action_fields: + field_html = _generate_single_field(field) + fields_html.append(field_html) + + return '\n'.join(fields_html) + + +def _generate_single_field(field: Dict[str, Any]) -> str: + """Generate HTML for a single form field with enhanced metadata.""" + field_name = field['name'] + field_type = field['type'] + required = field['required'] + placeholder = field.get('placeholder', '') + help_text = field.get('help_text', '') + choices = field.get('choices', []) + min_value = field.get('min_value') + max_value = field.get('max_value') + default_value = field.get('default_value') + + # Build label with required indicator + label_text = field_name.replace('_', ' ').title() + if required: + label_text += ' *' + + # Build input attributes + input_attrs = [] + if required: + input_attrs.append('required') + if placeholder: + input_attrs.append(f'placeholder="{placeholder}"') + if min_value is not None: + input_attrs.append(f'min="{min_value}"') + if max_value is not None: + input_attrs.append(f'max="{max_value}"') + if default_value is not None: + input_attrs.append(f'value="{default_value}"') + + attrs_str = ' '.join(input_attrs) + + if field_type == 'checkbox': + return f''' +
+
+ {f'{help_text}' if help_text else ''}
+
+ '''
+
+ elif field_type == 'select':
+ options_html = []
+ if not required:
+ options_html.append(f'')
+
+ for choice in choices:
+ selected = 'selected' if str(choice) == str(default_value) else ''
+ options_html.append(f'')
+
+ return f'''
+
+
+
+ {f'{help_text}' if help_text else ''}
+
+ '''
+
+ elif field_type == 'tensor':
+ return f'''
+
+
+
+ {help_text or 'Enter token IDs as comma-separated integers (e.g., 1,2,3,4,5)'}
+
+ '''
+
+ elif field_type == 'text' and ('message' in field_name.lower() or 'code' in field_name.lower()):
+ return f'''
+
+
+
+ {f'{help_text}' if help_text else ''}
+
+ '''
+
+ else:
+ return f'''
+
+
+
+ {f'{help_text}' if help_text else ''}
+
+ '''
diff --git a/src/core/http_env_client.py b/src/core/http_env_client.py
new file mode 100644
index 0000000000000000000000000000000000000000..dce46c8c5d6487792d6405f234bc71f35eef211c
--- /dev/null
+++ b/src/core/http_env_client.py
@@ -0,0 +1,250 @@
+"""
+core/runner_env.py
+Minimal HTTP-based environment client.
+- Talks to a single env worker exposing: POST /reset, POST /step
+
+Future hooks (commented below) for:
+- episode_id, seed on reset
+- request_id on step
+- custom headers (auth/trace)
+"""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from typing import Any, Dict, Generic, Optional, Type, TYPE_CHECKING, TypeVar
+
+import requests
+
+from .client_types import StepResult
+from .containers.runtime.uv_provider import UVProvider
+from .containers.runtime import LocalDockerProvider
+
+if TYPE_CHECKING:
+ from .containers.runtime import ContainerProvider
+
+ActT = TypeVar("ActT")
+ObsT = TypeVar("ObsT")
+EnvClientT = TypeVar("EnvClientT", bound="HTTPEnvClient")
+
+
+class HTTPEnvClient(ABC, Generic[ActT, ObsT]):
+ def __init__(
+ self,
+ base_url: str,
+ request_timeout_s: float = 15.0,
+ default_headers: Optional[Dict[str, str]] = None,
+ provider: Optional["ContainerProvider"] = None,
+ ):
+ self._base = base_url.rstrip("/")
+ self._timeout = float(request_timeout_s)
+ self._http = requests.Session()
+ self._headers = default_headers or {}
+ self._provider = provider
+
+ @classmethod
+ def from_docker_image(
+ cls: Type[EnvClientT],
+ image: str,
+ provider: Optional["ContainerProvider"] = None,
+ **kwargs: Any,
+ ) -> EnvClientT:
+ """
+ Create an environment client by spinning up a Docker container locally.
+
+ This is a development utility that:
+ 1. Starts a Docker container from the specified image
+ 2. Waits for the server to be ready
+ 3. Creates and returns a client instance connected to the container
+
+ Note:
+ The caller or a higher-level orchestrator manages the container
+ lifecycle. The container continues running until it is stopped.
+
+ Args:
+ image: Docker image name to run (e.g., "echo-env:latest")
+ provider: Container provider to use (defaults to
+ ``LocalDockerProvider``)
+ **kwargs: Additional arguments passed to
+ ``provider.start_container()`` (e.g., env_vars, port)
+
+ Returns:
+ An instance of the client class connected to the running container
+
+ Example:
+ >>> from envs.coding_env.client import CodingEnv
+ >>> from envs.coding_env.models import CodeAction
+ >>>
+ >>> # Create environment from image
+ >>> env = CodingEnv.from_docker_image("coding-env:latest")
+ >>>
+ >>> # Create environment with custom env vars
+ >>> env = CodingEnv.from_docker_image(
+ ... "coding-env:latest",
+ ... env_vars={"MY_VAR": "value"}
+ ... )
+ >>>
+ >>> # Use the environment
+ >>> result = env.reset()
+ >>> print(result.observation)
+ >>>
+ >>> step_result = env.step(CodeAction(code="print('hello')"))
+ >>> print(step_result.observation.stdout)
+ >>>
+ >>> # Cleanup (optional)
+ >>> env.close()
+ """
+
+ # Use default provider if none provided
+ if provider is None:
+ provider = LocalDockerProvider()
+
+ # 1. Start container with optional kwargs (e.g., env_vars, port)
+ base_url = provider.start_container(image, **kwargs)
+
+ # 2. Wait for server to be ready
+ provider.wait_for_ready(base_url)
+
+ # 3. Create and return client instance with provider reference
+ return cls(base_url=base_url, provider=provider)
+
+ @classmethod
+ def from_hub(
+ cls: Type[EnvClientT],
+ space_id: str,
+ *,
+ use_docker: bool = False,
+ provider: Optional["ContainerProvider"] = None,
+ host: str = "0.0.0.0",
+ port: Optional[int] = None,
+ reload: bool = False,
+ timeout_s: float = 60.0,
+ runner: Optional[UVProvider] = None,
+ project_url: Optional[str] = None,
+ connect_host: Optional[str] = None,
+ extra_env: Optional[Dict[str, str]] = None,
+ **provider_kwargs: Any,
+ ) -> EnvClientT:
+ """Create a client from a Hugging Face Space."""
+
+ if use_docker:
+ if provider is None:
+ provider = LocalDockerProvider()
+
+ tag = provider_kwargs.pop("tag", "latest")
+ image = provider_kwargs.pop(
+ "image",
+ f"registry.hf.space/{space_id.replace('/', '-')}:{tag}",
+ )
+
+ base_url = provider.start_container(image, **provider_kwargs)
+ provider.wait_for_ready(base_url, timeout_s=timeout_s)
+ return cls(base_url=base_url, provider=provider)
+
+ uv_runner = runner or UVProvider(
+ space_id=space_id,
+ host=host,
+ port=port,
+ reload=reload,
+ project_url=project_url,
+ connect_host=connect_host,
+ extra_env=extra_env,
+ )
+
+ non_docker_kwargs = dict(provider_kwargs)
+ env_vars = non_docker_kwargs.pop("env_vars", None)
+
+ base_url = uv_runner.start_container(
+ space_id,
+ port=port,
+ env_vars=env_vars,
+ **non_docker_kwargs,
+ )
+
+ try:
+ uv_runner.wait_for_ready(base_url, timeout_s=timeout_s)
+ except Exception:
+ uv_runner.stop_container()
+ raise
+
+ return cls(base_url=base_url, provider=uv_runner)
+
+ @abstractmethod
+ def _step_payload(self, action: ActT) -> dict:
+ """Convert an action to the JSON payload expected by the server."""
+ raise NotImplementedError
+
+ @abstractmethod
+ def _parse_result(self, payload: dict) -> StepResult[ObsT]:
+ """Convert a JSON response into :class:`StepResult`."""
+ raise NotImplementedError
+
+ @abstractmethod
+ def _parse_state(self, payload: dict) -> Any:
+ """Convert state JSON into a :class:`State` object."""
+ raise NotImplementedError
+
+ # ---------- Environment Server Interface Methods ----------
+ def reset(self) -> StepResult[ObsT]:
+ body: Dict[str, Any] = {}
+ # TODO: later:
+ # body["seed"] = seed
+ # body["episode_id"] = episode_id
+ r = self._http.post(
+ f"{self._base}/reset",
+ json=body,
+ headers=self._headers,
+ timeout=self._timeout,
+ )
+ r.raise_for_status()
+ return self._parse_result(r.json())
+
+ def step(self, action: ActT) -> StepResult[ObsT]:
+ body: Dict[str, Any] = {
+ "action": self._step_payload(action),
+ "timeout_s": int(self._timeout),
+ }
+ # TODO: later:
+ # body["request_id"] = str(uuid.uuid4())
+ # body["episode_id"] = current_episode_id
+ r = self._http.post(
+ f"{self._base}/step",
+ json=body,
+ headers=self._headers,
+ timeout=self._timeout,
+ )
+ r.raise_for_status()
+ return self._parse_result(r.json())
+
+ def state(self) -> Any:
+ """
+ Get the current environment state from the server.
+
+ Returns:
+ State object with environment state information (e.g.,
+ episode_id, step_count)
+
+ Example:
+ >>> client = EchoEnv.from_docker_image("echo-env:latest")
+ >>> result = client.reset()
+ >>> state = client.state()
+ >>> print(state.episode_id)
+ >>> print(state.step_count)
+ """
+ r = self._http.get(
+ f"{self._base}/state",
+ headers=self._headers,
+ timeout=self._timeout,
+ )
+ r.raise_for_status()
+ return self._parse_state(r.json())
+
+ def close(self) -> None:
+ """
+ Close the environment and clean up resources.
+
+ If this client was created via from_docker_image(), this will stop
+ and remove the associated container.
+ """
+ if self._provider is not None:
+ self._provider.stop_container()
diff --git a/src/core/pyproject.toml b/src/core/pyproject.toml
new file mode 100644
index 0000000000000000000000000000000000000000..32602f58cb6f066e6874f977f1838483a2ea931b
--- /dev/null
+++ b/src/core/pyproject.toml
@@ -0,0 +1,46 @@
+[build-system]
+requires = ["setuptools>=45", "wheel"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "openenv-core"
+version = "0.1.0"
+description = "Core components for OpenEnv - HTTP-based agentic environments"
+readme = "README.md"
+requires-python = ">=3.8"
+license = {text = "BSD-3-Clause"}
+authors = [
+ {name = "Meta Platforms, Inc.", email = "opensource@meta.com"}
+]
+keywords = ["environment", "agent", "http", "docker", "fastapi"]
+
+dependencies = [
+ "requests>=2.25.0",
+ "fastapi>=0.104.0",
+ "uvicorn>=0.24.0",
+]
+
+[project.optional-dependencies]
+dev = [
+ "pytest>=7.0.0",
+ "black>=23.0.0",
+ "ruff>=0.1.0",
+ "mypy>=1.0.0",
+]
+
+[project.urls]
+Homepage = "https://github.com/facebookresearch/OpenEnv"
+Repository = "https://github.com/facebookresearch/OpenEnv"
+Documentation = "https://github.com/facebookresearch/OpenEnv/blob/main/README.md"
+"Bug Tracker" = "https://github.com/facebookresearch/OpenEnv/issues"
+
+[tool.setuptools]
+py-modules = ["openenv_core.__init__", "openenv_core.http_env_client", "openenv_core.client_types"]
+packages = [
+ "openenv_core",
+ "openenv_core.containers",
+ "openenv_core.containers.runtime",
+ "openenv_core.env_server",
+ "openenv_core.tools"
+]
+package-dir = {"openenv_core" = "."}
diff --git a/src/core/tools/__init__.py b/src/core/tools/__init__.py
new file mode 100644
index 0000000000000000000000000000000000000000..034e7f0686b81bcdda4f30cf1e32b86cd4fa2bf5
--- /dev/null
+++ b/src/core/tools/__init__.py
@@ -0,0 +1,16 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+# All rights reserved.
+#
+# This source code is licensed under the BSD-style license found in the
+# LICENSE file in the root directory of this source tree.
+
+"""Core tools for code execution and other utilities."""
+
+from .git_server_client import GitServerClient, RepoInfo
+from .local_python_executor import PyExecutor
+
+__all__ = [
+ "PyExecutor",
+ "GitServerClient",
+ "RepoInfo",
+]
\ No newline at end of file
diff --git a/src/core/tools/git_server_client.py b/src/core/tools/git_server_client.py
new file mode 100644
index 0000000000000000000000000000000000000000..31b1ed4c1806287cffe43ad2f93d24f44329d575
--- /dev/null
+++ b/src/core/tools/git_server_client.py
@@ -0,0 +1,362 @@
+#!/usr/bin/env python3
+"""
+Git Server Client for connecting to external Gitea instance.
+
+This module provides a lightweight client for interacting with a shared
+Gitea service, optimized for task-based isolation where multiple environment
+instances share the same Gitea server but have isolated workspaces.
+"""
+
+import json
+import os
+import shutil
+import subprocess
+import time
+from dataclasses import dataclass
+from pathlib import Path
+from urllib.parse import urlparse
+
+
+@dataclass
+class RepoInfo:
+ """Information about a repository."""
+
+ name: str
+ url: str
+ commit: str
+ clone_url: str
+
+
+class GitServerClient:
+ """
+ Client for connecting to an external Gitea server.
+
+ This client is optimized for task-based isolation where:
+ - Multiple tasks share the same Gitea instance
+ - Each task has its own isolated workspace
+ - Fast reset() via git operations (no server restart)
+ - Repos are pre-migrated to Gitea once
+
+ Args:
+ gitea_url: URL of the Gitea server (e.g., "http://gitea:3000")
+ username: Gitea username for authentication
+ password: Gitea password for authentication
+ workspace_dir: Local workspace directory for cloning repos
+
+ Example:
+ >>> # Connect to shared Gitea (credentials from environment)
+ >>> import os
+ >>> client = GitServerClient(
+ ... gitea_url=os.getenv("GITEA_URL"),
+ ... username=os.getenv("GITEA_USERNAME"),
+ ... password=os.getenv("GITEA_PASSWORD")
+ ... )
+ >>> client.wait_for_ready()
+ >>> # Clone repo to workspace
+ >>> path = client.clone_to_workspace("my-repo", commit="abc123")
+ >>> # Fast reset to base state
+ >>> client.reset_workspace("my-repo", commit="abc123")
+ """
+
+ def __init__(
+ self,
+ gitea_url: str,
+ username: str,
+ password: str,
+ workspace_dir: str = "/workspace",
+ ):
+ """Initialize Git Server Client."""
+ self.gitea_url = gitea_url.rstrip("/")
+ self.username = username
+ self.password = password
+ self.workspace_dir = Path(workspace_dir)
+ self.is_ready = False
+
+ # Parse Gitea URL
+ parsed = urlparse(self.gitea_url)
+ self.domain = parsed.hostname or "localhost"
+ self.port = parsed.port or 3000
+
+ # Ensure workspace exists
+ os.makedirs(self.workspace_dir, exist_ok=True)
+
+ # Configure git credentials
+ self._configure_git()
+
+ def _configure_git(self):
+ """Configure git credentials for automatic authentication."""
+ home_dir = Path.home()
+
+ # Git config
+ git_config = f"""[user]
+ name = {self.username}
+ email = {self.username}@local.env
+[init]
+ defaultBranch = main
+[credential]
+ helper = store
+"""
+ gitconfig_path = home_dir / ".gitconfig"
+ gitconfig_path.write_text(git_config)
+
+ # Git credentials
+ git_credentials = f"http://{self.username}:{self.password}@{self.domain}:{self.port}\n"
+ gitcreds_path = home_dir / ".git-credentials"
+ gitcreds_path.write_text(git_credentials)
+ gitcreds_path.chmod(0o600)
+
+ def wait_for_ready(self, timeout: int = 30) -> bool:
+ """
+ Wait for Gitea server to be ready.
+
+ Args:
+ timeout: Maximum seconds to wait
+
+ Returns:
+ True if server is ready, False otherwise
+ """
+ start_time = time.time()
+ while time.time() - start_time < timeout:
+ try:
+ result = subprocess.run(
+ ["curl", "-sf", f"{self.gitea_url}/"],
+ capture_output=True,
+ timeout=5,
+ )
+ if result.returncode == 0:
+ self.is_ready = True
+ return True
+ except subprocess.TimeoutExpired:
+ pass
+ except Exception:
+ pass
+
+ time.sleep(1)
+
+ return False
+
+ def list_repositories(self) -> list[dict[str, str]]:
+ """
+ List all repositories in Gitea.
+
+ Returns:
+ List of repository information dictionaries
+ """
+ if not self.is_ready:
+ raise RuntimeError("Gitea server is not ready")
+
+ result = subprocess.run(
+ [
+ "curl",
+ "-s",
+ f"{self.gitea_url}/api/v1/user/repos",
+ "-u",
+ f"{self.username}:{self.password}",
+ ],
+ capture_output=True,
+ text=True,
+ )
+
+ if result.returncode != 0:
+ return []
+
+ try:
+ repos = json.loads(result.stdout)
+ return [
+ {
+ "name": repo["name"],
+ "full_name": repo["full_name"],
+ "clone_url": repo["clone_url"],
+ "description": repo.get("description", ""),
+ }
+ for repo in repos
+ ]
+ except (json.JSONDecodeError, KeyError):
+ return []
+
+ def clone_to_workspace(
+ self, repo_name: str, target_dir: str | None = None, commit: str = "main"
+ ) -> str:
+ """
+ Clone a repository to the workspace at a specific commit.
+
+ This creates a fresh clone optimized for task isolation.
+
+ Args:
+ repo_name: Name of repository to clone
+ target_dir: Target directory name (defaults to repo_name)
+ commit: Commit hash or branch to checkout
+
+ Returns:
+ Path to cloned repository
+
+ Raises:
+ RuntimeError: If clone fails
+ """
+ if not self.is_ready:
+ raise RuntimeError("Gitea server is not ready")
+
+ target_dir = target_dir or repo_name
+ target_path = self.workspace_dir / target_dir
+
+ # Remove existing directory if present
+ if target_path.exists():
+ shutil.rmtree(target_path)
+
+ clone_url = f"{self.gitea_url}/{self.username}/{repo_name}.git"
+
+ # Clone repository
+ result = subprocess.run(
+ ["git", "clone", clone_url, str(target_path)],
+ capture_output=True,
+ text=True,
+ )
+
+ if result.returncode != 0:
+ raise RuntimeError(f"Clone failed: {result.stderr}")
+
+ # Checkout specific commit
+ if commit != "main":
+ result = subprocess.run(
+ ["git", "checkout", commit],
+ cwd=str(target_path),
+ capture_output=True,
+ text=True,
+ )
+
+ if result.returncode != 0:
+ raise RuntimeError(f"Checkout failed: {result.stderr}")
+
+ return str(target_path)
+
+ def reset_workspace(self, repo_name: str, commit: str = "main") -> bool:
+ """
+ Fast reset of workspace to base state (optimized for task resets).
+
+ This is much faster than re-cloning. It:
+ 1. Checks out the target commit
+ 2. Resets to that commit (hard)
+ 3. Cleans untracked files
+
+ Args:
+ repo_name: Name of repository (directory in workspace)
+ commit: Commit hash or branch to reset to
+
+ Returns:
+ True if reset successful
+
+ Raises:
+ RuntimeError: If reset fails
+ """
+ repo_path = self.workspace_dir / repo_name
+
+ if not repo_path.exists():
+ raise RuntimeError(f"Repository not found in workspace: {repo_name}")
+
+ # Fetch latest (in case commit is new)
+ subprocess.run(
+ ["git", "fetch", "--all"],
+ cwd=str(repo_path),
+ capture_output=True,
+ )
+
+ # Checkout and hard reset to commit
+ result = subprocess.run(
+ ["git", "checkout", commit],
+ cwd=str(repo_path),
+ capture_output=True,
+ text=True,
+ )
+
+ if result.returncode != 0:
+ raise RuntimeError(f"Checkout failed: {result.stderr}")
+
+ result = subprocess.run(
+ ["git", "reset", "--hard", f"origin/{commit}" if commit != "main" else commit],
+ cwd=str(repo_path),
+ capture_output=True,
+ text=True,
+ )
+
+ if result.returncode != 0:
+ # Try without origin/ prefix
+ result = subprocess.run(
+ ["git", "reset", "--hard", commit],
+ cwd=str(repo_path),
+ capture_output=True,
+ text=True,
+ )
+ if result.returncode != 0:
+ raise RuntimeError(f"Reset failed: {result.stderr}")
+
+ # Clean untracked files and directories
+ subprocess.run(
+ ["git", "clean", "-fdx"],
+ cwd=str(repo_path),
+ capture_output=True,
+ )
+
+ return True
+
+ def execute_git_command(
+ self, command: str, working_dir: str = ""
+ ) -> tuple[int, str, str]:
+ """
+ Execute a git command in the workspace.
+
+ Args:
+ command: Git command to execute (without 'git' prefix)
+ working_dir: Working directory relative to workspace
+
+ Returns:
+ Tuple of (exit_code, stdout, stderr)
+ """
+ work_path = (
+ self.workspace_dir / working_dir if working_dir else self.workspace_dir
+ )
+
+ if not work_path.exists():
+ return (1, "", f"Working directory does not exist: {work_path}")
+
+ # Split command safely
+ cmd_parts = ["git"] + command.split()
+
+ result = subprocess.run(
+ cmd_parts,
+ cwd=str(work_path),
+ capture_output=True,
+ text=True,
+ )
+
+ return (result.returncode, result.stdout, result.stderr)
+
+ def get_current_commit(self, repo_name: str) -> str:
+ """
+ Get current commit hash of a workspace repository.
+
+ Args:
+ repo_name: Name of repository in workspace
+
+ Returns:
+ Commit hash
+ """
+ repo_path = self.workspace_dir / repo_name
+
+ if not repo_path.exists():
+ raise RuntimeError(f"Repository not found: {repo_name}")
+
+ result = subprocess.run(
+ ["git", "rev-parse", "HEAD"],
+ cwd=str(repo_path),
+ capture_output=True,
+ text=True,
+ )
+
+ if result.returncode != 0:
+ raise RuntimeError(f"Failed to get commit: {result.stderr}")
+
+ return result.stdout.strip()
+
+ def workspace_exists(self, repo_name: str) -> bool:
+ """Check if a repository exists in workspace."""
+ return (self.workspace_dir / repo_name).exists()
diff --git a/src/core/tools/local_python_executor.py b/src/core/tools/local_python_executor.py
new file mode 100644
index 0000000000000000000000000000000000000000..ba4477d52f3f0f295ec568dc21d807a711ea2cc5
--- /dev/null
+++ b/src/core/tools/local_python_executor.py
@@ -0,0 +1,105 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+# All rights reserved.
+#
+# This source code is licensed under the BSD-style license found in the
+# LICENSE file in the root directory of this source tree.
+
+"""
+Local Python Executor.
+
+This module provides functionality for executing Python code locally by wrapping
+the smolagents LocalPythonExecutor.
+"""
+
+from smolagents import LocalPythonExecutor
+
+from core.env_server.types import CodeExecResult
+
+
+class PyExecutor:
+ """
+ Wrapper around smolagents LocalPythonExecutor for executing Python code.
+
+ This class provides a simple interface to execute Python code in a subprocess
+ and capture the results including stdout, stderr, and exit code.
+
+ Args:
+ additional_imports: List of additional module imports to authorize.
+ For example: ["numpy", "pandas", "matplotlib"]
+ These will be added to the base authorized imports.
+
+ Example:
+ >>> # Basic usage with default imports
+ >>> executor = PyExecutor()
+ >>> result = executor.run("print('Hello, World!')")
+ >>> print(result.stdout) # "Hello, World!\n"
+ >>> print(result.exit_code) # 0
+ >>>
+ >>> # Usage with additional imports
+ >>> executor = PyExecutor(additional_imports=["numpy", "pandas"])
+ >>> result = executor.run("import numpy as np\\nprint(np.array([1, 2, 3]))")
+ >>> print(result.stdout) # "[1 2 3]\n"
+ """
+
+ def __init__(self, additional_imports: list[str] | None = None):
+ """
+ Initialize the PyExecutor with a LocalPythonExecutor instance.
+
+ Args:
+ additional_imports: List of additional module names to authorize for import.
+ Defaults to an empty list if not provided.
+ """
+ if additional_imports is None:
+ additional_imports = []
+ self._executor = LocalPythonExecutor(
+ additional_authorized_imports=additional_imports
+ )
+ # Initialize tools to make BASE_PYTHON_TOOLS available (including print)
+ self._executor.send_tools({})
+
+ def run(self, code: str) -> CodeExecResult:
+ """
+ Execute Python code and return the result.
+
+ Args:
+ code: Python code string to execute
+
+ Returns:
+ CodeExecResult containing stdout, stderr, and exit_code
+
+ Example:
+ >>> executor = PyExecutor()
+ >>> result = executor.run("x = 5 + 3\\nprint(x)")
+ >>> print(result.stdout) # "8\n"
+ >>> print(result.exit_code) # 0
+ >>>
+ >>> # Error handling
+ >>> result = executor.run("1 / 0")
+ >>> print(result.exit_code) # 1
+ >>> print(result.stderr) # Contains error message
+ """
+ try:
+ # Execute the code using LocalPythonExecutor
+ # LocalPythonExecutor returns a CodeOutput object with output, logs, is_final_answer
+ exec_result = self._executor(code)
+
+ # Extract the logs (which contain print outputs) as stdout
+ # The output field contains the return value of the code
+ stdout = exec_result.logs
+ stderr = ""
+ exit_code = 0 # Success
+
+ return CodeExecResult(
+ stdout=stdout,
+ stderr=stderr,
+ exit_code=exit_code,
+ )
+
+ except Exception as e:
+ # LocalPythonExecutor raises InterpreterError for various issues
+ # (syntax errors, forbidden operations, runtime errors, etc.)
+ return CodeExecResult(
+ stdout="",
+ stderr=str(e),
+ exit_code=1, # Non-zero indicates error
+ )
diff --git a/src/echo_cli.egg-info/PKG-INFO b/src/echo_cli.egg-info/PKG-INFO
new file mode 100644
index 0000000000000000000000000000000000000000..6cbab959152ba6fb623c604b9aa4fcd43e20b063
--- /dev/null
+++ b/src/echo_cli.egg-info/PKG-INFO
@@ -0,0 +1,153 @@
+Metadata-Version: 2.4
+Name: echo-cli
+Version: 0.1.0
+Summary: Add your description here
+Requires-Python: >=3.10
+Description-Content-Type: text/markdown
+
+---
+title: Echo Environment Server
+emoji: 🔊
+colorFrom: blue
+colorTo: yellow
+sdk: docker
+pinned: false
+app_port: 8000
+base_path: /web
+tags:
+ - openenv
+---
+
+# Echo Environment
+
+A simple test environment that echoes back messages. Perfect for testing the env APIs as well as demonstrating environment usage patterns.
+
+## Quick Start
+
+The simplest way to use the Echo environment is through the `EchoEnv` class:
+
+```python
+from envs.echo_env import EchoAction, EchoEnv
+
+try:
+ # Create environment from Docker image
+ echo_env = EchoEnv.from_docker_image("echo-env:latest")
+
+ # Reset
+ result = echo_env.reset()
+ print(f"Reset: {result.observation.echoed_message}")
+
+ # Send multiple messages
+ messages = ["Hello, World!", "Testing echo", "Final message"]
+
+ for msg in messages:
+ result = echo_env.step(EchoAction(message=msg))
+ print(f"Sent: '{msg}'")
+ print(f" → Echoed: '{result.observation.echoed_message}'")
+ print(f" → Length: {result.observation.message_length}")
+ print(f" → Reward: {result.reward}")
+
+finally:
+ # Always clean up
+ echo_env.close()
+```
+
+That's it! The `EchoEnv.from_docker_image()` method handles:
+- Starting the Docker container
+- Waiting for the server to be ready
+- Connecting to the environment
+- Container cleanup when you call `close()`
+
+## Building the Docker Image
+
+Before using the environment, you need to build the Docker image:
+
+```bash
+# From project root
+docker build -t echo-env:latest -f src/envs/echo_env/server/Dockerfile .
+```
+
+## Environment Details
+
+### Action
+**EchoAction**: Contains a single field
+- `message` (str) - The message to echo back
+
+### Observation
+**EchoObservation**: Contains the echo response and metadata
+- `echoed_message` (str) - The message echoed back
+- `message_length` (int) - Length of the message
+- `reward` (float) - Reward based on message length (length × 0.1)
+- `done` (bool) - Always False for echo environment
+- `metadata` (dict) - Additional info like step count
+
+### Reward
+The reward is calculated as: `message_length × 0.1`
+- "Hi" → reward: 0.2
+- "Hello, World!" → reward: 1.3
+- Empty message → reward: 0.0
+
+## Advanced Usage
+
+### Connecting to an Existing Server
+
+If you already have an Echo environment server running, you can connect directly:
+
+```python
+from envs.echo_env import EchoEnv
+
+# Connect to existing server
+echo_env = EchoEnv(base_url="