1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
|
import copy
from typing import List, cast
import httpx
from anthropic.types.beta import BetaDirectCaller, BetaToolUseBlock, BetaInputJSONDelta, BetaRawContentBlockDeltaEvent
from anthropic.types.tool_use_block import ToolUseBlock
from anthropic.types.beta.beta_usage import BetaUsage
from anthropic.lib.streaming._beta_messages import accumulate_event
from anthropic.types.beta.parsed_beta_message import ParsedBetaMessage
class TestPartialJson:
def test_trailing_strings_mode_header(self) -> None:
"""Test behavior differences with and without the beta header for JSON parsing."""
message = ParsedBetaMessage(
id="msg_123",
type="message",
role="assistant",
content=[
BetaToolUseBlock(
type="tool_use",
input={},
id="tool_123",
name="test_tool",
caller=BetaDirectCaller(type="direct"),
)
],
model="claude-sonnet-4-5",
stop_reason=None,
stop_sequence=None,
usage=BetaUsage(input_tokens=10, output_tokens=10),
)
# Test case 1: Complete JSON
complete_json = '{"key": "value"}'
event_complete = BetaRawContentBlockDeltaEvent(
type="content_block_delta",
index=0,
delta=BetaInputJSONDelta(type="input_json_delta", partial_json=complete_json),
)
# Both modes should handle complete JSON the same way
message1 = accumulate_event(
event=event_complete,
current_snapshot=copy.deepcopy(message),
request_headers=httpx.Headers({"some-header": "value"}),
)
message2 = accumulate_event(
event=event_complete,
current_snapshot=copy.deepcopy(message),
request_headers=httpx.Headers({"anthropic-beta": "fine-grained-tool-streaming-2025-05-14"}),
)
# Both should parse complete JSON correctly
assert cast(ToolUseBlock, message1.content[0]).input == {"key": "value"}
assert cast(ToolUseBlock, message2.content[0]).input == {"key": "value"}
# Test case 2: Incomplete JSON with trailing string that will be treated differently
# Here we want to create a situation where regular mode and trailing strings mode behave differently
incomplete_json = '{"items": ["item1", "item2"], "unfinished_field": "incomplete value'
event_incomplete = BetaRawContentBlockDeltaEvent(
type="content_block_delta",
index=0,
delta=BetaInputJSONDelta(type="input_json_delta", partial_json=incomplete_json),
)
# Without beta header (standard mode)
message_standard = accumulate_event(
event=event_incomplete,
current_snapshot=copy.deepcopy(message),
request_headers=httpx.Headers({"some-header": "value"}),
)
# With beta header (trailing strings mode)
message_trailing = accumulate_event(
event=event_incomplete,
current_snapshot=copy.deepcopy(message),
request_headers=httpx.Headers({"anthropic-beta": "fine-grained-tool-streaming-2025-05-14"}),
)
# Get the tool use blocks
standard_tool = cast(ToolUseBlock, message_standard.content[0])
trailing_tool = cast(ToolUseBlock, message_trailing.content[0])
# Both should have the valid complete part of the JSON
assert isinstance(standard_tool.input, dict)
assert isinstance(trailing_tool.input, dict)
standard_input = standard_tool.input # type: ignore
trailing_input = trailing_tool.input # type: ignore
# The input should have the items array in both cases
items_standard = cast(List[str], standard_input["items"])
items_trailing = cast(List[str], trailing_input["items"])
assert items_standard == ["item1", "item2"]
assert items_trailing == ["item1", "item2"]
# The key difference is how they handle the incomplete field:
# Standard mode should not include the incomplete field
assert "unfinished_field" not in standard_input
# Trailing strings mode should include the incomplete field
assert "unfinished_field" in trailing_input
assert trailing_input["unfinished_field"] == "incomplete value"
# test that with invalid JSON we throw the correct error
def test_partial_json_with_invalid_json(self) -> None:
"""Test that invalid JSON raises an error."""
message = ParsedBetaMessage(
id="msg_123",
type="message",
role="assistant",
content=[
BetaToolUseBlock(
type="tool_use",
input={},
id="tool_123",
name="test_tool",
caller=BetaDirectCaller(type="direct"),
)
],
model="claude-sonnet-4-5",
stop_reason=None,
stop_sequence=None,
usage=BetaUsage(input_tokens=10, output_tokens=10),
)
# Invalid JSON input
invalid_json = '{"key": "value", "incomplete_field": bad_value'
event_invalid = BetaRawContentBlockDeltaEvent(
type="content_block_delta",
index=0,
delta=BetaInputJSONDelta(type="input_json_delta", partial_json=invalid_json),
)
# Expect an error when trying to accumulate the invalid JSON
try:
accumulate_event(
event=event_invalid,
current_snapshot=copy.deepcopy(message),
request_headers=httpx.Headers({"anthropic-beta": "fine-grained-tool-streaming-2025-05-14"}),
)
raise AssertionError("Expected ValueError for invalid JSON, but no error was raised.")
except ValueError as e:
assert str(e).startswith(
"Unable to parse tool parameter JSON from model. Please retry your request or adjust your prompt."
)
except Exception as e:
raise AssertionError(f"Unexpected error type: {type(e).__name__} with message: {str(e)}") from e
|
