Core API Reference

The openenv-core package provides the core abstractions for building and running environments. For an end-to-end tutorial on building environments with OpenEnv, see the building an environment guide.

Core runtime (core)

Environment server primitives

Message

Bases: TypedDict

A message in a conversation.

Compatible with Huggingface chat template format.

ModelTokenizer

Bases: 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.

apply_chat_template(conversation, tokenize=True, return_tensors=None, **kwargs)

Apply a chat template to format and optionally tokenize a conversation.

Parameters:

Name	Type	Description	Default
conversation	list[Message]	List of message dictionaries with 'role' and 'content'	required
tokenize	bool	Whether to tokenize the output	True
return_tensors	str | None	Format for returned tensors ('pt' for PyTorch)	None
**kwargs	Any	Additional arguments	{}

Returns:

Type	Description
Any	Formatted and optionally tokenized conversation

decode(token_ids, skip_special_tokens=False, **kwargs)

Decode token IDs back to text.

Parameters:

Name	Type	Description	Default
token_ids	Any	Token IDs to decode	required
skip_special_tokens	bool	Whether to skip special tokens in output	False
**kwargs	Any	Additional arguments	{}

Returns:

Type	Description
str	Decoded text string

Transform

Bases: 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.

__call__(observation) abstractmethod

Transform an observation.

Parameters:

Name	Type	Description	Default
observation	Observation	The input observation	required

Returns:

Type	Description
Observation	The transformed observation

Environment

Bases: ABC

Base class for all environment servers following Gym/Gymnasium API.

Parameters:

Name	Type	Description	Default
transform	Transform | None	Optional transform to apply to observations	None

state abstractmethod property

Get the current environment state.

reset() abstractmethod

Reset the environment and return initial observation.

step(action) abstractmethod

Take a step in the environment.

HTTP server utilities

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.

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)

__init__(env, action_cls, observation_cls)

Initialize HTTP server wrapper.

Parameters:

Name	Type	Description	Default
env	Environment	The Environment instance to wrap	required
action_cls	Type[Action]	The Action subclass this environment expects	required
observation_cls	Type[Observation]	The Observation subclass this environment returns	required

register_routes(app)

Register HTTP routes on a FastAPI application.

Parameters:

Name	Type	Description	Default
app	Any	FastAPI application instance	required

create_app(env, action_cls, observation_cls, env_name=None)

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.

Parameters:

Name	Type	Description	Default
env	Environment	The Environment instance to serve	required
action_cls	Type[Action]	The Action subclass this environment expects	required
observation_cls	Type[Observation]	The Observation subclass this environment returns	required
env_name	Optional[str]	Optional environment name for README loading	None

Returns:

Type	Description
Any	FastAPI application instance with or without web interface and README integration

create_fastapi_app(env, action_cls, observation_cls)

Create a FastAPI application with routes for the given environment.

Parameters:

Name	Type	Description	Default
env	Environment	The Environment instance to serve	required
action_cls	Type[Action]	The Action subclass this environment expects	required
observation_cls	Type[Observation]	The Observation subclass this environment returns	required

Returns:

Type	Description
Any	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

Web interface helpers

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.

ActionLog dataclass

Log entry for an action taken.

EpisodeState dataclass

Current episode state for the web interface.

WebInterfaceManager

Manages the web interface for an environment.

connect_websocket(websocket) async

Connect a new WebSocket client.

disconnect_websocket(websocket) async

Disconnect a WebSocket client.

reset_environment() async

Reset the environment and update state.

step_environment(action_data) async

Execute a step in the environment and update state.

get_state()

Get current environment state.

load_environment_metadata(env, env_name=None)

Load environment metadata including README content.

Parameters:

Name	Type	Description	Default
env	Environment	The environment instance	required
env_name	Optional[str]	Optional environment name for README file lookup	None

Returns:

Type	Description
EnvironmentMetadata	EnvironmentMetadata with loaded information

create_web_interface_app(env, action_cls, observation_cls, env_name=None)

Create a FastAPI application with web interface for the given environment.

Parameters:

Name	Type	Description	Default
env	Environment	The Environment instance to serve	required
action_cls	Type[Action]	The Action subclass this environment expects	required
observation_cls	Type[Observation]	The Observation subclass this environment returns	required
env_name	Optional[str]	Optional environment name for README loading	None

Returns:

Type	Description
FastAPI	FastAPI application instance with web interface

get_web_interface_html(action_cls, metadata=None)

Generate the HTML for the web interface.

Client contracts

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)

HTTPEnvClient

