nemo_gym.server_utils

Module Contents

Classes

Name	Description
`BaseServer`	All instances of BaseServer are queryable using ServerClient.
`GlobalAIOHTTPAsyncClientConfig`	-
`HeadServer`	-
`ProfilingMiddlewareConfig`	-
`ProfilingMiddlewareInputConfig`	-
`ServerClient`	-
`ServerInstanceDisplayConfig`	-
`SimpleServer`	-
`UvicornLoggingConfig`	-
`_PickleSafeRequestInfo`	-

Functions

Name	Description
`_has_injected_global_config_env`	-
`_make_keepalive_socket_factory`	-
`_redacted_url`	Strip the query string from a URL before it becomes a span attribute.
`_request_with_retries`	-
`_telemetry_server_type`	Return the Gym server-type name for server_cls, or None if it is not one.
`_traced_request`	`_request_with_retries` wrapped in a CLIENT span, with `traceparent` injected.
`apply_rollout_prefix`	Append a rollout prefix to a model-server root URL.
`get_global_aiohttp_client`	-
`get_nemo_gym_fastapi_num_workers`	-
`get_response_json`	-
`get_server_url`	-
`global_aiohttp_client_exit`	-
`initialize_ray`	Initialize ray cluster in a process.
`is_global_aiohttp_client_request_debug_enabled`	-
`is_global_aiohttp_client_setup`	-
`is_nemo_gym_fastapi_entrypoint`	-
`is_nemo_gym_fastapi_worker`	-
`maybe_ray_cluster_exit`	-
`raise_for_status`	-
`request`	Make an outbound HTTP call through Gym’s shared aiohttp client.
`rollout_path_prefix`	Return the leading model-server path prefix for a rollout, if available.
`set_global_aiohttp_client`	-
`set_is_nemo_gym_fastapi_entrypoint`	-
`set_is_nemo_gym_fastapi_worker`	-
`set_nemo_gym_fastapi_num_workers`	-
`setup_server_client`	-

Data

DEFAULT_HEAD_SERVER_PORT

DISCONNECTED_CLIENT_OS_HELP_TEXT

DISCONNECTED_CLIENT_OS_PRINT_INTERVAL

IS_NEMO_GYM_FASTAPI_ENTRYPOINT_KEY_NAME

IS_NEMO_GYM_FASTAPI_WORKER_KEY_NAME

MAX_NUM_TRIES

NEMO_GYM_FASTAPI_NUM_WORKERS

SESSION_ID_KEY

ServerStatus

_GLOBAL_AIOHTTP_CLIENT

_GLOBAL_AIOHTTP_CLIENT_REQUEST_DEBUG

_NEMO_GYM_STARTED_RAY_CLUSTER

_NUM_CLIENT_OS_ERROR

_NUM_SERVER_DISCONNECTED_ERROR

_TELEMETRY_SERVER_TYPE_BY_BASE

API

class nemo_gym.server_utils.BaseServer()

Bases: BaseModel

All instances of BaseServer are queryable using ServerClient.

config

BaseRunServerInstanceConfig

nemo_gym.server_utils.BaseServer.load_config_from_global_config() -> nemo_gym.config_types.BaseRunServerInstanceConfig

classmethod

nemo_gym.server_utils.BaseServer.setup_liveness(
    app: fastapi.FastAPI
) -> None

class nemo_gym.server_utils.GlobalAIOHTTPAsyncClientConfig()

Bases: BaseModel

global_aiohttp_client_request_debug

bool = False

global_aiohttp_connector_limit

int = 100 * 1024

global_aiohttp_connector_limit_per_host

int = 1024

global_aiohttp_tcp_keepalive_idle_seconds

int

global_aiohttp_tcp_keepalive_interval_seconds

int

global_aiohttp_tcp_keepalive_probes

int

class nemo_gym.server_utils.HeadServer()

Bases: BaseServer

_cached_yaml

Optional[str] = None

_server_instances

List[dict] = []

config

BaseServerConfig

nemo_gym.server_utils.HeadServer.get_server_instances() -> typing.List[dict]

nemo_gym.server_utils.HeadServer.global_config_dict_yaml() -> str

async

nemo_gym.server_utils.HeadServer.invalidate_global_config_dict_yaml_cache() -> None

