fix(anthropic): support Claude Sonnet 5.5 compatibility (#40882)

Co-authored-by: Hunter Lovell <hntrl@users.noreply.github.com>
Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com>
Co-authored-by: ccurme <ccurme@users.noreply.github.com>
Co-authored-by: Chester Curme <chester.curme@gmail.com>
This commit is contained in:
authored and GitHub committed 2026-09-29 09:05:37 -04:00
1 parent 78a3cbcc6b
commit ce9066138d
10 files changed
+559 -19

No files matched your search

+1 -2
View File
@@ -617,8 +617,7 @@ class AIMessageChunk(AIMessage, BaseMessageChunk):
isinstance(block, dict)
and block.get("type")
in {"server_tool_call", "server_tool_call_chunk"}
and (args_str := block.get("args"))
and isinstance(args_str, str)
and isinstance(args_str := block.get("args") or "{}", str)
):
try:
args = json.loads(args_str)
@@ -286,8 +286,9 @@ def _convert_to_v1_from_anthropic(message: AIMessage) -> list[types.ContentBlock
args=chunk.get("args"),
type="tool_call_chunk",
)
if "caller" in block:
tool_call_chunk["extras"] = {"caller": block["caller"]}
for key in ("caller", "toolset_name"):
if key in block:
tool_call_chunk.setdefault("extras", {})[key] = block[key]
index = chunk.get("index")
if index is not None:
@@ -322,10 +323,9 @@ def _convert_to_v1_from_anthropic(message: AIMessage) -> list[types.ContentBlock
}
if "index" in block:
tool_call_block["index"] = block["index"]
if "caller" in block:
if "extras" not in tool_call_block:
tool_call_block["extras"] = {}
tool_call_block["extras"]["caller"] = block["caller"]
for key in ("caller", "toolset_name"):
if key in block:
tool_call_block.setdefault("extras", {})[key] = block[key]
yield tool_call_block
@@ -572,3 +572,34 @@ def test_convert_to_v1_from_anthropic_malformed_citations() -> None:
],
},
]
def test_toolset_namespace_in_content_blocks() -> None:
block = {
"type": "tool_use",
"id": "call_1",
"name": "click",
"input": {},
"toolset_name": "computer",
}
metadata = {"model_provider": "anthropic"}
message = AIMessage([block], response_metadata=metadata)
content_block = message.content_blocks[0]
assert content_block["type"] == "tool_call"
assert content_block["extras"]["toolset_name"] == "computer"
chunk = AIMessageChunk(
content=[{**block, "index": 0}],
tool_call_chunks=[
{
"type": "tool_call_chunk",
"id": "call_1",
"name": "click",
"args": "{}",
"index": 0,
}
],
response_metadata=metadata,
)
chunk_block = chunk.content_blocks[0]
assert chunk_block["type"] == "tool_call_chunk"
assert chunk_block["extras"]["toolset_name"] == "computer"
@@ -423,6 +423,18 @@ def test_content_blocks() -> None:
{"type": "server_tool_call", "name": "foo", "index": 0, "args": {"a": 1}}
]
# Server tool calls with no input stream no args
empty_args_chunk = AIMessageChunk(
content=[
{"type": "server_tool_call_chunk", "index": 0, "name": "foo", "args": ""}
]
) + AIMessageChunk(
content=[], chunk_position="last", response_metadata={"output_version": "v1"}
)
assert empty_args_chunk.content == [
{"type": "server_tool_call", "name": "foo", "index": 0, "args": {}}
]
# Test non-standard + non-standard
chunk_1 = AIMessageChunk(
content=[
+21
View File
@@ -31,6 +31,27 @@ As an open-source project in a rapidly developing field, we are extremely open t
For detailed information on how to contribute, see the [Contributing Guide](https://docs.langchain.com/oss/python/contributing/overview).
## Migrating to Claude Sonnet 5.5
```python
from langchain_anthropic import ChatAnthropic
model = ChatAnthropic(
model="claude-sonnet-5-5",
max_tokens=16000,
output_config={"effort": "medium"},
)
```
- Use `with_structured_output(schema, method="json_schema")` for native structured output. Sonnet 5.5 rejects forced tool choice (`"any"` or a tool name). Function-calling structured output does not force a call and raises a parsing error if the model answers without one.
- Thinking is adaptive by default. For no up-front thinking, use `thinking={"type": "between_tools"}` at `high` effort or below, with no additional thinking fields. `disabled` and budgeted `enabled` thinking are unsupported.
- Omit sampling settings; non-default `temperature`, `top_p`, and `top_k` are rejected. Budget output tokens for both thinking and text.
- Preserve signed thinking blocks, including empty ones, and keep history append-only. Use mid-conversation system messages to change instructions or tools rather than editing earlier turns.
- Progress updates can arrive as thinking blocks. Use adaptive thinking with `display="summarized"` or `display="updates"` to display them; the latter's beta header is added automatically.
- Computer use on the direct Claude API requires `computer_toolset_20260801`. Preserve the returned tool-use content: its `toolset_name` is retained on replay and copied to matching tool results. Remove the old fine-grained streaming beta when using toolsets.
See the [migration guide](https://platform.claude.com/docs/en/models/sonnet-5-5/migration-guide) for platform-specific restrictions, advisor pairings, and refusal/fallback behavior.
## Resources
- [LangChain Academy](https://academy.langchain.com/) — comprehensive, free courses on LangChain libraries and products, made by the LangChain team
@@ -129,8 +129,9 @@ def _convert_from_v1_to_anthropic(
"input": block.get("args", {}),
"id": block.get("id", ""),
}
if "caller" in block.get("extras", {}):
tool_use_block["caller"] = block["extras"]["caller"]
for key in ("caller", "toolset_name"):
if key in block.get("extras", {}):
tool_use_block[key] = block["extras"][key]
new_content.append(tool_use_block)
elif block["type"] == "tool_call_chunk":
@@ -147,6 +148,11 @@ def _convert_from_v1_to_anthropic(
"name": block.get("name", ""),
"input": input_,
"id": block.get("id", ""),
**{
key: block["extras"][key]
for key in ("caller", "toolset_name")
if key in block.get("extras", {})
},
}
)
@@ -720,6 +720,7 @@ def _format_messages(
"""Format messages for Anthropic's API."""
system: str | list[dict] | None = None
formatted_messages: list[dict] = []
toolsets: dict[str, str] = {}
merged_messages = _merge_messages(messages)
last_non_system_index = max(
(i for i, m in enumerate(merged_messages) if m.type != "system"),
@@ -795,11 +796,13 @@ def _format_messages(
for tc in message.tool_calls
if tc["id"] == block["id"]
]
content.extend(
_lc_tool_calls_to_anthropic_tool_use_blocks(
overlapping,
),
tool_blocks = _lc_tool_calls_to_anthropic_tool_use_blocks(
overlapping,
)
if toolset_name := block.get("toolset_name"):
for tool_block in tool_blocks:
tool_block["toolset_name"] = toolset_name
content.extend(tool_blocks)
else:
if tool_input := block.get("input"):
args = tool_input
@@ -818,6 +821,8 @@ def _format_messages(
)
if caller := block.get("caller"):
tool_use_block["caller"] = caller
if toolset_name := block.get("toolset_name"):
tool_use_block["toolset_name"] = toolset_name
content.append(tool_use_block)
elif block["type"] in ("server_tool_use", "mcp_tool_use"):
formatted_block = {
@@ -944,6 +949,8 @@ def _format_messages(
},
),
)
elif block["type"] == "advisor_tool_result":
content.append({k: v for k, v in block.items() if k != "index"})
else:
content.append(block)
else:
@@ -1036,6 +1043,16 @@ def _format_messages(
system = _format_system_content(pending.content, model=model)
_warn_system_message_hoisted(model)
pending_system = []
if isinstance(content, list):
for block in content:
if not isinstance(block, dict):
continue
if block.get("type") == "tool_use" and block.get("toolset_name"):
toolsets[block["id"]] = block["toolset_name"]
elif block.get("type") == "tool_result" and (
toolset_name := toolsets.get(block.get("tool_use_id", ""))
):
block.setdefault("toolset_name", toolset_name)
formatted_messages.append({"role": role, "content": content})
formatted_messages.extend(
@@ -1136,13 +1153,16 @@ def _supports_mid_conversation_system_messages(model: object) -> bool:
"claude-mythos-5",
"claude-opus-4-8",
"claude-opus-5",
"claude-sonnet-5-5",
)
)
def _supports_forced_tool_choice(model: str) -> bool:
"""Return whether the model accepts `tool_choice` types `any` and `tool`."""
return not model.startswith(("claude-fable-5-1", "claude-opus-5-5"))
return not model.startswith(
("claude-fable-5-1", "claude-opus-5-5", "claude-sonnet-5-5")
)
def _is_direct_anthropic_llm_type(llm_type: object) -> bool:
@@ -1793,7 +1813,8 @@ class ChatAnthropic(BaseChatModel):
output_config["effort"] = effort
is_fable_model = self.model.startswith("claude-fable-5")
if is_fable_model:
is_sonnet_55 = self.model.startswith("claude-sonnet-5-5")
if is_fable_model or is_sonnet_55:
top_k = request_config.get("top_k", self.top_k)
top_p = request_config.get("top_p", self.top_p)
temperature = request_config.get("temperature", self.temperature)
@@ -1819,7 +1840,7 @@ class ChatAnthropic(BaseChatModel):
raise ValueError(msg)
if (
(self.model.startswith("claude-opus-5") or is_fable_model)
(self.model.startswith("claude-opus-5") or is_fable_model or is_sonnet_55)
and isinstance(thinking, Mapping)
and thinking.get("type") == "enabled"
):
@@ -2331,6 +2352,12 @@ class ChatAnthropic(BaseChatModel):
warnings.warn("Received unexpected tool content block.", stacklevel=2)
content_block = event.content_block.model_dump()
if event.content_block.type == "advisor_tool_result":
content_block = {
key: content_block[key]
for key in ("type", "tool_use_id", "content", "cache_control")
if key in content_block
}
if "caller" in content_block and content_block["caller"] is None:
content_block.pop("caller")
content_block["index"] = event.index
@@ -2464,6 +2491,8 @@ class ChatAnthropic(BaseChatModel):
"stop_reason": event.delta.stop_reason,
"stop_sequence": event.delta.stop_sequence,
}
if stop_details := event.delta.model_dump().get("stop_details"):
response_metadata["stop_details"] = stop_details
if context_management := getattr(event, "context_management", None):
response_metadata["context_management"] = (
context_management.model_dump()
@@ -2882,8 +2911,8 @@ class ChatAnthropic(BaseChatModel):
- `'function_calling'` (default): Use forced tool calling to get
structured output. When `thinking` is enabled, or on models
that don't support forced tool use (Claude Opus 5.5, Claude
Fable 5.1), the tool call isn't forced, and a missing tool
call raises `OutputParserException`.
Fable 5.1, Claude Sonnet 5.5), the tool call isn't forced,
and a missing tool call raises `OutputParserException`.
- `'json_schema'`: Use Claude's dedicated
[structured output](https://platform.claude.com/docs/en/build-with-claude/structured-outputs)
feature.
@@ -3238,6 +3267,7 @@ class _AnthropicToolUse(TypedDict):
input: dict
id: str
caller: NotRequired[dict[str, Any]]
toolset_name: NotRequired[str]
def _lc_tool_calls_to_anthropic_tool_use_blocks(
@@ -488,4 +488,39 @@ _PROFILES: dict[str, dict[str, Any]] = {
],
"reasoning_effort_default": "high",
},
"claude-sonnet-5-5": {
"name": "Claude Sonnet 5.5",
"release_date": "2026-09-28",
"last_updated": "2026-09-28",
"open_weights": False,
"max_input_tokens": 1000000,
"max_output_tokens": 128000,
"text_inputs": True,
"image_inputs": True,
"audio_inputs": False,
"pdf_inputs": True,
"video_inputs": False,
"text_outputs": True,
"image_outputs": False,
"audio_outputs": False,
"video_outputs": False,
"reasoning_output": True,
"tool_calling": True,
"structured_output": True,
"attachment": True,
"temperature": False,
"image_url_inputs": True,
"pdf_tool_message": True,
"image_tool_message": True,
"tool_call_streaming": True,
"tool_choice": False,
"reasoning_effort_levels": [
"low",
"medium",
"high",
"xhigh",
"max",
],
"reasoning_effort_default": "high",
},
}
@@ -53,6 +53,12 @@ structured_output = true
reasoning_effort_levels = ["low", "medium", "high", "xhigh", "max"]
reasoning_effort_default = "medium"
[overrides."claude-sonnet-5-5"]
structured_output = true
tool_choice = false
reasoning_effort_levels = ["low", "medium", "high", "xhigh", "max"]
reasoning_effort_default = "high"
[overrides."claude-sonnet-5"]
structured_output = true
reasoning_effort_levels = ["low", "medium", "high", "xhigh", "max"]
@@ -0,0 +1,400 @@
from typing import Any, cast
from unittest.mock import MagicMock, patch
import pytest
from anthropic._models import construct_type
from anthropic.types import (
RawContentBlockStartEvent,
RawMessageDeltaEvent,
ThinkingBlock,
)
from anthropic.types.beta import BetaRawMessageStreamEvent
from langchain_core.exceptions import OutputParserException
from langchain_core.messages import (
AIMessage,
AIMessageChunk,
HumanMessage,
SystemMessage,
ToolMessage,
)
from langchain_core.runnables import RunnableBinding, RunnableSequence
from langchain_anthropic import ChatAnthropic
from langchain_anthropic.chat_models import _format_messages
MODEL = "claude-sonnet-5-5"
TOOL = {"name": "answer", "input_schema": {"type": "object", "properties": {}}}
def model(**kwargs: Any) -> ChatAnthropic:
return ChatAnthropic(model=MODEL, api_key="test", **kwargs)
@pytest.mark.parametrize(
("choice", "expected"),
[
("any", {"type": "any"}),
("answer", {"type": "tool", "name": "answer"}),
({"type": "any"}, {"type": "any"}),
({"type": "tool", "name": "answer"}, {"type": "tool", "name": "answer"}),
],
)
def test_forced_tool_choice_left_to_api(choice: Any, expected: dict[str, str]) -> None:
llm = model()
assert (
cast("RunnableBinding", llm.bind_tools([TOOL], tool_choice=choice)).kwargs[
"tool_choice"
]
== expected
)
assert (
llm._get_request_payload("hello", tool_choice=expected)["tool_choice"]
== expected
)
def test_tool_choice_and_structured_output() -> None:
llm = model()
assert cast("RunnableBinding", llm.bind_tools([TOOL], tool_choice="auto")).kwargs[
"tool_choice"
] == {"type": "auto"}
with pytest.warns(UserWarning, match="json_schema"):
structured = llm.with_structured_output(TOOL)
assert (
"tool_choice"
not in cast(
"RunnableBinding", cast("RunnableSequence", structured).first
).kwargs
)
native = llm.with_structured_output(
{"title": "Answer", "type": "object", "properties": {}}, method="json_schema"
)
assert (
cast("RunnableBinding", cast("RunnableSequence", native).first).kwargs[
"output_config"
]["format"]["type"]
== "json_schema"
)
older = ChatAnthropic(model="claude-sonnet-5", api_key="test")
assert cast("RunnableBinding", older.bind_tools([TOOL], tool_choice="any")).kwargs[
"tool_choice"
] == {"type": "any"}
@pytest.mark.parametrize(
"kwargs",
[
{"thinking": {"type": "disabled"}},
{"thinking": {"type": "enabled", "budget_tokens": 1024}},
{"temperature": 0},
{"top_p": 0.5},
{"top_k": 10},
],
)
def test_invalid_configuration(kwargs: dict[str, Any]) -> None:
with pytest.raises(ValueError):
model(**kwargs)._get_request_payload("hello")
with pytest.raises(ValueError):
model()._get_request_payload("hello", **kwargs)
@pytest.mark.parametrize("effort", ["low", "medium", "high", "xhigh", "max"])
def test_between_tools(effort: str) -> None:
payload = model(thinking={"type": "between_tools"})._get_request_payload(
"hello", effort=effort
)
assert payload["thinking"] == {"type": "between_tools"}
assert payload["output_config"] == {"effort": effort}
@pytest.mark.parametrize("extra", [{"display": "summarized"}, {"budget_tokens": 1024}])
def test_between_tools_extra_fields_left_to_api(extra: dict[str, Any]) -> None:
thinking = {"type": "between_tools", **extra}
assert (
model(thinking=thinking)._get_request_payload("hello")["thinking"] == thinking
)
def test_profile_and_defaults() -> None:
llm = model()
assert llm.max_tokens == 128000
assert llm.profile is not None
assert llm.profile["max_input_tokens"] == 1000000
assert llm.profile["structured_output"] is True
assert llm.profile["tool_choice"] is False
assert llm.profile["reasoning_effort_levels"] == [
"low",
"medium",
"high",
"xhigh",
"max",
]
assert llm.profile["reasoning_effort_default"] == "high"
payload = llm._get_request_payload("hello")
assert not {"temperature", "top_p", "top_k", "thinking"} & payload.keys()
assert llm._get_request_payload("hello", effort="medium")["thinking"] == {
"type": "adaptive",
"display": "summarized",
}
def test_mid_conversation_system() -> None:
system, messages = _format_messages(
[
SystemMessage("initial"),
HumanMessage("hello"),
SystemMessage("new instructions"),
AIMessage("answer"),
HumanMessage("next"),
],
model=MODEL,
)
assert system == "initial"
assert [m["role"] for m in messages] == ["user", "system", "assistant", "user"]
@pytest.mark.parametrize("standard", [False, True])
def test_toolset_round_trip(*, standard: bool) -> None:
content: list[str | dict[str, Any]] = [
{"type": "thinking", "thinking": "", "signature": "opaque-signature"},
{
"type": "tool_use",
"id": "call_1",
"name": "click",
"toolset_name": "computer",
"input": {"x": 1},
},
]
ai = AIMessage(
content=content,
tool_calls=[
{"name": "click", "id": "call_1", "args": {"x": 2}, "type": "tool_call"}
],
response_metadata={"model_provider": "anthropic"},
)
if standard:
ai = ai.model_copy(
update={
"content": ai.content_blocks,
"response_metadata": {
"model_provider": "anthropic",
"output_version": "v1",
},
}
)
payload = model()._get_request_payload(
[HumanMessage("click"), ai, ToolMessage("done", tool_call_id="call_1")]
)
assert payload["messages"][1]["content"][0] == content[0]
tool = payload["messages"][1]["content"][1]
assert tool["toolset_name"] == "computer"
assert tool["input"] == {"x": 2}
assert payload["messages"][2]["content"][0]["toolset_name"] == "computer"
def test_encrypted_advisor_streaming() -> None:
advisor_result = {
"type": "advisor_tool_result",
"tool_use_id": "srvtoolu_abc123",
"content": {
"type": "advisor_redacted_result",
"encrypted_content": "opaque-ciphertext",
},
}
event = cast(
RawContentBlockStartEvent,
construct_type(
type_=RawContentBlockStartEvent,
value={
"type": "content_block_start",
"index": 0,
"content_block": advisor_result,
},
),
)
llm = model()
chunk, _ = llm._make_message_chunk_from_anthropic_event(
event, stream_usage=True, coerce_content_to_string=False, block_start_event=None
)
assert chunk is not None
assert chunk.content == [{**advisor_result, "index": 0}]
payload = llm._get_request_payload(
[HumanMessage("help"), chunk, HumanMessage("continue")]
)
assert payload["messages"][1]["content"] == [advisor_result]
@pytest.mark.parametrize("output_version", ["v0", "v1"])
def test_encrypted_advisor_stream_aggregate_replay(output_version: str) -> None:
server_tool_use = {
"type": "server_tool_use",
"id": "srvtoolu_abc123",
"name": "advisor",
"input": {},
}
advisor_result = {
"type": "advisor_tool_result",
"tool_use_id": "srvtoolu_abc123",
"content": {
"type": "advisor_redacted_result",
"encrypted_content": "opaque-ciphertext",
"stop_reason": None,
},
}
raw_events = [
{
"type": "message_start",
"message": {
"id": "msg_1",
"type": "message",
"role": "assistant",
"model": MODEL,
"content": [],
"stop_reason": None,
"stop_sequence": None,
"usage": {"input_tokens": 10, "output_tokens": 1},
},
},
{"type": "content_block_start", "index": 0, "content_block": server_tool_use},
{"type": "content_block_stop", "index": 0},
{"type": "content_block_start", "index": 1, "content_block": advisor_result},
{"type": "content_block_stop", "index": 1},
{
"type": "content_block_start",
"index": 2,
"content_block": {"type": "text", "text": ""},
},
{
"type": "content_block_delta",
"index": 2,
"delta": {"type": "text_delta", "text": "Use a token bucket."},
},
{"type": "content_block_stop", "index": 2},
{
"type": "message_delta",
"delta": {"stop_reason": "end_turn", "stop_sequence": None},
"usage": {"output_tokens": 20},
},
{"type": "message_stop"},
]
events = [
construct_type(type_=BetaRawMessageStreamEvent, value=event)
for event in raw_events
]
llm = model(output_version=output_version).bind_tools(
[{"type": "advisor_20260301", "name": "advisor", "model": "claude-opus-5"}]
)
with patch.object(
ChatAnthropic, "_create", return_value=MagicMock(parse=lambda: iter(events))
):
chunks = [cast("AIMessageChunk", chunk) for chunk in llm.stream("help")]
full = chunks[0]
for chunk in chunks[1:]:
full += chunk
payload = model()._get_request_payload(
[HumanMessage("help"), full, HumanMessage("continue")]
)
assert payload["messages"][1]["content"] == [
server_tool_use,
advisor_result,
{"type": "text", "text": "Use a token bucket."},
]
def test_refusal_details_streaming() -> None:
event = RawMessageDeltaEvent.model_validate(
{
"type": "message_delta",
"delta": {
"stop_reason": "refusal",
"stop_sequence": None,
"stop_details": {"type": "refusal", "category": "cyber"},
},
"usage": {"output_tokens": 3},
}
)
chunk, _ = model()._make_message_chunk_from_anthropic_event(
event, stream_usage=True, coerce_content_to_string=False, block_start_event=None
)
assert chunk is not None
assert chunk.response_metadata["stop_details"]["category"] == "cyber"
def test_unforced_structured_output_requires_tool_call() -> None:
with pytest.warns(UserWarning, match="json_schema"):
structured = model().with_structured_output(TOOL)
check = cast("RunnableSequence", structured).steps[1]
with pytest.raises(OutputParserException):
check.invoke(AIMessage("No tool call"))
def test_mid_conversation_tool_change() -> None:
block = {
"type": "tool_addition",
"tool": {"type": "tool_reference", "name": "answer"},
}
_, messages = _format_messages(
[HumanMessage("hello"), SystemMessage([block]), AIMessage("answer")],
model=MODEL,
)
assert messages[1] == {"role": "system", "content": [block]}
def test_default_thinking_stream_preserves_signature() -> None:
event = RawContentBlockStartEvent(
type="content_block_start",
index=0,
content_block=ThinkingBlock(
type="thinking", thinking="", signature="opaque-signature"
),
)
chunk, _ = model()._make_message_chunk_from_anthropic_event(
event, stream_usage=True, coerce_content_to_string=True, block_start_event=None
)
assert chunk is not None
payload = model()._get_request_payload(
[HumanMessage("hello"), chunk, HumanMessage("next")]
)
assert payload["messages"][1]["content"] == [
{"type": "thinking", "thinking": "", "signature": "opaque-signature"}
]
def test_standard_tool_chunk_namespace_replay() -> None:
chunk = AIMessageChunk(
content=[
{
"type": "tool_use",
"name": "click",
"id": "call_1",
"input": {},
"toolset_name": "computer",
"index": 0,
}
],
tool_call_chunks=[
{
"type": "tool_call_chunk",
"name": "click",
"id": "call_1",
"args": "{}",
"index": 0,
}
],
response_metadata={"model_provider": "anthropic"},
)
chunk = chunk.model_copy(
update={
"content": chunk.content_blocks,
"response_metadata": {
"model_provider": "anthropic",
"output_version": "v1",
},
}
)
payload = model()._get_request_payload(
[HumanMessage("click"), chunk, ToolMessage("done", tool_call_id="call_1")]
)
assert payload["messages"][1]["content"][0]["toolset_name"] == "computer"
assert payload["messages"][2]["content"][0]["toolset_name"] == "computer"