nemo_gym.client.http_client

The shared aiohttp clients every Gym process uses to call HTTP endpoints, and requests with retries.

Module Contents

Classes

Name	Description
`GlobalAIOHTTPAsyncClientConfig`	-
`_PickleSafeRequestInfo`	-

Functions

Name	Description
`_close_detached_clients`	Close clients detached from a loop other than the running one, on their owner loop while it still runs.
`_detach_global_aiohttp_clients`	Clear the process-wide data and control clients and return the ones still open.
`_format_upstream_error_body`	Render a bounded upstream error body without hiding its useful traceback.
`_format_upstream_error_log`	-
`_make_keepalive_socket_factory`	-
`_redacted_url`	Strip the query string from a URL before it becomes a span attribute.
`_request_upload_complete`	Whether the one observed HTTP attempt finished sending its body, including EOF.
`_request_with_retries`	-
`_set_tcp_keepalive`	-
`_traced_request`	`_request_with_retries` wrapped in a CLIENT span, with `traceparent` injected.
`close_global_aiohttp_client`	Close the process-wide data and control clients on their owning event loop.
`format_exception_with_response_body`	Format an exception, appending its bounded HTTP response body when available.
`get_global_aiohttp_client`	-
`get_global_aiohttp_control_client`	The reserved pool for control calls, created on first use next to the data client.
`get_response_json`	-
`global_aiohttp_client_exit`	-
`is_global_aiohttp_client_request_debug_enabled`	-
`is_global_aiohttp_client_setup`	-
`raise_boundary_failure`	Raise a valid remote `NgFailureEvent` with its original IDs and explicit causes.
`raise_for_status`	-
`request`	Make an outbound HTTP call through Gym’s shared aiohttp client.
`set_global_aiohttp_client`	-

Data

DISCONNECTED_CLIENT_OS_HELP_TEXT

DISCONNECTED_CLIENT_OS_PRINT_INTERVAL

MAX_NUM_TRIES

_CLOSING_DETACHED_CLIENTS

_GLOBAL_AIOHTTP_CLIENT

_GLOBAL_AIOHTTP_CLIENT_CONFIG

_GLOBAL_AIOHTTP_CLIENT_LOOP

_GLOBAL_AIOHTTP_CLIENT_QUEUE_TELEMETRY

_GLOBAL_AIOHTTP_CLIENT_REQUEST_DEBUG

_GLOBAL_AIOHTTP_CONTROL_CLIENT

_NUM_CLIENT_OS_ERROR

_NUM_SERVER_DISCONNECTED_ERROR

_UPSTREAM_ERROR_LOG_BODY_CHARS

API

class nemo_gym.client.http_client.GlobalAIOHTTPAsyncClientConfig()

Bases: BaseModel

global_aiohttp_client_request_debug

bool = False

global_aiohttp_connector_limit

int

global_aiohttp_connector_limit_per_host

int

global_aiohttp_control_connector_limit

int = Field(default=256, ge=0)

global_aiohttp_control_connector_limit_per_host

int = Field(default=64, ge=0)

global_aiohttp_intended_concurrency

Optional[int]

global_aiohttp_intended_concurrency_per_host

Optional[int]

global_aiohttp_tcp_keepalive_idle_seconds

int

global_aiohttp_tcp_keepalive_interval_seconds

int

global_aiohttp_tcp_keepalive_probes

int

class nemo_gym.client.http_client._PickleSafeRequestInfo()

Bases: NamedTuple

headers

CIMultiDict[str]

method

str

real_url

str

url

str

nemo_gym.client.http_client._close_detached_clients(
    owner_loop: asyncio.AbstractEventLoop,
    clients: list[aiohttp.ClientSession]
) -> None

Close clients detached from a loop other than the running one, on their owner loop while it still runs.

nemo_gym.client.http_client._detach_global_aiohttp_clients() -> list[aiohttp.ClientSession]

Clear the process-wide data and control clients and return the ones still open.

nemo_gym.client.http_client._format_upstream_error_body(
    content: typing.Any
) -> str

Render a bounded upstream error body without hiding its useful traceback.

nemo_gym.client.http_client._format_upstream_error_log(
    server_name: str,
    error: aiohttp.ClientResponseError
) -> str

nemo_gym.client.http_client._make_keepalive_socket_factory(
    idle_seconds: int,
    interval_seconds: int,
    probes: int
)

nemo_gym.client.http_client._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.client.http_client._request_upload_complete(
    upload: aiohttp.UploadTracker | None
) -> bool | None

Whether the one observed HTTP attempt finished sending its body, including EOF.

Pending/cancelled state or an unobserved aiohttp resend is unknown; it cannot qualify a response timeout.

nemo_gym.client.http_client._request_with_retries(
    method: str,
    url: str,
    _internal: bool = False,
    _max_connection_retries: typing.Optional[int] = None,
    _server_name: typing.Optional[str] = None,
    _max_num_tries: typing.Optional[int] = None,
    _control: bool = False,
    _resend_after_disconnect: bool = True,
    _on_transport_failure: collections.abc.Callable[[Exception, bool | None], None] | None = None,
    _read_body: bool = False,
    _retry_other_errors: bool = True,
    kwargs: typing.Unpack[aiohttp.client._RequestOptions] = {}
) -> aiohttp.ClientResponse

async

