Skip to content
Open
Original file line number Diff line number Diff line change
Expand Up @@ -92,14 +92,17 @@ def _tool_result_to_content_blocks(content: list[ToolResultContent]) -> list[Con


def _flatten_messages_to_content(messages: Messages) -> list[ContentBlock]:
"""Flatten messages into a single ContentBlock list with role markers."""
"""Flatten messages into a single ContentBlock list with role markers.

Reasoning blocks are left out: their signatures are bound to the conversation that produced them.
"""
blocks: list[ContentBlock] = []
for message in messages:
blocks.append(ContentBlock(text=f"\n---\n[{message['role']}]"))
for block in message["content"]:
if "toolResult" in block:
blocks.extend(_tool_result_to_content_blocks(block["toolResult"]["content"]))
else:
elif "reasoningContent" not in block:
blocks.append(block)
return blocks

Expand Down
53 changes: 51 additions & 2 deletions strands-py/src/strands/models/anthropic.py
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,15 @@
from ..types.tools import ToolChoice, ToolChoiceToolDict, ToolSpec
from ._defaults import resolve_config_metadata
from ._validation import _has_location_source, _warn_on_deprecated_cache_tools, validate_config_keys
from .model import BaseModelConfig, CacheConfig, CacheToolsConfig, Model
from .model import (
_THINKING_BINDING_BETA,
BaseModelConfig,
CacheConfig,
CacheToolsConfig,
Model,
_is_thinking_binding_error,
_with_drop_block,
)

logger = logging.getLogger(__name__)

Expand Down Expand Up @@ -148,6 +156,7 @@ def __init__(self, *, client_args: dict[str, Any] | None = None, **model_config:

client_args = client_args or {}
self.client = anthropic.AsyncAnthropic(**client_args)
self._drop_stale_thinking_model_id: str | None = None

@override
def update_config(self, **model_config: Unpack[AnthropicConfig]) -> None: # type: ignore[override]
Expand Down Expand Up @@ -536,8 +545,25 @@ def format_request(
"tools": tools,
}

if self._drop_stale_thinking_model_id == self.config["model_id"]:
request = self._with_drop_block(request) or request

return request

def _with_drop_block(self, request: dict[str, Any]) -> dict[str, Any] | None:
Comment thread
lizradway marked this conversation as resolved.
Outdated
Comment thread
lizradway marked this conversation as resolved.
Outdated
"""Return the request with stale thinking blocks dropped, or None if it cannot change."""
thinking = _with_drop_block(request.get("thinking"))
if thinking is None:
return None
extra_headers = dict(request.get("extra_headers") or {})
# Request headers replace client headers of the same name, so betas set on the client are carried over.
configured = extra_headers.get("anthropic-beta") or self.client.default_headers.get("anthropic-beta")
betas = [beta.strip() for beta in configured.split(",")] if isinstance(configured, str) else []
if _THINKING_BINDING_BETA not in betas:
betas.append(_THINKING_BINDING_BETA)
Comment thread
lizradway marked this conversation as resolved.
Outdated
extra_headers["anthropic-beta"] = ",".join(beta for beta in betas if beta)
return {**request, "thinking": thinking, "extra_headers": extra_headers}

def _format_system_prompt(
self, system_prompt: str | None, system_prompt_content: list[SystemContentBlock] | None
) -> str | list[dict[str, Any]] | None:
Expand Down Expand Up @@ -937,6 +963,7 @@ async def stream(
usage: dict[str, int] = {}
block_index_offset = 0
continuations = 0
retry_with_drop_block = False
try:
while True:
async with self.client.messages.stream(**request) as stream:
Expand Down Expand Up @@ -1021,7 +1048,29 @@ async def stream(
if any(overflow_message in str(error).lower() for overflow_message in AnthropicModel.OVERFLOW_MESSAGES):
raise ContextWindowOverflowException(str(error)) from error

raise error
retry_with_drop_block = (
continuations == 0 and _is_thinking_binding_error(error) and self._with_drop_block(request) is not None
)
if not retry_with_drop_block:
raise error

if retry_with_drop_block:
# Stale blocks stay in history, so drop_block stays on for every later request to this model.
logger.warning(
"model_id=<%s> | history edit invalidated replayed thinking blocks, retrying with drop_block",
request["model"],
)
self._drop_stale_thinking_model_id = request["model"]
Comment thread
lizradway marked this conversation as resolved.
Outdated
async for chunk in self.stream(
Comment thread
lizradway marked this conversation as resolved.
messages,
tool_specs,
system_prompt,
tool_choice=tool_choice,
system_prompt_content=system_prompt_content,
**kwargs,
):
yield chunk
return

logger.debug("finished streaming response from model")

Expand Down
49 changes: 46 additions & 3 deletions strands-py/src/strands/models/bedrock.py
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,15 @@
from ._defaults import resolve_config_metadata
from ._strict_schema import ensure_strict_json_schema
from ._validation import _warn_on_deprecated_cache_tools, validate_config_keys
from .model import BaseModelConfig, CacheConfig, CacheToolsConfig, Model
from .model import (
_THINKING_BINDING_BETA,
BaseModelConfig,
CacheConfig,
CacheToolsConfig,
Model,
_is_thinking_binding_error,
_with_drop_block,
)

logger = logging.getLogger(__name__)

Expand Down Expand Up @@ -141,6 +149,17 @@ async def _next_stream_event(
_CANCEL_POLL_INTERVAL = 0.05


def _with_drop_block_fields(fields: dict[str, Any]) -> dict[str, Any] | None:
"""Return additional request fields that drop stale thinking blocks, or None if they cannot change."""
thinking = _with_drop_block(fields.get("thinking"))
if thinking is None:
return None
betas = list(fields.get("anthropic_beta") or [])
if _THINKING_BINDING_BETA not in betas:
betas.append(_THINKING_BINDING_BETA)
return {**fields, "anthropic_beta": betas, "thinking": thinking}


class BedrockModel(Model):
"""AWS Bedrock model provider implementation.

Expand Down Expand Up @@ -260,6 +279,7 @@ def __init__(
include_tool_result_status="auto",
)
self._tool_result_turn_separation_model_id: str | None = None
self._drop_stale_thinking_model_id: str | None = None
self.update_config(**model_config)

logger.debug("config=<%s> | initializing", self.config)
Expand Down Expand Up @@ -489,8 +509,31 @@ def _get_additional_request_fields(self, tool_choice: ToolChoice | None) -> dict
return {"additionalModelRequestFields": fields_without_thinking}
return {}

if self._drop_stale_thinking_model_id == self.config["model_id"]:
additional_fields = _with_drop_block_fields(additional_fields) or additional_fields

return {"additionalModelRequestFields": additional_fields}

def _converse(self, converse_method: Callable[..., Any], request: dict[str, Any]) -> Any:
"""Invoke the model, retrying once with drop_block when an edit before replayed thinking made it stale.

The retry turns drop_block on for every later request to this model, since the stale blocks stay in history.
"""
try:
return converse_method(**request)
except ClientError as error:
fields = request.get("additionalModelRequestFields") or {}
retry_fields = _with_drop_block_fields(fields) if _is_thinking_binding_error(error) else None
if retry_fields is None:
raise

logger.warning(
"model_id=<%s> | history edit invalidated replayed thinking blocks, retrying with drop_block",
request["modelId"],
)
self._drop_stale_thinking_model_id = request["modelId"]
return converse_method(**{**request, "additionalModelRequestFields": retry_fields})

def _apply_system_cache_ttl(
self, system_blocks: list[SystemContentBlock], tools_cache_point: list[dict[str, Any]]
) -> list[SystemContentBlock]:
Expand Down Expand Up @@ -1456,7 +1499,7 @@ def _stream(
converse_method = self.client.converse_stream if streaming else self.client.converse

try:
response = converse_method(**request)
response = self._converse(converse_method, request)
except ClientError as error:
error_details = error.response.get("Error", {})
if error_details.get(
Expand All @@ -1472,7 +1515,7 @@ def _stream(

logger.debug("model_id=<%s> | separating tool result and conversation turns", model_id)
request = {**request, "messages": separated_messages}
response = converse_method(**request)
response = self._converse(converse_method, request)
self._tool_result_turn_separation_model_id = model_id

logger.debug("got response from model")
Expand Down
21 changes: 21 additions & 0 deletions strands-py/src/strands/models/model.py
Original file line number Diff line number Diff line change
Expand Up @@ -122,6 +122,27 @@ def _estimate_tokens_with_heuristic(
return total


# Preserved-thinking Claude models bind each thinking block to the content before it. When that content
Comment thread
lizradway marked this conversation as resolved.
Outdated
# changes, enforcing accounts reject the request unless it asks the API to drop the stale blocks.
_THINKING_BINDING_BETA = "thinking-binding-controls-2026-08-01"
_THINKING_BINDING_ERROR = "bound to a different conversation"


def _is_thinking_binding_error(error: BaseException) -> bool:
"""Whether the provider rejected a replayed thinking block because history before it changed."""
return _THINKING_BINDING_ERROR in str(error)
Comment thread
lizradway marked this conversation as resolved.
Outdated


def _with_drop_block(thinking: Any) -> dict[str, Any] | None:
"""Return the thinking config with stale blocks dropped, or None if it cannot or should not change.
A caller-set ``block_binding`` is left as written.
"""
if not isinstance(thinking, dict) or "block_binding" in thinking:
return None
return {**thinking, "block_binding": {"prefix_mismatch_behavior": "drop_block"}}


class BaseModelConfig(TypedDict, total=False):
"""Base configuration shared by all model providers.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -91,6 +91,19 @@ def test_extracts_tool_result_content(self):
assert tru_result[0] == ContentBlock(text="\n---\n[user]")
assert tru_result[1] == ContentBlock(text="result data")

def test_excludes_reasoning(self):
messages = [
Message(
role="assistant",
content=[
ContentBlock(reasoningContent={"reasoningText": {"text": "plan", "signature": "sig"}}),
ContentBlock(text="hi"),
],
),
]
tru_result = _flatten_messages_to_content(messages)
assert tru_result == [ContentBlock(text="\n---\n[assistant]"), ContentBlock(text="hi")]


class TestSummarizeContent:
"""Tests for the summarize_content function."""
Expand Down
83 changes: 83 additions & 0 deletions strands-py/tests/strands/models/test_anthropic.py
Original file line number Diff line number Diff line change
Expand Up @@ -1055,6 +1055,89 @@ async def test_stream_bad_request_error(anthropic_client, model):
await anext(model.stream(messages))


def _thinking_binding_error():
return anthropic.BadRequestError(
"messages.1.content.0: Invalid `signature` in `thinking` block. "
"The block is bound to a different conversation.",
response=unittest.mock.Mock(),
body=None,
)


def _message_start_stream():
event = unittest.mock.Mock(type="message_start", model_dump=lambda: {"type": "message_start"})
final_message = unittest.mock.Mock(usage=unittest.mock.Mock(model_dump=lambda: {}))
return generate_mock_stream_context([event], final_message=final_message)


@pytest.mark.asyncio
async def test_stream_retries_with_drop_block_on_thinking_binding_error(anthropic_client, alist, messages, caplog):
anthropic_client.default_headers = {"anthropic-beta": "client-beta"}
anthropic_client.messages.stream.side_effect = [_thinking_binding_error(), _message_start_stream()]
model = AnthropicModel(model_id="m1", max_tokens=1, params={"thinking": {"type": "adaptive"}})

with caplog.at_level(logging.WARNING, logger="strands.models.anthropic"):
tru_events = await alist(model.stream(messages))

assert tru_events == [{"messageStart": {"role": "assistant"}}]
assert "retrying with drop_block" in caplog.text

first_request, retry_request = [call.kwargs for call in anthropic_client.messages.stream.call_args_list]
assert first_request["thinking"] == {"type": "adaptive"}
assert "extra_headers" not in first_request
exp_thinking = {"type": "adaptive", "block_binding": {"prefix_mismatch_behavior": "drop_block"}}
exp_headers = {"anthropic-beta": "client-beta,thinking-binding-controls-2026-08-01"}
assert retry_request["thinking"] == exp_thinking
assert retry_request["extra_headers"] == exp_headers

anthropic_client.messages.stream.reset_mock(side_effect=True)
anthropic_client.messages.stream.return_value = _message_start_stream()

await alist(model.stream(messages))

anthropic_client.messages.stream.assert_called_once()
assert anthropic_client.messages.stream.call_args.kwargs["thinking"] == exp_thinking
assert anthropic_client.messages.stream.call_args.kwargs["extra_headers"] == exp_headers


@pytest.mark.parametrize(
"params",
[
{"thinking": {"type": "adaptive", "block_binding": {"prefix_mismatch_behavior": "error"}}},
{"temperature": 1},
],
)
@pytest.mark.asyncio
async def test_stream_does_not_retry_thinking_binding_error_without_drop_block(anthropic_client, messages, params):
anthropic_client.messages.stream.side_effect = _thinking_binding_error()
model = AnthropicModel(model_id="m1", max_tokens=1, params=params)

with pytest.raises(anthropic.BadRequestError, match="bound to a different conversation"):
await anext(model.stream(messages))

anthropic_client.messages.stream.assert_called_once()
assert model._drop_stale_thinking_model_id is None


def test_format_request_drops_stale_thinking_only_for_remembered_model(anthropic_client, messages):
_ = anthropic_client
model = AnthropicModel(
model_id="m1",
max_tokens=1,
params={"thinking": {"type": "adaptive"}, "extra_headers": {"anthropic-beta": "other-beta"}},
)
model._drop_stale_thinking_model_id = "m1"

tru_request = model.format_request(messages)

assert tru_request["thinking"] == {"type": "adaptive", "block_binding": {"prefix_mismatch_behavior": "drop_block"}}
assert tru_request["extra_headers"] == {"anthropic-beta": "other-beta,thinking-binding-controls-2026-08-01"}

model.update_config(model_id="m2")

assert model.format_request(messages)["thinking"] == {"type": "adaptive"}


@pytest.mark.asyncio
async def test_structured_output(anthropic_client, model, test_output_model_cls, alist):
messages = [{"role": "user", "content": [{"text": "Generate a person"}]}]
Expand Down
Loading
Loading