> This page is for version 0.3.0.
> For other versions, use one of these documentation indexes:
> - Main (default): https://docs.nvidia.com/nemo/gym/main/llms.txt
> - 0.7.0: https://docs.nvidia.com/nemo/gym/v0.7.0/llms.txt
> - 0.6.0: https://docs.nvidia.com/nemo/gym/v0.6.0/llms.txt
> - 0.5.1: https://docs.nvidia.com/nemo/gym/v0.5.1/llms.txt
> - 0.5.0: https://docs.nvidia.com/nemo/gym/v0.5.0/llms.txt
> - 0.4.0: https://docs.nvidia.com/nemo/gym/v0.4.0/llms.txt
> - 0.3.0: https://docs.nvidia.com/nemo/gym/v0.3.0/llms.txt
> - 0.2.1: https://docs.nvidia.com/nemo/gym/v0.2.1/llms.txt

> For clean Markdown of any page, append .md to the page URL.
> For a complete documentation index, see https://docs.nvidia.com/nemo/gym/llms.txt.
> For AI client integration (Claude Code, Cursor, etc.), connect to the MCP server at https://docs.nvidia.com/nemo/gym/_mcp/server.

# nemo_gym.base_environment_server

Shared lifecycle for environment servers.

## Module Contents

### Classes