Bases: ABC, Generic[ActT, ObsT]

from_docker_image(image, provider=None, **kwargs) classmethod

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 container lifecycle management is left to the user or higher-level orchestration. The container will keep running until manually stopped.

Parameters:

Name	Type	Description	Default
image	str	Docker image name to run (e.g., "echo-env:latest")	required
provider	Optional['ContainerProvider']	Container provider to use (defaults to LocalDockerProvider)	None
**kwargs	Any	Additional arguments to pass to provider.start_container() (e.g., env_vars, port)	{}

Returns:

Type	Description
EnvClientT	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()

from_hub(repo_id, provider=None, **kwargs) classmethod

Create an environment client by pulling from a Hugging Face model hub.

state()

Get the current environment state from the server.

Returns:

Type	Description
Any	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)

close()

Close the environment and clean up resources.

If this client was created via from_docker_image(), this will stop and remove the associated container.

Shared dataclasses

StepResult dataclass

Bases: Generic[ObsT]

Represents the result of one environment step.

Attributes:

Name	Type	Description
observation	ObsT	The environment's observation after the action.
reward	Optional[float]	Scalar reward for this step (optional).
done	bool	Whether the episode is finished.

Container providers

Container provider abstractions for running environment servers.

This module provides a pluggable architecture for different container providers (local Docker, Kubernetes, cloud providers, etc.) to be used with HTTPEnvClient.

ContainerProvider

Bases: ABC

Abstract base class for container providers.

Providers implement this interface to support different container platforms: - LocalDockerProvider: Runs containers on local Docker daemon - KubernetesProvider: Runs containers in Kubernetes cluster - FargateProvider: Runs containers on AWS Fargate - CloudRunProvider: Runs containers on Google Cloud Run

The provider manages a single container lifecycle and provides the base URL for connecting to it.

Example

provider = LocalDockerProvider() base_url = provider.start_container("echo-env:latest") print(base_url) # http://localhost:8000

Use the environment via base_url

provider.stop_container()

start_container(image, port=None, env_vars=None, **kwargs) abstractmethod

Start a container from the specified image.

Parameters:

Name	Type	Description	Default
image	str	Container image name (e.g., "echo-env:latest")	required
port	Optional[int]	Port to expose (if None, provider chooses)	None
env_vars	Optional[Dict[str, str]]	Environment variables to pass to container	None
**kwargs	Any	Provider-specific options	{}

Returns:

Type	Description
str	Base URL to connect to the container (e.g., "http://localhost:8000")

Raises:

Type	Description
RuntimeError	If container fails to start

stop_container() abstractmethod

Stop and remove the running container.

This cleans up the container that was started by start_container().

wait_for_ready(base_url, timeout_s=30.0) abstractmethod

Wait for the container to be ready to accept requests.

This typically polls the /health endpoint until it returns 200.

Parameters:

Name	Type	Description	Default
base_url	str	Base URL of the container	required
timeout_s	float	Maximum time to wait	30.0

Raises:

Type	Description
TimeoutError	If container doesn't become ready in time

LocalDockerProvider

Bases: ContainerProvider

Container provider for local Docker daemon.

This provider runs containers on the local machine using Docker. Useful for development and testing.

Example

provider = LocalDockerProvider() base_url = provider.start_container("echo-env:latest")

Container running on http://localhost:

provider.stop_container()

__init__()

Initialize the local Docker provider.

start_container(image, port=None, env_vars=None, **kwargs)

Start a Docker container locally.

Parameters:

Name	Type	Description	Default
image	str	Docker image name	required
port	Optional[int]	Port to expose (if None, finds available port)	None
env_vars	Optional[Dict[str, str]]	Environment variables for the container	None
**kwargs	Any	Additional Docker run options	{}

Returns:

Type	Description
str	Base URL to connect to the container

stop_container()

Stop and remove the Docker container.

wait_for_ready(base_url, timeout_s=30.0)

Wait for container to be ready by polling /health endpoint.

Parameters:

Name	Type	Description	Default
base_url	str	Base URL of the container	required
timeout_s	float	Maximum time to wait	30.0

Raises:

Type	Description
TimeoutError	If container doesn't become ready

KubernetesProvider

Bases: ContainerProvider

Container provider for Kubernetes clusters.

This provider creates pods in a Kubernetes cluster and exposes them via services or port-forwarding.

Example

provider = KubernetesProvider(namespace="envtorch-dev") base_url = provider.start_container("echo-env:latest")

Pod running in k8s, accessible via service or port-forward

provider.stop_container()