A message in a conversation.

Compatible with Huggingface chat template format.

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.

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.

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

See rfcs/004-rubrics.md for rubric design details.

Server operation mode.

Server health status values.

WebSocket error codes for structured error handling.

Base class for all environment actions.

All action subclasses should inherit from this base class. Uses Pydantic for automatic validation and serialization.

Base class for all environment observations.

All observation subclasses should inherit from this base class. Uses Pydantic for automatic validation and serialization.

Request model for environment reset.

Response model for environment reset.

Request model for environment step.

Response model for environment step.

Base class for WebSocket messages with shared configuration.

Base class for environment state.

Represents internal environment state, separate from observations.

Result of code execution containing stdout, stderr, and exit code.

Metadata about an environment for documentation and UI purposes.

Response model for the combined schema endpoint.

Response model for health check endpoint.

WebSocket message to reset the environment.

WebSocket message to execute a step.

WebSocket message to request current state.

WebSocket message to close the session.

WebSocket response containing an observation.

WebSocket response containing environment state.

WebSocket response for errors.

Configuration for concurrent environment sessions.

Status of server capacity for concurrent sessions.

Information about an active session.

Base exception for all OpenEnv errors.

Raised when an environment is misconfigured for concurrent sessions.

This error is raised during server startup when max_concurrent_envs > 1 is specified for an environment that is not marked as SUPPORTS_CONCURRENT_SESSIONS.

Raised when the server cannot accept new sessions due to capacity limits.

This error is raised when a new WebSocket connection is attempted but the server has already reached max_concurrent_envs active sessions.

Raised when attempting to access a session that does not exist.

Raised when a session cannot be created.

Raised when the environment factory fails to create an instance.

HTTP server wrapper for Environment instances.

This class wraps an Environment and exposes its reset(), step(), and state methods as HTTP and WebSocket endpoints compatible with EnvClient.

The server expects:

• Action deserialization: Converts JSON dict to Action subclass
• Observation serialization: Converts Observation subclass to JSON dict

from core.env_server import HTTPEnvServer
from envs.coding_env.server import CodeExecutionEnvironment
from envs.coding_env.models import CodeAction, CodeObservation

# Pass environment class (factory pattern)
server = HTTPEnvServer(
    env=CodeExecutionEnvironment,
    action_cls=CodeAction,
    observation_cls=CodeObservation,
    max_concurrent_envs=4,
)

# Register routes with FastAPI
from fastapi import FastAPI
app = FastAPI()
server.register_routes(app)

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.

Create a FastAPI application with comprehensive documentation.

Log entry for an action taken.

Current episode state for the web interface.

Manages the web interface for an environment.

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

Convert JSON dict to Action instance using Pydantic validation.

MCP action types (list_tools, call_tool) are recognised automatically via the "type" discriminator field, regardless of the environment’s configured action_cls. All other payloads fall through to action_cls.model_validate().

For special cases (e.g., tensor fields, custom type conversions), use deserialize_action_with_preprocessing().

Convert JSON dict to Action instance with preprocessing for special types.

This version handles common type conversions needed for web interfaces:

• Converting lists/strings to tensors for ‘tokens’ field
• Converting string action_id to int
• Other custom preprocessing as needed

Convert Observation instance to JSON-compatible dict using Pydantic.

Combines multiple transforms into a single transform.

Default transform that passes through unchanged.

Configuration for a simple GET endpoint.

Register multiple GET endpoints from configuration.

Async environment client for persistent sessions.

This client maintains a persistent WebSocket connection to an environment server, enabling efficient multi-step interactions. Each client instance corresponds to a dedicated environment session on the server.

The client is async by default. For synchronous usage, use the .sync() method to get a SyncEnvClient wrapper.

Features:

• Lower latency for sequential interactions
• Session state is maintained server-side
• Better suited for long-running episodes
• Async by default for modern Python async/await patterns

Examples:

from envs.coding_env.client import CodingEnv

async with CodingEnv(base_url="ws://localhost:8000") as env:
    result = await env.reset(seed=42)
    while not result.done:
        action = agent.predict(result.observation)
        result = await env.step(action)

