Implementing Action and Observation Dataclasses with Pydantic Models in OpenEnv

OpenEnv leverages Pydantic BaseModel to define strict, validated Action and Observation dataclasses that enable type-safe communication between reinforcement learning clients and environment servers.

OpenEnv utilizes Pydantic-based dataclasses to enforce schema validation across distributed reinforcement learning workflows. Implementing Action and Observation dataclasses with Pydantic models in OpenEnv ensures that every interaction between client agents and environment servers follows a predictable, type-checked contract. These models reside in the core server types module and provide automatic serialization, validation, and documentation generation.

Understanding the Core Data Models

The foundation of OpenEnv's communication protocol rests on two primary Pydantic models defined in src/openenv/core/env_server/types.py. Both inherit from pydantic.BaseModel and enforce strict schema validation to prevent malformed data from propagating through the system.

The Action Model

The Action class represents a single step request sent from the client to the environment. According to the OpenEnv source code, this model includes:

• action_type: A string identifying the specific action category (e.g., "move", "click")
• payload: A dictionary containing action-specific parameters
• metadata: An optional dictionary for auxiliary data such as timestamps or episode IDs

Because Action subclasses BaseModel with strict validation enabled, unknown keys automatically raise validation errors. This prevents malformed messages from reaching the environment logic and ensures that all required fields are present before processing begins.

The Observation Model

The Observation class defines the standardized response returned after processing an Action. As implemented in src/openenv/core/env_server/types.py, it contains:

• observation_type: A string categorizing the response (e.g., "state", "reward", "done")
• data: A dictionary housing the environment state, reward value, and done flag
• info: An optional dictionary for auxiliary debugging information

This structure guarantees a consistent schema for downstream agents and facilitates seamless integration with logging and monitoring pipelines.

Client-Side Implementation

When implementing Action dataclasses with Pydantic models in OpenEnv, the client-side code constructs validated instances before transmission.

Creating and Sending Actions

The EnvClient class in src/openenv/core/env_client.py handles the construction and serialization of Action objects. Client implementations should build actions using the Pydantic model, then rely on the client's internal methods for transmission:

from openenv.core.env_client import EnvClient
from openenv.core.env_server.types import Action

# Initialize client connection

client = EnvClient("ws://localhost:8000/ws")

# Construct validated action instance

move_action = Action(
    action_type="move",
    payload={"direction": "north", "speed": 1.0},
    metadata={"episode_id": "ep-001", "timestamp": "2024-01-15T10:30:00Z"}
)

# Send action and receive Observation

observation = client.step(move_action)
print(observation.data["state"])  # Access validated response data

The step() method internally calls Action.json() to serialize the Pydantic model into a JSON string suitable for WebSocket or HTTP transport.

Server-Side Processing

On the server side, OpenEnv deserializes incoming requests using Pydantic's parsing methods and returns validated Observation instances.

Parsing Requests in FastAPI Endpoints

The server implementation in src/openenv/core/env_server/web_interface.py processes incoming actions using Action.parse_raw():

from fastapi import APIRouter
from openenv.core.env_server.types import Action, Observation

router = APIRouter()

@router.post("/step")
async def step_endpoint(raw_action: str):
    # Deserialize JSON into validated Pydantic model

    action = Action.parse_raw(raw_action)
    
    # Environment-specific processing logic

    new_state = process_environment_step(action)
    
    # Construct validated observation response

    observation = Observation(
        observation_type="state",
        data={
            "state": new_state,
            "reward": calculate_reward(action),
            "done": check_termination()
        },
        info={"processing_time_ms": 12.5}
    )
    
    return observation.json()

This approach ensures that any malformed JSON or missing required fields trigger Pydantic validation errors before reaching the core environment logic.

Logging and Monitoring

OpenEnv provides integrated logging utilities that leverage the Pydantic models' serialization capabilities. The ActionLog and ObservationLog classes in src/openenv/core/env_server/web_interface.py store interaction histories using the .dict() method:

from openenv.core.env_server.web_interface import ActionLog, ObservationLog

def log_interaction(action: Action, observation: Observation):
    # Convert Pydantic models to dictionaries for database storage

    ActionLog.create(action=action.dict())
    ObservationLog.create(observation=observation.dict())

Using .dict() rather than accessing raw attributes ensures that default values and nested structures are properly serialized for persistent storage.

Integration with Serialization Layer

The src/openenv/core/env_server/serialization.py module provides helper utilities that bridge Pydantic models with the transport layer. These utilities handle edge cases in JSON encoding and ensure that datetime objects in metadata fields are properly serialized according to ISO 8601 standards.

When implementing custom actions, extend the base models in types.py rather than modifying the serialization layer directly. This maintains compatibility with OpenEnv's automatic OpenAPI documentation generation and type-checked client code.

Summary

• Action and Observation dataclasses in OpenEnv inherit from pydantic.BaseModel and are defined in src/openenv/core/env_server/types.py
• Strict validation prevents unknown fields and ensures required parameters are present before processing
• Client-side usage involves constructing Action instances and calling .json() for serialization via EnvClient
• Server-side processing uses Action.parse_raw() for deserialization and returns Observation instances with consistent schemas
• Logging utilities leverage .dict() methods for database storage and interaction replay
• helper utilities in src/openenv/core/env_server/serialization.py manage transport-layer encoding

Frequently Asked Questions

What fields are required when creating an Action instance?

The Action model requires action_type (string) and payload (dictionary) fields. The metadata field is optional. Because the model uses strict Pydantic validation, attempting to instantiate an Action without these required fields or including undefined fields will raise a ValidationError before the code executes.

How does the Observation model handle environment resets versus steps?

The Observation model uses the observation_type field to distinguish between different response categories. For standard steps, this value is typically "state", while reset operations may use "reset" or "initial_state" depending on the environment implementation. The data dictionary always contains the current state representation, reward value, and done boolean flag.

Where are the Pydantic models actually defined in the OpenEnv repository?

The primary definitions reside in src/openenv/core/env_server/types.py. This file contains both the Action and Observation classes plus related type enums. The client module imports these definitions from the server types to ensure both sides of the communication use identical schemas.

How do I extend these models for custom environment actions?

Create subclasses of Action or Observation in your environment-specific module, adding fields with appropriate Pydantic type annotations. Ensure your server endpoints use these subclasses for parsing, or use Pydantic's Union types to accept multiple action variants. The serialization layer in src/openenv/core/env_server/serialization.py will handle the extended schemas automatically if they remain JSON-serializable.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}