From 1c1c8e6f47301b7e8abb2b996a18d0786e009944 Mon Sep 17 00:00:00 2001 From: Felipe Vieira Frujeri Date: Tue, 16 Jun 2026 18:51:56 +0000 Subject: [PATCH 1/4] Add a default /v1/messages (Anthropic Messages) route to the base Gym model server. Signed-off-by: Felipe Vieira Frujeri --- README.md | 1 + nemo_gym/anthropic_converter.py | 795 ++++++++++++++++++ nemo_gym/base_responses_api_model.py | 45 +- ...ng_gym_claude_code_agent_model_server.yaml | 48 ++ .../claude_code_agent/README.md | 18 +- tests/unit_tests/test_anthropic_converter.py | 520 ++++++++++++ .../test_anthropic_converter_egress.py | 382 +++++++++ .../test_responses_api_model_messages.py | 152 ++++ 8 files changed, 1958 insertions(+), 3 deletions(-) create mode 100644 nemo_gym/anthropic_converter.py create mode 100644 resources_servers/reasoning_gym/configs/reasoning_gym_claude_code_agent_model_server.yaml create mode 100644 tests/unit_tests/test_anthropic_converter.py create mode 100644 tests/unit_tests/test_anthropic_converter_egress.py create mode 100644 tests/unit_tests/test_responses_api_model_messages.py diff --git a/README.md b/README.md index eb53464dc0..afc51556a8 100644 --- a/README.md +++ b/README.md @@ -267,6 +267,7 @@ The Dataset column links to publicly available datasets (e.g., on HuggingFace). | Proof Verification | math | Proof verification scored against ground truth and meta-verifier agreement | - | - | - | - | proof_verification.yaml | - | | Rdkit Chemistry | knowledge | Molecular chemistry question answering: calculate properties of SMILES. Includes a mix of tool-use (python + rdkit) and no-tool-use questions. | Improve molecular reasoning and SMILES parsing. | ✓ | - | TBD | rdkit_chemistry.yaml | - | | Reasoning Gym | knowledge | Claude Code agent harness for reasoning gym tasks | Evaluate model capabilities in the Claude Code agent harness | ✓ | - | Creative Commons Attribution 4.0 International | reasoning_gym_claude_code_agent.yaml | Nemotron-RL-ReasoningGym-v1 | +| Reasoning Gym | knowledge | Claude Code agent harness for reasoning gym tasks, via a Gym model server's /v1/messages | Showcase Claude Code running against any Gym model backend | - | - | - | reasoning_gym_claude_code_agent_model_server.yaml | - | | Reasoning Gym | knowledge | LangGraph orchestrator agent compatible with resource servers that do not use tools; enables diverse agent training data and test time scaling vs a simple agent, extensible to use tools or other agent architectures | Iterative test time scaling for improved performance in reasoning tasks | ✓ | - | Apache 2.0 | orchestrator_agent.yaml | - | | Reasoning Gym | knowledge | LangGraph parallel thinking agent compatible with resource servers that do not use tools; enables diverse agent training data and test time scaling vs a simple agent, extensible to use tools or other agent architectures | Iterative test time scaling for improved performance in reasoning tasks | ✓ | - | Apache 2.0 | parallel_thinking_agent.yaml | - | | Reasoning Gym | knowledge | LangGraph reflection agent compatible with resource servers that do not use tools; provides iterative reflection for diverse agent training data and test time scaling, extensible to use tools or other agent architectures | Iterative test time scaling for improved performance in reasoning tasks | ✓ | - | Apache 2.0 | reflection_agent.yaml | - | diff --git a/nemo_gym/anthropic_converter.py b/nemo_gym/anthropic_converter.py new file mode 100644 index 0000000000..6fce5eddf3 --- /dev/null +++ b/nemo_gym/anthropic_converter.py @@ -0,0 +1,795 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""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. +""" + +import base64 +import binascii +import json +from time import time +from typing import Any, Dict, Iterator, List, Optional +from uuid import uuid4 + +from nemo_gym.openai_utils import ( + NeMoGymEasyInputMessage, + NeMoGymFunctionCallOutput, + NeMoGymResponse, + NeMoGymResponseCreateParamsNonStreaming, + NeMoGymResponseFunctionToolCall, + NeMoGymResponseInputTokensDetails, + NeMoGymResponseOutputMessage, + NeMoGymResponseOutputText, + NeMoGymResponseOutputTokensDetails, + NeMoGymResponseReasoningItem, + NeMoGymResponseUsage, + NeMoGymSummary, +) + + +SUPPORTED_ANTHROPIC_IMAGE_MEDIA_TYPES = {"image/jpeg", "image/png", "image/gif", "image/webp"} + + +class AnthropicConverter: + ############################################################################ + # Egress: NeMo Gym Responses -> Anthropic Messages request + ############################################################################ + def responses_to_anthropic( + self, + body: NeMoGymResponseCreateParamsNonStreaming, + model: str, + max_tokens: int, + thinking: Optional[Dict[str, Any]], + thinking_budget_tokens: Optional[int], + extra_body: Dict[str, Any], + ) -> Dict[str, Any]: + body_dict = body.model_dump(exclude_unset=True) + anthropic_body = dict(extra_body) + anthropic_body.update( + { + "model": model, + "max_tokens": body_dict.pop("max_output_tokens", None) or max_tokens, + "messages": [], + } + ) + + system_parts = [] + if body.instructions: + system_parts.append(body.instructions) + + response_input = body_dict.pop("input") + input_items = self._normalize_input(response_input) + for item in input_items: + item_type = item.get("type") or "message" + if item_type == "message": + self._append_message_item(item, anthropic_body["messages"], system_parts) + elif item_type == "reasoning": + self._append_content( + anthropic_body["messages"], + "assistant", + self._reasoning_item_to_anthropic_blocks(item), + ) + elif item_type == "function_call": + self._append_content( + anthropic_body["messages"], + "assistant", + [self._function_call_to_tool_use(item)], + ) + elif item_type == "function_call_output": + self._append_content( + anthropic_body["messages"], + "user", + [ + { + "type": "tool_result", + "tool_use_id": item["call_id"], + "content": item["output"], + } + ], + ) + else: + raise NotImplementedError(f"Unsupported Responses API item type for Anthropic: {item_type}") + + if system_parts: + anthropic_body["system"] = self._system_parts_to_anthropic_blocks(system_parts) + + self._copy_sampling_params(body_dict, anthropic_body) + self._validate_sampling_params_for_model(model, anthropic_body) + self._copy_tools(body_dict, anthropic_body) + self._copy_tool_choice(body_dict, anthropic_body) + self._copy_thinking_params( + anthropic_body=anthropic_body, + thinking=thinking, + thinking_budget_tokens=thinking_budget_tokens, + ) + + return anthropic_body + + # ---- egress-only policy (see module boundary note) ---- + def _copy_thinking_params( + self, + anthropic_body: Dict[str, Any], + thinking: Optional[Dict[str, Any]], + thinking_budget_tokens: Optional[int], + ) -> None: + configured_sources = sum( + source_is_set + for source_is_set in ( + "thinking" in anthropic_body, + thinking is not None, + thinking_budget_tokens is not None, + ) + ) + if configured_sources > 1: + raise ValueError( + "Configure Anthropic thinking in only one place: thinking, thinking_budget_tokens, or extra_body." + ) + + if thinking is not None: + anthropic_body["thinking"] = thinking + elif thinking_budget_tokens is not None: + anthropic_body["thinking"] = { + "type": "enabled", + "budget_tokens": thinking_budget_tokens, + } + + def _validate_sampling_params_for_model(self, model: str, anthropic_body: Dict[str, Any]) -> None: + if not self._model_disallows_sampling_params(model): + return + configured_sampling_params = [ + param for param in ("temperature", "top_p", "top_k") if anthropic_body.get(param) is not None + ] + if configured_sampling_params: + raise ValueError( + f"{model} does not support configurable sampling parameters; omit {configured_sampling_params}." + ) + + def _model_disallows_sampling_params(self, model: str) -> bool: + return any(model_id in model for model_id in ("claude-opus-4-7", "claude-opus-4-8")) + + ############################################################################ + # Egress: Anthropic Messages response -> NeMo Gym Responses + ############################################################################ + def anthropic_to_responses( + self, + anthropic_response: Dict[str, Any], + request_body: NeMoGymResponseCreateParamsNonStreaming, + model: str, + ) -> NeMoGymResponse: + output = self._anthropic_content_to_output_items(anthropic_response.get("content", [])) + if not output: + self._flush_text_output([""], output) + + usage = self._usage_to_responses_usage(anthropic_response.get("usage")) + stop_reason = anthropic_response.get("stop_reason") + incomplete_details = self._incomplete_details_from_stop_reason(stop_reason) + + return NeMoGymResponse( + id=f"resp_{uuid4().hex}", + created_at=int(time()), + model=model, + object="response", + output=[item.model_dump() for item in output], + tool_choice=request_body.tool_choice, + parallel_tool_calls=request_body.parallel_tool_calls, + tools=request_body.tools, + temperature=request_body.temperature, + top_p=request_body.top_p, + background=request_body.background, + max_output_tokens=request_body.max_output_tokens, + max_tool_calls=request_body.max_tool_calls, + previous_response_id=request_body.previous_response_id, + prompt=request_body.prompt, + reasoning=request_body.reasoning, + service_tier=request_body.service_tier, + text=request_body.text, + top_logprobs=request_body.top_logprobs, + truncation=request_body.truncation, + metadata=request_body.metadata, + instructions=request_body.instructions, + user=request_body.user, + incomplete_details=incomplete_details, + usage=usage, + ) + + def _anthropic_content_to_output_items(self, content: List[Dict[str, Any]]) -> List[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). + """ + output: List[Any] = [] + pending_text: List[str] = [] + for block in content: + block_type = block.get("type") + if block_type == "text": + pending_text.append(block.get("text", "")) + elif block_type == "thinking": + self._flush_text_output(pending_text, output) + output.append( + NeMoGymResponseReasoningItem( + id=f"rs_{uuid4().hex}", + summary=[ + NeMoGymSummary( + text=block.get("thinking") or block.get("text", ""), + type="summary_text", + ) + ], + encrypted_content=block.get("signature"), + ) + ) + elif block_type == "tool_use": + self._flush_text_output(pending_text, output) + output.append( + NeMoGymResponseFunctionToolCall( + arguments=json.dumps(block.get("input", {})), + call_id=block["id"], + name=block["name"], + id=block["id"], + status="completed", + ) + ) + else: + raise NotImplementedError(f"Unsupported Anthropic content block type: {block_type}") + + self._flush_text_output(pending_text, output) + return output + + def _incomplete_details_from_stop_reason(self, stop_reason: Optional[str]) -> Optional[Dict[str, str]]: + if stop_reason in ("max_tokens", "model_context_window_exceeded"): + return {"reason": "max_output_tokens"} + if stop_reason == "refusal": + return {"reason": "content_filter"} + return None + + ############################################################################ + # Ingress: Anthropic Messages request -> NeMo Gym Responses + ############################################################################ + def anthropic_request_to_responses( + self, anthropic_body: Dict[str, Any] + ) -> 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``. + """ + params: Dict[str, Any] = {"input": self._anthropic_messages_to_input_items(anthropic_body)} + + instructions = self._anthropic_system_to_instructions(anthropic_body.get("system")) + if instructions: + params["instructions"] = instructions + + if anthropic_body.get("model") is not None: + params["model"] = anthropic_body["model"] + if anthropic_body.get("max_tokens") is not None: + params["max_output_tokens"] = anthropic_body["max_tokens"] + if anthropic_body.get("temperature") is not None: + params["temperature"] = anthropic_body["temperature"] + if anthropic_body.get("top_p") is not None: + params["top_p"] = anthropic_body["top_p"] + + tools = self._anthropic_tools_to_responses(anthropic_body.get("tools")) + if tools: + params["tools"] = tools + tool_choice = self._anthropic_tool_choice_to_responses(anthropic_body.get("tool_choice")) + if tool_choice is not None: + params["tool_choice"] = tool_choice + + return NeMoGymResponseCreateParamsNonStreaming(**params) + + def _anthropic_system_to_instructions(self, system: Any) -> str: + if system is None: + return "" + if isinstance(system, str): + return system + return "\n".join(block["text"] for block in system if block.get("type") == "text" and block.get("text")) + + def _anthropic_messages_to_input_items(self, anthropic_body: Dict[str, Any]) -> List[Any]: + items: List[Any] = [] + for message in anthropic_body.get("messages", []): + role = message["role"] + content = message.get("content", "") + if isinstance(content, str): + items.append(NeMoGymEasyInputMessage(role=role, content=content, type="message")) + continue + self._append_anthropic_blocks_as_items(role, content, items) + return items + + def _append_anthropic_blocks_as_items(self, role: str, blocks: List[Dict[str, Any]], items: List[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. + """ + pending_parts: List[Dict[str, Any]] = [] + + def flush_message() -> None: + if not pending_parts: + return + if len(pending_parts) == 1 and pending_parts[0]["type"] == "input_text": + items.append(NeMoGymEasyInputMessage(role=role, content=pending_parts[0]["text"], type="message")) + else: + items.append(NeMoGymEasyInputMessage(role=role, content=list(pending_parts), type="message")) + pending_parts.clear() + + for block in blocks: + block_type = block.get("type") + if block_type == "text": + pending_parts.append({"type": "input_text", "text": block.get("text", "")}) + elif block_type == "image": + pending_parts.append(self._anthropic_image_to_input_part(block)) + elif block_type == "tool_use": + flush_message() + items.append( + NeMoGymResponseFunctionToolCall( + arguments=json.dumps(block.get("input", {})), + call_id=block["id"], + name=block["name"], + id=block["id"], + status="completed", + type="function_call", + ) + ) + elif block_type == "tool_result": + flush_message() + items.append( + NeMoGymFunctionCallOutput( + call_id=block["tool_use_id"], + output=self._anthropic_tool_result_content_to_text(block.get("content", "")), + type="function_call_output", + ) + ) + elif block_type == "thinking": + flush_message() + items.append( + NeMoGymResponseReasoningItem( + id=f"rs_{uuid4().hex}", + summary=[NeMoGymSummary(text=block.get("thinking", ""), type="summary_text")], + encrypted_content=block.get("signature"), + type="reasoning", + ) + ) + else: + raise NotImplementedError(f"Unsupported Anthropic content block type for ingress: {block_type}") + flush_message() + + def _anthropic_image_to_input_part(self, block: Dict[str, Any]) -> Dict[str, Any]: + source = block.get("source") or {} + if source.get("type") != "base64": + raise NotImplementedError("Anthropic ingress supports base64 image sources only.") + media_type = source["media_type"] + if media_type not in SUPPORTED_ANTHROPIC_IMAGE_MEDIA_TYPES: + raise ValueError( + f"Unsupported Anthropic image media type. Supported types: " + f"{sorted(SUPPORTED_ANTHROPIC_IMAGE_MEDIA_TYPES)}." + ) + return { + "type": "input_image", + "image_url": self._build_image_data_url(media_type, source["data"]), + "detail": "auto", + } + + def _anthropic_tool_result_content_to_text(self, content: Any) -> str: + if isinstance(content, str): + return content + texts = [] + for block in content: + if block.get("type") == "text": + texts.append(block.get("text", "")) + else: + raise NotImplementedError( + f"Unsupported Anthropic tool_result content block for ingress: {block.get('type')}" + ) + return "\n".join(texts) + + def _anthropic_tools_to_responses(self, tools: Any) -> List[Dict[str, Any]]: + if not tools: + return [] + responses_tools = [] + for tool in tools: + responses_tools.append( + { + "type": "function", + "name": tool["name"], + "description": tool.get("description"), + "parameters": tool.get("input_schema") or {"type": "object", "properties": {}}, + "strict": False, + } + ) + return responses_tools + + def _anthropic_tool_choice_to_responses(self, tool_choice: Any) -> Any: + if tool_choice is None: + return None + choice_type = tool_choice.get("type") + if choice_type == "auto": + return "auto" + if choice_type == "none": + return "none" + if choice_type == "any": + return "required" + if choice_type == "tool": + return {"type": "function", "name": tool_choice["name"]} + raise NotImplementedError(f"Unsupported Anthropic tool_choice for ingress: {tool_choice}") + + def _build_image_data_url(self, media_type: str, data: str) -> str: + return f"data:{media_type};base64,{data}" + + ############################################################################ + # Ingress: NeMo Gym Responses -> Anthropic Messages response (+ SSE) + ############################################################################ + def responses_to_anthropic_response(self, response: NeMoGymResponse, model: str) -> Dict[str, 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. + """ + content: List[Dict[str, Any]] = [] + has_tool_use = False + for item in self._iter_output_dicts(response): + item_type = item.get("type") or "message" + if item_type == "message": + content.extend(self._output_message_to_anthropic_blocks(item)) + elif item_type == "reasoning": + content.extend(self._reasoning_item_to_anthropic_blocks(item)) + elif item_type == "function_call": + content.append(self._function_call_to_tool_use(item)) + has_tool_use = True + else: + raise NotImplementedError(f"Unsupported Responses output item for Anthropic response: {item_type}") + + usage = response.usage.model_dump() if response.usage is not None else None + return { + "id": f"msg_{uuid4().hex}", + "type": "message", + "role": "assistant", + "model": model, + "content": content, + "stop_reason": self._stop_reason_from_response(response, has_tool_use), + "stop_sequence": None, + "usage": { + "input_tokens": (usage or {}).get("input_tokens", 0), + "output_tokens": (usage or {}).get("output_tokens", 0), + }, + } + + def _iter_output_dicts(self, response: NeMoGymResponse) -> List[Dict[str, Any]]: + items = [] + for item in response.output or []: + items.append(item if isinstance(item, dict) else item.model_dump()) + return items + + def _output_message_to_anthropic_blocks(self, item: Dict[str, Any]) -> List[Dict[str, Any]]: + blocks = [] + for part in item.get("content", []): + part_type = part.get("type") + if part_type == "output_text": + blocks.append({"type": "text", "text": part.get("text", "")}) + elif part_type == "refusal": + blocks.append({"type": "text", "text": part.get("refusal", "")}) + else: + raise NotImplementedError(f"Unsupported output_text part for Anthropic response: {part_type}") + return blocks + + def _stop_reason_from_response(self, response: NeMoGymResponse, has_tool_use: bool) -> str: + incomplete = response.incomplete_details + reason = incomplete.reason if incomplete is not None else None + if reason == "max_output_tokens": + return "max_tokens" + if reason == "content_filter": + return "refusal" + if has_tool_use: + return "tool_use" + return "end_turn" + + def anthropic_response_to_sse(self, anthropic_response: Dict[str, Any]) -> 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``. + """ + content = anthropic_response.get("content", []) + usage = anthropic_response.get("usage", {}) + + message_shell = {k: v for k, v in anthropic_response.items() if k != "content"} + message_shell["content"] = [] + message_shell.setdefault("usage", {}) + yield self._sse_event("message_start", {"type": "message_start", "message": message_shell}) + + for index, block in enumerate(content): + yield self._sse_event( + "content_block_start", + {"type": "content_block_start", "index": index, "content_block": self._empty_block_shell(block)}, + ) + for delta in self._block_deltas(block): + yield self._sse_event( + "content_block_delta", {"type": "content_block_delta", "index": index, "delta": delta} + ) + yield self._sse_event("content_block_stop", {"type": "content_block_stop", "index": index}) + + yield self._sse_event( + "message_delta", + { + "type": "message_delta", + "delta": { + "stop_reason": anthropic_response.get("stop_reason"), + "stop_sequence": anthropic_response.get("stop_sequence"), + }, + "usage": {"output_tokens": usage.get("output_tokens", 0)}, + }, + ) + yield self._sse_event("message_stop", {"type": "message_stop"}) + + def _empty_block_shell(self, block: Dict[str, Any]) -> Dict[str, Any]: + block_type = block.get("type") + if block_type == "text": + return {"type": "text", "text": ""} + if block_type == "thinking": + return {"type": "thinking", "thinking": ""} + if block_type == "tool_use": + return {"type": "tool_use", "id": block["id"], "name": block["name"], "input": {}} + raise NotImplementedError(f"Unsupported Anthropic block for SSE synthesis: {block_type}") + + def _block_deltas(self, block: Dict[str, Any]) -> List[Dict[str, Any]]: + block_type = block.get("type") + if block_type == "text": + return [{"type": "text_delta", "text": block.get("text", "")}] + if block_type == "thinking": + return [{"type": "thinking_delta", "thinking": block.get("thinking", "")}] + if block_type == "tool_use": + return [{"type": "input_json_delta", "partial_json": json.dumps(block.get("input", {}))}] + raise NotImplementedError(f"Unsupported Anthropic block for SSE synthesis: {block_type}") + + def _sse_event(self, event_type: str, data: Dict[str, Any]) -> str: + return f"event: {event_type}\ndata: {json.dumps(data)}\n\n" + + ############################################################################ + # Shared structural helpers + ############################################################################ + def _normalize_input(self, response_input: Any) -> List[Dict[str, Any]]: + if isinstance(response_input, str): + return [NeMoGymEasyInputMessage(content=response_input, role="user").model_dump(exclude_unset=True)] + return [ + item.model_dump(exclude_unset=True) if hasattr(item, "model_dump") else item for item in response_input + ] + + def _append_message_item( + self, + item: Dict[str, Any], + messages: List[Dict[str, Any]], + system_parts: List[str], + ) -> None: + role = item["role"] + content = item.get("content", "") + if role in ("system", "developer"): + system_parts.append(self._content_to_text(content)) + return + if role not in ("user", "assistant"): + raise NotImplementedError(f"Unsupported Responses API role for Anthropic: {role}") + self._append_content(messages, role, self._content_to_anthropic_blocks(content, role)) + + def _append_content( + self, + messages: List[Dict[str, Any]], + role: str, + content_blocks: List[Dict[str, Any]], + ) -> None: + if messages and messages[-1]["role"] == role: + messages[-1]["content"].extend(content_blocks) + else: + messages.append({"role": role, "content": content_blocks}) + + def _content_to_anthropic_blocks(self, content: Any, role: str) -> List[Dict[str, Any]]: + if isinstance(content, str): + return [{"type": "text", "text": content}] + blocks = [] + for part in content: + part_type = part.get("type") + if part_type in ("input_text", "output_text", "text"): + blocks.append({"type": "text", "text": part["text"]}) + elif part_type == "input_image" and role == "user": + blocks.append(self._input_image_to_anthropic_block(part)) + elif part_type == "refusal" and role == "assistant": + blocks.append({"type": "text", "text": part["refusal"]}) + else: + raise NotImplementedError(f"Unsupported content part for Anthropic: {part_type}") + return blocks + + def _input_image_to_anthropic_block(self, part: Dict[str, Any]) -> Dict[str, Any]: + image_url = part.get("image_url") + if isinstance(image_url, dict): + image_url = image_url.get("url") + if not isinstance(image_url, str): + raise ValueError("Responses input_image.image_url must be a base64 data URL string.") + + media_type, data = self._parse_image_data_url(image_url) + return { + "type": "image", + "source": { + "type": "base64", + "media_type": media_type, + "data": data, + }, + } + + def _parse_image_data_url(self, image_url: str) -> tuple[str, str]: + if not image_url.startswith("data:"): + raise ValueError("Anthropic image inputs require base64 data URLs; remote image URLs are not supported.") + + header, separator, data = image_url.partition(",") + if not separator or not data: + raise ValueError("Responses input_image.image_url must include base64 image data.") + + metadata = header[len("data:") :].split(";") + media_type = metadata[0].lower() + if media_type == "image/jpg": + media_type = "image/jpeg" + if "base64" not in metadata[1:]: + raise ValueError("Responses input_image.image_url must be base64 encoded.") + if media_type not in SUPPORTED_ANTHROPIC_IMAGE_MEDIA_TYPES: + raise ValueError( + "Unsupported Anthropic image media type. Supported types: " + f"{sorted(SUPPORTED_ANTHROPIC_IMAGE_MEDIA_TYPES)}." + ) + + try: + base64.b64decode(data, validate=True) + except binascii.Error as exc: + raise ValueError("Responses input_image.image_url contains invalid base64 image data.") from exc + + return media_type, data + + def _content_to_text(self, content: Any) -> str: + if isinstance(content, str): + return content + texts = [] + for part in content: + part_type = part.get("type") + if part_type in ("input_text", "output_text", "text"): + texts.append(part["text"]) + else: + raise NotImplementedError(f"Unsupported system content part for Anthropic: {part_type}") + return "\n".join(texts) + + def _system_parts_to_anthropic_blocks(self, system_parts: List[str]) -> List[Dict[str, str]]: + return [{"type": "text", "text": text} for text in system_parts if text] + + def _reasoning_item_to_anthropic_blocks(self, item: Dict[str, Any]) -> List[Dict[str, Any]]: + blocks = [] + for summary in item.get("summary", []): + block = { + "type": "thinking", + "thinking": summary["text"], + } + if item.get("encrypted_content"): + block["signature"] = item["encrypted_content"] + blocks.append(block) + return blocks + + def _function_call_to_tool_use(self, item: Dict[str, Any]) -> Dict[str, Any]: + return { + "type": "tool_use", + "id": item["call_id"], + "name": item["name"], + "input": self._json_object_from_arguments(item["arguments"]), + } + + def _json_object_from_arguments(self, arguments: str) -> Dict[str, Any]: + parsed = json.loads(arguments or "{}") + if not isinstance(parsed, dict): + raise ValueError(f"Anthropic tool_use input must be a JSON object, got {type(parsed).__name__}") + return parsed + + def _copy_sampling_params(self, body_dict: Dict[str, Any], anthropic_body: Dict[str, Any]) -> None: + for source, target in ( + ("temperature", "temperature"), + ("top_p", "top_p"), + ): + value = body_dict.get(source) + if value is not None: + anthropic_body[target] = value + + def _copy_tools(self, body_dict: Dict[str, Any], anthropic_body: Dict[str, Any]) -> None: + tools = body_dict.get("tools") or [] + if not tools: + return + + anthropic_tools = [] + for tool in tools: + if tool.get("type") != "function": + raise NotImplementedError(f"Unsupported Responses API tool type for Anthropic: {tool.get('type')}") + anthropic_tool = { + "name": tool["name"], + "input_schema": tool.get("parameters") or {"type": "object", "properties": {}}, + } + if tool.get("description"): + anthropic_tool["description"] = tool["description"] + anthropic_tools.append(anthropic_tool) + anthropic_body["tools"] = anthropic_tools + + def _copy_tool_choice(self, body_dict: Dict[str, Any], anthropic_body: Dict[str, Any]) -> None: + tool_choice = body_dict.get("tool_choice") + if tool_choice is None: + return + if isinstance(tool_choice, str): + if tool_choice == "required": + anthropic_body["tool_choice"] = {"type": "any"} + elif tool_choice in ("auto", "none"): + anthropic_body["tool_choice"] = {"type": tool_choice} + else: + raise NotImplementedError(f"Unsupported tool_choice for Anthropic: {tool_choice}") + elif isinstance(tool_choice, dict) and tool_choice.get("type") == "function": + anthropic_body["tool_choice"] = {"type": "tool", "name": tool_choice["name"]} + else: + raise NotImplementedError(f"Unsupported tool_choice for Anthropic: {tool_choice}") + + def _flush_text_output(self, pending_text: List[str], output: List[Any]) -> None: + if not pending_text: + return + output.append( + NeMoGymResponseOutputMessage( + id=f"msg_{uuid4().hex}", + content=[ + NeMoGymResponseOutputText( + annotations=[], + text="".join(pending_text), + ) + ], + role="assistant", + status="completed", + type="message", + ) + ) + pending_text.clear() + + def _usage_to_responses_usage(self, usage: Optional[Dict[str, Any]]) -> Optional[NeMoGymResponseUsage]: + if usage is None: + return None + input_tokens = usage.get("input_tokens", 0) + output_tokens = usage.get("output_tokens", 0) + return NeMoGymResponseUsage( + input_tokens=input_tokens, + input_tokens_details=NeMoGymResponseInputTokensDetails( + cached_tokens=usage.get("cache_read_input_tokens", 0) + ), + output_tokens=output_tokens, + output_tokens_details=NeMoGymResponseOutputTokensDetails(reasoning_tokens=0), + total_tokens=input_tokens + output_tokens, + ) diff --git a/nemo_gym/base_responses_api_model.py b/nemo_gym/base_responses_api_model.py index e20f14c579..b969afd563 100644 --- a/nemo_gym/base_responses_api_model.py +++ b/nemo_gym/base_responses_api_model.py @@ -12,10 +12,13 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. +import inspect from abc import abstractmethod -from fastapi import Body, FastAPI +from fastapi import Body, FastAPI, Request +from fastapi.responses import StreamingResponse +from nemo_gym.anthropic_converter import AnthropicConverter from nemo_gym.openai_utils import ( NeMoGymChatCompletion, NeMoGymChatCompletionCreateParamsNonStreaming, @@ -25,6 +28,10 @@ from nemo_gym.server_utils import BaseRunServerInstanceConfig, BaseServer, SimpleServer +# Stateless; shared by every model server's default /v1/messages handler. +_ANTHROPIC_CONVERTER = AnthropicConverter() + + class BaseResponsesAPIModelConfig(BaseRunServerInstanceConfig): pass @@ -43,6 +50,12 @@ def setup_webserver(self) -> FastAPI: app.post("/v1/responses")(self.responses) + # Every Gym model server speaks the Anthropic Messages API by default, mapping + # Messages <-> Responses around its own responses() implementation. This lets blackbox + # harnesses that require an Anthropic endpoint (e.g. the Claude Code CLI) target any + # model server directly. + app.post("/v1/messages")(self.messages) + return app @abstractmethod @@ -54,3 +67,33 @@ async def chat_completions( @abstractmethod async def responses(self, body: NeMoGymResponseCreateParamsNonStreaming = Body()) -> NeMoGymResponse: pass + + async def messages(self, request: Request, body: dict = Body()): + """Default Anthropic Messages <-> Responses mapping shared by every Gym model server. + + Translates the inbound Anthropic Messages request to the Responses API, delegates to this + server's own ``responses()`` (so it reuses whatever backend the server has), and maps the + result back to an Anthropic Messages response. When the client requested ``stream: true`` + (the Claude Code CLI always does), the complete response is re-emitted as a synthesized + Anthropic SSE event stream. Servers may override this for native Messages handling. + """ + params = _ANTHROPIC_CONVERTER.anthropic_request_to_responses(body) + response = await self._invoke_responses(request, params) + model_name = body.get("model") or response.model + anthropic_response = _ANTHROPIC_CONVERTER.responses_to_anthropic_response(response, model=model_name) + if body.get("stream"): + return StreamingResponse( + _ANTHROPIC_CONVERTER.anthropic_response_to_sse(anthropic_response), + media_type="text/event-stream", + ) + return anthropic_response + + async def _invoke_responses( + self, request: Request, params: NeMoGymResponseCreateParamsNonStreaming + ) -> NeMoGymResponse: + # responses() signatures vary across servers: some take a leading `request`, some only + # `body`. Dispatch on whichever this server declares so the default messages() works for + # all of them. + if "request" in inspect.signature(self.responses).parameters: + return await self.responses(request=request, body=params) + return await self.responses(body=params) diff --git a/resources_servers/reasoning_gym/configs/reasoning_gym_claude_code_agent_model_server.yaml b/resources_servers/reasoning_gym/configs/reasoning_gym_claude_code_agent_model_server.yaml new file mode 100644 index 0000000000..1e56bc95cf --- /dev/null +++ b/resources_servers/reasoning_gym/configs/reasoning_gym_claude_code_agent_model_server.yaml @@ -0,0 +1,48 @@ +# Showcase: claude_code_agent against a NeMo Gym model server's /v1/messages (not Anthropic). +# +# Every Gym model server exposes /v1/messages (default Messages <-> Responses mapping on +# SimpleResponsesAPIModel). The agent's `model_server` ref resolves ANTHROPIC_BASE_URL to the +# model server; the CLI appends /v1/messages. Needs NO anthropic_* env vars. +# +# Compose with any model server as `policy_model`, e.g. a vLLM OpenAI-compatible endpoint: +# ng_run "+config_paths=[\ +# resources_servers/reasoning_gym/configs/reasoning_gym_claude_code_agent_model_server.yaml,\ +# responses_api_models/vllm_model/configs/vllm_model.yaml]" +# ng_collect_rollouts \ +# +agent_name=reasoning_gym_claude_code_agent_model_server \ +# +input_jsonl_fpath=resources_servers/reasoning_gym/data/example.jsonl \ +# +output_jsonl_fpath=claude_code_via_model_server_rollout.jsonl +limit=1 + +reasoning_gym: + resources_servers: + reasoning_gym: + entrypoint: app.py + domain: knowledge + verified: false + description: Claude Code agent harness for reasoning gym tasks, via a Gym model server's /v1/messages + value: Showcase Claude Code running against any Gym model backend + +reasoning_gym_claude_code_agent_model_server: + responses_api_agents: + claude_code_agent: + entrypoint: app.py + resources_server: + type: resources_servers + name: reasoning_gym + model_server: + type: responses_api_models + name: policy_model + concurrency: 32 + model: ${policy_model_name} + anthropic_api_key: EMPTY # pragma: allowlist secret + anthropic_base_url: null + max_turns: 30 + timeout: 300 + thinking: disabled + system_prompt: | + You are a precise reasoning assistant. You have access to Bash to run Python for calculations. + For every problem: think step by step, use code to verify when helpful, and state your final answer clearly. + datasets: + - name: example + type: example + jsonl_fpath: resources_servers/reasoning_gym/data/example.jsonl diff --git a/responses_api_agents/claude_code_agent/README.md b/responses_api_agents/claude_code_agent/README.md index fb6c65f5e2..ac8c9c2200 100644 --- a/responses_api_agents/claude_code_agent/README.md +++ b/responses_api_agents/claude_code_agent/README.md @@ -26,12 +26,24 @@ anthropic_base_url: http://localhost:8000 ### Launch -No model server is needed for basic eval. To extend this agent to training, a model server should be developed that handles messages endpoint. For evals with the current version, just pass the resources server config, which includes the agent server config, as is the current standard in NeMo Gym: +For a quick eval against Anthropic's API (or any endpoint set via `anthropic_base_url`), pass the resources server config, which includes the agent server config: ```bash ng_run "+config_paths=[resources_servers/reasoning_gym/configs/reasoning_gym_claude_code_agent.yaml]" ``` +#### Against a Gym model server + +Every Gym model server now exposes `POST /v1/messages` (a default Messages ↔ Responses mapping on `SimpleResponsesAPIModel`), so Claude Code can run against any backend Gym serves — vLLM, OpenAI, an inference provider. Set the agent's `model_server` ref to that server (it takes precedence over `anthropic_base_url`); the harness resolves `ANTHROPIC_BASE_URL` to it and the CLI appends `/v1/messages`. + +`reasoning_gym_claude_code_agent_model_server.yaml` wires the agent's `model_server` ref to `policy_model`. Compose it with any model server (here a vLLM serving `policy_model`): + +```bash +ng_run "+config_paths=[resources_servers/reasoning_gym/configs/reasoning_gym_claude_code_agent_model_server.yaml,responses_api_models/vllm_model/configs/vllm_model.yaml]" +``` + +This path needs only the model server's `policy_base_url`, `policy_api_key`, and `policy_model_name` (in `env.yaml` or as `+` overrides) — no `anthropic_*` vars. + ### Run the agent ```bash @@ -42,11 +54,13 @@ ng_collect_rollouts \ +limit=1 ``` +For the model-server config above, use `+agent_name=reasoning_gym_claude_code_agent_model_server`. + ## Description The agent runs `claude -p` as an async subprocess for each request. Claude Code handles all tool execution (Bash, file read/write) internally. The agent parses the stream-json output into NeMoGym output items and forwards the response to a resources server for verification. -Claude Code talks to the model via the Anthropic Messages API (`/v1/messages`). This means it can connect to Anthropic's API directly, or to any local endpoint that implements `/v1/messages` such as vLLM or Ollama. It does not go through a Gym model server, but that is the next step to extend this integration to training and additional features. +Claude Code talks to the model via the Anthropic Messages API (`/v1/messages`). This means it can connect to Anthropic's API directly, to any local endpoint that implements `/v1/messages` (vLLM, Ollama), or — via the agent's `model_server` ref — to any NeMo Gym model server, since every Gym model server now serves `/v1/messages` by mapping Messages ↔ Responses around its own `responses()` backend. The agent runs with `--bare`, which skips auto-discovery of hooks, skills, plugins, MCP servers, auto memory, and CLAUDE.md so each scripted call starts clean and fast; Claude still has access to Bash, file read, and file edit tools. To enable MCP servers or skills, remove `--bare` and add the relevant flags in `app.py`'s `_run_claude_code` command list. diff --git a/tests/unit_tests/test_anthropic_converter.py b/tests/unit_tests/test_anthropic_converter.py new file mode 100644 index 0000000000..4a290387d1 --- /dev/null +++ b/tests/unit_tests/test_anthropic_converter.py @@ -0,0 +1,520 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Tests for the ingress (inverse) direction of the shared AnthropicConverter. + +The egress direction (Responses -> Anthropic request, Anthropic response -> Responses) is +covered by responses_api_models/anthropic_model/tests/test_app.py. These tests cover the new +inverse direction used by an Anthropic Messages ingress proxy, plus round-trips that guard the +two directions against drift. +""" + +import json + +from nemo_gym.anthropic_converter import AnthropicConverter +from nemo_gym.openai_utils import NeMoGymResponseCreateParamsNonStreaming + + +PNG_DATA_URL = "data:image/png;base64,aGVsbG8=" # "hello" + + +def _converter() -> AnthropicConverter: + return AnthropicConverter() + + +class TestAnthropicRequestToResponses: + def test_system_string_and_user_text(self) -> None: + params = _converter().anthropic_request_to_responses( + { + "model": "m", + "system": "Be concise.", + "max_tokens": 256, + "temperature": 0.5, + "top_p": 0.9, + "messages": [{"role": "user", "content": "Hello"}], + } + ) + assert params.instructions == "Be concise." + assert params.model == "m" + assert params.max_output_tokens == 256 + assert params.temperature == 0.5 + assert params.top_p == 0.9 + assert len(params.input) == 1 + assert params.input[0].role == "user" + assert params.input[0].content == "Hello" + + def test_system_block_list_is_joined(self) -> None: + params = _converter().anthropic_request_to_responses( + { + "system": [ + {"type": "text", "text": "Answer concisely."}, + {"type": "text", "text": "Use JSON."}, + ], + "max_tokens": 10, + "messages": [{"role": "user", "content": "hi"}], + } + ) + assert params.instructions == "Answer concisely.\nUse JSON." + + def test_no_system_leaves_instructions_unset(self) -> None: + params = _converter().anthropic_request_to_responses( + {"max_tokens": 10, "messages": [{"role": "user", "content": "hi"}]} + ) + assert params.instructions is None + + def test_user_text_and_image_blocks(self) -> None: + params = _converter().anthropic_request_to_responses( + { + "max_tokens": 10, + "messages": [ + { + "role": "user", + "content": [ + {"type": "text", "text": "What is this?"}, + { + "type": "image", + "source": {"type": "base64", "media_type": "image/png", "data": "aGVsbG8="}, + }, + ], + } + ], + } + ) + content = params.input[0].content + assert content[0] == {"type": "input_text", "text": "What is this?"} + assert content[1]["type"] == "input_image" + assert content[1]["image_url"] == PNG_DATA_URL + + def test_assistant_tool_use_becomes_function_call(self) -> None: + params = _converter().anthropic_request_to_responses( + { + "max_tokens": 10, + "messages": [ + { + "role": "assistant", + "content": [ + {"type": "text", "text": "calling"}, + {"type": "tool_use", "id": "toolu_1", "name": "lookup", "input": {"city": "Paris"}}, + ], + } + ], + } + ) + # text message, then function_call + assert params.input[0].role == "assistant" + assert params.input[0].content == "calling" + fc = params.input[1] + assert fc.type == "function_call" + assert fc.call_id == "toolu_1" + assert fc.name == "lookup" + assert json.loads(fc.arguments) == {"city": "Paris"} + + def test_tool_result_becomes_function_call_output(self) -> None: + params = _converter().anthropic_request_to_responses( + { + "max_tokens": 10, + "messages": [ + { + "role": "user", + "content": [{"type": "tool_result", "tool_use_id": "toolu_1", "content": "Sunny"}], + } + ], + } + ) + out = params.input[0] + assert out.type == "function_call_output" + assert out.call_id == "toolu_1" + assert out.output == "Sunny" + + def test_tool_result_block_list_content_is_flattened(self) -> None: + params = _converter().anthropic_request_to_responses( + { + "max_tokens": 10, + "messages": [ + { + "role": "user", + "content": [ + { + "type": "tool_result", + "tool_use_id": "toolu_1", + "content": [{"type": "text", "text": "a"}, {"type": "text", "text": "b"}], + } + ], + } + ], + } + ) + assert params.input[0].output == "a\nb" + + def test_thinking_block_becomes_reasoning_item(self) -> None: + params = _converter().anthropic_request_to_responses( + { + "max_tokens": 10, + "messages": [ + { + "role": "assistant", + "content": [{"type": "thinking", "thinking": "hmm", "signature": "sig-1"}], + } + ], + } + ) + item = params.input[0] + assert item.type == "reasoning" + assert item.summary[0].text == "hmm" + assert item.encrypted_content == "sig-1" + + def test_tools_and_tool_choice_variants(self) -> None: + conv = _converter() + params = conv.anthropic_request_to_responses( + { + "max_tokens": 10, + "messages": [{"role": "user", "content": "x"}], + "tools": [{"name": "f", "description": "d", "input_schema": {"type": "object", "properties": {}}}], + "tool_choice": {"type": "any"}, + } + ) + assert params.tools[0]["type"] == "function" + assert params.tools[0]["name"] == "f" + assert params.tools[0]["parameters"] == {"type": "object", "properties": {}} + assert params.tool_choice == "required" + + assert conv._anthropic_tool_choice_to_responses({"type": "auto"}) == "auto" + assert conv._anthropic_tool_choice_to_responses({"type": "none"}) == "none" + assert conv._anthropic_tool_choice_to_responses({"type": "tool", "name": "f"}) == { + "type": "function", + "name": "f", + } + assert conv._anthropic_tool_choice_to_responses(None) is None + + def test_unsupported_block_raises(self) -> None: + import pytest + + with pytest.raises(NotImplementedError): + _converter().anthropic_request_to_responses( + { + "max_tokens": 10, + "messages": [{"role": "user", "content": [{"type": "video", "data": "x"}]}], + } + ) + + def test_unsupported_tool_choice_raises(self) -> None: + import pytest + + with pytest.raises(NotImplementedError): + _converter()._anthropic_tool_choice_to_responses({"type": "weird"}) + + def test_unsupported_image_source_raises(self) -> None: + import pytest + + with pytest.raises(NotImplementedError): + _converter()._anthropic_image_to_input_part({"source": {"type": "url", "url": "http://x"}}) + + def test_unsupported_image_media_type_raises(self) -> None: + import pytest + + with pytest.raises(ValueError): + _converter()._anthropic_image_to_input_part( + {"source": {"type": "base64", "media_type": "image/tiff", "data": "x"}} + ) + + def test_unsupported_tool_result_block_raises(self) -> None: + import pytest + + with pytest.raises(NotImplementedError): + _converter()._anthropic_tool_result_content_to_text([{"type": "image", "source": {}}]) + + +class TestResponsesToAnthropicResponse: + def _response_from_anthropic(self, anthropic_response: dict): + conv = _converter() + request_body = NeMoGymResponseCreateParamsNonStreaming(input="hi") + return conv, conv.anthropic_to_responses(anthropic_response, request_body=request_body, model="m") + + def test_text_and_tool_use_and_stop_reason(self) -> None: + conv, resp = self._response_from_anthropic( + { + "content": [ + {"type": "text", "text": "Hello"}, + {"type": "tool_use", "id": "toolu_1", "name": "f", "input": {"a": 1}}, + ], + "stop_reason": "tool_use", + "usage": {"input_tokens": 5, "output_tokens": 7}, + } + ) + out = conv.responses_to_anthropic_response(resp, model="m") + assert out["role"] == "assistant" + assert out["model"] == "m" + assert out["content"][0] == {"type": "text", "text": "Hello"} + tool_use = out["content"][1] + assert tool_use["type"] == "tool_use" + assert tool_use["id"] == "toolu_1" + assert tool_use["input"] == {"a": 1} + assert out["stop_reason"] == "tool_use" + assert out["usage"] == {"input_tokens": 5, "output_tokens": 7} + + def test_reasoning_becomes_thinking_block(self) -> None: + conv, resp = self._response_from_anthropic( + { + "content": [ + {"type": "thinking", "thinking": "step", "signature": "sig"}, + {"type": "text", "text": "ok"}, + ], + "stop_reason": "end_turn", + "usage": {"input_tokens": 1, "output_tokens": 2}, + } + ) + out = conv.responses_to_anthropic_response(resp, model="m") + thinking = out["content"][0] + assert thinking["type"] == "thinking" + assert thinking["thinking"] == "step" + assert thinking["signature"] == "sig" + assert out["stop_reason"] == "end_turn" + + def test_max_tokens_stop_reason(self) -> None: + conv, resp = self._response_from_anthropic( + { + "content": [{"type": "text", "text": "x"}], + "stop_reason": "max_tokens", + "usage": {"input_tokens": 1, "output_tokens": 1}, + } + ) + out = conv.responses_to_anthropic_response(resp, model="m") + assert out["stop_reason"] == "max_tokens" + + def test_refusal_stop_reason(self) -> None: + conv, resp = self._response_from_anthropic( + { + "content": [{"type": "text", "text": "x"}], + "stop_reason": "refusal", + "usage": {"input_tokens": 1, "output_tokens": 1}, + } + ) + out = conv.responses_to_anthropic_response(resp, model="m") + assert out["stop_reason"] == "refusal" + + def test_missing_usage_defaults_to_zero(self) -> None: + conv = _converter() + request_body = NeMoGymResponseCreateParamsNonStreaming(input="hi") + resp = conv.anthropic_to_responses( + {"content": [{"type": "text", "text": "x"}], "stop_reason": "end_turn"}, + request_body=request_body, + model="m", + ) + out = conv.responses_to_anthropic_response(resp, model="m") + assert out["usage"] == {"input_tokens": 0, "output_tokens": 0} + + +class TestAnthropicResponseToSSE: + def _events(self, anthropic_response: dict): + raw = list(_converter().anthropic_response_to_sse(anthropic_response)) + parsed = [] + for chunk in raw: + lines = chunk.strip().split("\n") + event_type = lines[0].removeprefix("event: ") + data = json.loads(lines[1].removeprefix("data: ")) + parsed.append((event_type, data)) + return parsed + + def test_event_ordering_and_framing(self) -> None: + events = self._events( + { + "id": "msg_1", + "type": "message", + "role": "assistant", + "model": "m", + "content": [ + {"type": "text", "text": "hi"}, + {"type": "tool_use", "id": "toolu_1", "name": "f", "input": {"a": 1}}, + ], + "stop_reason": "tool_use", + "stop_sequence": None, + "usage": {"input_tokens": 3, "output_tokens": 4}, + } + ) + types = [t for t, _ in events] + assert types == [ + "message_start", + "content_block_start", + "content_block_delta", + "content_block_stop", + "content_block_start", + "content_block_delta", + "content_block_stop", + "message_delta", + "message_stop", + ] + # message_start carries an empty content list + assert events[0][1]["message"]["content"] == [] + # text delta + assert events[2][1]["delta"] == {"type": "text_delta", "text": "hi"} + # tool_use input arrives as input_json_delta + assert events[5][1]["delta"]["type"] == "input_json_delta" + assert json.loads(events[5][1]["delta"]["partial_json"]) == {"a": 1} + # message_delta carries stop_reason + output usage + assert events[7][1]["delta"]["stop_reason"] == "tool_use" + assert events[7][1]["usage"] == {"output_tokens": 4} + + def test_thinking_block_delta(self) -> None: + events = self._events( + { + "id": "msg_1", + "role": "assistant", + "model": "m", + "content": [{"type": "thinking", "thinking": "ponder", "signature": "s"}], + "stop_reason": "end_turn", + "usage": {"input_tokens": 1, "output_tokens": 1}, + } + ) + delta_events = [d for t, d in events if t == "content_block_delta"] + assert delta_events[0]["delta"] == {"type": "thinking_delta", "thinking": "ponder"} + + def test_unsupported_block_for_sse_raises(self) -> None: + import pytest + + with pytest.raises(NotImplementedError): + list(_converter().anthropic_response_to_sse({"content": [{"type": "image"}], "usage": {}})) + + +class TestRoundTrips: + def test_request_round_trip_preserves_messages_system_tools(self) -> None: + conv = _converter() + original = { + "model": "claude-sonnet-4-6", + "system": "Be helpful.", + "max_tokens": 100, + "messages": [ + {"role": "user", "content": [{"type": "text", "text": "weather?"}]}, + { + "role": "assistant", + "content": [{"type": "tool_use", "id": "toolu_1", "name": "lookup", "input": {"city": "Paris"}}], + }, + {"role": "user", "content": [{"type": "tool_result", "tool_use_id": "toolu_1", "content": "Sunny"}]}, + ], + "tools": [ + { + "name": "lookup", + "description": "Look up weather.", + "input_schema": {"type": "object", "properties": {}}, + } + ], + "tool_choice": {"type": "auto"}, + } + params = conv.anthropic_request_to_responses(original) + rebuilt = conv.responses_to_anthropic( + body=params, + model="claude-sonnet-4-6", + max_tokens=100, + thinking=None, + thinking_budget_tokens=None, + extra_body={}, + ) + assert rebuilt["system"] == [{"type": "text", "text": "Be helpful."}] + assert rebuilt["messages"] == original["messages"] + assert rebuilt["tools"] == [ + {"name": "lookup", "description": "Look up weather.", "input_schema": {"type": "object", "properties": {}}} + ] + assert rebuilt["tool_choice"] == {"type": "auto"} + + def test_response_round_trip_preserves_content(self) -> None: + conv = _converter() + request_body = NeMoGymResponseCreateParamsNonStreaming(input="hi") + anthropic_response = { + "content": [ + {"type": "text", "text": "Hello"}, + {"type": "tool_use", "id": "toolu_1", "name": "f", "input": {"a": 1}}, + ], + "stop_reason": "tool_use", + "usage": {"input_tokens": 5, "output_tokens": 7}, + } + resp = conv.anthropic_to_responses(anthropic_response, request_body=request_body, model="m") + rebuilt = conv.responses_to_anthropic_response(resp, model="m") + assert rebuilt["content"] == anthropic_response["content"] + assert rebuilt["stop_reason"] == "tool_use" + + +class TestSharedHelperBranches: + """Cover egress/shared helper branches now owned by this module.""" + + def test_empty_anthropic_content_yields_empty_message(self) -> None: + conv = _converter() + request_body = NeMoGymResponseCreateParamsNonStreaming(input="hi") + resp = conv.anthropic_to_responses( + {"content": [], "usage": {"input_tokens": 1, "output_tokens": 0}}, request_body, "m" + ) + out = conv.responses_to_anthropic_response(resp, model="m") + assert out["content"] == [{"type": "text", "text": ""}] + assert out["stop_reason"] == "end_turn" + + def test_output_message_refusal_becomes_text_block(self) -> None: + blocks = _converter()._output_message_to_anthropic_blocks({"content": [{"type": "refusal", "refusal": "no"}]}) + assert blocks == [{"type": "text", "text": "no"}] + + def test_output_message_unsupported_part_raises(self) -> None: + import pytest + + with pytest.raises(NotImplementedError): + _converter()._output_message_to_anthropic_blocks({"content": [{"type": "weird"}]}) + + def test_egress_assistant_refusal_block(self) -> None: + blocks = _converter()._content_to_anthropic_blocks([{"type": "refusal", "refusal": "no"}], "assistant") + assert blocks == [{"type": "text", "text": "no"}] + + def test_egress_image_url_dict_form(self) -> None: + block = _converter()._input_image_to_anthropic_block( + {"type": "input_image", "image_url": {"url": PNG_DATA_URL}} + ) + assert block["source"]["media_type"] == "image/png" + assert block["source"]["data"] == "aGVsbG8=" + + def test_egress_image_url_non_string_raises(self) -> None: + import pytest + + with pytest.raises(ValueError): + _converter()._input_image_to_anthropic_block({"type": "input_image", "image_url": 123}) + + def test_parse_image_data_url_jpg_normalized_and_validations(self) -> None: + import pytest + + conv = _converter() + media_type, data = conv._parse_image_data_url("data:image/jpg;base64,aGVsbG8=") + assert media_type == "image/jpeg" and data == "aGVsbG8=" + + with pytest.raises(ValueError): # no base64 data + conv._parse_image_data_url("data:image/png;base64,") + with pytest.raises(ValueError): # not declared base64 + conv._parse_image_data_url("data:image/png,aGVsbG8=") + with pytest.raises(ValueError): # unsupported media type + conv._parse_image_data_url("data:image/tiff;base64,aGVsbG8=") + with pytest.raises(ValueError): # invalid base64 payload + conv._parse_image_data_url("data:image/png;base64,!!!notb64!!!") + + def test_content_to_text_list_and_unsupported(self) -> None: + import pytest + + conv = _converter() + assert conv._content_to_text([{"type": "input_text", "text": "a"}, {"type": "text", "text": "b"}]) == "a\nb" + with pytest.raises(NotImplementedError): + conv._content_to_text([{"type": "input_image", "image_url": "x"}]) + + def test_json_object_from_arguments_rejects_non_object(self) -> None: + import pytest + + with pytest.raises(ValueError): + _converter()._json_object_from_arguments("[1, 2]") + + def test_copy_tool_choice_required_maps_to_any(self) -> None: + conv = _converter() + anthropic_body: dict = {} + conv._copy_tool_choice({"tool_choice": "required"}, anthropic_body) + assert anthropic_body["tool_choice"] == {"type": "any"} diff --git a/tests/unit_tests/test_anthropic_converter_egress.py b/tests/unit_tests/test_anthropic_converter_egress.py new file mode 100644 index 0000000000..882c856086 --- /dev/null +++ b/tests/unit_tests/test_anthropic_converter_egress.py @@ -0,0 +1,382 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Egress-direction tests for the shared AnthropicConverter (Responses -> Anthropic request, +Anthropic response -> Responses). Mirrors the converter coverage that the egress anthropic_model +server's test suite provides on the #1546 branch, kept here so the shared converter module is +fully covered on this (ingress) branch where the egress server is absent.""" + +import json + +import pytest + +from nemo_gym.anthropic_converter import AnthropicConverter +from nemo_gym.openai_utils import NeMoGymResponseCreateParamsNonStreaming + + +class TestAnthropicConverter: + def test_responses_to_anthropic_maps_messages_tools_and_thinking(self) -> None: + converter = AnthropicConverter() + body = NeMoGymResponseCreateParamsNonStreaming( + input=[ + { + "type": "message", + "role": "developer", + "content": "Be concise.", + }, + { + "type": "message", + "role": "user", + "content": [{"type": "input_text", "text": "What is the weather?"}], + }, + { + "type": "reasoning", + "id": "rs_123", + "summary": [{"type": "summary_text", "text": "Need weather data."}], + "encrypted_content": "signature_123", + }, + { + "type": "function_call", + "call_id": "toolu_123", + "name": "get_weather", + "arguments": '{"city": "San Francisco"}', + }, + { + "type": "function_call_output", + "call_id": "toolu_123", + "output": '{"temperature": 65}', + }, + ], + instructions="You are helpful.", + max_output_tokens=512, + temperature=0.2, + tools=[ + { + "type": "function", + "name": "get_weather", + "description": "Get weather.", + "parameters": { + "type": "object", + "properties": {"city": {"type": "string"}}, + "required": ["city"], + }, + "strict": True, + } + ], + tool_choice={"type": "function", "name": "get_weather"}, + ) + + actual = converter.responses_to_anthropic( + body=body, + model="claude-sonnet-4-20250514", + max_tokens=4096, + thinking=None, + thinking_budget_tokens=1024, + extra_body={"metadata": {"request_id": "abc"}}, + ) + + assert actual == { + "metadata": {"request_id": "abc"}, + "model": "claude-sonnet-4-20250514", + "max_tokens": 512, + "messages": [ + { + "role": "user", + "content": [{"type": "text", "text": "What is the weather?"}], + }, + { + "role": "assistant", + "content": [ + { + "type": "thinking", + "thinking": "Need weather data.", + "signature": "signature_123", + }, + { + "type": "tool_use", + "id": "toolu_123", + "name": "get_weather", + "input": {"city": "San Francisco"}, + }, + ], + }, + { + "role": "user", + "content": [ + { + "type": "tool_result", + "tool_use_id": "toolu_123", + "content": '{"temperature": 65}', + } + ], + }, + ], + "system": [ + {"type": "text", "text": "You are helpful."}, + {"type": "text", "text": "Be concise."}, + ], + "temperature": 0.2, + "tools": [ + { + "name": "get_weather", + "description": "Get weather.", + "input_schema": { + "type": "object", + "properties": {"city": {"type": "string"}}, + "required": ["city"], + }, + } + ], + "tool_choice": {"type": "tool", "name": "get_weather"}, + "thinking": {"type": "enabled", "budget_tokens": 1024}, + } + + def test_anthropic_to_responses_maps_text_thinking_tools_and_usage(self) -> None: + converter = AnthropicConverter() + request_body = NeMoGymResponseCreateParamsNonStreaming(input="hello") + + response = converter.anthropic_to_responses( + anthropic_response={ + "id": "msg_123", + "type": "message", + "role": "assistant", + "model": "claude-sonnet-4-20250514", + "content": [ + { + "type": "thinking", + "thinking": "I should call a tool.", + "signature": "signature_123", + }, + {"type": "text", "text": "Let me check."}, + { + "type": "tool_use", + "id": "toolu_123", + "name": "get_weather", + "input": {"city": "San Francisco"}, + }, + ], + "stop_reason": "tool_use", + "usage": {"input_tokens": 10, "output_tokens": 20, "cache_read_input_tokens": 3}, + }, + request_body=request_body, + model="claude-sonnet-4-20250514", + ) + + assert response.model == "claude-sonnet-4-20250514" + assert response.output[0].type == "reasoning" + assert response.output[0].summary[0].text == "I should call a tool." + assert response.output[0].encrypted_content == "signature_123" + assert response.output[1].type == "message" + assert response.output[1].content[0].text == "Let me check." + assert response.output[2].type == "function_call" + assert response.output[2].call_id == "toolu_123" + assert response.output[2].name == "get_weather" + assert json.loads(response.output[2].arguments) == {"city": "San Francisco"} + assert response.usage.input_tokens == 10 + assert response.usage.output_tokens == 20 + assert response.usage.total_tokens == 30 + assert response.usage.input_tokens_details.cached_tokens == 3 + + def test_anthropic_to_responses_maps_stop_reasons_to_incomplete_details(self) -> None: + converter = AnthropicConverter() + request_body = NeMoGymResponseCreateParamsNonStreaming(input="hello") + + base_response = { + "id": "msg_123", + "type": "message", + "role": "assistant", + "model": "claude-sonnet-4-20250514", + "content": [{"type": "text", "text": "Partial response."}], + } + + max_tokens_response = converter.anthropic_to_responses( + anthropic_response=base_response | {"stop_reason": "max_tokens"}, + request_body=request_body, + model="claude-sonnet-4-20250514", + ) + assert max_tokens_response.incomplete_details.reason == "max_output_tokens" + + context_response = converter.anthropic_to_responses( + anthropic_response=base_response | {"stop_reason": "model_context_window_exceeded"}, + request_body=request_body, + model="claude-sonnet-4-20250514", + ) + assert context_response.incomplete_details.reason == "max_output_tokens" + + refusal_response = converter.anthropic_to_responses( + anthropic_response=base_response | {"stop_reason": "refusal"}, + request_body=request_body, + model="claude-sonnet-4-20250514", + ) + assert refusal_response.incomplete_details.reason == "content_filter" + + tool_use_response = converter.anthropic_to_responses( + anthropic_response=base_response | {"stop_reason": "tool_use"}, + request_body=request_body, + model="claude-sonnet-4-20250514", + ) + assert tool_use_response.incomplete_details is None + + def test_responses_to_anthropic_maps_typed_adaptive_thinking(self) -> None: + converter = AnthropicConverter() + body = NeMoGymResponseCreateParamsNonStreaming(input="Hello") + + actual = converter.responses_to_anthropic( + body=body, + model="claude-opus-4-8", + max_tokens=1024, + thinking={"type": "adaptive"}, + thinking_budget_tokens=None, + extra_body={}, + ) + + assert actual["thinking"] == {"type": "adaptive"} + + def test_responses_to_anthropic_maps_input_image_data_url(self) -> None: + converter = AnthropicConverter() + body = NeMoGymResponseCreateParamsNonStreaming( + input=[ + { + "type": "message", + "role": "user", + "content": [ + {"type": "input_text", "text": "What is in this image?"}, + { + "type": "input_image", + "image_url": "data:image/png;base64,iVBORw0KGgo=", + "detail": "high", + }, + ], + } + ] + ) + + actual = converter.responses_to_anthropic( + body=body, + model="claude-sonnet-4-20250514", + max_tokens=1024, + thinking=None, + thinking_budget_tokens=None, + extra_body={}, + ) + + assert actual["messages"] == [ + { + "role": "user", + "content": [ + {"type": "text", "text": "What is in this image?"}, + { + "type": "image", + "source": { + "type": "base64", + "media_type": "image/png", + "data": "iVBORw0KGgo=", + }, + }, + ], + } + ] + + def test_responses_to_anthropic_rejects_remote_image_url(self) -> None: + converter = AnthropicConverter() + body = NeMoGymResponseCreateParamsNonStreaming( + input=[ + { + "type": "message", + "role": "user", + "content": [ + { + "type": "input_image", + "image_url": "https://example.com/image.png", + "detail": "high", + } + ], + } + ] + ) + + with pytest.raises(ValueError, match="base64 data URLs"): + converter.responses_to_anthropic( + body=body, + model="claude-sonnet-4-20250514", + max_tokens=1024, + thinking=None, + thinking_budget_tokens=None, + extra_body={}, + ) + + def test_responses_to_anthropic_rejects_invalid_image_data_url(self) -> None: + converter = AnthropicConverter() + body = NeMoGymResponseCreateParamsNonStreaming( + input=[ + { + "type": "message", + "role": "user", + "content": [ + { + "type": "input_image", + "image_url": "data:image/png;base64,not valid base64", + "detail": "high", + } + ], + } + ] + ) + + with pytest.raises(ValueError, match="invalid base64"): + converter.responses_to_anthropic( + body=body, + model="claude-sonnet-4-20250514", + max_tokens=1024, + thinking=None, + thinking_budget_tokens=None, + extra_body={}, + ) + + def test_responses_to_anthropic_rejects_ambiguous_thinking_config(self) -> None: + converter = AnthropicConverter() + body = NeMoGymResponseCreateParamsNonStreaming(input="Hello") + + with pytest.raises(ValueError, match="Configure Anthropic thinking in only one place"): + converter.responses_to_anthropic( + body=body, + model="claude-opus-4-8", + max_tokens=1024, + thinking={"type": "adaptive"}, + thinking_budget_tokens=1024, + extra_body={}, + ) + + def test_responses_to_anthropic_rejects_opus_4_8_sampling_params(self) -> None: + converter = AnthropicConverter() + + with pytest.raises(ValueError, match="does not support configurable sampling"): + converter.responses_to_anthropic( + body=NeMoGymResponseCreateParamsNonStreaming(input="Hello", temperature=0.2), + model="claude-opus-4-8", + max_tokens=1024, + thinking={"type": "adaptive"}, + thinking_budget_tokens=None, + extra_body={}, + ) + + with pytest.raises(ValueError, match="does not support configurable sampling"): + converter.responses_to_anthropic( + body=NeMoGymResponseCreateParamsNonStreaming(input="Hello"), + model="us/aws/anthropic/eccn-claude-opus-4-8", + max_tokens=1024, + thinking={"type": "adaptive"}, + thinking_budget_tokens=None, + extra_body={"top_k": 5}, + ) diff --git a/tests/unit_tests/test_responses_api_model_messages.py b/tests/unit_tests/test_responses_api_model_messages.py new file mode 100644 index 0000000000..f326c4b33e --- /dev/null +++ b/tests/unit_tests/test_responses_api_model_messages.py @@ -0,0 +1,152 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Tests for the default ``/v1/messages`` route on ``SimpleResponsesAPIModel``. + +Every Gym model server inherits an Anthropic Messages endpoint that maps Messages <-> Responses +around the server's own ``responses()``. These tests use minimal fake servers to exercise the +default mapping for both ``responses()`` signatures (with and without a leading ``request``). +""" + +from time import time +from unittest.mock import MagicMock +from uuid import uuid4 + +from fastapi import Body, Request +from fastapi.testclient import TestClient + +from nemo_gym.base_responses_api_model import BaseResponsesAPIModelConfig, SimpleResponsesAPIModel +from nemo_gym.openai_utils import ( + NeMoGymChatCompletion, + NeMoGymChatCompletionCreateParamsNonStreaming, + NeMoGymResponse, + NeMoGymResponseCreateParamsNonStreaming, +) +from nemo_gym.server_utils import ServerClient + + +def _build_response(text: str, model: str = "downstream-model") -> NeMoGymResponse: + return NeMoGymResponse( + id=f"resp_{uuid4().hex}", + created_at=int(time()), + model=model, + object="response", + output=[ + { + "type": "message", + "id": f"msg_{uuid4().hex}", + "role": "assistant", + "status": "completed", + "content": [{"type": "output_text", "text": text, "annotations": []}], + } + ], + tool_choice="auto", + parallel_tool_calls=True, + tools=[], + ) + + +class _BodyOnlyModel(SimpleResponsesAPIModel): + """A server whose responses() takes only `body` (like openai_model).""" + + config: BaseResponsesAPIModelConfig + last_params: object = None + model_config = {"arbitrary_types_allowed": True} + + async def responses(self, body: NeMoGymResponseCreateParamsNonStreaming = Body()) -> NeMoGymResponse: + object.__setattr__(self, "last_params", body) + return _build_response("hi from body-only") + + async def chat_completions( + self, body: NeMoGymChatCompletionCreateParamsNonStreaming = Body() + ) -> NeMoGymChatCompletion: + raise NotImplementedError + + +class _RequestAwareModel(SimpleResponsesAPIModel): + """A server whose responses() also takes `request` (like vllm_model / azure).""" + + config: BaseResponsesAPIModelConfig + saw_request: bool = False + model_config = {"arbitrary_types_allowed": True} + + async def responses( + self, request: Request, body: NeMoGymResponseCreateParamsNonStreaming = Body() + ) -> NeMoGymResponse: + object.__setattr__(self, "saw_request", isinstance(request, Request)) + return _build_response("hi from request-aware") + + async def chat_completions( + self, body: NeMoGymChatCompletionCreateParamsNonStreaming = Body() + ) -> NeMoGymChatCompletion: + raise NotImplementedError + + +def _config() -> BaseResponsesAPIModelConfig: + return BaseResponsesAPIModelConfig(host="0.0.0.0", port=8099, entrypoint="", name="") + + +def _client(model_cls) -> TestClient: + server = model_cls(config=_config(), server_client=MagicMock(spec=ServerClient)) + return TestClient(server.setup_webserver()), server + + +class TestDefaultMessagesRoute: + def test_messages_route_registered_alongside_openai_routes(self) -> None: + server = _BodyOnlyModel(config=_config(), server_client=MagicMock(spec=ServerClient)) + paths = {route.path for route in server.setup_webserver().routes} + assert {"/v1/messages", "/v1/responses", "/v1/chat/completions"} <= paths + + def test_body_only_responses_signature(self) -> None: + client, server = _client(_BodyOnlyModel) + resp = client.post( + "/v1/messages", + json={"model": "claude-x", "max_tokens": 32, "messages": [{"role": "user", "content": "hello"}]}, + ) + assert resp.status_code == 200 + data = resp.json() + assert data["role"] == "assistant" + assert data["content"] == [{"type": "text", "text": "hi from body-only"}] + assert data["model"] == "claude-x" # request model echoed back + # the inbound Anthropic request was translated to Responses params before delegating + assert server.last_params.input[0].content == "hello" + assert server.last_params.max_output_tokens == 32 + + def test_request_aware_responses_signature(self) -> None: + client, server = _client(_RequestAwareModel) + resp = client.post( + "/v1/messages", + json={"model": "claude-x", "max_tokens": 8, "messages": [{"role": "user", "content": "hi"}]}, + ) + assert resp.status_code == 200 + assert resp.json()["content"] == [{"type": "text", "text": "hi from request-aware"}] + assert server.saw_request is True # request was forwarded to responses() + + def test_streaming_returns_anthropic_sse(self) -> None: + client, _ = _client(_BodyOnlyModel) + resp = client.post( + "/v1/messages", + json={ + "model": "claude-x", + "max_tokens": 8, + "stream": True, + "messages": [{"role": "user", "content": "hi"}], + }, + ) + assert resp.status_code == 200 + assert resp.headers["content-type"].startswith("text/event-stream") + body = resp.text + assert "event: message_start" in body + assert "event: content_block_delta" in body + assert "event: message_stop" in body From 0232bbcb9928b13746fef09387b865c595c99eef Mon Sep 17 00:00:00 2001 From: Lin Jia Date: Wed, 17 Jun 2026 16:11:45 -0700 Subject: [PATCH 2/4] Use Anthropic SDK Pydantic types (types-only) in the Messages converter MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adopt the official anthropic SDK types for the /v1/messages <-> Responses mapping instead of hand-rolled dicts: validate the egress Anthropic response via Message.model_validate and hint requests with MessageCreateParams. Types only — the SDK client is never used (it uses httpx, whose O(n^2) connection pooling hangs at high concurrency); all transport stays on aiohttp. Pin anthropic<=0.109.2, mirroring the openai pin. Add converter test coverage for empty output, empty-text system lists, and system-role messages. Co-Authored-By: Claude Opus 4.8 (1M context) Signed-off-by: Lin Jia --- nemo_gym/anthropic_converter.py | 53 +++++++++---- pyproject.toml | 9 +++ tests/unit_tests/test_anthropic_converter.py | 78 ++++++++++++++++++++ uv.lock | 30 ++++++++ 4 files changed, 154 insertions(+), 16 deletions(-) diff --git a/nemo_gym/anthropic_converter.py b/nemo_gym/anthropic_converter.py index 6fce5eddf3..d70c7e2234 100644 --- a/nemo_gym/anthropic_converter.py +++ b/nemo_gym/anthropic_converter.py @@ -46,6 +46,13 @@ from typing import Any, Dict, Iterator, List, Optional from uuid import uuid4 +# Types only — never the `anthropic` client. The client uses httpx (O(n^2) connection +# pooling at high concurrency); all transport in Gym stays on aiohttp via server_utils. +# MessageCreateParams (request) is a TypedDict used purely as a hint; Message (response) +# is a BaseModel used to validate what we emit. +from anthropic.types import Message +from anthropic.types.message_create_params import MessageCreateParams + from nemo_gym.openai_utils import ( NeMoGymEasyInputMessage, NeMoGymFunctionCallOutput, @@ -281,12 +288,17 @@ def _incomplete_details_from_stop_reason(self, stop_reason: Optional[str]) -> Op # Ingress: Anthropic Messages request -> NeMo Gym Responses ############################################################################ def anthropic_request_to_responses( - self, anthropic_body: Dict[str, Any] + self, anthropic_body: MessageCreateParams ) -> 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. """ params: Dict[str, Any] = {"input": self._anthropic_messages_to_input_items(anthropic_body)} @@ -459,6 +471,11 @@ def responses_to_anthropic_response(self, response: NeMoGymResponse, model: str) 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 the Anthropic SDK's ``Message`` model + (catches 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). """ content: List[Dict[str, Any]] = [] has_tool_use = False @@ -475,19 +492,22 @@ def responses_to_anthropic_response(self, response: NeMoGymResponse, model: str) raise NotImplementedError(f"Unsupported Responses output item for Anthropic response: {item_type}") usage = response.usage.model_dump() if response.usage is not None else None - return { - "id": f"msg_{uuid4().hex}", - "type": "message", - "role": "assistant", - "model": model, - "content": content, - "stop_reason": self._stop_reason_from_response(response, has_tool_use), - "stop_sequence": None, - "usage": { - "input_tokens": (usage or {}).get("input_tokens", 0), - "output_tokens": (usage or {}).get("output_tokens", 0), - }, - } + message = Message.model_validate( + { + "id": f"msg_{uuid4().hex}", + "type": "message", + "role": "assistant", + "model": model, + "content": content, + "stop_reason": self._stop_reason_from_response(response, has_tool_use), + "stop_sequence": None, + "usage": { + "input_tokens": (usage or {}).get("input_tokens", 0), + "output_tokens": (usage or {}).get("output_tokens", 0), + }, + } + ) + return message.model_dump(mode="json", exclude_none=True) def _iter_output_dicts(self, response: NeMoGymResponse) -> List[Dict[str, Any]]: items = [] @@ -694,12 +714,13 @@ def _system_parts_to_anthropic_blocks(self, system_parts: List[str]) -> List[Dic def _reasoning_item_to_anthropic_blocks(self, item: Dict[str, Any]) -> List[Dict[str, Any]]: blocks = [] for summary in item.get("summary", []): + # Anthropic's ThinkingBlock requires a signature; open-model backends don't + # produce one, so default to "" (the synthesized SSE never emits it anyway). block = { "type": "thinking", "thinking": summary["text"], + "signature": item.get("encrypted_content") or "", } - if item.get("encrypted_content"): - block["signature"] = item["encrypted_content"] blocks.append(block) return blocks diff --git a/pyproject.toml b/pyproject.toml index ac4f4ffacc..1999dbc6e6 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -88,6 +88,15 @@ dependencies = [ # License: Apache 2.0 https://github.com/openai/openai-python/blob/a8258744cbecf51321587fc870e8920bd2c07809/LICENSE "openai<=2.7.2", + # Anthropic: We leverage the Anthropic Messages schemas (request/response Python types) for the + # /v1/messages <-> Responses mapping in nemo_gym/anthropic_converter.py. Types only — we never + # use the anthropic client (it uses httpx, whose O(n^2) connection pooling hangs at high + # concurrency); all transport stays on aiohttp. Upper-bounded since the version bumps frequently, + # mirroring how we pin openai. + # Updated Tue Jun 16, 2026 with anthropic<=0.109.2 + # License: MIT https://github.com/anthropics/anthropic-sdk-python/blob/v0.109.2/LICENSE + "anthropic<=0.109.2", + # tqdm: Used for progress tracking on batch operations. # Updated Fri Jul 25, 2025 with tqdm==4.67.1 # License: MIT https://github.com/tqdm/tqdm/blob/0ed5d7f18fa3153834cbac0aa57e8092b217cc16/LICENCE diff --git a/tests/unit_tests/test_anthropic_converter.py b/tests/unit_tests/test_anthropic_converter.py index 4a290387d1..6fe997705a 100644 --- a/tests/unit_tests/test_anthropic_converter.py +++ b/tests/unit_tests/test_anthropic_converter.py @@ -22,6 +22,8 @@ import json +from anthropic.types import Message + from nemo_gym.anthropic_converter import AnthropicConverter from nemo_gym.openai_utils import NeMoGymResponseCreateParamsNonStreaming @@ -73,6 +75,33 @@ def test_no_system_leaves_instructions_unset(self) -> None: ) assert params.instructions is None + def test_system_list_without_text_leaves_instructions_unset(self) -> None: + # A system list that contributes no usable text (empty-text blocks) yields no instructions. + params = _converter().anthropic_request_to_responses( + { + "system": [{"type": "text", "text": ""}], + "max_tokens": 10, + "messages": [{"role": "user", "content": "hi"}], + } + ) + assert params.instructions is None + + def test_system_role_message_passes_through(self) -> None: + # Anthropic allows a "system" role inside messages (distinct from the top-level system + # param); it is forwarded as a system input item rather than dropped or merged. + params = _converter().anthropic_request_to_responses( + { + "max_tokens": 10, + "messages": [ + {"role": "system", "content": "stay terse"}, + {"role": "user", "content": "hi"}, + ], + } + ) + assert params.input[0].role == "system" + assert params.input[0].content == "stay terse" + assert params.input[1].role == "user" + def test_user_text_and_image_blocks(self) -> None: params = _converter().anthropic_request_to_responses( { @@ -314,6 +343,55 @@ def test_missing_usage_defaults_to_zero(self) -> None: out = conv.responses_to_anthropic_response(resp, model="m") assert out["usage"] == {"input_tokens": 0, "output_tokens": 0} + def test_reasoning_without_signature_defaults_to_empty(self) -> None: + # Open-model reasoning carries no Anthropic signature, but the typed Message build + # requires one — default it to "" rather than dropping the block or crashing. + conv, resp = self._response_from_anthropic( + { + "content": [{"type": "thinking", "thinking": "step"}], + "stop_reason": "end_turn", + "usage": {"input_tokens": 1, "output_tokens": 1}, + } + ) + out = conv.responses_to_anthropic_response(resp, model="m") + assert out["content"][0] == {"type": "thinking", "thinking": "step", "signature": ""} + + def test_output_validates_as_anthropic_message(self) -> None: + # Regression guard: the builder must emit an object the Anthropic SDK accepts as a Message + # (this is what the internal Message.model_validate enforces on every response). + conv, resp = self._response_from_anthropic( + { + "content": [ + {"type": "text", "text": "hi"}, + {"type": "tool_use", "id": "toolu_1", "name": "f", "input": {"a": 1}}, + ], + "stop_reason": "tool_use", + "usage": {"input_tokens": 2, "output_tokens": 3}, + } + ) + out = conv.responses_to_anthropic_response(resp, model="m") + message = Message.model_validate(out) # raises if our output drifts from the SDK schema + assert message.stop_reason == "tool_use" + assert message.content[1].input == {"a": 1} + + def test_empty_output_yields_empty_content(self) -> None: + # Defensive: a downstream response carrying no output items maps to empty content, + # which is still a valid Anthropic Message. (Realistic empty responses arrive as an + # empty message item and are rendered as a single empty text block instead — see + # TestSharedHelperBranches.test_empty_anthropic_content_yields_empty_message.) + conv, resp = self._response_from_anthropic( + { + "content": [{"type": "text", "text": "x"}], + "stop_reason": "end_turn", + "usage": {"input_tokens": 1, "output_tokens": 0}, + } + ) + resp = resp.model_copy(update={"output": []}) + out = conv.responses_to_anthropic_response(resp, model="m") + assert out["content"] == [] + assert out["stop_reason"] == "end_turn" + Message.model_validate(out) # empty content is still a valid Message + class TestAnthropicResponseToSSE: def _events(self, anthropic_response: dict): diff --git a/uv.lock b/uv.lock index c34d4a2c0c..bd1f60fd21 100644 --- a/uv.lock +++ b/uv.lock @@ -196,6 +196,25 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/78/b6/6307fbef88d9b5ee7421e68d78a9f162e0da4900bc5f5793f6d3d0e34fb8/annotated_types-0.7.0-py3-none-any.whl", hash = "sha256:1f02e8b43a8fbbc3f3e0d4f0f4bfc8131bcb4eebe8849b8e5c773f3a1c582a53", size = 13643, upload-time = "2024-05-20T21:33:24.1Z" }, ] +[[package]] +name = "anthropic" +version = "0.109.2" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "anyio" }, + { name = "distro" }, + { name = "docstring-parser" }, + { name = "httpx" }, + { name = "jiter" }, + { name = "pydantic" }, + { name = "sniffio" }, + { name = "typing-extensions" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/1b/b7/9a8e2f79011e89dd6eeb599c27332aed765dac9d6fbee3a55e68e4e3ec25/anthropic-0.109.2.tar.gz", hash = "sha256:d37db299597c7bc124b49b767ff135f1e6456b64af2b2fad4b63b2a1df333cf0", size = 927559, upload-time = "2026-06-15T17:30:25.024Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/d1/f2/bee5de8a2699fc8a3cce34d61c7a2626a2c310ddde7ea5611327eb0ddbe9/anthropic-0.109.2-py3-none-any.whl", hash = "sha256:e0fb4ca5df0ed983248c9c6c3242adc81d9cfddb8725902da53698554117abac", size = 923800, upload-time = "2026-06-15T17:30:23.124Z" }, +] + [[package]] name = "antlr4-python3-runtime" version = "4.9.3" @@ -536,6 +555,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/e3/26/57c6fb270950d476074c087527a558ccb6f4436657314bfb6cdf484114c4/docker-7.1.0-py3-none-any.whl", hash = "sha256:c96b93b7f0a746f9e77d325bcfb87422a3d8bd4f03136ae8a85b37f1898d5fc0", size = 147774, upload-time = "2024-05-23T11:13:55.01Z" }, ] +[[package]] +name = "docstring-parser" +version = "0.18.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/e0/4d/f332313098c1de1b2d2ff91cf2674415cc7cddab2ca1b01ae29774bd5fdf/docstring_parser-0.18.0.tar.gz", hash = "sha256:292510982205c12b1248696f44959db3cdd1740237a968ea1e2e7a900eeb2015", size = 29341, upload-time = "2026-04-14T04:09:19.867Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/a7/5f/ed01f9a3cdffbd5a008556fc7b2a08ddb1cc6ace7effa7340604b1d16699/docstring_parser-0.18.0-py3-none-any.whl", hash = "sha256:b3fcbed555c47d8479be0796ef7e19c2670d428d72e96da63f3a40122860374b", size = 22484, upload-time = "2026-04-14T04:09:18.638Z" }, +] + [[package]] name = "docutils" version = "0.21.2" @@ -1374,6 +1402,7 @@ name = "nemo-gym" source = { editable = "." } dependencies = [ { name = "aiohttp" }, + { name = "anthropic" }, { name = "datasets" }, { name = "devtools" }, { name = "fastapi" }, @@ -1432,6 +1461,7 @@ docs = [ [package.metadata] requires-dist = [ { name = "aiohttp", specifier = ">=3.13.3" }, + { name = "anthropic", specifier = "<=0.109.2" }, { name = "coverage", extras = ["toml"], marker = "extra == 'dev'" }, { name = "datasets" }, { name = "devtools" }, From c16946c80ca29fc13f8538c59c21ea3a526db8e9 Mon Sep 17 00:00:00 2001 From: Lin Jia Date: Thu, 18 Jun 2026 16:25:40 -0700 Subject: [PATCH 3/4] docs(claude_code_agent): add /v1/messages smoke test and model-server note Co-Authored-By: Claude Opus 4.8 (1M context) Signed-off-by: Lin Jia --- .../claude_code_agent/README.md | 20 +++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/responses_api_agents/claude_code_agent/README.md b/responses_api_agents/claude_code_agent/README.md index ac8c9c2200..23929745b4 100644 --- a/responses_api_agents/claude_code_agent/README.md +++ b/responses_api_agents/claude_code_agent/README.md @@ -44,6 +44,8 @@ ng_run "+config_paths=[resources_servers/reasoning_gym/configs/reasoning_gym_cla This path needs only the model server's `policy_base_url`, `policy_api_key`, and `policy_model_name` (in `env.yaml` or as `+` overrides) — no `anthropic_*` vars. +Use `vllm_model` for OpenAI-compatible **chat** endpoints (vLLM, NVIDIA build, most providers) — it forwards to `/chat/completions`. `openai_model` forwards to the OpenAI **Responses** API (`/responses`), which only OpenAI/Azure implement, so it 404s against chat-only providers. + ### Run the agent ```bash @@ -56,6 +58,24 @@ ng_collect_rollouts \ For the model-server config above, use `+agent_name=reasoning_gym_claude_code_agent_model_server`. +### Smoke test + +Check the `/v1/messages` proxy and the real-CLI seam without a full rollout. Launch a model server, then take its URL from the `ng_run` log (`'url': 'http://127.0.0.1:'`): + +```bash +ng_run "+config_paths=[responses_api_models/vllm_model/configs/vllm_model.yaml]" \ + +policy_base_url=https://integrate.api.nvidia.com/v1 \ + '+policy_api_key=${oc.env:NVIDIA_API_KEY}' +policy_model_name=meta/llama-3.1-8b-instruct + +# 1. proxy speaks Anthropic Messages (add "stream": true for the SSE path): +curl $URL/v1/messages -H 'content-type: application/json' \ + -d '{"model":"x","max_tokens":64,"messages":[{"role":"user","content":"2+2?"}]}' + +# 2. the real Claude Code CLI runs against it: +ANTHROPIC_BASE_URL=$URL ANTHROPIC_AUTH_TOKEN=local \ + claude -p --output-format stream-json --max-turns 2 --model meta/llama-3.1-8b-instruct -- "What is 2+2?" +``` + ## Description The agent runs `claude -p` as an async subprocess for each request. Claude Code handles all tool execution (Bash, file read/write) internally. The agent parses the stream-json output into NeMoGym output items and forwards the response to a resources server for verification. From 8f347adb4a26b11815d2bb2be610659c83251da5 Mon Sep 17 00:00:00 2001 From: Lin Jia Date: Mon, 22 Jun 2026 14:28:52 -0700 Subject: [PATCH 4/4] feat: add Anthropic Messages boundary types (anthropic_utils) Add nemo_gym/anthropic_utils.py with the two boundary types for the Anthropic Messages API, mirroring openai_utils.py's wrapping strategy but deliberately minimal (no *ForTraining hierarchy, no async client): - NeMoGymAnthropicMessageCreateParamsNonStreaming: request validator, a BaseModel copy of the SDK's MessageCreateParams TypedDict with extra="forbid" and Iterable fields overridden to List for strict server-side validation at the ingress proxy. - NeMoGymAnthropicMessage: response validator, a thin subclass of the SDK's Message used to validate what we emit. Wire NeMoGymAnthropicMessage into AnthropicConverter's response validation path and add unit tests. Co-Authored-By: Claude Opus 4.8 (1M context) Signed-off-by: Lin Jia --- nemo_gym/anthropic_converter.py | 13 +-- nemo_gym/anthropic_utils.py | 112 +++++++++++++++++++++++ tests/unit_tests/test_anthropic_utils.py | 75 +++++++++++++++ 3 files changed, 194 insertions(+), 6 deletions(-) create mode 100644 nemo_gym/anthropic_utils.py create mode 100644 tests/unit_tests/test_anthropic_utils.py diff --git a/nemo_gym/anthropic_converter.py b/nemo_gym/anthropic_converter.py index d70c7e2234..ad3e7c6d82 100644 --- a/nemo_gym/anthropic_converter.py +++ b/nemo_gym/anthropic_converter.py @@ -48,11 +48,11 @@ # Types only — never the `anthropic` client. The client uses httpx (O(n^2) connection # pooling at high concurrency); all transport in Gym stays on aiohttp via server_utils. -# MessageCreateParams (request) is a TypedDict used purely as a hint; Message (response) -# is a BaseModel used to validate what we emit. -from anthropic.types import Message +# MessageCreateParams (request) is a TypedDict used purely as a hint; NeMoGymAnthropicMessage +# (response) is the BaseModel used to validate what we emit. from anthropic.types.message_create_params import MessageCreateParams +from nemo_gym.anthropic_utils import NeMoGymAnthropicMessage from nemo_gym.openai_utils import ( NeMoGymEasyInputMessage, NeMoGymFunctionCallOutput, @@ -472,8 +472,9 @@ def responses_to_anthropic_response(self, response: NeMoGymResponse, model: str) 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 the Anthropic SDK's ``Message`` model - (catches malformed blocks / bad stop_reason / missing fields at the boundary), then + 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). """ @@ -492,7 +493,7 @@ def responses_to_anthropic_response(self, response: NeMoGymResponse, model: str) raise NotImplementedError(f"Unsupported Responses output item for Anthropic response: {item_type}") usage = response.usage.model_dump() if response.usage is not None else None - message = Message.model_validate( + message = NeMoGymAnthropicMessage.model_validate( { "id": f"msg_{uuid4().hex}", "type": "message", diff --git a/nemo_gym/anthropic_utils.py b/nemo_gym/anthropic_utils.py new file mode 100644 index 0000000000..40402bd6fb --- /dev/null +++ b/nemo_gym/anthropic_utils.py @@ -0,0 +1,112 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""NeMo Gym Pydantic types for the Anthropic Messages API boundary. + +This mirrors the wrapping strategy in ``nemo_gym/openai_utils.py``, but deliberately stays +small. Anthropic is only a *wire format* at the proxy boundary — the internal canonical +representation is the Responses API (the ``NeMoGym*`` types in ``openai_utils``). So unlike +``openai_utils``, this module does not replicate the ``*ForTraining`` hierarchies or an async +client (the ``anthropic`` SDK is never used as a client — it uses httpx, whose O(n^2) +connection pooling hangs at high concurrency). Types only. + +Two boundary types, one per direction: + +* :class:`NeMoGymAnthropicMessageCreateParamsNonStreaming` — the **request** validator. The + SDK's ``MessageCreateParams`` is a ``TypedDict`` with no runtime validation, so we copy it as + a ``BaseModel`` for strict server-side validation at the ingress proxy endpoint, exactly as + ``NeMoGymResponseCreateParamsNonStreaming`` does for the Responses API. +* :class:`NeMoGymAnthropicMessage` — the **response** validator. A thin subclass of the SDK's + ``Message`` (already a ``BaseModel``), used to validate what we emit, mirroring + ``NeMoGymResponse(Response)``. +""" + +from typing import List, Literal, Optional, Union + +from anthropic.types import ( + CacheControlEphemeralParam, + MessageParam, + MetadataParam, + ModelParam, + OutputConfigParam, + TextBlockParam, + ThinkingConfigParam, + ToolChoiceParam, + ToolUnionParam, +) +from anthropic.types import Message as AnthropicMessage +from pydantic import BaseModel, ConfigDict, Field + + +######################################## +# Messages API request +######################################## + + +class NeMoGymAnthropicMessageCreateParamsNonStreaming(BaseModel): + """Copy of ``anthropic.types.message_create_params.MessageCreateParamsBase`` as a BaseModel. + + The SDK ships this as a ``TypedDict`` (no runtime validation). We need server-side + validation at the ingress proxy, so we re-declare it here, mirroring + ``NeMoGymResponseCreateParamsNonStreaming``. + + The ``Iterable`` fields (``messages``, ``tools``, and the list arm of ``system``) are + overridden to ``List`` so Pydantic eagerly validates them into real, re-iterable, + indexable, JSON-serializable lists rather than single-use lazy ``ValidatorIterator``s. + + Note on ``extra="forbid"``: this matches the strict Responses-API policy and rejects + unknown fields. Anthropic occasionally introduces beta body fields ahead of an SDK bump; + if the ingress client (e.g. the Claude Code CLI) sends one, relax this to ``ignore``. + """ + + model_config = ConfigDict(extra="forbid") + + # Required by the Anthropic API. + max_tokens: int + messages: List[MessageParam] + model: ModelParam + + cache_control: Optional[CacheControlEphemeralParam] = None + container: Optional[str] = None + inference_geo: Optional[str] = None + metadata: Optional[MetadataParam] = None + output_config: Optional[OutputConfigParam] = None + service_tier: Optional[Literal["auto", "standard_only"]] = None + stop_sequences: Optional[List[str]] = None + system: Optional[Union[str, List[TextBlockParam]]] = None + temperature: Optional[float] = None + thinking: Optional[ThinkingConfigParam] = None + tool_choice: Optional[ToolChoiceParam] = None + tools: Optional[List[ToolUnionParam]] = Field(default=None) + top_k: Optional[int] = None + top_p: Optional[float] = None + # We synthesize SSE from a complete response; we never proxy a true upstream stream. + stream: Optional[Literal[False]] = None + + +######################################## +# Messages API response +######################################## + + +class NeMoGymAnthropicMessage(AnthropicMessage): + """Thin subclass of the SDK's ``Message`` response model, used to validate what we emit. + + ``Message.content`` is already typed as ``List[...]`` in the pinned ``anthropic`` version, + so no iterable override is needed here; the subclass exists for symmetry with + ``NeMoGymResponse`` and to give the egress/ingress response path a single ``NeMoGym*`` + validation point. + """ + + pass diff --git a/tests/unit_tests/test_anthropic_utils.py b/tests/unit_tests/test_anthropic_utils.py new file mode 100644 index 0000000000..d9454cb967 --- /dev/null +++ b/tests/unit_tests/test_anthropic_utils.py @@ -0,0 +1,75 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Tests for the Anthropic Messages API boundary types in nemo_gym/anthropic_utils.py.""" + +import pytest +from pydantic import ValidationError + +from nemo_gym.anthropic_utils import ( + NeMoGymAnthropicMessage, + NeMoGymAnthropicMessageCreateParamsNonStreaming, +) + + +class TestNeMoGymAnthropicMessageCreateParamsNonStreaming: + def test_validates_request_and_eagerly_materializes_iterables(self) -> None: + req = NeMoGymAnthropicMessageCreateParamsNonStreaming.model_validate( + { + "max_tokens": 1024, + "model": "claude-opus-4-8", + "messages": [{"role": "user", "content": "hi"}], + "system": "be terse", + "tools": [{"name": "get_weather", "description": "w", "input_schema": {"type": "object"}}], + "stop_sequences": ["STOP"], + "temperature": 0.2, + } + ) + # Iterable fields are real lists, not single-use lazy ValidatorIterators. + assert isinstance(req.messages, list) + assert isinstance(req.tools, list) + assert len(req.messages) == 1 + + def test_rejects_unknown_fields(self) -> None: + with pytest.raises(ValidationError): + NeMoGymAnthropicMessageCreateParamsNonStreaming.model_validate( + {"max_tokens": 1, "model": "claude-opus-4-8", "messages": [], "bogus": 1} + ) + + def test_requires_max_tokens(self) -> None: + with pytest.raises(ValidationError): + NeMoGymAnthropicMessageCreateParamsNonStreaming.model_validate( + {"model": "claude-opus-4-8", "messages": []} + ) + + +class TestNeMoGymAnthropicMessage: + def test_validates_response_and_subclasses_sdk_message(self) -> None: + from anthropic.types import Message + + msg = NeMoGymAnthropicMessage.model_validate( + { + "id": "msg_1", + "type": "message", + "role": "assistant", + "model": "claude-opus-4-8", + "content": [{"type": "text", "text": "hello"}], + "stop_reason": "end_turn", + "stop_sequence": None, + "usage": {"input_tokens": 3, "output_tokens": 2}, + } + ) + assert isinstance(msg, Message) + assert isinstance(msg.content, list) + assert msg.content[0].text == "hello"