env = CodingEnv(base_url="ws://localhost:8000").sync()
with env:
    result = env.reset(seed=42)
    result = env.step(action)

# Pull and run from HF Docker registry
env = await MyEnv.from_env("openenv/echo-env")

# Run locally with UV (clones the space)
env = await MyEnv.from_env("openenv/echo-env", use_docker=False)

# Run from a local checkout
env = await MyEnv.from_env(
    "openenv/echo-env",
    use_docker=False,
    project_path="/path/to/local/checkout"
)

async_client = GenericEnvClient(base_url="http://localhost:8000")
sync_client = async_client.sync()

with sync_client:
    result = sync_client.reset()
    result = sync_client.step({"code": "print('hello')"})

Synchronous wrapper around an async EnvClient.

This class provides a synchronous interface to an async EnvClient, making it easier to use in synchronous code or to stop async from “infecting” the entire call stack.

The wrapper executes async operations on a dedicated background event loop so connection state remains bound to a single loop.

For guaranteed resource cleanup, use with SyncEnvClient(...) or call close() explicitly. __del__ is best-effort only and may not run reliably (for example, during interpreter shutdown).

# From an async client
async_client = GenericEnvClient(base_url="http://localhost:8000")
sync_client = async_client.sync()

# Use synchronous context manager
with sync_client:
    result = sync_client.reset()
    result = sync_client.step({"action": "test"})

Environment client that works with raw dictionaries instead of typed classes.

This client doesn’t require installing environment-specific packages, making it ideal for:

• Connecting to remote servers without installing their packages
• Quick prototyping and testing
• Environments where type safety isn’t needed
• Security-conscious scenarios where you don’t want to run remote code

The trade-off is that you lose type safety and IDE autocomplete for actions and observations. Instead of typed objects, you work with plain dictionaries.

# Direct connection to a running server (no installation needed)
with GenericEnvClient(base_url="http://localhost:8000") as env:
    result = env.reset()
    result = env.step({"code": "print('hello')"})
    print(result.observation)  # Dict[str, Any]
    print(result.observation.get("output"))

# From local Docker image
env = GenericEnvClient.from_docker_image("coding-env:latest")
result = env.reset()
result = env.step({"code": "x = 1 + 2"})
env.close()

# From HuggingFace Hub (pulls Docker image, no pip install)
env = GenericEnvClient.from_env("user/my-env", use_docker=True)
result = env.reset()
env.close()

GenericEnvClient inherits from_docker_image() and from_env() from EnvClient, so you can use it with Docker containers and HuggingFace Spaces without any package installation.

A dictionary subclass for creating actions when using GenericEnvClient.

This provides a semantic wrapper around dictionaries to make code more readable when working with GenericEnvClient. It behaves exactly like a dict but signals intent that this is an action for an environment.

# Without GenericAction (works fine)
env.step({"code": "print('hello')"})

# With GenericAction (more explicit)
action = GenericAction(code="print('hello')")
env.step(action)

# With multiple fields
action = GenericAction(code="x = 1", timeout=30, metadata={"tag": "test"})
env.step(action)

GenericAction is just a dict with a constructor that accepts keyword arguments. It’s provided for symmetry with typed Action classes and to make code more readable.

A single tool/function call returned by the model.

Normalized response from an LLM, with optional tool calls.

Abstract base for LLM endpoint clients.

Subclass and implement complete() for your protocol.

Client for OpenAI-compatible APIs.

Works with: OpenAI, vLLM, TGI, Ollama, HuggingFace Inference API, or any endpoint that speaks the OpenAI chat completions format.

Client for Anthropic’s Messages API.

Requires the anthropic package (lazy-imported at construction time).

Create an LLM client for a hosted provider.

Represents the result of one environment step.

Base class for environments that expose tools via MCP (Model Context Protocol).

MCPEnvironment bridges FastMCP servers with OpenEnv’s Gym-style API, allowing agents to discover and invoke MCP tools through the standard step() interface.

The class automatically handles:

