diff --git a/litellm/llms/anthropic/chat/handler.py b/litellm/llms/anthropic/chat/handler.py
index ed4ed295dd3..66c560013fb 100644
--- a/litellm/llms/anthropic/chat/handler.py
+++ b/litellm/llms/anthropic/chat/handler.py
@@ -11,7 +11,6 @@ from types import MappingProxyType
 from typing import TYPE_CHECKING, Any, Final, Literal, Union, cast
 
 import httpx
-from pydantic import TypeAdapter
 
 import litellm
 import litellm.litellm_core_utils
@@ -27,6 +26,8 @@ from litellm.litellm_core_utils.json_fragment_accumulator import JSONFragmentAcc
 from litellm.llms.base_llm.base_utils import (
     RESPONSE_FORMAT_ENVELOPE_NAME_KEYS,
     ResponseFormatSchema,
+    answer_json_text,
+    decoded_answer_object,
     names_a_response_format_property,
     partial_response_format_answer,
     response_format_answer_json,
@@ -91,9 +92,6 @@ def _loads_stream_chunk(payload: str) -> dict[str, object]:
     return json.loads(payload)
 
 
-_DECODED_JSON: Final = TypeAdapter(object)
-
-
 _EnvelopeVerdict = Literal["undecided", "pass_through", "hold"]
 _FIRST_OBJECT_KEY: Final = re.compile(r'^\s*\{\s*"((?:[^"\\]|\\.)*)"\s*:')
 _OBJECT_KEY_PREFIX: Final = re.compile(r'^\s*(?:\{\s*(?:"(?:[^"\\]|\\.)*\\?(?:"\s*)?)?)?$')
@@ -175,12 +173,19 @@ def _prefix_verdict(buffered_json: str, schema: ResponseFormatSchema, tool_name:
     return _first_value_verdict(buffered_json[first_key.end() :], schema, tool_name)
 
 
+def _unwrapped_values_answer(arguments: str) -> str:
+    payload_object: Final = decoded_answer_object(arguments)
+    values: Final = None if payload_object is None else payload_object.get("values")
+    return arguments if values is None else answer_json_text(values, ensure_ascii=True)
+
+
 def _held_answer(held: str, schema: ResponseFormatSchema) -> str:
-    try:
-        payload: Final = _DECODED_JSON.validate_python(json.loads(held))
-    except (json.JSONDecodeError, RecursionError):
-        return held
-    contents: Final = response_format_envelope_contents(payload, schema, RESPONSE_FORMAT_TOOL_NAME)
+    payload_object: Final = decoded_answer_object(held)
+    contents: Final = (
+        None
+        if payload_object is None
+        else response_format_envelope_contents(payload_object, schema, RESPONSE_FORMAT_TOOL_NAME)
+    )
     return held if contents is None else response_format_answer_json(contents)
 
 
@@ -736,6 +741,7 @@ class ModelResponseIterator:
         self._withheld_answer_close: str = ""
         self._streamed_answer_keys: frozenset[str] | None = None
         self._held_prose: str = ""
+        self._schemaless_input_started: bool = False
 
         # For handling partial JSON chunks from fragmentation
         # See: https://github.com/BerriAI/litellm/issues/17473
@@ -1315,15 +1321,19 @@ class ModelResponseIterator:
 
         # Convert tool to content if we're tracking a response_format tool
         if self.is_response_format_tool:
-            message: Final = AnthropicConfig.convert_tool_response_to_message(tool_calls=[tool_use])
-            if message is not None:
-                text = message.content or ""
-                tool_use = None
-                # Track that we converted a response_format tool
-                self.converted_response_format_tool = True
+            self.converted_response_format_tool = True
+            return self._stream_schemaless_fragment(tool_use), None
 
         return text, tool_use
 
+    def _stream_schemaless_fragment(self, tool_use: ChatCompletionToolCallChunk) -> str:
+        fragment: Final = tool_use.get("function", {}).get("arguments") or ""
+        if tool_use.get("id") is not None:
+            self._schemaless_input_started = False
+        is_first_piece: Final = not self._schemaless_input_started
+        self._schemaless_input_started = self._schemaless_input_started or bool(fragment.strip())
+        return _unwrapped_values_answer(fragment) if is_first_piece else fragment
+
     def _stream_response_format_fragment(
         self, tool_use: ChatCompletionToolCallChunk, schema: ResponseFormatSchema
     ) -> str:
diff --git a/litellm/llms/anthropic/chat/transformation.py b/litellm/llms/anthropic/chat/transformation.py
index 775eb6eda39..e0d313175b7 100644
--- a/litellm/llms/anthropic/chat/transformation.py
+++ b/litellm/llms/anthropic/chat/transformation.py
@@ -72,7 +72,8 @@ from litellm.litellm_core_utils.reasoning_effort_utils import OPAQUE_VALUE, warn
 from litellm.litellm_core_utils.tool_fields import tool_params_without_hosted_mcp
 from litellm.llms.base_llm.base_utils import (
     ResponseFormatSchema,
-    as_json_object,
+    answer_json_text,
+    decoded_answer_object,
     response_format_answer_json,
     response_format_schema,
     split_response_format_answer,
@@ -3416,22 +3417,21 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
         )
         if split_answer is not None:
             return litellm.Message(content=response_format_answer_json(split_answer))
-        try:
-            payload: Final = _DECODED_JSON.validate_python(json.loads(json_mode_content_str))
-        except (json.JSONDecodeError, RecursionError):
-            # json decode error does occur, return the original tool response str
+        payload_object: Final = decoded_answer_object(json_mode_content_str)
+        if payload_object is None:
             return litellm.Message(content=json_mode_content_str)
         if response_format_schema is not None:
             return litellm.Message(
                 content=response_format_answer_json(
-                    unwrap_response_format_envelope(payload, response_format_schema, RESPONSE_FORMAT_TOOL_NAME)
+                    unwrap_response_format_envelope(payload_object, response_format_schema, RESPONSE_FORMAT_TOOL_NAME)
                 )
             )
         # a lot of the times the `values` key is not present in the tool response
         # relevant issue: https://github.com/BerriAI/litellm/issues/6741
-        payload_object: Final = as_json_object(payload)
-        values: Final = None if payload_object is None else payload_object.get("values")
-        return litellm.Message(content=json.dumps(payload if values is None else values))
+        values: Final = payload_object.get("values")
+        return litellm.Message(
+            content=answer_json_text(payload_object if values is None else values, ensure_ascii=True)
+        )
 
     @classmethod
     def convert_tool_response_to_message(
diff --git a/litellm/llms/base_llm/base_utils.py b/litellm/llms/base_llm/base_utils.py
index ef8d080027f..45f623730a8 100644
--- a/litellm/llms/base_llm/base_utils.py
+++ b/litellm/llms/base_llm/base_utils.py
@@ -3,9 +3,12 @@ Utility functions for base LLM classes.
 """
 
 import copy
+import functools
 import itertools
 import json
+import math
 import re
+import secrets
 from abc import ABC, abstractmethod
 from collections.abc import Iterator, Mapping, Sequence
 from dataclasses import dataclass
@@ -166,13 +169,10 @@ def convert_tool_response_to_message(
             args: Final = _DECODED_JSON.validate_python(json.loads(json_mode_content_str))
             args_object: Final = as_str_mapping(args)
             if args_object is not None and (values := args_object.get("values")) is not None:
-                _message = Message(content=json.dumps(values))
-                return _message
-            else:
-                # a lot of the times the `values` key is not present in the tool response
-                # relevant issue: https://github.com/BerriAI/litellm/issues/6741
-                _message = Message(content=json.dumps(args))
-                return _message
+                return Message(content=json.dumps(values))
+            # a lot of the times the `values` key is not present in the tool response
+            # relevant issue: https://github.com/BerriAI/litellm/issues/6741
+            return Message(content=json.dumps(args))
     except json.JSONDecodeError:
         # json decode error does occur, return the original tool response str
         return Message(content=json_mode_content_str)
@@ -205,6 +205,60 @@ def as_json_object(value: object) -> Mapping[str, JsonValue] | None:
         return None
 
 
+@dataclass(frozen=True, slots=True)
+class SourceNumber:
+    text: str
+
+
+_JsonScalar: TypeAlias = str | bool | int | float | None
+AnswerValue: TypeAlias = SourceNumber | _JsonScalar | Sequence["AnswerValue"] | Mapping[str, "AnswerValue"]
+
+
+def as_answer_object(value: AnswerValue) -> Mapping[str, AnswerValue] | None:
+    return value if isinstance(value, Mapping) else None
+
+
+def _source_float(text: str) -> float | SourceNumber:
+    value: Final = float(text)
+    written_as_zero: Final = not any(digit in "123456789" for digit in text.lower().partition("e")[0])
+    return value if math.isfinite(value) and (value != 0 or written_as_zero) else SourceNumber(text)
+
+
+def _source_int(text: str) -> int | SourceNumber:
+    try:
+        return int(text)
+    except ValueError:
+        return SourceNumber(text)
+
+
+def _is_decoded_json_object(
+    decoded: object,
+) -> TypeIs[Mapping[str, AnswerValue]]:  # guard-ok: json.loads with the source-number hooks yields only answer values
+    return isinstance(decoded, dict)
+
+
+def decoded_answer_object(text: str) -> Mapping[str, AnswerValue] | None:
+    try:
+        decoded: Final = _DECODED_JSON.validate_python(
+            json.loads(text, parse_float=_source_float, parse_int=_source_int)
+        )
+    except (json.JSONDecodeError, RecursionError):
+        return None
+    return decoded if _is_decoded_json_object(decoded) else None
+
+
+def _marked_source_number(marker: str, number: SourceNumber) -> str:
+    return marker + number.text
+
+
+def answer_json_text(value: AnswerValue, *, ensure_ascii: bool) -> str:
+    marker: Final = secrets.token_hex(16)
+    encoded: Final = json.dumps(
+        value, ensure_ascii=ensure_ascii, default=functools.partial(_marked_source_number, marker)
+    )
+    return re.sub(f'"{marker}([^"]*)"', r"\1", encoded)
+
+
 def _names(value: object) -> frozenset[str]:
     try:
         return _NAMES.validate_python(value)
@@ -241,20 +295,15 @@ RESPONSE_FORMAT_ENVELOPE_NAME_KEYS: Final = frozenset({"name", "tool_name"})
 
 @dataclass(frozen=True, slots=True)
 class _EnvelopeLayer:
-    inner: Mapping[str, JsonValue]
+    inner: Mapping[str, AnswerValue]
     names_the_tool: bool
 
 
-def _json_object_or_encoded_object(value: JsonValue) -> Mapping[str, JsonValue] | None:
-    if not isinstance(value, str):
-        return as_json_object(value)
-    try:
-        return as_json_object(_DECODED_JSON.validate_python(json.loads(value)))
-    except (json.JSONDecodeError, RecursionError):
-        return None
+def _answer_object_or_encoded_object(value: AnswerValue) -> Mapping[str, AnswerValue] | None:
+    return decoded_answer_object(value) if isinstance(value, str) else as_answer_object(value)
 
 
-def _label_keys(payload_object: Mapping[str, JsonValue], schema: ResponseFormatSchema) -> frozenset[str]:
+def _label_keys(payload_object: Mapping[str, AnswerValue], schema: ResponseFormatSchema) -> frozenset[str]:
     if not schema.required_names:
         return frozenset()
     return frozenset(
@@ -263,7 +312,7 @@ def _label_keys(payload_object: Mapping[str, JsonValue], schema: ResponseFormatS
 
 
 def _envelope_layer(
-    payload_object: Mapping[str, JsonValue], schema: ResponseFormatSchema, tool_name: str
+    payload_object: Mapping[str, AnswerValue], schema: ResponseFormatSchema, tool_name: str
 ) -> _EnvelopeLayer | None:
     name_keys: Final = frozenset(
         key for key in RESPONSE_FORMAT_ENVELOPE_NAME_KEYS if payload_object.get(key) == tool_name
@@ -273,12 +322,12 @@ def _envelope_layer(
     if len(wrapper_keys) != 1 or wrapper_keys[0] in schema.declared_names:
         return None
     wrapped: Final = payload_object[wrapper_keys[0]]
-    inner: Final = _json_object_or_encoded_object(wrapped) if name_keys else as_json_object(wrapped)
+    inner: Final = _answer_object_or_encoded_object(wrapped) if name_keys else as_answer_object(wrapped)
     return None if inner is None else _EnvelopeLayer(inner=inner, names_the_tool=bool(name_keys))
 
 
 def _envelope_layers(
-    payload_object: Mapping[str, JsonValue], schema: ResponseFormatSchema, tool_name: str
+    payload_object: Mapping[str, AnswerValue], schema: ResponseFormatSchema, tool_name: str
 ) -> Iterator[_EnvelopeLayer]:
     layer = _envelope_layer(payload_object, schema, tool_name)  # rebind-ok: the recursion gate needs a loop
     while layer is not None:
@@ -287,8 +336,8 @@ def _envelope_layers(
 
 
 def _schema_object_inside_envelope(
-    payload_object: Mapping[str, JsonValue], schema: ResponseFormatSchema, tool_name: str
-) -> Mapping[str, JsonValue] | None:
+    payload_object: Mapping[str, AnswerValue], schema: ResponseFormatSchema, tool_name: str
+) -> Mapping[str, AnswerValue] | None:
     walk: Final = _envelope_layers(payload_object, schema, tool_name)
     layers: Final = tuple(walk if schema.declared_names else itertools.islice(walk, 1))
     schema_object: Final = next((layer.inner for layer in layers if _shares_a_schema_key(layer.inner, schema)), None)
@@ -297,14 +346,13 @@ def _schema_object_inside_envelope(
     return next((layer.inner for layer in layers if layer.names_the_tool), None)
 
 
-def _shares_a_schema_key(candidate: Mapping[str, JsonValue], schema: ResponseFormatSchema) -> bool:
+def _shares_a_schema_key(candidate: Mapping[str, AnswerValue], schema: ResponseFormatSchema) -> bool:
     return not schema.accepts_a_lone_undeclared_key and not schema.declared_names.isdisjoint(candidate)
 
 
-def response_format_envelope_contents(payload: object, schema: ResponseFormatSchema, tool_name: str) -> object | None:
-    payload_object: Final = as_json_object(payload)
-    if payload_object is None:
-        return None
+def response_format_envelope_contents(
+    payload_object: Mapping[str, AnswerValue], schema: ResponseFormatSchema, tool_name: str
+) -> AnswerValue:
     schema_object: Final = _schema_object_inside_envelope(payload_object, schema, tool_name)
     if schema_object is not None:
         return schema_object
@@ -314,29 +362,31 @@ def response_format_envelope_contents(payload: object, schema: ResponseFormatSch
 _LONE_SURROGATE: Final = re.compile("[\ud800-\udfff]")
 
 
-def response_format_answer_json(value: object) -> str:
-    text: Final = json.dumps(value, ensure_ascii=False)
-    return json.dumps(value) if _LONE_SURROGATE.search(text) else text
+def response_format_answer_json(value: AnswerValue) -> str:
+    text: Final = answer_json_text(value, ensure_ascii=False)
+    return answer_json_text(value, ensure_ascii=True) if _LONE_SURROGATE.search(text) else text
 
 
-def unwrap_response_format_envelope(payload: object, schema: ResponseFormatSchema, tool_name: str) -> object:
-    contents: Final = response_format_envelope_contents(payload, schema, tool_name)
-    return payload if contents is None else contents
+def unwrap_response_format_envelope(
+    payload_object: Mapping[str, AnswerValue], schema: ResponseFormatSchema, tool_name: str
+) -> AnswerValue:
+    contents: Final = response_format_envelope_contents(payload_object, schema, tool_name)
+    return payload_object if contents is None else contents
 
 
 def _field_names(
-    blocks: tuple[Mapping[str, JsonValue], ...], key: str, schema: ResponseFormatSchema
+    blocks: tuple[Mapping[str, AnswerValue], ...], key: str, schema: ResponseFormatSchema
 ) -> tuple[str, ...] | None:
     values: Final = (block[key] for block in blocks)
     names: Final = tuple(value for value in values if isinstance(value, str) and value in schema.declared_names)
     return names if len(frozenset(names)) == len(blocks) else None
 
 
-def _is_a_label_every_block_repeats(blocks: tuple[Mapping[str, JsonValue], ...], key: str) -> bool:
+def _is_a_label_every_block_repeats(blocks: tuple[Mapping[str, AnswerValue], ...], key: str) -> bool:
     return isinstance(blocks[0][key], str) and all(block[key] == blocks[0][key] for block in blocks)
 
 
-def _value_key(blocks: tuple[Mapping[str, JsonValue], ...], keys: frozenset[str]) -> str | None:
+def _value_key(blocks: tuple[Mapping[str, AnswerValue], ...], keys: frozenset[str]) -> str | None:
     if len(keys) == 1:
         return next(iter(keys))
     varying: Final = tuple(key for key in keys if not _is_a_label_every_block_repeats(blocks, key))
@@ -344,8 +394,8 @@ def _value_key(blocks: tuple[Mapping[str, JsonValue], ...], keys: frozenset[str]
 
 
 def _split_answer_fields(
-    blocks: tuple[Mapping[str, JsonValue], ...], schema: ResponseFormatSchema
-) -> Mapping[str, JsonValue] | None:
+    blocks: tuple[Mapping[str, AnswerValue], ...], schema: ResponseFormatSchema
+) -> Mapping[str, AnswerValue] | None:
     keys: Final = frozenset(blocks[0])
     if not keys.isdisjoint(schema.declared_names) or any(frozenset(block) != keys for block in blocks):
         return None
@@ -360,8 +410,8 @@ def _split_answer_fields(
 
 
 def _merged_partial_answers(
-    blocks: tuple[Mapping[str, JsonValue], ...], schema: ResponseFormatSchema
-) -> Mapping[str, JsonValue] | None:
+    blocks: tuple[Mapping[str, AnswerValue], ...], schema: ResponseFormatSchema
+) -> Mapping[str, AnswerValue] | None:
     keys: Final = frozenset(itertools.chain.from_iterable(blocks))
     if not keys <= schema.declared_names or len(keys) != sum(len(block) for block in blocks):
         return None
@@ -370,8 +420,8 @@ def _merged_partial_answers(
 
 def split_response_format_answer(
     encoded_blocks: Sequence[str], schema: ResponseFormatSchema
-) -> Mapping[str, JsonValue] | None:
-    decoded: Final = tuple(_json_object_or_encoded_object(encoded) for encoded in encoded_blocks)
+) -> Mapping[str, AnswerValue] | None:
+    decoded: Final = tuple(decoded_answer_object(encoded) for encoded in encoded_blocks)
     blocks: Final = tuple(block for block in decoded if block is not None)
     if len(blocks) < 2 or len(blocks) != len(decoded):
         return None
@@ -379,13 +429,15 @@ def split_response_format_answer(
     return named_fields if named_fields is not None else _merged_partial_answers(blocks, schema)
 
 
-def partial_response_format_answer(encoded_block: str, schema: ResponseFormatSchema) -> Mapping[str, JsonValue] | None:
-    block: Final = _json_object_or_encoded_object(encoded_block)
+def partial_response_format_answer(
+    encoded_block: str, schema: ResponseFormatSchema
+) -> Mapping[str, AnswerValue] | None:
+    block: Final = decoded_answer_object(encoded_block)
     return block if block is not None and frozenset(block) <= schema.declared_names else None
 
 
 def names_a_response_format_property(encoded_block: str, schema: ResponseFormatSchema) -> bool:
-    block: Final = _json_object_or_encoded_object(encoded_block)
+    block: Final = decoded_answer_object(encoded_block)
     if block is None:
         return False
     return any(isinstance(value, str) and value in schema.declared_names for value in block.values())
diff --git a/tests/integration/providers/test_vertex_anthropic_structured_output_wire.py b/tests/integration/providers/test_vertex_anthropic_structured_output_wire.py
index 95e07593b2a..edecdf9e785 100644
--- a/tests/integration/providers/test_vertex_anthropic_structured_output_wire.py
+++ b/tests/integration/providers/test_vertex_anthropic_structured_output_wire.py
@@ -2097,6 +2097,29 @@ def test_vertex_chat_malformed_response_format_schema_keeps_the_tool_input(
         assert gateway.request("GET", "/health/liveliness").status_code == 200
 
 
+@pytest.mark.parametrize("marker_hex", ["e66127078448420083852775e765081e", "e6612707841e0000000000000000051e"])
+def test_vertex_chat_malformed_schema_stream_keeps_numeric_fragments_verbatim(
+    gateway: Gateway, marker_hex: str
+) -> None:
+    with wire_server(_peer) as wire, gateway.scenario() as scenario:
+        model: Final = _vertex_deployment(gateway, scenario, wire.url, _TOOL_MODEL)
+        response: Final = gateway.request(
+            "POST",
+            "/v1/chat/completions",
+            _chat_body(
+                model,
+                f"wrapped {marker_hex}",
+                stream=True,
+                response_format={
+                    "type": "json_schema",
+                    "json_schema": {"name": "risk", "schema": {"type": "object", "properties": None}},
+                },
+            ),
+        )
+        assert _stream_text(response) == _tool_text("wrapped", marker_hex), response.text
+        _one_delivery(wire, marker_hex)
+
+
 @pytest.mark.parametrize("status", [400, 403, 404])
 def test_vertex_chat_peer_error_reaches_the_caller_and_the_deployment_keeps_serving(
     gateway: Gateway, status: int
diff --git a/tests/unit/llms/anthropic/chat/test_anthropic_chat_handler.py b/tests/unit/llms/anthropic/chat/test_anthropic_chat_handler.py
index 2fb4eb9931c..c11fd58f84e 100644
--- a/tests/unit/llms/anthropic/chat/test_anthropic_chat_handler.py
+++ b/tests/unit/llms/anthropic/chat/test_anthropic_chat_handler.py
@@ -1,4 +1,5 @@
 import json
+import sys
 import threading
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from itertools import chain
@@ -33,8 +34,14 @@ from tests.unit.llms.anthropic.anthropic_docs_refusal import ANTHROPIC_DOCS_REFU
 
 def _text_delta_stream(text: str) -> bytes:
     events: Final = (
-        ("content_block_start", {"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}}),
-        ("content_block_delta", {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": text}}),
+        (
+            "content_block_start",
+            {"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}},
+        ),
+        (
+            "content_block_delta",
+            {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": text}},
+        ),
     )
     return "".join(
         f"event: {name}\ndata: {json.dumps(payload, ensure_ascii=False)}\n\n" for name, payload in events
@@ -460,7 +467,9 @@ def test_anthropic_sync_completion_sends_a_link_that_serves_a_pdf_as_a_url_docum
     def serve(client: object, url: str, **kwargs: object) -> httpx.Response:
         if url == pdf_link:
             return _served_pdf_slice(url)
-        return httpx.Response(200, content=b"\x89PNG", headers={"content-type": "image/png"}, request=httpx.Request("GET", url))
+        return httpx.Response(
+            200, content=b"\x89PNG", headers={"content-type": "image/png"}, request=httpx.Request("GET", url)
+        )
 
     monkeypatch.setattr(image_handling, "safe_get", serve)
     client: Final = HTTPHandler(client=httpx.Client(transport=httpx.MockTransport(_anthropic_reply(captured))))
@@ -3210,6 +3219,81 @@ def _streamed_json_mode_contents(
     return tuple(chunk.choices[0].delta.content or "" for chunk in chunks)
 
 
+def _schemaless_streamed_json_mode_contents(fragments: tuple[str, ...]) -> str:
+    iterator: Final = ModelResponseIterator(streaming_response=iter(()), sync_stream=True, json_mode=True)
+    chunks: Final = tuple(iterator.chunk_parser(event) for event in _json_tool_events(fragments))
+    assert all(chunk.choices[0].delta.tool_calls is None for chunk in chunks)
+    assert chunks[-1].choices[0].finish_reason == "stop"
+    return "".join(chunk.choices[0].delta.content or "" for chunk in chunks)
+
+
+@pytest.mark.parametrize(
+    "fragments",
+    [
+        ('{"summary":"', "48420083852775e76508", '"}'),
+        ('{"summary":"', "1e5", '"}'),
+        ('{"outer":', '{"values": [3, 5]}', "}"),
+    ],
+)
+def test_streamed_json_mode_without_schema_preserves_argument_fragments(fragments: tuple[str, ...]) -> None:
+    assert _schemaless_streamed_json_mode_contents(fragments) == "".join(fragments)
+
+
+_DIGITS_PAST_THE_INT_LIMIT: Final = "1" * (sys.get_int_max_str_digits() + 1)
+
+
+@pytest.mark.parametrize(
+    ("fragments", "expected"),
+    [
+        (('{"values": [3, 5]}',), "[3, 5]"),
+        ((" ", '{"values": [3, 5]}', " "), " [3, 5] "),
+        (('{"count": 1e400}',), '{"count": 1e400}'),
+        (('{"values": [1e400]}',), "[1e400]"),
+        pytest.param(
+            ('{"values": [' + _DIGITS_PAST_THE_INT_LIMIT + "]}",),
+            "[" + _DIGITS_PAST_THE_INT_LIMIT + "]",
+            id="values-int-over-digit-limit",
+        ),
+        pytest.param(
+            ('{"n": ' + _DIGITS_PAST_THE_INT_LIMIT + "}",),
+            '{"n": ' + _DIGITS_PAST_THE_INT_LIMIT + "}",
+            id="int-over-digit-limit",
+        ),
+        (('{"n": 1e5}',), '{"n": 1e5}'),
+        (('{"values": false}',), "false"),
+        (('{"values": null}',), '{"values": null}'),
+        (("[1, 2]",), "[1, 2]"),
+        (('"x"',), '"x"'),
+        (("5",), "5"),
+        pytest.param(
+            ('{"values": ' + "[" * 300 + "]" * 300 + "}",),
+            "[" * 300 + "]" * 300,
+            id="values-nested-300-levels",
+        ),
+        pytest.param(
+            ('{"values": ' + "[" * 100000 + "]" * 100000 + "}",),
+            '{"values": ' + "[" * 100000 + "]" * 100000 + "}",
+            id="nesting-over-recursion-limit",
+        ),
+    ],
+)
+def test_streamed_json_mode_without_schema_unwraps_values_only_from_a_one_piece_input(
+    fragments: tuple[str, ...], expected: str
+) -> None:
+    assert _schemaless_streamed_json_mode_contents(fragments) == expected
+
+
+def test_streamed_json_mode_without_schema_unwraps_each_json_tool_blocks_one_piece_input() -> None:
+    json_tool_block: Final = {"type": "tool_use", "id": "toolu_1", "name": RESPONSE_FORMAT_TOOL_NAME, "input": {}}
+    events: Final = _message_events(
+        (json_tool_block, ({"type": "input_json_delta", "partial_json": '{"values": [1]}'},)),
+        ({**json_tool_block, "id": "toolu_2"}, ({"type": "input_json_delta", "partial_json": '{"values": [2]}'},)),
+    )
+    iterator: Final = ModelResponseIterator(streaming_response=iter(()), sync_stream=True, json_mode=True)
+    chunks: Final = tuple(iterator.chunk_parser(event) for event in events)
+    assert "".join(chunk.choices[0].delta.content or "" for chunk in chunks) == "[1][2]"
+
+
 def test_streamed_json_mode_envelope_is_unwrapped_when_the_block_ends():
     envelope = json.dumps({"name": RESPONSE_FORMAT_TOOL_NAME, "parameters": _STREAMED_RISK_OBJECT})
     fragments = (envelope[:9], envelope[9:40], envelope[40:])
@@ -3532,6 +3616,36 @@ def test_streamed_json_mode_held_answer_without_an_envelope_keeps_the_model_text
     assert contents[:-2] == ("",) * (len(contents) - 2)
 
 
+@pytest.mark.parametrize(
+    ("answer", "expected"),
+    [
+        pytest.param(
+            '{"name":"json_tool_call","parameters":{"tier":1e400,"summary":"x"}}',
+            '{"tier": 1e400, "summary": "x"}',
+            id="number_past_a_float",
+        ),
+        pytest.param(
+            '{"name":"json_tool_call","parameters":{"tier":-' + _DIGITS_PAST_THE_INT_LIMIT + ',"summary":"x"}}',
+            '{"tier": -' + _DIGITS_PAST_THE_INT_LIMIT + ', "summary": "x"}',
+            id="negative_integer_past_the_digit_limit",
+        ),
+        pytest.param('{"values":[1e400, 2.5]}', "[1e400, 2.5]", id="values_number_past_a_float"),
+        pytest.param(
+            '{"name":"json_tool_call","parameters":{"tier":1e-400,"summary":"x"}}',
+            '{"tier": 1e-400, "summary": "x"}',
+            id="number_below_a_float",
+        ),
+        pytest.param(
+            '{"values":[1e400, 1e-400, 0.0e5]}', "[1e400, 1e-400, 0.0]", id="values_numbers_past_a_float_both_ways"
+        ),
+    ],
+)
+def test_streamed_json_mode_held_answer_keeps_out_of_range_numbers_as_the_model_wrote_them(answer: str, expected: str):
+    contents = _streamed_json_mode_contents((answer[:9], answer[9:30], answer[30:]))
+
+    assert "".join(contents) == expected
+
+
 def test_envelope_verdict_holds_a_prefix_that_stays_undecided_past_the_limit():
     schema = response_format_schema(_STREAMED_RISK_SCHEMA)
     assert schema is not None
@@ -3733,6 +3847,41 @@ def test_streamed_json_mode_answer_split_into_one_partial_object_per_field_arriv
     assert contents[3] == blocks[0][:-1]
 
 
+@pytest.mark.parametrize(
+    ("blocks", "expected"),
+    [
+        pytest.param(
+            ('{"tier": 1e400}', '{"summary": "x"}'), '{"tier": 1e400, "summary": "x"}', id="number_past_a_float_first"
+        ),
+        pytest.param(
+            ('{"summary": "x"}', '{"tier": ' + _DIGITS_PAST_THE_INT_LIMIT + "}"),
+            '{"summary": "x", "tier": ' + _DIGITS_PAST_THE_INT_LIMIT + "}",
+            id="integer_past_the_digit_limit_last",
+        ),
+        pytest.param(
+            (
+                '{"$PARAMETER_NAME": "tier", "$PARAMETER_VALUE": 1e400}',
+                '{"$PARAMETER_NAME": "summary", "$PARAMETER_VALUE": "x"}',
+            ),
+            '{"tier": 1e400, "summary": "x"}',
+            id="named_field_number_past_a_float_first",
+        ),
+        pytest.param(
+            (
+                ' {"$PARAMETER_NAME":"summary","$PARAMETER_VALUE":"Jos\u00e9"} ',
+                '{"$PARAMETER_NAME":"tier","$PARAMETER_VALUE":' + _DIGITS_PAST_THE_INT_LIMIT + "}",
+            ),
+            '{"summary": "Jos\u00e9", "tier": ' + _DIGITS_PAST_THE_INT_LIMIT + "}",
+            id="compact_named_field_integer_past_the_digit_limit_last",
+        ),
+    ],
+)
+def test_streamed_json_mode_split_fields_keep_out_of_range_numbers_as_the_model_wrote_them(
+    blocks: tuple[str, ...], expected: str
+):
+    assert "".join(_streamed_json_tool_blocks(blocks)) == expected
+
+
 @pytest.mark.parametrize(
     "blocks",
     [
diff --git a/tests/unit/llms/anthropic/chat/test_anthropic_chat_transformation.py b/tests/unit/llms/anthropic/chat/test_anthropic_chat_transformation.py
index ae72af200d4..bf093f1a0d8 100644
--- a/tests/unit/llms/anthropic/chat/test_anthropic_chat_transformation.py
+++ b/tests/unit/llms/anthropic/chat/test_anthropic_chat_transformation.py
@@ -2,6 +2,7 @@ import copy
 import itertools
 import json
 import logging
+import sys
 from collections.abc import Iterator, Mapping
 from itertools import chain
 from typing import Final, cast
@@ -1459,9 +1460,9 @@ def test_dict_fields_take_the_unforced_json_tool_on_a_default_thinking_model(loc
 
     assert "output_format" not in optional_params
     assert "tool_choice" not in optional_params
-    assert [
-        (tool["name"], tool["input_schema"]["properties"]["field"]) for tool in optional_params["tools"]
-    ] == [("json_tool_call", {"type": "object", "additionalProperties": {"type": "integer"}})]
+    assert [(tool["name"], tool["input_schema"]["properties"]["field"]) for tool in optional_params["tools"]] == [
+        ("json_tool_call", {"type": "object", "additionalProperties": {"type": "integer"}})
+    ]
 
 
 def test_native_structured_output_uses_bundled_capability_when_remote_map_lags(
@@ -3827,7 +3828,12 @@ def _parsed_assistant_message(content: list[dict[str, object]], stop_reason: str
     return _parsed_choice(content, stop_reason).message
 
 
-_WEATHER_TOOL_USE: Final = {"type": "tool_use", "id": "toolu_01Weather", "name": "get_weather", "input": {"city": "Paris"}}
+_WEATHER_TOOL_USE: Final = {
+    "type": "tool_use",
+    "id": "toolu_01Weather",
+    "name": "get_weather",
+    "input": {"city": "Paris"},
+}
 
 
 def _web_search_turn(last_block: Mapping[str, object]) -> list[dict[str, object]]:
@@ -3963,7 +3969,11 @@ _PARIS_ANSWER: Final[Mapping[str, str]] = {"type": "text", "text": _PARIS_TEXT}
     [
         pytest.param(_web_search_turn(_PARIS_ANSWER), "end_turn", [], "stop", id="web_search_then_answer"),
         pytest.param(
-            _web_search_turn(_WEATHER_TOOL_USE), "tool_use", ["toolu_01Weather"], "tool_calls", id="web_search_then_tool"
+            _web_search_turn(_WEATHER_TOOL_USE),
+            "tool_use",
+            ["toolu_01Weather"],
+            "tool_calls",
+            id="web_search_then_tool",
         ),
         pytest.param([_WEATHER_TOOL_USE], "tool_use", ["toolu_01Weather"], "tool_calls", id="client_tool_only"),
     ],
@@ -8574,6 +8584,166 @@ def test_convert_tool_response_to_message_keeps_arguments_too_deep_to_decode(arg
     assert message.content == arguments
 
 
+@pytest.mark.parametrize("arguments", ["[1, 2]", '"x"', "5"])
+def test_convert_tool_response_to_message_keeps_a_non_object_answer_as_written(arguments: str) -> None:
+    message: Final = AnthropicConfig.convert_tool_response_to_message(
+        tool_calls=[
+            ChatCompletionToolCallChunk(
+                id="toolu_1",
+                type="function",
+                function=ChatCompletionToolCallFunctionChunk(name=RESPONSE_FORMAT_TOOL_NAME, arguments=arguments),
+                index=0,
+            )
+        ],
+        response_format_schema=None,
+    )
+
+    assert message is not None
+    assert message.content == arguments
+
+
+_DEEPLY_NESTED: Final = "[" * 300 + "]" * 300
+
+
+@pytest.mark.parametrize(
+    "arguments, schema, expected",
+    [
+        pytest.param('{"values": ' + _DEEPLY_NESTED + "}", None, _DEEPLY_NESTED, id="values_answer"),
+        pytest.param(
+            '{"name": "json_tool_call", "parameters": {"tier": ' + _DEEPLY_NESTED + ', "summary": "x"}}',
+            _RISK_SCHEMA,
+            '{"tier": ' + _DEEPLY_NESTED + ', "summary": "x"}',
+            id="envelope",
+        ),
+    ],
+)
+def test_convert_tool_response_to_message_unwraps_a_deeply_nested_answer(
+    arguments: str, schema: dict[str, object] | None, expected: str
+) -> None:
+    message = AnthropicConfig.convert_tool_response_to_message(
+        tool_calls=[
+            ChatCompletionToolCallChunk(
+                id="toolu_1",
+                type="function",
+                function=ChatCompletionToolCallFunctionChunk(name=RESPONSE_FORMAT_TOOL_NAME, arguments=arguments),
+                index=0,
+            )
+        ],
+        response_format_schema=None if schema is None else _schema(schema),
+    )
+
+    assert message is not None
+    assert message.content == expected
+
+
+_DIGITS_PAST_THE_INT_LIMIT: Final = "1" * (sys.get_int_max_str_digits() + 1)
+
+
+@pytest.mark.parametrize(
+    "arguments, schema, expected",
+    [
+        pytest.param('{"values": [1e400, "\u00e9"]}', None, '[1e400, "\\u00e9"]', id="values_number_past_a_float"),
+        pytest.param(
+            '{"values": [' + _DIGITS_PAST_THE_INT_LIMIT + "]}",
+            None,
+            "[" + _DIGITS_PAST_THE_INT_LIMIT + "]",
+            id="values_integer_past_the_digit_limit",
+        ),
+        pytest.param('{"count":-1e400}', None, '{"count": -1e400}', id="schemaless_answer_number_past_a_float"),
+        pytest.param(
+            '{"values": [1e400, -1E-400, 0e-400]}',
+            None,
+            "[1e400, -1E-400, 0.0]",
+            id="values_numbers_past_a_float_both_ways",
+        ),
+        pytest.param(
+            '{"name": "json_tool_call", "parameters": {"tier": 2.5e-400, "summary": "x"}}',
+            _RISK_SCHEMA,
+            '{"tier": 2.5e-400, "summary": "x"}',
+            id="envelope_number_below_a_float",
+        ),
+        pytest.param(
+            '{"name": "json_tool_call", "parameters": {"tier": 1e400, "summary": "\u00e9"}}',
+            _RISK_SCHEMA,
+            '{"tier": 1e400, "summary": "\u00e9"}',
+            id="envelope_number_past_a_float",
+        ),
+        pytest.param(
+            '{"tier":' + _DIGITS_PAST_THE_INT_LIMIT + ',"summary":"x"}',
+            _RISK_SCHEMA,
+            '{"tier": ' + _DIGITS_PAST_THE_INT_LIMIT + ', "summary": "x"}',
+            id="answer_integer_past_the_digit_limit",
+        ),
+    ],
+)
+def test_convert_tool_response_to_message_keeps_out_of_range_numbers_as_the_model_wrote_them(
+    arguments: str, schema: dict[str, object] | None, expected: str
+):
+    message = AnthropicConfig.convert_tool_response_to_message(
+        tool_calls=[
+            ChatCompletionToolCallChunk(
+                id="toolu_1",
+                type="function",
+                function=ChatCompletionToolCallFunctionChunk(name=RESPONSE_FORMAT_TOOL_NAME, arguments=arguments),
+                index=0,
+            )
+        ],
+        response_format_schema=None if schema is None else _schema(schema),
+    )
+
+    assert message is not None
+    assert message.content == expected
+
+
+@pytest.mark.parametrize(
+    "blocks, expected",
+    [
+        pytest.param(
+            ('{"tier": 1e400}', '{"summary": "x"}'), '{"tier": 1e400, "summary": "x"}', id="number_past_a_float_first"
+        ),
+        pytest.param(
+            ('{"summary": "x"}', '{"tier": ' + _DIGITS_PAST_THE_INT_LIMIT + "}"),
+            '{"summary": "x", "tier": ' + _DIGITS_PAST_THE_INT_LIMIT + "}",
+            id="integer_past_the_digit_limit_last",
+        ),
+        pytest.param(
+            (
+                '{"$PARAMETER_NAME": "tier", "$PARAMETER_VALUE": 1e400}',
+                '{"$PARAMETER_NAME": "summary", "$PARAMETER_VALUE": "x"}',
+            ),
+            '{"tier": 1e400, "summary": "x"}',
+            id="named_field_number_past_a_float_first",
+        ),
+        pytest.param(
+            (
+                ' {"$PARAMETER_NAME":"summary","$PARAMETER_VALUE":"Jos\u00e9"} ',
+                '{"$PARAMETER_NAME":"tier","$PARAMETER_VALUE":' + _DIGITS_PAST_THE_INT_LIMIT + "}",
+            ),
+            '{"summary": "Jos\u00e9", "tier": ' + _DIGITS_PAST_THE_INT_LIMIT + "}",
+            id="compact_named_field_integer_past_the_digit_limit_last",
+        ),
+    ],
+)
+def test_convert_tool_response_to_message_joins_split_fields_keeping_out_of_range_numbers_as_written(
+    blocks: tuple[str, ...], expected: str
+):
+    message = AnthropicConfig.convert_tool_response_to_message(
+        tool_calls=[
+            ChatCompletionToolCallChunk(
+                id=f"toolu_{index}",
+                type="function",
+                function=ChatCompletionToolCallFunctionChunk(name=RESPONSE_FORMAT_TOOL_NAME, arguments=block),
+                index=index,
+            )
+            for index, block in enumerate(blocks)
+        ],
+        response_format_schema=_schema(_RISK_SCHEMA),
+    )
+
+    assert message is not None
+    assert message.content == expected
+
+
 @pytest.mark.parametrize(
     "document",
     [
@@ -8836,6 +9006,12 @@ def test_vertex_claude_json_answer_split_into_one_tool_block_per_field_returns_o
             {"name": "unknown", "email": "unknown"},
             id="one_partial_object_per_field",
         ),
+        pytest.param(
+            _CONTACT_SCHEMA,
+            ({}, {"name": "unknown"}, {"email": "unknown"}),
+            {"name": "unknown", "email": "unknown"},
+            id="empty_object_then_one_partial_object_per_field",
+        ),
         pytest.param(
             _OPTIONAL_CONTACT_SCHEMA,
             ({"name": "Ada", "email": "ada@example.com"}, {"phone": "555-0100"}),
@@ -8854,7 +9030,7 @@ def test_json_tool_blocks_that_each_name_one_schema_property_become_one_object(
     )
 
     assert replacement is not None
-    assert json.loads(replacement.content) == expected_content
+    assert replacement.content == json.dumps(expected_content)
     assert remaining_tool_calls == []
     assert extra_content is None