| Name                                                                                           | Description                                                     |
| ---------------------------------------------------------------------------------------------- | --------------------------------------------------------------- |
| [`BaseEnvironmentServer`](#nemo_gym-base_environment_server-BaseEnvironmentServer)             | Expose a typed episode protocol with shared limits and cleanup. |
| [`BaseEnvironmentServerConfig`](#nemo_gym-base_environment_server-BaseEnvironmentServerConfig) | Configure protocol-neutral episode limits.                      |
| [`CleanupContext`](#nemo_gym-base_environment_server-CleanupContext)                           | Hold bounded process-local cleanup callbacks for one episode.   |
| [`CleanupHandle`](#nemo_gym-base_environment_server-CleanupHandle)                             | Close one registered participant at a protocol boundary.        |
| [`HandledEpisodeError`](#nemo_gym-base_environment_server-HandledEpisodeError)                 | Carry a failure that belongs in the episode response.           |
| [`_CleanupEntry`](#nemo_gym-base_environment_server-_CleanupEntry)                             | -                                                               |

### Data

[`CleanupCallback`](#nemo_gym-base_environment_server-CleanupCallback)

[`EpisodeRequestT`](#nemo_gym-base_environment_server-EpisodeRequestT)

[`EpisodeResponseT`](#nemo_gym-base_environment_server-EpisodeResponseT)

[`LOGGER`](#nemo_gym-base_environment_server-LOGGER)

### API

```python
class nemo_gym.base_environment_server.BaseEnvironmentServer()
```

**Bases:** [SimpleServer](/nemo/gym/nemo-gym/nemo_gym/server_utils#nemo_gym-server_utils-SimpleServer), `Generic[EpisodeRequestT, EpisodeResponseT]`

Expose a typed episode protocol with shared limits and cleanup.

**`_admission`** `Semaphore | None = None`

---

**`_checkpoint`** `EnvironmentParticipant | None = None`

---

**`checkpoint_boundaries`** `bool = False`

---

**`config`** `BaseEnvironmentServerConfig`

---

**`request_model`** `type[EpisodeRequestT]`

---

**`response_model`** `type[EpisodeResponseT]`

---

```python
nemo_gym.base_environment_server.BaseEnvironmentServer._unhandled_failure_response(
    request: nemo_gym.base_environment_server.EpisodeRequestT,
    error: Exception
) -> nemo_gym.base_environment_server.EpisodeResponseT
```

```python
nemo_gym.base_environment_server.BaseEnvironmentServer.aggregate_metrics(
    body: nemo_gym.config_types.AggregateMetricsRequest = Body()
) -> nemo_gym.config_types.AggregateMetrics
```

async

abstract

Aggregate per-rollout scores into task-level metrics.

```python
nemo_gym.base_environment_server.BaseEnvironmentServer.checkpoint_boundary(
    request: nemo_gym.base_environment_server.EpisodeRequestT,
    state: dict[str, pydantic.JsonValue] | collections.abc.Callable[[], dict[str, pydantic.JsonValue]]
) -> None
```

async

Record that a protocol step completed; `state` names the next step and what it needs.

While a checkpoint is open the episode parks here until resume.
A replacement attempt
that continues a checkpoint receives the latest recorded `state` from `checkpoint_continuation`.
Pass a function when building `state` costs something: it is called only with checkpointing on.

```python
nemo_gym.base_environment_server.BaseEnvironmentServer.checkpoint_continuation(
    request: nemo_gym.base_environment_server.EpisodeRequestT
) -> dict[str, pydantic.JsonValue] | None
```

Return the protocol state this attempt continues from a checkpoint, once, or `None`.

```python
nemo_gym.base_environment_server.BaseEnvironmentServer.checkpoint_restart(
    request: nemo_gym.base_environment_server.EpisodeRequestT
) -> None
```

async

Mark this episode as a restart: a server it uses cannot capture its part, such as a restart-only agent.

Call it when a seed reply says so (`nemo_gym._checkpoint.steps.seed_restarts`).
The episode then never holds up a checkpoint and is never exported; after a crash,
the controller starts it over from its input.

```python
nemo_gym.base_environment_server.BaseEnvironmentServer.checkpoint_step(
    request: nemo_gym.base_environment_server.EpisodeRequestT,
    mode: nemo_gym._checkpoint.steps.StepMode
) -> contextlib.AbstractAsyncContextManager[None]
```

Run one protocol step in `wait` or `replay` mode (see `nemo_gym._checkpoint.steps`).

Use `replay` only for a step that is safe to run again after a crash
and does not change checkpointed state; everything else waits.

```python
nemo_gym.base_environment_server.BaseEnvironmentServer.failure_response(
    request: nemo_gym.base_environment_server.EpisodeRequestT,
    failure: nemo_gym.episode_types.EpisodeFailure
) -> nemo_gym.base_environment_server.EpisodeResponseT
```

```python
nemo_gym.base_environment_server.BaseEnvironmentServer.model_post_init(
    context: typing.Any
) -> None
```

```python
nemo_gym.base_environment_server.BaseEnvironmentServer.run(
    request: nemo_gym.base_environment_server.EpisodeRequestT,
    cleanup: nemo_gym.base_environment_server.CleanupContext
) -> nemo_gym.base_environment_server.EpisodeResponseT
```

async

abstract

Run one concrete environment protocol.

```python
nemo_gym.base_environment_server.BaseEnvironmentServer.run_request(
    request: nemo_gym.base_environment_server.EpisodeRequestT
) -> nemo_gym.base_environment_server.EpisodeResponseT
```

async

```python
nemo_gym.base_environment_server.BaseEnvironmentServer.setup_environment_checkpoint(
    app: fastapi.FastAPI
) -> None
```

Take part in partial-rollout checkpoints when they are enabled.

Every episode is tracked.
A protocol that records boundaries (`checkpoint_boundaries`) continues after a restore;
the episodes of a protocol that records none are restarts,
which never hold up a checkpoint and start over from their input after a crash.
Call this from any `setup_webserver` that builds its own app.

```python
nemo_gym.base_environment_server.BaseEnvironmentServer.setup_webserver() -> fastapi.FastAPI
```

```python
nemo_gym.base_environment_server.BaseEnvironmentServer.validate_response_identity(
    request: nemo_gym.base_environment_server.EpisodeRequestT,
    response: nemo_gym.base_environment_server.EpisodeResponseT
) -> None
```

staticmethod

```python
class nemo_gym.base_environment_server.BaseEnvironmentServerConfig()
```

**Bases:** [BaseRunServerInstanceConfig](/nemo/gym/nemo-gym/nemo_gym/config_types#nemo_gym-config_types-BaseRunServerInstanceConfig)

Configure protocol-neutral episode limits.

**`cleanup_timeout_seconds`** `PositiveFloat`

---

**`default_episode_timeout_seconds`** `PositiveFloat | None = None`

---

**`max_concurrent_episodes`** `PositiveInt | None = None`

---

**`model_config`** `= ConfigDict(extra='forbid')`

---

**`queue_timeout_seconds`** `PositiveFloat | None = None`

---

```python
nemo_gym.base_environment_server.BaseEnvironmentServerConfig.validate_queue_timeout() -> typing_extensions.Self
```

```python
class nemo_gym.base_environment_server.CleanupContext(
    episode_id: nemo_gym.episode_types.EpisodeId,
    cleanup_timeout_seconds: float,
    _cleanups: list[nemo_gym.base_environment_server._CleanupEntry] = list()
)
```

Dataclass

Hold bounded process-local cleanup callbacks for one episode.

Callbacks:

* are process-local Python objects, although they may issue remote close requests;
* run sequentially in LIFO order, each with its own cleanup timeout;
* must be idempotent because a timed-out remote request may have succeeded;
* are lost on process or host failure, so remote owners need expiry or reaping.

Callback failures or timeouts do not prevent later owners from tearing down
their resources. Total cleanup is bounded by the number of registered callbacks
times the per-callback timeout. Cleanup has no
durable retry after this context is discarded.

**`_cleanups`** `list[_CleanupEntry] = field(default_factory=list)`

---

**`cleanup_timeout_seconds`** `float`

---

**`episode_id`** `EpisodeId`

---

```python
nemo_gym.base_environment_server.CleanupContext.aclose() -> None
```

async

```python
nemo_gym.base_environment_server.CleanupContext.register_cleanup(
    name: str,
    callback: nemo_gym.base_environment_server.CleanupCallback
) -> nemo_gym.base_environment_server.CleanupHandle
```

```python
class nemo_gym.base_environment_server.CleanupHandle(
    entry: nemo_gym.base_environment_server._CleanupEntry,
    timeout_seconds: float
)
```

Dataclass

Close one registered participant at a protocol boundary.

**`entry`** `_CleanupEntry`

---

**`timeout_seconds`** `float`

---

```python
nemo_gym.base_environment_server.CleanupHandle.close() -> None
```

async

Run this idempotent callback once.

The episode timeout bounds calls made during the protocol. The context's
cleanup timeout bounds callbacks left for final unwinding.

```python
class nemo_gym.base_environment_server.HandledEpisodeError(
    failure: nemo_gym.episode_types.EpisodeFailure
)
```

Exception

**Bases:** `Exception`

Carry a failure that belongs in the episode response.

```python
class nemo_gym.base_environment_server._CleanupEntry(
    name: str,
    callback: nemo_gym.base_environment_server.CleanupCallback,
    active: bool = True,
    lock: asyncio.Lock = asyncio.Lock()
)
```

Dataclass

**`active`** `bool = True`

---

**`callback`** `CleanupCallback`

---

**`lock`** `Lock = field(default_factory=(asyncio.Lock))`

---

**`name`** `str`

---

```python
nemo_gym.base_environment_server._CleanupEntry.close() -> None
```

async

```python
nemo_gym.base_environment_server.CleanupCallback = Callable[[], Awaitable[None]]
```

```python
nemo_gym.base_environment_server.EpisodeRequestT = TypeVar('EpisodeRequestT', bound=(BaseEpisodeRequest[Any]))
```

```python
nemo_gym.base_environment_server.EpisodeResponseT = TypeVar('EpisodeResponseT', bound=(BaseEpisodeResponse[Any]))
```

```python
nemo_gym.base_environment_server.LOGGER = logging.getLogger(__name__)
```