• ListToolsAction: Returns available tools from the MCP server
• CallToolAction: Invokes a specific tool with arguments

All other actions are delegated to the abstract _step_impl() method, which subclasses must implement.

from fastmcp import FastMCP

mcp = FastMCP("calculator")

@mcp.tool()
def add(a: int, b: int) -> int:
    return a + b

env = MyMCPEnvironment(mcp)
obs = env.step(ListToolsAction())
obs.tools[0].name  # 'add'

Standard JSON-RPC 2.0 error codes.

See: https://www.jsonrpc.org/specification#error_object

Supported MCP method names.

JSON-RPC 2.0 error object.

See: https://www.jsonrpc.org/specification#error_object

JSON-RPC 2.0 request object.

See: https://www.jsonrpc.org/specification#request_object

JSON-RPC 2.0 response object.

Per JSON-RPC 2.0 spec, a response has either ‘result’ or ‘error’, not both. This model excludes None values during serialization to comply with the spec.

See: https://www.jsonrpc.org/specification#response_object

Strongly typed MCP tool specification.

Follows the MCP ToolSpec format for tool discovery. See: https://modelcontextprotocol.io/specification/2025-06-18/server/tools

Types of errors that can occur during tool execution.

Structured error for tool execution failures.

This is used for transport/framework errors, NOT for errors returned by the tool itself (those go in the result field).

Request list of available tools from the environment.

This action triggers MCP’s tools/list operation and returns all available tools with their schemas. Does NOT require reset() to be called first.

Call a specific tool via MCP.

This action triggers MCP’s tools/call operation with the specified tool name and arguments.

Response containing available tools.

Returned when processing a ListToolsAction.

Response from tool execution.

Contains the tool’s result or an error if the call failed. Tool-specific errors (from the tool itself) are included in the result. Transport/framework errors use the error field.

WebSocket message for MCP JSON-RPC requests.

Allows direct MCP access via WebSocket for production inference, bypassing the step() API.

WebSocket response for MCP JSON-RPC.

Contains the JSON-RPC response from the MCP server.

Base class for MCP clients with tool discovery.

This class provides the common list_tools() method for discovering available tools from an MCP-enabled environment. Subclasses implement specific interaction patterns (tool-calling or CodeAct).

tools = await env.list_tools()
for tool in tools:
    print(f"{tool.name}: {tool.description}")

Async client for tool-calling style MCP interactions.

Each step invokes a single tool. Use this for traditional function-calling agent patterns where the agent decides which tool to call next.

This client provides convenience methods for tool discovery and invocation:

• list_tools(): Get all available tools with their schemas
• call_tool(name, **kwargs): Invoke a tool by name with arguments

async with MCPToolClient(base_url="http://localhost:8000") as env:
    # Reset the environment
    await env.reset()

    # Discover available tools
    tools = await env.list_tools()
    print([t.name for t in tools])  # ['echo_message', 'echo_with_length']

    # Call a tool directly
    result = await env.call_tool("echo_message", message="Hello!")
    print(result)  # "Hello!"

    # Or use the full action interface
    from openenv.core.env_server.mcp_types import CallToolAction
    step_result = await env.step(CallToolAction(
        tool_name="echo_with_length",
        arguments={"message": "Test"}
    ))
    print(step_result.observation.result)

env = MCPToolClient(base_url="http://localhost:8000").sync()
with env:
    tools = env.list_tools()
    result = env.call_tool("echo_message", message="Hello!")

result = await env.call_tool("add", a=5, b=3)
print(result)  # 8

result = await env.call_tool("greet", name="Claude")
print(result)  # "Hello, Claude!"

tool = await env.get_tool("echo_message")
if tool:
    print(tool.description)
    print(tool.input_schema)

Abstract base class for reward computation.

A Rubric computes a reward signal from an action and observation. Subclasses implement forward() to define the reward logic.

class MyRubric(Rubric):
    def forward(self, action, observation) -> float:
        return 1.0 if action.valid else 0.0

rubric = MyRubric()
reward = rubric(action, observation)

