nemo_gym.anthropic_converter

Bidirectional converter between NeMo Gym Responses API objects and Anthropic Messages.

This module is the single source of truth for the Anthropic <-> Responses mapping. It is shared by two opposite-direction consumers:

Egress (responses_api_models/anthropic_model): NeMo Gym is the client and Anthropic is the backend. Uses responses_to_anthropic (request) and anthropic_to_responses (response).
Ingress (an Anthropic-Messages proxy, e.g. for the Claude Code CLI): an Anthropic client talks to NeMo Gym, which forwards to a downstream Gym model server. Uses anthropic_request_to_responses (request), responses_to_anthropic_response (response), and anthropic_response_to_sse (synthesize Anthropic SSE from a complete response).

The converter is transport-free and SDK-free: pure dict/Pydantic in, pure dict/Pydantic out. All HTTP stays in the servers via nemo_gym.server_utils.request() (the anthropic SDK is avoided because it uses httpx, whose O(n^2) connection pooling hangs at high concurrency).

Boundary note: a few methods here implement egress-only policy (Anthropic-API-as-backend concerns) rather than structural mapping: _validate_sampling_params_for_model, _model_disallows_sampling_params, and the thinking-config handling in _copy_thinking_params. They are invoked only on the egress responses_to_anthropic path; ingress never calls them (an open-model backend has none of those restrictions). Relocating them into the egress server is a deliberate follow-up, kept out of this refactor to avoid changing the egress contract.

Module Contents

Classes

Name	Description
`AnthropicConverter`	-

Data

SUPPORTED_ANTHROPIC_IMAGE_MEDIA_TYPES

API

class nemo_gym.anthropic_converter.AnthropicConverter()

nemo_gym.anthropic_converter.AnthropicConverter._anthropic_content_to_output_items(
    content: typing.List[typing.Dict[str, typing.Any]]
) -> typing.List[typing.Any]

Anthropic assistant content blocks -> ordered Responses output items.

Shared by egress anthropic_to_responses and ingress anthropic_request_to_responses (for assistant turns in the input trajectory).

nemo_gym.anthropic_converter.AnthropicConverter._anthropic_image_to_input_part(
    block: typing.Dict[str, typing.Any]
) -> typing.Dict[str, typing.Any]

nemo_gym.anthropic_converter.AnthropicConverter._anthropic_messages_to_input_items(
    anthropic_body: typing.Dict[str, typing.Any]
) -> typing.List[typing.Any]

nemo_gym.anthropic_converter.AnthropicConverter._anthropic_system_to_instructions(
    system: typing.Any
) -> str

nemo_gym.anthropic_converter.AnthropicConverter._anthropic_tool_choice_to_responses(
    tool_choice: typing.Any
) -> typing.Any

nemo_gym.anthropic_converter.AnthropicConverter._anthropic_tool_result_content_to_text(
    content: typing.Any
) -> str

nemo_gym.anthropic_converter.AnthropicConverter._anthropic_tools_to_responses(
    tools: typing.Any
) -> typing.List[typing.Dict[str, typing.Any]]

nemo_gym.anthropic_converter.AnthropicConverter._append_anthropic_blocks_as_items(
    role: str,
    blocks: typing.List[typing.Dict[str, typing.Any]],
    items: typing.List[typing.Any]
) -> None

Translate one Anthropic message’s content blocks into ordered Responses items.

Text/image blocks group into a single message item; tool_use, tool_result, and thinking blocks each become their own item, preserving order.

nemo_gym.anthropic_converter.AnthropicConverter._append_content(
    messages: typing.List[typing.Dict[str, typing.Any]],
    role: str,
    content_blocks: typing.List[typing.Dict[str, typing.Any]]
) -> None

nemo_gym.anthropic_converter.AnthropicConverter._append_message_item(
    item: typing.Dict[str, typing.Any],
    messages: typing.List[typing.Dict[str, typing.Any]],
    system_parts: typing.List[str]
) -> None

