> 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.responses_streaming

Streaming Responses-dialect support shared by every Gym model server.

Blackbox harnesses that speak the OpenAI Responses API over SSE (e.g. the Codex CLI) send
requests the strict `NeMoGymResponseCreateParamsNonStreaming` model would reject: a
`stream: true` flag, client-bookkeeping fields (`client_metadata`, `prompt_cache_key`),
and `namespace` tool specs — functions grouped under a namespace that the model calls back
with separate `namespace` + `name` fields on the `function_call` item. Backends behind a
Gym model server (chat-completions endpoints, vLLM) only understand flat function tools, so
this module provides:

* the request-side sanitizer that flattens namespace tools into plain functions (joined as
  `<namespace>__<name>`), rewrites replayed namespaced calls in the input history the same
  way, and drops the fields the params model does not know;
* the response-side synthesizer that re-emits a complete `NeMoGymResponse` as the minimal
  Responses SSE event sequence streaming clients require (`response.created` ->
  `response.output_item.done` per output item -> `response.completed`), splitting flattened
  function-call names back into `namespace` + `name` on the way out.

## Module Contents

### Functions

| Name                                                                                                       | Description                                                                               |
| ---------------------------------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------- |
| [`_delete_loc`](#nemo_gym-responses_streaming-_delete_loc)                                                 | Delete the value at a pydantic error loc from a nested dict/list structure.               |
| [`_input_message_text`](#nemo_gym-responses_streaming-_input_message_text)                                 | -                                                                                         |
| [`_is_system_like_message`](#nemo_gym-responses_streaming-_is_system_like_message)                         | -                                                                                         |
| [`_sse_event`](#nemo_gym-responses_streaming-_sse_event)                                                   | -                                                                                         |
| [`_tool_valid`](#nemo_gym-responses_streaming-_tool_valid)                                                 | Whether a tool spec validates against the params model's tool union; failures are logged. |
| [`flatten_namespace_tools`](#nemo_gym-responses_streaming-flatten_namespace_tools)                         | Flatten `namespace` tool specs into plain function tools with joined names.               |
| [`sanitize_streaming_responses_body`](#nemo_gym-responses_streaming-sanitize_streaming_responses_body)     | Map a streaming-dialect request body onto the strict non-streaming params shape.          |
| [`synthesize_responses_failure_sse`](#nemo_gym-responses_streaming-synthesize_responses_failure_sse)       | Emit a terminal Responses SSE stream for a backend failure.                               |
| [`synthesize_responses_sse`](#nemo_gym-responses_streaming-synthesize_responses_sse)                       | Re-emit a complete Responses API response object as an SSE event stream.                  |
| [`validate_streaming_responses_params`](#nemo_gym-responses_streaming-validate_streaming_responses_params) | Validate a sanitized streaming-dialect body, pruning fields newer than the params model.  |

### Data

[`LOG`](#nemo_gym-responses_streaming-LOG)

[`NAMESPACE_TOOL_DELIMITER`](#nemo_gym-responses_streaming-NAMESPACE_TOOL_DELIMITER)

[`NamespaceMap`](#nemo_gym-responses_streaming-NamespaceMap)

[`_INPUT_ITEM_ADAPTER`](#nemo_gym-responses_streaming-_INPUT_ITEM_ADAPTER)

[`_PARAM_FIELDS`](#nemo_gym-responses_streaming-_PARAM_FIELDS)

[`_TOOL_ADAPTER`](#nemo_gym-responses_streaming-_TOOL_ADAPTER)

### API

```python
nemo_gym.responses_streaming._delete_loc(
    body: typing.Any,
    loc: tuple
) -> bool
```

Delete the value at a pydantic error loc from a nested dict/list structure.

Returns False when the loc cannot be walked literally (e.g. it contains a union-arm label
rather than a real key), in which case nothing is deleted.

```python
nemo_gym.responses_streaming._input_message_text(
    item: dict[str, typing.Any]
) -> str
```

```python
nemo_gym.responses_streaming._is_system_like_message(
    item: typing.Any
) -> bool
```

```python
nemo_gym.responses_streaming._sse_event(
    payload: dict[str, typing.Any]
) -> str
```

```python
nemo_gym.responses_streaming._tool_valid(
    tool: typing.Any
) -> bool
```

Whether a tool spec validates against the params model's tool union; failures are logged.

```python
nemo_gym.responses_streaming.flatten_namespace_tools(
    tools: typing.Any
) -> tuple[list[typing.Any], nemo_gym.responses_streaming.NamespaceMap]
```

Flatten `namespace` tool specs into plain function tools with joined names.

```python
nemo_gym.responses_streaming.sanitize_streaming_responses_body(
    body: dict[str, typing.Any]
) -> tuple[dict[str, typing.Any], nemo_gym.responses_streaming.NamespaceMap]
```

Map a streaming-dialect request body onto the strict non-streaming params shape.

Returns the cleaned body dict (ready for `NeMoGymResponseCreateParamsNonStreaming`
validation) and the namespace map needed to restore namespaced call names in the
synthesized SSE response. Tool entries that still fail per-entry validation after
flattening are dropped with a warning rather than failing the whole request, since a
harness's exotic tool is better lost than the rollout.

```python
nemo_gym.responses_streaming.synthesize_responses_failure_sse(
    message: str,
    code: str = 'server_error'
) -> typing.Iterator[str]
```

Emit a terminal Responses SSE stream for a backend failure.

Once the streaming contract is committed (HTTP 200 + `text/event-stream`), a `responses()`
error can no longer surface as an HTTP 500. Streaming clients (e.g. Codex) expect a terminal
`response.failed` event; without one they see a truncated stream and cannot tell an
application error from a transport failure. Emitting `response.failed` lets the client report
a clean turn failure, and lets model-call capture classify it as an upstream error (its
terminal-SSE table maps `response.failed` to an error) rather than a stream truncation.

```python
nemo_gym.responses_streaming.synthesize_responses_sse(
    response_json: dict[str, typing.Any],
    ns_map: typing.Optional[nemo_gym.responses_streaming.NamespaceMap] = None
) -> typing.Iterator[str]
```

Re-emit a complete Responses API response object as an SSE event stream.

Streaming clients build their view of the turn from `response.output_item.done` events and
treat `response.completed` (which carries the response id and usage) as the terminal event,
so those two are the required minimum; `response.created` is included for clients that wait
for an acknowledgement before reading items.

```python
nemo_gym.responses_streaming.validate_streaming_responses_params(
    body: dict[str, typing.Any]
) -> nemo_gym.openai_utils.NeMoGymResponseCreateParamsNonStreaming
```

Validate a sanitized streaming-dialect body, pruning fields newer than the params model.

Harness wire formats evolve faster than the pinned OpenAI SDK types (e.g. Codex sending
`reasoning.context`, which the SDK's `Reasoning` model forbids). Any field pydantic flags
as `extra_forbidden` is removed and validation retried, so only errors that cannot be fixed
by dropping an unknown field surface to the client.

```python
nemo_gym.responses_streaming.LOG = logging.getLogger(__name__)
```

```python
nemo_gym.responses_streaming.NAMESPACE_TOOL_DELIMITER = '__'
```

```python
nemo_gym.responses_streaming.NamespaceMap = dict[str, tuple[str, str]]
```

```python
nemo_gym.responses_streaming._INPUT_ITEM_ADAPTER = TypeAdapter(NeMoGymResponseInputItem)
```

```python
nemo_gym.responses_streaming._PARAM_FIELDS = frozenset(NeMoGymResponseCreateParamsNonStreaming.model_fields)
```

```python
nemo_gym.responses_streaming._TOOL_ADAPTER = TypeAdapter(ToolParam)
```