Single-Step Environment

Build a complete environment end-to-end, from scaffolding to RL-ready rollouts.

Prerequisites

Complete Getting Started before starting.

How It Works

NeMo Gym uses a decoupled three-component architecture: the Agent Server orchestrates the loop, the Model Server runs inference, and the Resources Server provides tools and verification. All three are async FastAPI servers communicating over HTTP, which allows many rollouts to run concurrently across episodes. See Environment Components for the full architecture and diagram.

In most cases, the Resources Server is where your changes go: define your tool endpoints and a verify() method that returns a reward. NeMo Gym ships several pre-built agent servers (simple_agent, swe_agents, etc.) and model servers (openai_model, vllm_model) that you can use as-is, or you can bring your own.

1. Scaffolding

Resource servers live in the resources_servers/ directory. Scaffold a weather server that provides weather information to models:

gym env init --resources-server my_weather_tool

This generates the following structure along with a paired simple agent configuration:

resources_servers/my_weather_tool/
+-- __init__.py
+-- app.py                      # Main server implementation and verifier fixture
+-- configs/
|   +-- my_weather_tool.yaml    # Configuration files
+-- data/
|   +-- .gitignore              # Data directory for examples/datasets
+-- example.jsonl               # Scaffolded example task
+-- tests/
|   +-- __init__.py
|   +-- test_app.py             # Verifier fixture test
|   +-- verifier_cases.jsonl    # Verifier test cases
+-- requirements.txt            # Python dependencies
+-- README.md                   # Documentation

2. Task Preparation

Understanding the task is the first step in designing the environment itself.

Every environment starts with task data — the scenarios your model will practice on. Task data is stored in JSONL format (one JSON object per line), where each line represents a single training example. To get started, it’s not atypical for a domain-expert to hand-craft a few examples from scratch. Once the environment is developed and tested with these examples, you can scale up by collecting more data or using synthetic data generation using libraries like NeMo Data Designer.

JSONL Format

Each line contains a responses_create_params object with the conversation messages, tool definitions, and any ground-truth metadata needed for verification:

{
  "responses_create_params": {
    "input": [
      {"role": "system", "content": "You are a helpful weather assistant."},
      {"role": "user", "content": "What's the weather in San Francisco?"}
    ],
    "tools": [
      {
        "type": "function",
        "name": "get_weather",
        "description": "Get weather for a city.",
        "parameters": {
          "type": "object",
          "properties": {"city": {"type": "string", "description": "City name"}},
          "required": ["city"],
          "additionalProperties": false
        },
        "strict": true
      }
    ],
    "parallel_tool_calls": false
  }
}

Create Data

Create resources_servers/my_weather_tool/data/example.jsonl with five weather examples:

{"responses_create_params": {"input": [{"role": "user", "content": "What's the weather in San Francisco?"}], "tools": [{"type": "function", "name": "get_weather", "description": "Get weather for a city.", "parameters": {"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"], "additionalProperties": false}, "strict": true}]}}
{"responses_create_params": {"input": [{"role": "user", "content": "Tell me the weather in New York"}], "tools": [{"type": "function", "name": "get_weather", "description": "Get weather for a city.", "parameters": {"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"], "additionalProperties": false}, "strict": true}]}}
{"responses_create_params": {"input": [{"role": "user", "content": "How's the weather in Seattle?"}], "tools": [{"type": "function", "name": "get_weather", "description": "Get weather for a city.", "parameters": {"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"], "additionalProperties": false}, "strict": true}]}}
{"responses_create_params": {"input": [{"role": "user", "content": "What is the current weather in Boston?"}], "tools": [{"type": "function", "name": "get_weather", "description": "Get weather for a city.", "parameters": {"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"], "additionalProperties": false}, "strict": true}]}}
{"responses_create_params": {"input": [{"role": "user", "content": "Can you check the weather in Chicago?"}], "tools": [{"type": "function", "name": "get_weather", "description": "Get weather for a city.", "parameters": {"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"], "additionalProperties": false}, "strict": true}]}}

3. Environment Design

This section covers the key aspects of building the environment itself: building or using an existing Agent server, creating the Resources Server, and writing tool and verification logic.

3.1 Agent Server

While this tutorial is about a single-step environment, it still can use the built-in simple_agent, which handles even multi-step tool calling out of the box. No custom agent code is needed. Here is simplified pseudocode showing the core flow (actual implementation):

# run() — episode lifecycle
async def run(self, request, body):
    await resources_server.seed_session(body)      # initialize env state
    response = await self.responses(body)           # multi-step agent loop
    return await resources_server.verify(response)  # compute reward