nemo_gym.anthropic_converter.AnthropicConverter._block_deltas(
    block: typing.Dict[str, typing.Any]
) -> typing.List[typing.Dict[str, typing.Any]]

nemo_gym.anthropic_converter.AnthropicConverter._build_image_data_url(
    media_type: str,
    data: str
) -> str

nemo_gym.anthropic_converter.AnthropicConverter._content_to_anthropic_blocks(
    content: typing.Any,
    role: str
) -> typing.List[typing.Dict[str, typing.Any]]

nemo_gym.anthropic_converter.AnthropicConverter._content_to_text(
    content: typing.Any
) -> str

nemo_gym.anthropic_converter.AnthropicConverter._copy_sampling_params(
    body_dict: typing.Dict[str, typing.Any],
    anthropic_body: typing.Dict[str, typing.Any]
) -> None

nemo_gym.anthropic_converter.AnthropicConverter._copy_thinking_params(
    anthropic_body: typing.Dict[str, typing.Any],
    thinking: typing.Optional[typing.Dict[str, typing.Any]],
    thinking_budget_tokens: typing.Optional[int]
) -> None

nemo_gym.anthropic_converter.AnthropicConverter._copy_tool_choice(
    body_dict: typing.Dict[str, typing.Any],
    anthropic_body: typing.Dict[str, typing.Any]
) -> None

nemo_gym.anthropic_converter.AnthropicConverter._copy_tools(
    body_dict: typing.Dict[str, typing.Any],
    anthropic_body: typing.Dict[str, typing.Any]
) -> None

nemo_gym.anthropic_converter.AnthropicConverter._empty_block_shell(
    block: typing.Dict[str, typing.Any]
) -> typing.Dict[str, typing.Any]

nemo_gym.anthropic_converter.AnthropicConverter._flush_text_output(
    pending_text: typing.List[str],
    output: typing.List[typing.Any]
) -> None

nemo_gym.anthropic_converter.AnthropicConverter._function_call_to_tool_use(
    item: typing.Dict[str, typing.Any]
) -> typing.Dict[str, typing.Any]

nemo_gym.anthropic_converter.AnthropicConverter._incomplete_details_from_stop_reason(
    stop_reason: typing.Optional[str]
) -> typing.Optional[typing.Dict[str, str]]

nemo_gym.anthropic_converter.AnthropicConverter._input_image_to_anthropic_block(
    part: typing.Dict[str, typing.Any]
) -> typing.Dict[str, typing.Any]

nemo_gym.anthropic_converter.AnthropicConverter._iter_output_dicts(
    response: nemo_gym.openai_utils.NeMoGymResponse
) -> typing.List[typing.Dict[str, typing.Any]]

nemo_gym.anthropic_converter.AnthropicConverter._json_object_from_arguments(
    arguments: str
) -> typing.Dict[str, typing.Any]

nemo_gym.anthropic_converter.AnthropicConverter._model_disallows_sampling_params(
    model: str
) -> bool

nemo_gym.anthropic_converter.AnthropicConverter._normalize_input(
    response_input: typing.Any
) -> typing.List[typing.Dict[str, typing.Any]]

nemo_gym.anthropic_converter.AnthropicConverter._output_message_to_anthropic_blocks(
    item: typing.Dict[str, typing.Any]
) -> typing.List[typing.Dict[str, typing.Any]]

nemo_gym.anthropic_converter.AnthropicConverter._parse_image_data_url(
    image_url: str
) -> tuple[str, str]

nemo_gym.anthropic_converter.AnthropicConverter._reasoning_item_to_anthropic_blocks(
    item: typing.Dict[str, typing.Any]
) -> typing.List[typing.Dict[str, typing.Any]]

nemo_gym.anthropic_converter.AnthropicConverter._sse_event(
    event_type: str,
    data: typing.Dict[str, typing.Any]
) -> str