nemo_gym.client.http_client._set_tcp_keepalive(
    sock: socket.socket,
    idle_seconds: int,
    interval_seconds: int,
    probes: int
) -> None

nemo_gym.client.http_client._traced_request(
    method: str,
    url: str,
    _internal: bool = False,
    _max_connection_retries: typing.Optional[int] = None,
    _server_name: typing.Optional[str] = None,
    _max_num_tries: typing.Optional[int] = None,
    _control: bool = False,
    _resend_after_disconnect: bool = True,
    _on_transport_failure: collections.abc.Callable[[Exception, bool | None], None] | None = None,
    _read_body: bool = False,
    _retry_other_errors: bool = True,
    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.client.http_client.close_global_aiohttp_client() -> None

async

Close the process-wide data and control clients on their owning event loop.

nemo_gym.client.http_client.format_exception_with_response_body(
    error: BaseException
) -> str

Format an exception, appending its bounded HTTP response body when available.

str(ClientResponseError) names only the status, reason phrase, and URL. The upstream server’s own reason, such as an agent’s AgentSeedRejectedError detail, is in the body that raise_for_status keeps on response_content.

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

nemo_gym.client.http_client.get_global_aiohttp_control_client() -> aiohttp.ClientSession

The reserved pool for control calls, created on first use next to the data client.

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

async

nemo_gym.client.http_client.global_aiohttp_client_exit()

nemo_gym.client.http_client.is_global_aiohttp_client_request_debug_enabled() -> bool

nemo_gym.client.http_client.is_global_aiohttp_client_setup() -> bool

nemo_gym.client.http_client.raise_boundary_failure(
    response: aiohttp.ClientResponse,
    content: bytes | None = None
) -> None

async

Raise a valid remote NgFailureEvent with its original IDs and explicit causes.

error.http_error retains this hop’s status, body and request information. Without a final remote failure, raise the original ClientResponseError. Call raise_for_status where callers require that HTTP exception type even for typed replies, such as a model client’s retry/routing logic.

nemo_gym.client.http_client.raise_for_status(
    response: aiohttp.ClientResponse,
    content: typing.Optional[bytes] = None
) -> None

async

nemo_gym.client.http_client.request(
    method: str,
    url: str,
    _internal: bool = False,
    _max_connection_retries: typing.Optional[int] = None,
    _server_name: typing.Optional[str] = None,
    _max_num_tries: typing.Optional[int] = None,
    _control: bool = False,
    _resend_after_disconnect: bool = True,
    _on_transport_failure: collections.abc.Callable[[Exception, bool | None], None] | None = None,
    _read_body: bool = False,
    _retry_other_errors: bool = True,
    kwargs: typing.Unpack[aiohttp.client._RequestOptions] = {}
) -> aiohttp.ClientResponse

async

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

The private transport-failure observer runs before retry decisions and must not raise. It is not forwarded to aiohttp; cancellation bypasses it.

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.

_server_name is the bounded logical destination label for pool metrics: a configured ServerClient server name, remote_agent_service for the remote agent’s external service, or None for the fallback label external. It is retained across retries and redirects and is not forwarded to aiohttp.

_max_num_tries caps this call’s total attempts on every exception path. It replaces the default generic-error limit (MAX_NUM_TRIES attempts for external calls, unbounded for internal ones). A _max_connection_retries limit still applies as well.

_resend_after_disconnect=False raises a dropped connection instead of sending the request again, because the server may already have received it. A connection that was never made is still retried.

_retry_other_errors=False resends only a dropped or refused connection (ServerDisconnectedError, ClientOSError) within the caller’s bound; any other error, a timeout or a body cut off mid-read included, is raised at once.

_read_body=True reads a success reply’s body before returning, but only within a _max_num_tries or _max_connection_retries bound and only when resending is allowed, so a reply cut off mid-body is sent again within that bound. Otherwise, and for an error reply, the response is returned unread: the caller’s own read gets the complete body or raises the original error, so the caller’s failure handling (for collection, its rollout attempt budget) decides whether the request runs again.

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

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

nemo_gym.client.http_client.DISCONNECTED_CLIENT_OS_PRINT_INTERVAL: int = 100

nemo_gym.client.http_client.MAX_NUM_TRIES = 3

nemo_gym.client.http_client._CLOSING_DETACHED_CLIENTS: set[Task] = set()

nemo_gym.client.http_client._GLOBAL_AIOHTTP_CLIENT: Union[None, ClientSession] = None

nemo_gym.client.http_client._GLOBAL_AIOHTTP_CLIENT_CONFIG: Optional[GlobalAIOHTTPAsyncClientConfig] = None

nemo_gym.client.http_client._GLOBAL_AIOHTTP_CLIENT_LOOP: Optional[AbstractEventLoop] = None

nemo_gym.client.http_client._GLOBAL_AIOHTTP_CLIENT_QUEUE_TELEMETRY: bool = False

nemo_gym.client.http_client._GLOBAL_AIOHTTP_CLIENT_REQUEST_DEBUG: bool = False

nemo_gym.client.http_client._GLOBAL_AIOHTTP_CONTROL_CLIENT: Union[None, ClientSession] = None

nemo_gym.client.http_client._NUM_CLIENT_OS_ERROR: int = 0

nemo_gym.client.http_client._NUM_SERVER_DISCONNECTED_ERROR: int = 0

nemo_gym.client.http_client._UPSTREAM_ERROR_LOG_BODY_CHARS = 2000