# responses() — multi-step tool loop
async def responses(self, body):
    while True:
        model_response = await model_server.responses(conversation)
        tool_calls = [o for o in model_response.output if o.type == "function_call"]
        if not tool_calls:  # model produced a final text response
            break
        for call in tool_calls:
            result = await resources_server.post(f"/{call.name}", call.arguments)
            conversation.append(result)
    return model_response

This tutorial uses simple_agent. For other patterns (multi-turn correction, custom orchestration), see the other agents in responses_api_agents/, or build your own by extending SimpleResponsesAPIAgent.

3.2 Resources Server

While the agent handles orchestration, the Resources Server is where you define what makes your environment unique. It is the backbone of tool-based interactions in NeMo Gym.

It provides:

• Tool implementations — APIs that models can call
• Verification logic — reward computation for RL
• Session state — per-episode state management (for stateful environments)

Some agents may come with predefined tools, and you can use the Resources Server to supplement them with additional external tools. When building a new environment, prefer defining tools in the Resources Server rather than the Agent Server. This separation lets multiple agents share the same tool logic without duplicating it.

The scaffold has already done some useful setup for you: the generated app.py includes a stateless verifier, a verifier mixin, and a VERIFIER_FIXTURE used by the generated tests. Rather than starting over, you can build on that foundation by adding the weather tool and changing the verifier to reward get_weather calls.

Open resources_servers/my_weather_tool/app.py, keep the generated SPDX license header at the top, and update the rest of the file to match the following implementation:

from pathlib import Path
from typing import ClassVar
from fastapi import FastAPI
from pydantic import BaseModel
from nemo_gym.base_resources_server import (
    BaseResourcesServerConfig,
    BaseVerifyRequest,
    BaseVerifyResponse,
    ReverifyMode,
    SimpleResourcesServer,
)
from nemo_gym.verifier_fixture import VerifierFixture
class MyWeatherToolResourcesServerConfig(BaseResourcesServerConfig):
    # Verification only uses the request and this server configuration,
    # so saved rollouts can be safely verified again later.
    REVERIFY_MODE: ClassVar[ReverifyMode] = ReverifyMode.STATELESS
class GetWeatherRequest(BaseModel):
    """Input accepted by the get_weather tool."""
    city: str
class GetWeatherResponse(BaseModel):
    """Weather information returned to the agent."""
    city: str
    weather_description: str
class MyWeatherToolVerifyRequest(BaseVerifyRequest):
    """The standard verification request is sufficient for this environment."""
    pass
class MyWeatherToolVerifier:
    async def verify(self, body: MyWeatherToolVerifyRequest) -> BaseVerifyResponse:
        # Reward the rollout when the model called the weather tool.
        used_tool = any(
            output.type == "function_call" and output.name == "get_weather" for output in body.response.output
        )
        return BaseVerifyResponse(**body.model_dump(), reward=float(used_tool))
class MyWeatherToolResourcesServer(MyWeatherToolVerifier, SimpleResourcesServer):
    config: MyWeatherToolResourcesServerConfig
    def setup_webserver(self) -> FastAPI:
        """Add the weather tool route to the standard resources server routes."""
        app = super().setup_webserver()
        app.post("/get_weather")(self.get_weather)
        return app
    async def get_weather(self, body: GetWeatherRequest) -> GetWeatherResponse:
        """Return example weather data for the requested city."""
        # A production environment could call a weather service here.
        return GetWeatherResponse(city=body.city, weather_description=f"The weather in {body.city} is cold.")
# Keep verification testable without starting the web server.
VERIFIER_FIXTURE = VerifierFixture(
    server_factory=MyWeatherToolVerifier,
    request_model=MyWeatherToolVerifyRequest,
    cases_path=Path(__file__).parent / "tests" / "verifier_cases.jsonl",
)
if __name__ == "__main__":
    MyWeatherToolResourcesServer.run_webserver()

Most of this should look familiar from the generated file. The important pieces to keep are REVERIFY_MODE, the separate MyWeatherToolVerifier mixin, and VERIFIER_FIXTURE. Together, they keep scoring independent of the web server and connect it to the generated test cases.

Key Components

3.3 Verification Logic

The verify() function is the heart of your RL environment — it computes the reward signal that drives model training. In this example, verification is simple: return 1.0 if the model called the get_weather tool, 0.0 otherwise. Real environments will have more sophisticated logic, but the principle is the same — inspect the model’s output and score it.

class MyWeatherToolVerifier:
    async def verify(self, body: MyWeatherToolVerifyRequest) -> BaseVerifyResponse:
        # Look through the model's output for a call to get_weather.
        used_tool = any(
            output.type == "function_call" and output.name == "get_weather" for output in body.response.output
        )
        # A matching tool call earns 1.0; every other response earns 0.0.
        return BaseVerifyResponse(**body.model_dump(), reward=float(used_tool))