Clear the serialized global config cache.

nemo_gym.server_utils.HeadServer.run_webserver() -> typing.Tuple[uvicorn.Server, threading.Thread, nemo_gym.server_utils.HeadServer]

classmethod

nemo_gym.server_utils.HeadServer.set_server_instances(
    instances: typing.List
) -> None

nemo_gym.server_utils.HeadServer.setup_webserver() -> fastapi.FastAPI

class nemo_gym.server_utils.ProfilingMiddlewareConfig()

Bases: ProfilingMiddlewareInputConfig

profiling_enabled

bool = False

class nemo_gym.server_utils.ProfilingMiddlewareInputConfig()

Bases: BaseModel

profiling_results_dirpath

Optional[str] = None

class nemo_gym.server_utils.ServerClient()

Bases: BaseModel

_server_base_urls

dict[str, str] = PrivateAttr(default_factory=dict)

global_config_dict

DictConfig

head_server_config

BaseServerConfig

model_config

= ConfigDict(arbitrary_types_allowed=True)

nemo_gym.server_utils.ServerClient._build_server_base_url(
    server_config_dict: omegaconf.OmegaConf
) -> str

nemo_gym.server_utils.ServerClient._resolve_base_url(
    server_name: str
) -> str

nemo_gym.server_utils.ServerClient.get(
    server_name: str,
    url_path: str,
    kwargs: typing.Unpack[aiohttp.client._RequestOptions] = {}
) -> aiohttp.ClientResponse

async

Parameters:

server_name

str

str The name of the server you are trying to call.

url_path

str

str The URL path in the server you are trying to call e.g. “/v1/responses”.

nemo_gym.server_utils.ServerClient.load_from_global_config(
    head_server_config: typing.Optional[nemo_gym.config_types.BaseServerConfig] = None
) -> nemo_gym.server_utils.ServerClient

classmethod

Build a client from the fully resolved global config.

Gym-launched server processes reuse the config injected by their parent. Other processes fetch the config from the head server.

nemo_gym.server_utils.ServerClient.load_head_server_config() -> nemo_gym.config_types.BaseServerConfig

classmethod

nemo_gym.server_utils.ServerClient.poll_for_status(
    server_name: str
) -> nemo_gym.server_utils.ServerStatus

nemo_gym.server_utils.ServerClient.post(
    server_name: str,
    url_path: str,
    kwargs: typing.Unpack[aiohttp.client._RequestOptions] = {}
) -> aiohttp.ClientResponse

async

Parameters:

server_name

str

str The name of the server you are trying to call.

url_path

str

str The URL path in the server you are trying to call e.g. “/v1/responses”.

nemo_gym.server_utils.ServerClient.request(
    server_name: str,
    url_path: str,
    method: str,
    kwargs: typing.Unpack[aiohttp.client._RequestOptions] = {}
) -> aiohttp.ClientResponse

async

class nemo_gym.server_utils.ServerInstanceDisplayConfig()

Bases: BaseModel

config_path

Optional[str] = None

dir_path

Optional[Path] = None

entrypoint

Optional[str] = None

host

Optional[str] = None

name

Optional[str] = None

pid

Optional[int] = None

port

Optional[int] = None

process_name

Optional[str] = None

server_type

Optional[str] = None

start_time

Optional[float] = None

status

Optional[ServerStatus] = None

uptime_seconds

Optional[float] = None

url

Optional[str] = None

class nemo_gym.server_utils.SimpleServer()

Bases: BaseServer

server_client

ServerClient

nemo_gym.server_utils.SimpleServer.get_session_middleware_key() -> str

nemo_gym.server_utils.SimpleServer.instrument_app_for_telemetry(
    app: fastapi.FastAPI
) -> None

Apply OTel FastAPI auto-instrumentation to app, if telemetry is exporting.

Gives every server the inbound half of context propagation plus dimensioned http.server.* metrics, which is why Gym does not use nemo-lens’s undimensioned gym.server.request_duration_ms (see nemo_gym/telemetry/metrics.py).

A server whose instrumentation could not be applied keeps serving; it just does not extract inbound trace context.

nemo_gym.server_utils.SimpleServer.prefix_server_logs() -> None