nemo_gym.anthropic_converter.AnthropicConverter._stop_reason_from_response(
    response: nemo_gym.openai_utils.NeMoGymResponse,
    has_tool_use: bool
) -> str

nemo_gym.anthropic_converter.AnthropicConverter._system_parts_to_anthropic_blocks(
    system_parts: typing.List[str]
) -> typing.List[typing.Dict[str, str]]

nemo_gym.anthropic_converter.AnthropicConverter._usage_to_responses_usage(
    usage: typing.Optional[typing.Dict[str, typing.Any]]
) -> typing.Optional[nemo_gym.openai_utils.NeMoGymResponseUsage]

nemo_gym.anthropic_converter.AnthropicConverter._validate_sampling_params_for_model(
    model: str,
    anthropic_body: typing.Dict[str, typing.Any]
) -> None

nemo_gym.anthropic_converter.AnthropicConverter.anthropic_request_to_responses(
    anthropic_body: anthropic.types.message_create_params.MessageCreateParams
) -> nemo_gym.openai_utils.NeMoGymResponseCreateParamsNonStreaming

Inverse of responses_to_anthropic (the request direction).

Parses an inbound Anthropic Messages request into Responses create params so it can be forwarded to a downstream Gym model server’s /v1/responses.

anthropic_body is hinted with the Anthropic SDK’s native MessageCreateParams (a TypedDict union, so it accepts stream: true). It’s a type hint only — at runtime the value is the raw request dict; we read fields defensively so the proxy stays permissive toward unknown / future-beta fields the Claude Code CLI may send.

nemo_gym.anthropic_converter.AnthropicConverter.anthropic_response_to_sse(
    anthropic_response: typing.Dict[str, typing.Any]
) -> typing.Iterator[str]

Synthesize an Anthropic Messages SSE stream from a complete response object.

The downstream call is non-streaming; this fakes the event sequence the Claude Code CLI expects: message_start -> per-block (content_block_start -> content_block_delta -> content_block_stop) -> message_delta -> message_stop.

nemo_gym.anthropic_converter.AnthropicConverter.anthropic_to_responses(
    anthropic_response: typing.Dict[str, typing.Any],
    request_body: nemo_gym.openai_utils.NeMoGymResponseCreateParamsNonStreaming,
    model: str
) -> nemo_gym.openai_utils.NeMoGymResponse

nemo_gym.anthropic_converter.AnthropicConverter.responses_to_anthropic(
    body: nemo_gym.openai_utils.NeMoGymResponseCreateParamsNonStreaming,
    model: str,
    max_tokens: int,
    thinking: typing.Optional[typing.Dict[str, typing.Any]],
    thinking_budget_tokens: typing.Optional[int],
    extra_body: typing.Dict[str, typing.Any]
) -> typing.Dict[str, typing.Any]

nemo_gym.anthropic_converter.AnthropicConverter.responses_to_anthropic_response(
    response: nemo_gym.openai_utils.NeMoGymResponse,
    model: str
) -> typing.Dict[str, typing.Any]

Inverse of anthropic_to_responses (the response direction).

Renders a downstream /v1/responses result as a complete Anthropic Messages response object (non-streaming shape). Token-id / logprob fields are intentionally dropped here; they are carried out-of-band by the ingress server’s side channel.

The assembled object is validated by constructing NeMoGymAnthropicMessage (a thin subclass of the Anthropic SDK’s Message) — catching malformed blocks / bad stop_reason / missing fields at the boundary — then returned as a JSON dict for the SSE synthesizer and the non-streaming JSON response. exclude_none keeps the lean Anthropic shape (drops null SDK-only fields).

nemo_gym.anthropic_converter.SUPPORTED_ANTHROPIC_IMAGE_MEDIA_TYPES = {'image/jpeg', 'image/png', 'image/gif', 'image/webp'}