This first verifier intentionally checks only tool usage, not whether the city argument is correct. Starting with a small, predictable reward function makes it easier to confirm that the complete environment works before adding more sophisticated scoring. Jump to Advanced: Verification Patterns at the end of this tutorial for more examples.

Configure - Wiring the pieces together

Open resources_servers/my_weather_tool/configs/my_weather_tool.yaml. This file contains both the resource server and its paired simple agent configuration.

Update the domain field from other to agent:

my_weather_tool_resources_server:
  resources_servers:
    my_weather_tool:
      entrypoint: app.py
      domain: agent  # Change from 'other' to match your use case
      verified: false
      description: Single-step weather tool calling
my_weather_tool_simple_agent:
  responses_api_agents:
    simple_agent:
      entrypoint: app.py
      resources_server:
        type: resources_servers
        name: my_weather_tool_resources_server
      model_server:
        type: responses_api_models
        name: policy_model
      datasets:
      - name: example
        type: example
        jsonl_fpath: resources_servers/my_weather_tool/data/example.jsonl
      # The scaffold also generates train/validation dataset entries
      # (which can declare a `source:` block — the unified dataset source).
      # Those are omitted here since we only have example data at this stage.

The domain field categorizes your resource server and is required. Common values: math, coding, agent, knowledge, instruction_following, long_context, safety, games, e2e, other.

The agent entry references the resource server and model server by name, wiring all three components together.

4. Add Dependencies (Optional)

If your server needs external packages, add them to requirements.txt:

-e nemo-gym[dev] @ ../../
# Add any other dependencies here

5. Write Tests

You will test both parts of the resources server: the reward behavior and the weather tool itself. The scaffold has already connected tests/test_app.py to VERIFIER_FIXTURE, so begin by replacing resources_servers/my_weather_tool/tests/verifier_cases.jsonl with three cases tailored to the new verifier:

{"name":"used weather tool","kind":"full_reward","request":{"responses_create_params":{"input":"What's the weather in San Francisco?"},"response":{"id":"fixture_response","created_at":0,"model":"fixture","object":"response","output":[{"id":"fixture_call","type":"function_call","call_id":"fixture_call","name":"get_weather","arguments":"{\"city\":\"San Francisco\"}","status":"completed"}],"parallel_tool_calls":false,"tool_choice":"auto","tools":[]}},"expected_reward":1.0}
{"name":"did not use weather tool","kind":"zero_reward","request":{"responses_create_params":{"input":"What's the weather in San Francisco?"},"response":{"id":"fixture_response","created_at":0,"model":"fixture","object":"response","output":[{"id":"fixture_message","type":"message","role":"assistant","status":"completed","content":[{"type":"output_text","text":"It's cold.","annotations":[]}]}],"parallel_tool_calls":false,"tool_choice":"auto","tools":[]}},"expected_reward":0.0}
{"name":"missing response","kind":"malformed","request":{"responses_create_params":{"input":"What's the weather in San Francisco?"}},"expected_error":"response"}

The full-reward case contains a get_weather function call, the zero-reward case contains only an assistant message, and the malformed case confirms that requests without a response are rejected. These cases let the fixture validate the request schema and reward behavior without starting the Gym services.

Next, update resources_servers/my_weather_tool/tests/test_app.py. Keep the generated SPDX license header and fixture test, then add a focused test for the weather tool:

import asyncio
from unittest.mock import MagicMock
from nemo_gym.server_utils import ServerClient
from nemo_gym.verifier_fixture import exercise_verifier_fixture
from ..app import (
    VERIFIER_FIXTURE,
    GetWeatherRequest,
    MyWeatherToolResourcesServer,
    MyWeatherToolResourcesServerConfig,
)
def test_verifier_fixture() -> None:
    """Run the full-reward, zero-reward, and malformed verifier cases."""
    asyncio.run(
        exercise_verifier_fixture(
            VERIFIER_FIXTURE,
            reward_range=(0.0, 1.0),
            higher_is_better=True,
            determinism="unknown",
        )
    )
def test_get_weather() -> None:
    """Return weather information for the requested city."""
    config = MyWeatherToolResourcesServerConfig(
        host="0.0.0.0",
        port=8080,
        entrypoint="",
        name="my_weather_tool",
    )
    server = MyWeatherToolResourcesServer(
        config=config,
        server_client=MagicMock(spec=ServerClient),
    )
    response = asyncio.run(server.get_weather(GetWeatherRequest(city="San Francisco")))
    assert response.city == "San Francisco"
    assert response.weather_description == "The weather in San Francisco is cold."

The fixture test protects the reward contract, while test_get_weather checks the tool response directly. Keeping those responsibilities separate makes failures easier to understand as the environment grows.