Child rubrics are auto-registered when assigned as attributes, enabling hierarchical composition and introspection.

Run rubrics in order, fail-fast on zero.

Runs child rubrics in order. If any returns 0, stops immediately and returns 0. This implements hierarchical gating patterns where syntax checks run before execution checks.

rubric = Sequential(
    Gate(Compiles()),
    Gate(PassesTests(), threshold=0.5),
    WeightedSum([PassesTests(), StyleRubric()], weights=[0.7, 0.3])
)

Threshold wrapper - returns 0 if child score is below threshold.

Useful for hard constraints like “must pass 50% of tests”.

rubric = Gate(PassesTests(), threshold=0.5)
# Returns PassesTests() score if >= 0.5, else 0.0

Weighted combination of child rubrics.

Standard aggregation pattern for multi-criteria evaluation.

rubric = WeightedSum(
    [PassesTests(), StyleRubric()],
    weights=[0.7, 0.3]
)

Container for dynamic lists of rubrics.

Analogous to nn.ModuleList. Does not define aggregation - use within a parent rubric that implements custom logic.

class MultiGameRubric(Rubric):
    def __init__(self, games: List[str]):
        super().__init__()
        self.games = RubricList([GameRubric(g) for g in games])

    def forward(self, action, obs) -> float:
        return self.games[obs.game_index](action, obs)

Container for named rubrics with keyed access.

Analogous to nn.ModuleDict. Enables keyed access for multi-task environments where different tasks require different rubrics.

class AtariRubric(Rubric):
    def __init__(self):
        super().__init__()
        self.games = RubricDict({
            "pong": PongRubric(),
            "breakout": BreakoutRubric(),
            "space_invaders": SpaceInvadersRubric(),
        })

    def forward(self, action, obs) -> float:
        return self.games[obs.game_id](action, obs)

# Access: env.rubric.games["pong"]

Abstract base for rubrics that score based on full trajectories.

Subclasses implement:

• score_trajectory(): Compute final score from trajectory
• compute_step_rewards(): Define credit assignment strategy

The call method accumulates steps and returns rewards according to the subclass’s implementation.

IMPORTANT: Trajectories are stored in CPU memory to avoid GPU pressure. Environments with GPU tensors in observations must move them to CPU before returning from step().

Known limitation: Very long episodes (thousands of steps) may consume significant CPU memory. For such cases, consider streaming rubrics.

class WinLossRubric(TrajectoryRubric):
    def score_trajectory(self, trajectory):
        _, final_obs = trajectory[-1]
        return 1.0 if final_obs.metadata.get('won') else 0.0

    def compute_step_rewards(self):
        # Equal credit to all steps
        score = self.score_trajectory(self._trajectory)
        return [score] * len(self._trajectory)

rubric = WinLossRubric()
for action, obs in episode:
    reward = rubric(action, obs)  # 0.0 until done
step_rewards = rubric.compute_step_rewards()  # Credit assignment

TrajectoryRubric with exponential discounting for credit assignment.

r_t = gamma^(T-1-t) * R_final

With gamma=0.99, later steps get higher reward (they’re “closer” to the outcome). With gamma=1.0, all steps get equal reward. With gamma=0.0, only the final step gets reward.

This is the standard temporal discounting used in reinforcement learning, applied retroactively once the episode outcome is known.

class ChessRubric(ExponentialDiscountingTrajectoryRubric):
    def score_trajectory(self, trajectory):
        _, final_obs = trajectory[-1]
        outcome = final_obs.metadata.get('winner')
        if outcome == 'agent': return 1.0
        elif outcome == 'opponent': return 0.0
        else: return 0.5  # Draw

rubric = ChessRubric(gamma=0.99)
reward = rubric(action, obs)  # 0.0 until done, then final score
step_rewards = rubric.compute_step_rewards()  # Discounted per-step rewards

Rubric that uses an LLM to evaluate agent actions/observations.

The prompt template is formatted with {action} and {observation} placeholders. The LLM response is parsed for a numeric score.

Information about a repository.

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

import os

# Connect to shared Gitea (credentials from environment)
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")

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.

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()