nemo_gym.server_utils.SimpleServer.run_webserver() -> typing.Optional[fastapi.FastAPI]

classmethod

nemo_gym.server_utils.SimpleServer.set_ulimit(
    target_soft_limit: int = 65535
)

nemo_gym.server_utils.SimpleServer.setup_exception_middleware(
    app: fastapi.FastAPI
) -> None

nemo_gym.server_utils.SimpleServer.setup_profiling(
    app: fastapi.FastAPI,
    profiling_config: nemo_gym.server_utils.ProfilingMiddlewareConfig
) -> None

nemo_gym.server_utils.SimpleServer.setup_session_middleware(
    app: fastapi.FastAPI
) -> None

nemo_gym.server_utils.SimpleServer.setup_telemetry() -> None

Initialise this process’s nemo-lens telemetry. Idempotent, once per process.

Every Gym server is its own process with its own providers — there is no parent handle to inherit, only the NEMO_GYM_OTEL_* environment the orchestrator exported before spawning it. A failure here is logged and swallowed: telemetry must never stop a server from serving.

nemo_gym.server_utils.SimpleServer.setup_webserver() -> fastapi.FastAPI

abstract

class nemo_gym.server_utils.UvicornLoggingConfig()

Bases: BaseModel

uvicorn_logging_show_200_ok

bool = False

class nemo_gym.server_utils._PickleSafeRequestInfo()

Bases: NamedTuple

headers

CIMultiDict[str]

method

str

real_url

str

url

str

nemo_gym.server_utils._has_injected_global_config_env() -> bool

nemo_gym.server_utils._make_keepalive_socket_factory(
    idle_seconds: int,
    interval_seconds: int,
    probes: int
)

nemo_gym.server_utils._redacted_url(
    url: str
) -> str

Strip the query string from a URL before it becomes a span attribute.

Query strings in Gym carry API keys and, on rollout-prefixed routes, task content. The path is the useful part for a trace; the query is a leak waiting to happen.

nemo_gym.server_utils._request_with_retries(
    method: str,
    url: str,
    _internal: bool = False,
    _max_connection_retries: typing.Optional[int] = None,
    kwargs: typing.Unpack[aiohttp.client._RequestOptions] = {}
) -> aiohttp.ClientResponse

async

nemo_gym.server_utils._telemetry_server_type(
    server_cls: typing.Type
) -> typing.Optional[str]

Return the Gym server-type name for server_cls, or None if it is not one.

nemo_gym.server_utils._traced_request(
    method: str,
    url: str,
    _internal: bool = False,
    _max_connection_retries: typing.Optional[int] = None,
    kwargs: typing.Unpack[aiohttp.client._RequestOptions] = {}
) -> aiohttp.ClientResponse

async

_request_with_retries wrapped in a CLIENT span, with traceparent injected.

Uses nemo_gym.telemetry.spans.client_span rather than nemo-lens’s managed_span because the latter cannot set SpanKind — see that module for why an INTERNAL span is wrong on a cross-service hop.

Injection happens inside the span so the header carries this span as the parent — the receiving server’s FastAPI instrumentation extracts it and its SERVER span becomes our child. That edge is what makes one rollout a single trace across three processes.

Retries reuse the same span rather than starting one per attempt: the caller asked for one logical request, and the retry count is recorded as an attribute instead.

nemo_gym.server_utils.apply_rollout_prefix(
    base_url: str,
    rollout_id: typing.Optional[str],
    token_capture: bool = False
) -> str

Append a rollout prefix to a model-server root URL.

nemo_gym.server_utils.get_global_aiohttp_client(
    global_config_dict_parser_config: typing.Optional[nemo_gym.global_config.GlobalConfigDictParserConfig] = None,
    global_config_dict_parser_cls: typing.Type[nemo_gym.global_config.GlobalConfigDictParser] = GlobalConfigDictParser
) -> aiohttp.ClientSession

nemo_gym.server_utils.get_nemo_gym_fastapi_num_workers() -> int

nemo_gym.server_utils.get_response_json(
    response: aiohttp.ClientResponse
) -> typing.Any

async

nemo_gym.server_utils.get_server_url(
    server_name: str
) -> str

nemo_gym.server_utils.global_aiohttp_client_exit()