Run the tests:

gym env test --resources-server my_weather_tool

For detailed test output:

cd resources_servers/my_weather_tool
source .venv/bin/activate
pytest -v

6. Run & Validate

Run the Servers

Start the servers:

gym env start \
    --model-type openai_model \
    --resources-server my_weather_tool

gym env start reads the config files and starts all three components from the architecture diagram:

1. Agent Server (my_weather_tool_simple_agent) — the simple_agent that orchestrates the seed → model → tool → verify loop
2. Model Server (openai_model) — proxies LLM inference requests to the OpenAI API
3. Resources Server (my_weather_tool_resources_server) — serves your get_weather tool endpoint and verify() logic

Configure API Keys

Configure your OpenAI API key in env.yaml (located in the repository root). The env.yaml is never committed to Git and is designed to hold secrets like API keys:

openai_api_key: ???
policy_api_key: ${openai_api_key}
policy_base_url: https://api.openai.com/v1
policy_model_name: gpt-4o-mini

export OPENAI_API_KEY="sk-your-key-here"  # pragma: allowlist secret

Test with Client (Optional)

You can do a quick spot-check by pointing the built-in client at your agent. Inside responses_api_agents/simple_agent/client.py, change the server name to my_weather_tool_simple_agent, then run:

python responses_api_agents/simple_agent/client.py

Collect Rollouts

Before training, you collect rollouts to validate that your environment works end-to-end and to profile and establish a baseline. Each rollout runs a task through the full agent loop (prompt → model → tool calls → verification) and records the complete interaction along with the reward. This serves two purposes:

1. Validation — confirm your tools, verification logic, and data produce sensible rewards. If a strong model scores near zero, something is likely wrong with your environment.
2. Baselining — measure pass rates across models to understand task difficulty before training begins.

With your servers still running, collect rollouts against your example inputs:

gym eval run --no-serve \
    --agent my_weather_tool_simple_agent \
    --input resources_servers/my_weather_tool/data/example.jsonl \
    --output resources_servers/my_weather_tool/data/example_rollouts.jsonl \
    --limit null \
    --num-repeats null \
    --concurrency null

7. Train with RL

Once you’ve collected rollouts and validated your environment, run training with your preferred RL framework:

8. Update Documentation

Update resources_servers/my_weather_tool/README.md with licensing and usage information:

# My Weather Tool Resource Server
A simple weather information resource server demonstrating tool calling.
## Description
This resource server provides a `get_weather` tool that returns weather information for cities.
## Data
- Example data: Five synthetic weather queries
## Licensing Information
**Code**: Apache 2.0
**Data**: Apache 2.0 (synthetic examples)
## Dependencies
- nemo_gym: Apache 2.0

Summary

You’ve learned how to:

• Initialize a resources server with gym env init
• Prepare task data in JSONL format
• Implement tool endpoints and verification logic
• Configure the required domain field and wire components together
• Write and run tests
• Run servers, validate with a client, and collect rollouts
• Update documentation with licensing information

Advanced: Verification Patterns

For a catalog of common scoring approaches and when to use each, see Verification Patterns.

from nemo_gym.base_resources_server import BaseVerifyRequest, BaseVerifyResponse
class MultiStepVerifyRequest(BaseVerifyRequest):
    """Custom request model that carries ground-truth data for verification."""
    expected_values: list[int]
async def verify(self, body: MultiStepVerifyRequest) -> BaseVerifyResponse:
    """Extract and validate multi-step results."""
    expected = body.expected_values  # Available because we declared it above
    # Parse the final tool call output
    actual = []
    for output in reversed(body.response.output):
        if output.type == "function_call" and output.name == "submit_answer":
            import json
            actual = json.loads(output.arguments).get("values", [])
            break
    # Compute accuracy metrics
    accuracy = expected == actual
    set_overlap = len(set(actual) & set(expected)) / len(expected) if expected else 0
    return BaseVerifyResponse(
        **body.model_dump(),
        reward=float(accuracy),
    )

Troubleshooting

Domain validation error

If you encounter the error "A domain is required for resource servers", ensure the domain field is set in your config YAML file.

Import errors

Ensure you are running commands from the repository root directory and have installed dependencies:

uv sync

Server does not start

Check that:

• Port is not already in use
• Configuration file syntax is valid YAML
• All imports in app.py are correct

Tests fail

Ensure:

• You are in the correct Python environment
• All dependencies are installed
• Test file imports match your actual file structure

Debugging server behavior

Check server status and logs:

# View running servers
gym env status
# For detailed logs, run the server directly:
cd resources_servers/my_weather_tool
source .venv/bin/activate
python app.py

Server logs appear in the terminal where gym env start was executed.