diff --git a/python/packages/anthropic/agent_framework_anthropic/_chat_client.py b/python/packages/anthropic/agent_framework_anthropic/_chat_client.py index accc9aa0098..3a6c4dbd585 100644 --- a/python/packages/anthropic/agent_framework_anthropic/_chat_client.py +++ b/python/packages/anthropic/agent_framework_anthropic/_chat_client.py @@ -5,7 +5,7 @@ import asyncio import logging import sys -from collections.abc import AsyncIterable, Awaitable, Callable, Mapping, Sequence +from collections.abc import AsyncIterable, Awaitable, Callable, Mapping, Sequence, Set from dataclasses import dataclass, field from typing import Any, ClassVar, Final, Generic, Literal, TypedDict, cast @@ -284,6 +284,7 @@ class _AnthropicRequestState: active_call_id: str | None = None active_call_content_type: str | None = None + active_call_is_hosted_function: bool = False tool_name_aliases: dict[str, str] = field(default_factory=dict[str, str]) @@ -901,9 +902,11 @@ def _prepare_messages_for_anthropic(self, messages: Sequence[Message]) -> list[d else: msgs = list(messages) + pending_hosted_calls = self._pending_hosted_function_calls(msgs) + result: list[dict[str, Any]] = [] for msg in msgs: - result.extend(self._prepare_message_groups_for_anthropic(msg)) + result.extend(self._prepare_message_groups_for_anthropic(msg, pending_hosted_calls)) # Anthropic requires the conversation to end with a user message. # Append a synthetic user turn so chained agent outputs work as @@ -913,9 +916,43 @@ def _prepare_messages_for_anthropic(self, messages: Sequence[Message]) -> list[d return result - def _prepare_message_groups_for_anthropic(self, message: Message) -> list[dict[str, Any]]: + @staticmethod + def _pending_hosted_function_calls(messages: Sequence[Message]) -> set[int]: + """Return the identities (``id()``) of hosted function calls that have no result yet. + + A hosted (server tool) call without a result is still running on Anthropic's side, for example + when Claude called a local tool in parallel. It has to be sent back as ``server_tool_use``, or + Anthropic expects a local ``tool_result`` for it. + + A ``call_id`` is not unique across a transcript: a later occurrence may reuse it. Calls and results + are therefore correlated in transcript order. A result settles the open call occurrence with the + same ``call_id`` whose ``Content.id`` it carries, and otherwise the earliest open occurrence with + that ``call_id``. A result never settles a call that appears after it. + """ + open_calls_by_call_id: dict[str, list[Content]] = {} + for message in messages: + for content in message.contents: + if not content.call_id: + continue + if content.type == "function_call": + open_calls_by_call_id.setdefault(content.call_id, []).append(content) + elif content.type == "function_result" and (open_calls := open_calls_by_call_id.get(content.call_id)): + settled_index = next( + ( + index + for index, call in enumerate(open_calls) + if content.id is not None and call.id == content.id + ), + 0, + ) + open_calls.pop(settled_index) + return {id(call) for open_calls in open_calls_by_call_id.values() for call in open_calls if call.hosted} + + def _prepare_message_groups_for_anthropic( + self, message: Message, pending_hosted_calls: Set[int] | None = None + ) -> list[dict[str, Any]]: """Prepare a Message and split Anthropic content blocks into valid role groups.""" - prepared_message = self._prepare_message_for_anthropic(message) + prepared_message = self._prepare_message_for_anthropic(message, pending_hosted_calls) content = prepared_message.get("content") if not isinstance(content, list): return [prepared_message] @@ -964,11 +1001,17 @@ def _message_has_tool_use(self, message: dict[str, Any]) -> bool: return False - def _prepare_message_for_anthropic(self, message: Message) -> dict[str, Any]: + def _prepare_message_for_anthropic( + self, message: Message, pending_hosted_calls: Set[int] | None = None + ) -> dict[str, Any]: """Prepare a Message for the Anthropic client. Args: message: The Message to convert. + pending_hosted_calls: Identities (``id()``) of the hosted (server tool) call contents that have + no result yet, from ``_pending_hosted_function_calls``. These are sent as ``server_tool_use`` + blocks. An identity rather than a ``call_id`` is used because a later occurrence may reuse + the ``call_id`` of a call that already has its result. Returns: A dictionary representing the message in Anthropic format. @@ -1001,8 +1044,9 @@ def _prepare_message_for_anthropic(self, message: Message) -> dict[str, Any]: else: logger.debug(f"Ignoring unsupported data content media type: {content.media_type} for now") case "function_call": + pending_hosted_call = content.hosted and id(content) in (pending_hosted_calls or ()) a_content.append({ - "type": "tool_use", + "type": "server_tool_use" if pending_hosted_call else "tool_use", "id": content.call_id, "name": content.name, "input": content.parse_arguments(), @@ -1411,6 +1455,7 @@ def _parse_contents_from_anthropic( case "tool_use" | "mcp_tool_use" | "server_tool_use": request_state.active_call_id = content_block.id request_state.active_call_content_type = content_block.type + request_state.active_call_is_hosted_function = False if content_block.type == "mcp_tool_use": contents.append( Content.from_mcp_server_tool_call( @@ -1437,6 +1482,7 @@ def _parse_contents_from_anthropic( ) else: resolved_tool_name = request_state.tool_name_aliases.get(content_block.name, content_block.name) + request_state.active_call_is_hosted_function = content_block.type == "server_tool_use" contents.append( Content.from_function_call( call_id=content_block.id, @@ -1682,16 +1728,20 @@ def _parse_contents_from_anthropic( ) ) case "input_json_delta": - # Skip argument deltas for MCP and server tools — execution is handled server-side. - if request_state.active_call_content_type in ("mcp_tool_use", "server_tool_use"): - pass - else: + # MCP and code execution server tool calls are parsed into their own content types, so + # their argument deltas are skipped. Argument deltas of other server tool calls (such as + # web search) are kept as hosted, informational-only fragments that merge into the call, + # so a pending call is replayed with its full input. + hosted_call = request_state.active_call_is_hosted_function + if hosted_call or request_state.active_call_content_type not in ("mcp_tool_use", "server_tool_use"): call_id = request_state.active_call_id or "" contents.append( Content.from_function_call( call_id=call_id, name="", arguments=content_block.partial_json, + informational_only=hosted_call, + hosted=hosted_call, merge_mode="append" if streaming else None, raw_representation=content_block, ) diff --git a/python/packages/anthropic/tests/test_anthropic_client.py b/python/packages/anthropic/tests/test_anthropic_client.py index e0bf7aed4ac..8dd3ea7c542 100644 --- a/python/packages/anthropic/tests/test_anthropic_client.py +++ b/python/packages/anthropic/tests/test_anthropic_client.py @@ -39,6 +39,7 @@ from anthropic.types.beta import ( BetaMessage, BetaMessageDeltaUsage, + BetaServerToolUseBlock, BetaTextBlock, BetaToolUseBlock, BetaUsage, @@ -825,6 +826,161 @@ def test_prepare_messages_for_anthropic_without_system( assert result[2]["content"] == "Continue" +def _parallel_local_and_server_tool_call_contents(client: AnthropicClient) -> list[Content]: + """Parse a response in which Claude calls a local tool and a server tool in parallel.""" + return client._parse_contents_from_anthropic( + [ + BetaToolUseBlock(type="tool_use", id="toolu_local", name="local_value", input={}), + BetaServerToolUseBlock( + type="server_tool_use", id="srvtoolu_search", name="web_search", input={"query": "release"} + ), + ], + _AnthropicRequestState(), + ) + + +@pytest.mark.parametrize("reload_history", [False, True], ids=["in_memory", "reloaded"]) +def test_prepare_messages_preserves_parallel_server_tool_use( + mock_anthropic_client: MagicMock, + reload_history: bool, +) -> None: + """A pending server tool call is sent back as server_tool_use next to the local tool result.""" + client = create_test_anthropic_client(mock_anthropic_client) + messages = [ + Message(role="user", contents=["Call local_value and search the web."]), + Message(role="assistant", contents=_parallel_local_and_server_tool_call_contents(client)), + Message(role="tool", contents=[Content.from_function_result(call_id="toolu_local", result="local-ok")]), + ] + if reload_history: + messages = [Message.from_dict(message.to_dict()) for message in messages] + + result = client._prepare_messages_for_anthropic(messages) + + assert [message["role"] for message in result] == ["user", "assistant", "user"] + assert result[1]["content"] == [ + {"type": "tool_use", "id": "toolu_local", "name": "local_value", "input": {}}, + {"type": "server_tool_use", "id": "srvtoolu_search", "name": "web_search", "input": {"query": "release"}}, + ] + assert [(block["type"], block["tool_use_id"]) for block in result[2]["content"]] == [("tool_result", "toolu_local")] + + +def test_prepare_messages_keeps_completed_server_tool_call_paired_with_its_result( + mock_anthropic_client: MagicMock, +) -> None: + """A server tool call that already has its result keeps the existing tool_use/tool_result pairing.""" + client = create_test_anthropic_client(mock_anthropic_client) + contents = _parallel_local_and_server_tool_call_contents(client) + search_result = Content.from_function_result(call_id="srvtoolu_search", result="search results") + messages = [ + Message(role="user", contents=["Call local_value and search the web."]), + Message(role="assistant", contents=contents), + Message(role="tool", contents=[Content.from_function_result(call_id="toolu_local", result="local-ok")]), + Message(role="assistant", contents=[search_result, Content.from_text("Both done.")]), + ] + + result = client._prepare_messages_for_anthropic(messages) + + assert [block["type"] for block in result[1]["content"]] == ["tool_use", "tool_use"] + tool_result_ids = [ + block["tool_use_id"] + for message in result + if message["role"] == "user" and isinstance(message["content"], list) + for block in message["content"] + if block["type"] == "tool_result" + ] + assert tool_result_ids == ["toolu_local", "srvtoolu_search"] + + +@pytest.mark.parametrize("reload_history", [False, True], ids=["in_memory", "reloaded"]) +def test_prepare_messages_keeps_later_server_tool_call_pending_when_call_id_is_reused( + mock_anthropic_client: MagicMock, + reload_history: bool, +) -> None: + """An earlier result for a reused call_id does not settle a later pending server tool call.""" + client = create_test_anthropic_client(mock_anthropic_client) + messages = [ + Message(role="user", contents=["Search the web."]), + Message( + role="assistant", + contents=[ + Content.from_function_call( + call_id="srvtoolu_search", name="web_search", arguments={"query": "first"}, hosted=True + ), + Content.from_function_result(call_id="srvtoolu_search", result="first results"), + Content.from_text("First search done."), + ], + ), + Message(role="user", contents=["Call local_value and search again."]), + Message(role="assistant", contents=_parallel_local_and_server_tool_call_contents(client)), + Message(role="tool", contents=[Content.from_function_result(call_id="toolu_local", result="local-ok")]), + ] + if reload_history: + messages = [Message.from_dict(message.to_dict()) for message in messages] + + result = client._prepare_messages_for_anthropic(messages) + + blocks = [ + (block["type"], block.get("id") or block.get("tool_use_id")) + for m in result + for block in m["content"] + if isinstance(m["content"], list) and block["type"] != "text" + ] + assert blocks == [ + ("tool_use", "srvtoolu_search"), + ("tool_result", "srvtoolu_search"), + ("tool_use", "toolu_local"), + ("server_tool_use", "srvtoolu_search"), + ("tool_result", "toolu_local"), + ] + + +def test_prepare_messages_settles_the_server_tool_call_occurrence_named_by_the_result_id( + mock_anthropic_client: MagicMock, +) -> None: + """A result carrying Content.id settles that occurrence, not the earliest one sharing its call_id.""" + client = create_test_anthropic_client(mock_anthropic_client) + messages = [ + Message(role="user", contents=["Search the web twice."]), + Message( + role="assistant", + contents=[ + Content.from_function_call( + call_id="srvtoolu_search", id="occurrence_1", name="web_search", arguments={}, hosted=True + ), + Content.from_function_call( + call_id="srvtoolu_search", id="occurrence_2", name="web_search", arguments={}, hosted=True + ), + Content.from_function_result(call_id="srvtoolu_search", id="occurrence_2", result="results"), + ], + ), + ] + + result = client._prepare_messages_for_anthropic(messages) + + assert [block["type"] for block in result[1]["content"]] == ["server_tool_use", "tool_use"] + + +def test_prepare_messages_does_not_settle_a_server_tool_call_with_an_earlier_result( + mock_anthropic_client: MagicMock, +) -> None: + """A result that precedes every call with its call_id leaves a later server tool call pending.""" + client = create_test_anthropic_client(mock_anthropic_client) + messages = [ + Message(role="user", contents=["Search the web."]), + Message(role="tool", contents=[Content.from_function_result(call_id="srvtoolu_search", result="stale")]), + Message( + role="assistant", + contents=[ + Content.from_function_call(call_id="srvtoolu_search", name="web_search", arguments={}, hosted=True) + ], + ), + ] + + result = client._prepare_messages_for_anthropic(messages) + + assert [block["type"] for block in result[-1]["content"]] == ["server_tool_use"] + + def test_prepare_messages_for_anthropic_does_not_append_after_tool_use( mock_anthropic_client: MagicMock, ) -> None: @@ -1845,14 +2001,13 @@ def test_parse_contents_from_anthropic_input_json_delta_no_duplicate_name( assert result[0].arguments == '"San Francisco"}' -def test_parse_contents_server_tool_use_input_json_delta_ignored( +def test_parse_contents_server_tool_use_input_json_delta_is_hosted_and_informational( mock_anthropic_client: MagicMock, ) -> None: - """Regression test: input_json_delta events are ignored after a server_tool_use block. + """input_json_delta events after a server_tool_use block stay hosted, informational-only fragments. - Server-managed tools have their execution handled server-side, so streaming - input_json_delta events must not produce Content.from_function_call(name='') - entries that would cause Anthropic API 400 errors on subsequent turns. + The fragments carry the server tool call's ``call_id`` and an empty name, so they merge into the + call and give it the streamed input without making it a local function call. """ client = create_test_anthropic_client(mock_anthropic_client) request_state = _AnthropicRequestState() @@ -1864,31 +2019,58 @@ def test_parse_contents_server_tool_use_input_json_delta_ignored( server_tool_content.name = "web_search" server_tool_content.input = {} - result = client._parse_contents_from_anthropic([server_tool_content], request_state) + result = client._parse_contents_from_anthropic([server_tool_content], request_state, streaming=True) # server_tool_use falls through to informational-only function_call (not mcp_tool_use / code_execution) assert len(result) == 1 assert result[0].type == "function_call" assert result[0].informational_only is True assert result[0].hosted is True assert request_state.active_call_content_type == "server_tool_use" + call = result[0] + + for partial_json in ('{"query": "latest', ' news"}'): + delta_content = MagicMock() + delta_content.type = "input_json_delta" + delta_content.partial_json = partial_json + + result = client._parse_contents_from_anthropic([delta_content], request_state, streaming=True) + assert len(result) == 1 + assert result[0].type == "function_call" + assert result[0].call_id == "srvtool_abc" + assert result[0].name == "" + assert result[0].arguments == partial_json + assert result[0].informational_only is True + assert result[0].hosted is True + call += result[0] + + assert call.name == "web_search" + assert call.informational_only is True + assert call.hosted is True + assert call.parse_arguments() == {"query": "latest news"} + + +def test_parse_contents_mcp_and_code_execution_input_json_delta_ignored( + mock_anthropic_client: MagicMock, +) -> None: + """input_json_delta events after MCP and code execution tool blocks produce no content.""" + client = create_test_anthropic_client(mock_anthropic_client) - # input_json_delta events after server_tool_use must be silently ignored - delta_content = MagicMock() - delta_content.type = "input_json_delta" - delta_content.partial_json = '{"query": "latest news"}' + for block_type, name in (("mcp_tool_use", "search_docs"), ("server_tool_use", "code_execution")): + request_state = _AnthropicRequestState() + tool_content = MagicMock() + tool_content.type = block_type + tool_content.id = "call_abc" + tool_content.name = name + tool_content.server_name = "docs" + tool_content.input = {} + assert client._parse_contents_from_anthropic([tool_content], request_state, streaming=True) - result = client._parse_contents_from_anthropic([delta_content], request_state) - assert result == [], "input_json_delta after server_tool_use should produce no content, but got: %r" % result + delta_content = MagicMock() + delta_content.type = "input_json_delta" + delta_content.partial_json = '{"query": "latest news"}' - # A second delta must also be ignored - delta_content_2 = MagicMock() - delta_content_2.type = "input_json_delta" - delta_content_2.partial_json = '{"extra": true}' - - result = client._parse_contents_from_anthropic([delta_content_2], request_state) - assert result == [], ( - "subsequent input_json_delta after server_tool_use should also be ignored, but got: %r" % result - ) + result = client._parse_contents_from_anthropic([delta_content], request_state, streaming=True) + assert result == [], f"input_json_delta after {block_type} {name} should produce no content: {result!r}" # Stream Processing Tests @@ -2279,6 +2461,67 @@ async def create(**kwargs: Any) -> Any: assert response.text == "done" +async def test_streaming_parallel_server_tool_use_replays_streamed_input( + mock_anthropic_client: MagicMock, +) -> None: + """A streamed pending server tool call is replayed with the input that arrived through argument deltas.""" + executions = 0 + + @tool(name="local_value", approval_mode="never_require") + def local_value() -> str: + nonlocal executions + executions += 1 + return "local-ok" + + async def parallel_tool_events() -> Any: + yield _message_start_event("tool-response") + yield _tool_start_event(block_type="tool_use", call_id="toolu_local", name="local_value", index=0) + yield _content_block_stop_event(index=0) + yield _tool_start_event(block_type="server_tool_use", call_id="srvtoolu_search", name="web_search", index=1) + yield _argument_delta_event('{"query": "agent framework') + yield _argument_delta_event(' release"}') + yield _content_block_stop_event(index=1) + yield _message_delta_event("tool_use") + yield _message_stop_event() + + async def final_events() -> Any: + yield _message_start_event("final-response") + yield _message_delta_event("end_turn") + yield _message_stop_event() + + requests: list[dict[str, Any]] = [] + + async def create(**kwargs: Any) -> Any: + requests.append(kwargs) + return parallel_tool_events() if len(requests) == 1 else final_events() + + client = create_test_anthropic_client(mock_anthropic_client) + mock_anthropic_client.beta.messages.create.side_effect = create + stream = client.get_response( + [Message(role="user", contents=["Call local_value and search the web."])], + options={"tools": [client.get_web_search_tool(), local_value], "max_tokens": 64}, + stream=True, + ) + _ = [update async for update in stream] + await stream.get_final_response() + + assert executions == 1 + assert len(requests) == 2 + replayed_blocks = [ + block + for message in requests[1]["messages"] + if isinstance(message["content"], list) + for block in message["content"] + if block["type"] in ("tool_use", "server_tool_use", "tool_result") + ] + assert [(block["type"], block.get("id") or block.get("tool_use_id")) for block in replayed_blocks] == [ + ("tool_use", "toolu_local"), + ("server_tool_use", "srvtoolu_search"), + ("tool_result", "toolu_local"), + ] + assert replayed_blocks[1]["input"] == {"query": "agent framework release"} + + @pytest.mark.parametrize( ("close_block", "stop_reason", "include_message_stop"), [ @@ -2500,12 +2743,16 @@ async def consume_primary() -> tuple[list[ChatResponseUpdate], ChatResponse]: assert primary_calls[0].call_id == "shared-call" assert primary_calls[0].parse_arguments() == {"command": "pwd"} - hosted_calls = [content for content in hosted_contents if content.type == "function_call"] - assert len(hosted_calls) == 1 - assert hosted_calls[0].name == "web_search" - assert hosted_calls[0].call_id == "shared-call" - assert hosted_calls[0].arguments == {} - assert hosted_calls[0].informational_only is True + assert not any(content.hosted or content.informational_only for content in primary_fragments) + + hosted_fragments = [content for content in hosted_contents if content.type == "function_call"] + assert [ + (content.name, content.call_id, content.arguments, content.hosted, content.informational_only) + for content in hosted_fragments + ] == [ + ("web_search", "shared-call", {}, True, True), + ("", "shared-call", '{"query":"framework"}', True, True), + ] async def test_concurrent_streams_keep_approval_requests_request_local(