nemo_gym.server_utils.initialize_ray() -> None

Initialize ray cluster in a process. We store the Ray address in the global config dict so that child processes can connect to it. This avoids the need to start a new Ray cluster in each child process. Note: This function will modify the global config dict - update ray_head_node_address

nemo_gym.server_utils.is_global_aiohttp_client_request_debug_enabled() -> bool

nemo_gym.server_utils.is_global_aiohttp_client_setup() -> bool

nemo_gym.server_utils.is_nemo_gym_fastapi_entrypoint(
    file: str
) -> bool

nemo_gym.server_utils.is_nemo_gym_fastapi_worker() -> bool

nemo_gym.server_utils.maybe_ray_cluster_exit()

nemo_gym.server_utils.raise_for_status(
    response: aiohttp.ClientResponse
) -> None

async

nemo_gym.server_utils.request(
    method: str,
    url: str,
    _internal: bool = False,
    _max_connection_retries: typing.Optional[int] = None,
    kwargs: typing.Unpack[aiohttp.client._RequestOptions] = {}
) -> aiohttp.ClientResponse

async

Make an outbound HTTP call through Gym’s shared aiohttp client.

This is the only place Gym talks to another server, so it is also the only place trace context has to be injected: every agent -> model and agent -> resources hop goes through here. CLAUDE.md bans httpx precisely to keep it that way.

nemo_gym.server_utils.rollout_path_prefix(
    rollout_id: typing.Optional[str],
    token_capture: bool = False
) -> str

Return the leading model-server path prefix for a rollout, if available.

nemo_gym.server_utils.set_global_aiohttp_client(
    cfg: nemo_gym.server_utils.GlobalAIOHTTPAsyncClientConfig
) -> aiohttp.ClientSession

nemo_gym.server_utils.set_is_nemo_gym_fastapi_entrypoint(
    file: str
) -> None

nemo_gym.server_utils.set_is_nemo_gym_fastapi_worker() -> None

nemo_gym.server_utils.set_nemo_gym_fastapi_num_workers(
    num_workers: int
) -> None

nemo_gym.server_utils.setup_server_client(
    head_server_config: typing.Optional[nemo_gym.config_types.BaseServerConfig] = None
) -> nemo_gym.server_utils.ServerClient

nemo_gym.server_utils.DEFAULT_HEAD_SERVER_PORT = 11000

nemo_gym.server_utils.DISCONNECTED_CLIENT_OS_HELP_TEXT = "We've run into this issue in two different scenarios previously:\n1. Too many o...

nemo_gym.server_utils.DISCONNECTED_CLIENT_OS_PRINT_INTERVAL: int = 100

nemo_gym.server_utils.IS_NEMO_GYM_FASTAPI_ENTRYPOINT_KEY_NAME = 'IS_NEMO_GYM_FASTAPI_ENTRYPOINT'

nemo_gym.server_utils.IS_NEMO_GYM_FASTAPI_WORKER_KEY_NAME = 'IS_NEMO_GYM_FASTAPI_WORKER'

nemo_gym.server_utils.MAX_NUM_TRIES = 3

nemo_gym.server_utils.NEMO_GYM_FASTAPI_NUM_WORKERS = 'NEMO_GYM_FASTAPI_NUM_WORKERS'

nemo_gym.server_utils.SESSION_ID_KEY = 'session_id'

nemo_gym.server_utils.ServerStatus = Union[Literal['success'], Literal['connection_error'], Literal['timeout'], Liter...

nemo_gym.server_utils._GLOBAL_AIOHTTP_CLIENT: Union[None, ClientSession] = None

nemo_gym.server_utils._GLOBAL_AIOHTTP_CLIENT_REQUEST_DEBUG: bool = False

nemo_gym.server_utils._NEMO_GYM_STARTED_RAY_CLUSTER: bool = False

nemo_gym.server_utils._NUM_CLIENT_OS_ERROR: int = 0

nemo_gym.server_utils._NUM_SERVER_DISCONNECTED_ERROR: int = 0

nemo_gym.server_utils._TELEMETRY_SERVER_TYPE_BY_BASE = {'SimpleResourcesServer': 'resources_servers', 'SimpleResponsesAPIAgent': 'respo...