diff --git a/python/packages/anthropic/agent_framework_anthropic/_chat_client.py b/python/packages/anthropic/agent_framework_anthropic/_chat_client.py index cefc50b122..aeb4822411 100644 --- a/python/packages/anthropic/agent_framework_anthropic/_chat_client.py +++ b/python/packages/anthropic/agent_framework_anthropic/_chat_client.py @@ -5,6 +5,7 @@ import logging import sys from collections.abc import AsyncIterable, Awaitable, Callable, Mapping, Sequence +from dataclasses import dataclass, field from typing import Any, ClassVar, Final, Generic, Literal, TypedDict, cast from agent_framework import ( @@ -255,6 +256,15 @@ def _map_finish_reason(stop_reason: str | None) -> FinishReason | None: return FinishReason(FINISH_REASON_MAP.get(stop_reason, stop_reason)) +@dataclass +class _AnthropicRequestState: + """Mutable parsing state scoped to one Anthropic request.""" + + active_call_id: str | None = None + active_call_content_type: str | None = None + tool_name_aliases: dict[str, str] = field(default_factory=dict[str, str]) + + class AnthropicSettings(TypedDict, total=False): """Anthropic Project settings. @@ -408,10 +418,6 @@ class MyOptions(AnthropicChatOptions, total=False): self.anthropic_client = anthropic_client self.additional_beta_flags = additional_beta_flags or [] self.model = model_setting - # streaming requires tracking the last function call ID, name, and content type - self._last_call_id_name: tuple[str, str] | None = None - self._last_call_content_type: str | None = None - self._tool_name_aliases: dict[str, str] = {} # region Static factory methods for hosted tools @@ -583,7 +589,8 @@ def _inner_get_response( **kwargs: Any, ) -> Awaitable[ChatResponse] | ResponseStream[ChatResponseUpdate, ChatResponse]: # prepare - run_options = self._prepare_options(messages, options, **kwargs) + request_state = _AnthropicRequestState() + run_options = self._prepare_options(messages, options, request_state, **kwargs) if stream: # Streaming mode @@ -595,7 +602,7 @@ async def _stream() -> AsyncIterable[ChatResponseUpdate]: mark_feature_used(FeatureIndex.ANTHROPIC) try: async for chunk in await self.anthropic_client.beta.messages.create(**run_options, stream=True): - parsed_chunk = self._process_stream_event(chunk, emitted_usage) + parsed_chunk = self._process_stream_event(chunk, emitted_usage, request_state=request_state) if parsed_chunk: yield parsed_chunk except AgentFrameworkException: @@ -614,7 +621,7 @@ async def _get_response() -> ChatResponse: raise except Exception as ex: raise _wrap_anthropic_error(ex) from ex - return self._process_message(message, options) + return self._process_message(message, options, request_state=request_state) return _get_response() @@ -624,6 +631,8 @@ def _prepare_options( self, messages: Sequence[Message], options: Mapping[str, Any], + request_state: _AnthropicRequestState | None = None, + /, **kwargs: Any, ) -> dict[str, Any]: """Create run options for the Anthropic client based on messages and options. @@ -631,11 +640,13 @@ def _prepare_options( Args: messages: The list of chat messages. options: The options dict. + request_state: State used to parse the response for this request. kwargs: Additional keyword arguments. Returns: A dictionary of run options for the Anthropic client. """ + request_state = request_state or _AnthropicRequestState() # Start with a copy of options, excluding keys we handle separately run_options: dict[str, Any] = { k: v @@ -697,7 +708,7 @@ def _prepare_options( run_options["metadata"] = metadata # tools, mcp servers and tool choice - if tools_config := self._prepare_tools_for_anthropic(options): + if tools_config := self._prepare_tools_for_anthropic(options, request_state=request_state): run_options.update(tools_config) # response_format - emit Anthropic's GA ``output_config.format`` shape. @@ -1014,7 +1025,11 @@ def _prepare_message_for_anthropic(self, message: Message) -> dict[str, Any]: "content": a_content, } - def _prepare_tools_for_anthropic(self, options: Mapping[str, Any]) -> dict[str, Any] | None: + def _prepare_tools_for_anthropic( + self, + options: Mapping[str, Any], + request_state: _AnthropicRequestState | None = None, + ) -> dict[str, Any] | None: """Prepare tools and tool choice configuration for the Anthropic API request. Converts FunctionTool to Anthropic format. MCP tools are routed to separate @@ -1022,12 +1037,14 @@ def _prepare_tools_for_anthropic(self, options: Mapping[str, Any]) -> dict[str, Args: options: The options dict containing tools and tool choice settings. + request_state: State used to parse the response for this request. Returns: A dictionary with tools, mcp_servers, and tool_choice configuration, or None if empty. """ from agent_framework._types import validate_tool_mode + request_state = request_state or _AnthropicRequestState() result: dict[str, Any] = {} tools = options.get("tools") @@ -1076,9 +1093,9 @@ def _prepare_tools_for_anthropic(self, options: Mapping[str, Any]) -> dict[str, result["tools"] = tool_list if mcp_server_list: result["mcp_servers"] = mcp_server_list - self._tool_name_aliases = tool_name_aliases + request_state.tool_name_aliases = tool_name_aliases else: - self._tool_name_aliases = {} + request_state.tool_name_aliases = {} # Process tool choice if options.get("tool_choice") is None: @@ -1101,7 +1118,7 @@ def _prepare_tools_for_anthropic(self, options: Mapping[str, Any]) -> dict[str, api_tool_name = next( ( api_name - for api_name, local_name in self._tool_name_aliases.items() + for api_name, local_name in request_state.tool_name_aliases.items() if local_name == required_name ), required_name, @@ -1124,12 +1141,18 @@ def _prepare_tools_for_anthropic(self, options: Mapping[str, Any]) -> dict[str, # region Response Processing Methods - def _process_message(self, message: BetaMessage, options: Mapping[str, Any]) -> ChatResponse: + def _process_message( + self, + message: BetaMessage, + options: Mapping[str, Any], + request_state: _AnthropicRequestState | None = None, + ) -> ChatResponse: """Process the response from the Anthropic client. Args: message: The message returned by the Anthropic client. options: The options dict used for the request. + request_state: State used to parse this response. Returns: A ChatResponse object containing the processed response. @@ -1139,7 +1162,7 @@ def _process_message(self, message: BetaMessage, options: Mapping[str, Any]) -> messages=[ Message( role="assistant", - contents=self._parse_contents_from_anthropic(message.content), + contents=self._parse_contents_from_anthropic(message.content, request_state=request_state), raw_representation=message, ) ], @@ -1151,7 +1174,10 @@ def _process_message(self, message: BetaMessage, options: Mapping[str, Any]) -> ) def _process_stream_event( - self, event: BetaRawMessageStreamEvent, emitted_usage: dict[str, int] | None = None + self, + event: BetaRawMessageStreamEvent, + emitted_usage: dict[str, int] | None = None, + request_state: _AnthropicRequestState | None = None, ) -> ChatResponseUpdate | None: """Process a streaming event from the Anthropic client. @@ -1161,6 +1187,7 @@ def _process_stream_event( emitted, used to convert Anthropic's cumulative usage snapshots into increments (see ``_incremental_usage``). Pass ``None`` for a one-off event to keep the snapshot unchanged. + request_state: State used to parse this stream. Returns: A ChatResponseUpdate object containing the processed update. @@ -1177,7 +1204,7 @@ def _process_stream_event( role="assistant", response_id=event.message.id, contents=[ - *self._parse_contents_from_anthropic(event.message.content), + *self._parse_contents_from_anthropic(event.message.content, request_state=request_state), *usage_details, ], model=event.message.model, @@ -1201,13 +1228,13 @@ def _process_stream_event( case "message_stop": logger.debug("Received message_stop event; no content to process.") case "content_block_start": - contents = self._parse_contents_from_anthropic([event.content_block]) + contents = self._parse_contents_from_anthropic([event.content_block], request_state=request_state) return ChatResponseUpdate( contents=contents, raw_representation=event, ) case "content_block_delta": - contents = self._parse_contents_from_anthropic([event.delta]) + contents = self._parse_contents_from_anthropic([event.delta], request_state=request_state) return ChatResponseUpdate( contents=contents, raw_representation=event, @@ -1270,8 +1297,10 @@ def _incremental_usage(cumulative: UsageDetails, emitted: dict[str, int] | None) def _parse_contents_from_anthropic( self, content: Sequence[BetaContentBlock | BetaRawContentBlockDelta | BetaTextBlock], + request_state: _AnthropicRequestState | None = None, ) -> list[Content]: """Parse contents from the Anthropic message.""" + request_state = request_state or _AnthropicRequestState() contents: list[Content] = [] for content_block in content: match content_block.type: @@ -1284,8 +1313,8 @@ def _parse_contents_from_anthropic( ) ) case "tool_use" | "mcp_tool_use" | "server_tool_use": - self._last_call_id_name = (content_block.id, content_block.name) - self._last_call_content_type = content_block.type + request_state.active_call_id = content_block.id + request_state.active_call_content_type = content_block.type if content_block.type == "mcp_tool_use": contents.append( Content.from_mcp_server_tool_call( @@ -1310,7 +1339,7 @@ def _parse_contents_from_anthropic( ) ) else: - resolved_tool_name = self._tool_name_aliases.get(content_block.name, content_block.name) + resolved_tool_name = request_state.tool_name_aliases.get(content_block.name, content_block.name) contents.append( Content.from_function_call( call_id=content_block.id, @@ -1321,11 +1350,10 @@ def _parse_contents_from_anthropic( ) ) case "mcp_tool_result": - call_id, _ = self._last_call_id_name or (None, None) parsed_output: list[Content] | None = None if content_block.content: if isinstance(content_block.content, list): - parsed_output = self._parse_contents_from_anthropic(content_block.content) + parsed_output = self._parse_contents_from_anthropic(content_block.content, request_state) elif isinstance(content_block.content, (str, bytes)): parsed_output = [ Content.from_text( @@ -1334,7 +1362,7 @@ def _parse_contents_from_anthropic( ) ] else: - parsed_output = self._parse_contents_from_anthropic([content_block.content]) + parsed_output = self._parse_contents_from_anthropic([content_block.content], request_state) contents.append( Content.from_mcp_server_tool_result( call_id=content_block.tool_use_id, @@ -1343,7 +1371,6 @@ def _parse_contents_from_anthropic( ) ) case "web_search_tool_result" | "web_fetch_tool_result": - call_id, _ = self._last_call_id_name or (None, None) contents.append( Content.from_function_result( call_id=content_block.tool_use_id, @@ -1548,10 +1575,10 @@ def _parse_contents_from_anthropic( ) case "input_json_delta": # Skip argument deltas for MCP and server tools — execution is handled server-side. - if self._last_call_content_type in ("mcp_tool_use", "server_tool_use"): + if request_state.active_call_content_type in ("mcp_tool_use", "server_tool_use"): pass else: - call_id = self._last_call_id_name[0] if self._last_call_id_name else "" + call_id = request_state.active_call_id or "" contents.append( Content.from_function_call( call_id=call_id, diff --git a/python/packages/anthropic/tests/test_anthropic_client.py b/python/packages/anthropic/tests/test_anthropic_client.py index 133656e197..c671076505 100644 --- a/python/packages/anthropic/tests/test_anthropic_client.py +++ b/python/packages/anthropic/tests/test_anthropic_client.py @@ -1,4 +1,5 @@ # Copyright (c) Microsoft. All rights reserved. +import asyncio import os import re from pathlib import Path @@ -16,8 +17,10 @@ ChatResponseUpdate, Content, FunctionInvocationLayer, + FunctionTool, InlineSkill, Message, + ResponseStream, SkillFrontmatter, SkillsProvider, SupportsChatGetResponse, @@ -41,7 +44,7 @@ from pydantic import BaseModel, Field from agent_framework_anthropic import AnthropicChatOptions, AnthropicClient, RawAnthropicClient -from agent_framework_anthropic._chat_client import AnthropicSettings +from agent_framework_anthropic._chat_client import AnthropicSettings, _AnthropicRequestState from agent_framework_anthropic._feature_usage import FeatureIndex # Test constants @@ -75,8 +78,6 @@ def create_test_anthropic_client( # Set attributes directly client.anthropic_client = mock_anthropic_client client.model = model or anthropic_settings["chat_model"] - client._last_call_id_name = None - client._tool_name_aliases = {} client.additional_properties = {} cast(Any, client).middleware = None client.additional_beta_flags = [] @@ -598,32 +599,26 @@ def test_streaming_replay_preserves_empty_signed_thinking_block( client = create_test_anthropic_client(mock_anthropic_client) events: list[BetaRawMessageStreamEvent] = [ - BetaRawContentBlockStartEvent.model_validate( - { - "type": "content_block_start", - "index": 0, - "content_block": {"type": "thinking", "thinking": "", "signature": ""}, - } - ), - BetaRawContentBlockDeltaEvent.model_validate( - { - "type": "content_block_delta", - "index": 0, - "delta": {"type": "signature_delta", "signature": "synthetic-signature"}, - } - ), - BetaRawContentBlockStartEvent.model_validate( - { - "type": "content_block_start", - "index": 1, - "content_block": { - "type": "tool_use", - "id": "toolu_test", - "name": "lookup", - "input": {}, - }, - } - ), + BetaRawContentBlockStartEvent.model_validate({ + "type": "content_block_start", + "index": 0, + "content_block": {"type": "thinking", "thinking": "", "signature": ""}, + }), + BetaRawContentBlockDeltaEvent.model_validate({ + "type": "content_block_delta", + "index": 0, + "delta": {"type": "signature_delta", "signature": "synthetic-signature"}, + }), + BetaRawContentBlockStartEvent.model_validate({ + "type": "content_block_start", + "index": 1, + "content_block": { + "type": "tool_use", + "id": "toolu_test", + "name": "lookup", + "input": {}, + }, + }), ] updates = [client._process_stream_event(event) for event in events] @@ -1768,15 +1763,16 @@ def test_parse_contents_from_anthropic_input_json_delta_no_duplicate_name( ag-ui from emitting duplicate ToolCallStartEvents. """ client = create_test_anthropic_client(mock_anthropic_client) + request_state = _AnthropicRequestState() - # First, simulate a tool_use event that sets _last_call_id_name + # First, simulate a tool_use event that establishes the active call. tool_use_content = MagicMock() tool_use_content.type = "tool_use" tool_use_content.id = "call_123" tool_use_content.name = "get_weather" tool_use_content.input = {} - result = client._parse_contents_from_anthropic([tool_use_content]) + result = client._parse_contents_from_anthropic([tool_use_content], request_state) assert len(result) == 1 assert result[0].type == "function_call" assert result[0].call_id == "call_123" @@ -1787,7 +1783,7 @@ def test_parse_contents_from_anthropic_input_json_delta_no_duplicate_name( delta_content_1.type = "input_json_delta" delta_content_1.partial_json = '{"location":' - result = client._parse_contents_from_anthropic([delta_content_1]) + result = client._parse_contents_from_anthropic([delta_content_1], request_state) assert len(result) == 1 assert result[0].type == "function_call" assert result[0].call_id == "call_123" @@ -1799,7 +1795,7 @@ def test_parse_contents_from_anthropic_input_json_delta_no_duplicate_name( delta_content_2.type = "input_json_delta" delta_content_2.partial_json = '"San Francisco"}' - result = client._parse_contents_from_anthropic([delta_content_2]) + result = client._parse_contents_from_anthropic([delta_content_2], request_state) assert len(result) == 1 assert result[0].type == "function_call" assert result[0].call_id == "call_123" @@ -1817,27 +1813,28 @@ def test_parse_contents_server_tool_use_input_json_delta_ignored( entries that would cause Anthropic API 400 errors on subsequent turns. """ client = create_test_anthropic_client(mock_anthropic_client) + request_state = _AnthropicRequestState() - # Simulate a server_tool_use event that sets _last_call_content_type + # Simulate a server_tool_use event that establishes the hosted call type. server_tool_content = MagicMock() server_tool_content.type = "server_tool_use" server_tool_content.id = "srvtool_abc" server_tool_content.name = "web_search" server_tool_content.input = {} - result = client._parse_contents_from_anthropic([server_tool_content]) + result = client._parse_contents_from_anthropic([server_tool_content], request_state) # 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 client._last_call_content_type == "server_tool_use" # type: ignore[attr-defined] + assert request_state.active_call_content_type == "server_tool_use" # 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"}' - result = client._parse_contents_from_anthropic([delta_content]) + 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 # A second delta must also be ignored @@ -1845,7 +1842,7 @@ def test_parse_contents_server_tool_use_input_json_delta_ignored( delta_content_2.type = "input_json_delta" delta_content_2.partial_json = '{"extra": true}' - result = client._parse_contents_from_anthropic([delta_content_2]) + 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 ) @@ -2049,6 +2046,300 @@ async def _raise_after_first_event() -> Any: pass +def _message_start_event(response_id: str) -> MagicMock: + event = MagicMock() + event.type = "message_start" + event.message.id = response_id + event.message.role = "assistant" + event.message.model = "claude-test" + event.message.content = [] + event.message.stop_reason = None + event.message.usage = None + return event + + +def _tool_start_event(*, block_type: str, call_id: str, name: str) -> MagicMock: + event = MagicMock() + event.type = "content_block_start" + event.content_block.type = block_type + event.content_block.id = call_id + event.content_block.name = name + event.content_block.input = {} + return event + + +def _argument_delta_event(partial_json: str) -> MagicMock: + event = MagicMock() + event.type = "content_block_delta" + event.delta.type = "input_json_delta" + event.delta.partial_json = partial_json + return event + + +def _message_delta_event(stop_reason: str) -> MagicMock: + event = MagicMock() + event.type = "message_delta" + event.delta.stop_reason = stop_reason + event.usage = None + return event + + +async def test_concurrent_streams_isolate_tool_state_when_peer_fails( + mock_anthropic_client: MagicMock, +) -> None: + """A failing hosted-tool stream must not affect a concurrent local-tool stream.""" + client = create_test_anthropic_client(mock_anthropic_client) + + def run_primary_shell(command: str) -> str: + return command + + def run_peer_shell(command: str) -> str: + return command + + primary_tool = client.get_shell_tool(func=run_primary_shell, approval_mode="never_require") + peer_tool = client.get_shell_tool(func=run_peer_shell, approval_mode="never_require") + hosted_tool = client.get_mcp_tool(name="hosted-search", url="https://example.com/mcp") + primary_started = asyncio.Event() + hosted_started = asyncio.Event() + hosted_failed = asyncio.Event() + + async def primary_events() -> Any: + yield _message_start_event("primary-response") + yield _tool_start_event(block_type="tool_use", call_id="shared-call", name="bash") + primary_started.set() + await asyncio.wait_for(hosted_started.wait(), timeout=1) + yield _argument_delta_event('{"command":') + await asyncio.wait_for(hosted_failed.wait(), timeout=1) + yield _argument_delta_event('"pwd"}') + yield _message_delta_event("tool_use") + + async def hosted_events() -> Any: + await asyncio.wait_for(primary_started.wait(), timeout=1) + yield _message_start_event("hosted-response") + yield _tool_start_event(block_type="server_tool_use", call_id="shared-call", name="web_search") + hosted_started.set() + yield _argument_delta_event('{"query":"framework"}') + hosted_failed.set() + raise RuntimeError("peer stream failed") + + async def create(**kwargs: Any) -> Any: + if kwargs.get("mcp_servers"): + return hosted_events() + return primary_events() + + mock_anthropic_client.beta.messages.create.side_effect = create + primary_stream = client._inner_get_response( # type: ignore[attr-defined] + messages=[Message(role="user", contents=["primary"])], + options={"tools": [primary_tool], "max_tokens": 64}, + stream=True, + ) + hosted_stream = client._inner_get_response( # type: ignore[attr-defined] + messages=[Message(role="user", contents=["hosted"])], + options={"tools": [peer_tool, hosted_tool], "max_tokens": 64}, + stream=True, + ) + assert isinstance(primary_stream, ResponseStream) + assert isinstance(hosted_stream, ResponseStream) + + async def consume_primary() -> tuple[list[ChatResponseUpdate], ChatResponse]: + updates = [update async for update in primary_stream] + return updates, await primary_stream.get_final_response() + + primary_task = asyncio.create_task(consume_primary()) + hosted_contents: list[Content] = [] + with pytest.raises(ChatClientException, match="Anthropic"): + async for update in hosted_stream: + hosted_contents.extend(update.contents) + + primary_updates, primary_response = await asyncio.wait_for(primary_task, timeout=1) + primary_fragments = [ + content for update in primary_updates for content in update.contents if content.type == "function_call" + ] + assert [(content.name, content.call_id, content.arguments) for content in primary_fragments] == [ + (primary_tool.name, "shared-call", {}), + ("", "shared-call", '{"command":'), + ("", "shared-call", '"pwd"}'), + ] + primary_calls = [ + content + for message in primary_response.messages + for content in message.contents + if content.type == "function_call" + ] + assert len(primary_calls) == 1 + assert primary_calls[0].name == primary_tool.name + 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 + + +async def test_concurrent_streams_keep_approval_requests_request_local( + mock_anthropic_client: MagicMock, +) -> None: + """Concurrent approval-required shell calls must retain their request's tool identity.""" + client = create_test_anthropic_client(mock_anthropic_client) + + def first_shell(command: str) -> str: + return command + + def second_shell(command: str) -> str: + return command + + first_tool = client.get_shell_tool(func=first_shell, approval_mode="always_require") + second_tool = client.get_shell_tool(func=second_shell, approval_mode="always_require") + requests_ready = asyncio.Event() + request_count = 0 + + async def approval_events(response_id: str, command: str) -> Any: + yield _message_start_event(response_id) + yield _tool_start_event(block_type="tool_use", call_id="shared-approval-call", name="bash") + yield _argument_delta_event('{"command":') + yield _argument_delta_event(f'"{command}"}}') + yield _message_delta_event("tool_use") + + async def create(**kwargs: Any) -> Any: + nonlocal request_count + prompt = kwargs["messages"][0]["content"][0]["text"] + request_count += 1 + if request_count == 2: + requests_ready.set() + await asyncio.wait_for(requests_ready.wait(), timeout=1) + return approval_events(f"{prompt}-response", prompt) + + mock_anthropic_client.beta.messages.create.side_effect = create + first_stream = client.get_response( + messages=[Message(role="user", contents=["first"])], + options={"tools": [first_tool], "max_tokens": 64}, + stream=True, + ) + second_stream = client.get_response( + messages=[Message(role="user", contents=["second"])], + options={"tools": [second_tool], "max_tokens": 64}, + stream=True, + ) + + async def consume( + stream: ResponseStream[ChatResponseUpdate, ChatResponse], + ) -> tuple[list[ChatResponseUpdate], ChatResponse]: + updates = [update async for update in stream] + return updates, await stream.get_final_response() + + (first_updates, first_response), (second_updates, second_response) = await asyncio.gather( + consume(first_stream), consume(second_stream) + ) + + def assert_request_owned( + updates: list[ChatResponseUpdate], + response: ChatResponse, + expected_tool: FunctionTool, + expected_command: str, + ) -> tuple[Content, Content]: + update_contents = [content for update in updates for content in update.contents] + named_update_calls = [ + content for content in update_contents if content.type == "function_call" and content.name + ] + update_requests = [content for content in update_contents if content.type == "function_approval_request"] + assert named_update_calls + assert {content.name for content in named_update_calls} == {expected_tool.name} + assert len(update_requests) == 1 + + response_contents = [content for message in response.messages for content in message.contents] + response_calls = [content for content in response_contents if content.type == "function_call"] + response_requests = [content for content in response_contents if content.type == "function_approval_request"] + assert len(response_calls) == 1 + assert len(response_requests) == 1 + function_call = response_calls[0] + approval_request = response_requests[0] + assert function_call.name == expected_tool.name + assert function_call.call_id == "shared-approval-call" + assert function_call.parse_arguments() == {"command": expected_command} + assert approval_request.function_call is not None + assert approval_request.function_call.name == expected_tool.name + assert approval_request.function_call.call_id == "shared-approval-call" + assert approval_request.function_call.parse_arguments() == {"command": expected_command} + assert approval_request.id == function_call.id + return function_call, approval_request + + first_call, _ = assert_request_owned(first_updates, first_response, first_tool, "first") + second_call, _ = assert_request_owned(second_updates, second_response, second_tool, "second") + assert first_call.id != second_call.id + + +async def test_concurrent_non_streaming_responses_keep_tool_aliases_request_local( + mock_anthropic_client: MagicMock, +) -> None: + """Concurrent non-streaming responses must retain their request's shell alias.""" + client = create_test_anthropic_client(mock_anthropic_client) + + def first_shell(command: str) -> str: + return command + + def second_shell(command: str) -> str: + return command + + first_tool = client.get_shell_tool(func=first_shell, approval_mode="never_require") + second_tool = client.get_shell_tool(func=second_shell, approval_mode="never_require") + requests_ready = asyncio.Event() + request_count = 0 + + def tool_response(response_id: str, command: str) -> MagicMock: + message = MagicMock(spec=BetaMessage) + message.id = response_id + message.model = "claude-test" + message.content = [ + BetaToolUseBlock( + type="tool_use", + id="shared-non-stream-call", + name="bash", + input={"command": command}, + ) + ] + message.usage = None + message.stop_reason = "tool_use" + return message + + async def create(**kwargs: Any) -> Any: + nonlocal request_count + prompt = kwargs["messages"][0]["content"][0]["text"] + request_count += 1 + if request_count == 2: + requests_ready.set() + await asyncio.wait_for(requests_ready.wait(), timeout=1) + return tool_response(f"{prompt}-response", prompt) + + async def get_response(prompt: str, function_tool: FunctionTool) -> ChatResponse: + response = client._inner_get_response( # type: ignore[attr-defined] + messages=[Message(role="user", contents=[prompt])], + options={"tools": [function_tool], "max_tokens": 64}, + ) + assert not isinstance(response, ResponseStream) + return await response + + mock_anthropic_client.beta.messages.create.side_effect = create + first_response, second_response = await asyncio.gather( + get_response("first", first_tool), + get_response("second", second_tool), + ) + + def assert_response_owned(response: ChatResponse, expected_tool: FunctionTool, command: str) -> None: + calls = [ + content for message in response.messages for content in message.contents if content.type == "function_call" + ] + assert len(calls) == 1 + assert calls[0].name == expected_tool.name + assert calls[0].call_id == "shared-non-stream-call" + assert calls[0].parse_arguments() == {"command": command} + + assert_response_owned(first_response, first_tool, "first") + assert_response_owned(second_response, second_tool, "second") + + def test_process_stream_event_message_start_sets_assistant_role(mock_anthropic_client: MagicMock) -> None: """Test that message_start streaming event sets role='assistant'. @@ -2699,7 +2990,6 @@ def test_parse_contents_mcp_tool_result_list_content( ) -> None: """Test parsing MCP tool result with list content.""" client = create_test_anthropic_client(mock_anthropic_client) - client._last_call_id_name = ("call_123", "test_tool") # Create mock MCP tool result with list content mock_text_block = MagicMock() @@ -2722,7 +3012,6 @@ def test_parse_contents_mcp_tool_result_string_content( ) -> None: """Test parsing MCP tool result with string content.""" client = create_test_anthropic_client(mock_anthropic_client) - client._last_call_id_name = ("call_123", "test_tool") # Create mock MCP tool result with string content mock_block = MagicMock() @@ -2741,7 +3030,6 @@ def test_parse_contents_mcp_tool_result_bytes_content( ) -> None: """Test parsing MCP tool result with bytes content.""" client = create_test_anthropic_client(mock_anthropic_client) - client._last_call_id_name = ("call_123", "test_tool") # Create mock MCP tool result with bytes content mock_block = MagicMock() @@ -2760,7 +3048,6 @@ def test_parse_contents_mcp_tool_result_object_content( ) -> None: """Test parsing MCP tool result with object content.""" client = create_test_anthropic_client(mock_anthropic_client) - client._last_call_id_name = ("call_123", "test_tool") # Create mock MCP tool result with object content mock_content_obj = MagicMock() @@ -2783,7 +3070,6 @@ def test_parse_contents_web_search_tool_result( ) -> None: """Test parsing web search tool result.""" client = create_test_anthropic_client(mock_anthropic_client) - client._last_call_id_name = ("call_789", "web_search") # Create mock web search tool result mock_block = MagicMock() @@ -2800,7 +3086,6 @@ def test_parse_contents_web_search_tool_result( def test_parse_contents_web_fetch_tool_result(mock_anthropic_client: MagicMock) -> None: """Test parsing web fetch tool result.""" client = create_test_anthropic_client(mock_anthropic_client) - client._last_call_id_name = ("call_101", "web_fetch") # Create mock web fetch tool result mock_block = MagicMock() @@ -3128,7 +3413,6 @@ def test_parse_code_execution_result_with_error( ) -> None: """Test parsing code execution result with error.""" client = create_test_anthropic_client(mock_anthropic_client) - client._last_call_id_name = ("call_code1", "code_execution_tool") # Create mock code execution result with error from anthropic.types.beta.beta_code_execution_tool_result_error import ( @@ -3153,7 +3437,6 @@ def test_parse_code_execution_result_with_stdout( ) -> None: """Test parsing code execution result with stdout.""" client = create_test_anthropic_client(mock_anthropic_client) - client._last_call_id_name = ("call_code2", "code_execution_tool") # Create mock code execution result with stdout mock_content = MagicMock() @@ -3177,7 +3460,6 @@ def test_parse_code_execution_result_with_stderr( ) -> None: """Test parsing code execution result with stderr.""" client = create_test_anthropic_client(mock_anthropic_client) - client._last_call_id_name = ("call_code3", "code_execution_tool") # Create mock code execution result with stderr mock_content = MagicMock() @@ -3201,7 +3483,6 @@ def test_parse_code_execution_result_with_files( ) -> None: """Test parsing code execution result with file outputs.""" client = create_test_anthropic_client(mock_anthropic_client) - client._last_call_id_name = ("call_code4", "code_execution_tool") # Create mock file output mock_file = MagicMock() @@ -3232,7 +3513,6 @@ def test_parse_bash_execution_result_with_stdout( ) -> None: """Test parsing bash execution result with stdout.""" client = create_test_anthropic_client(mock_anthropic_client) - client._last_call_id_name = ("call_bash2", "bash_code_execution") # Create mock bash execution result with stdout mock_content = MagicMock() @@ -3264,7 +3544,6 @@ def test_parse_bash_execution_result_with_stderr( ) -> None: """Test parsing bash execution result with stderr.""" client = create_test_anthropic_client(mock_anthropic_client) - client._last_call_id_name = ("call_bash3", "bash_code_execution") # Create mock bash execution result with stderr mock_content = MagicMock() @@ -3298,7 +3577,6 @@ def test_parse_bash_execution_result_with_error( ) client = create_test_anthropic_client(mock_anthropic_client) - client._last_call_id_name = ("call_bash_err", "bash_code_execution") mock_error = MagicMock(spec=BetaBashCodeExecutionToolResultError) mock_error.error_code = "execution_time_exceeded" @@ -3324,7 +3602,6 @@ def test_parse_bash_execution_result_with_error( def test_parse_text_editor_result_error(mock_anthropic_client: MagicMock) -> None: """Test parsing text editor result with error.""" client = create_test_anthropic_client(mock_anthropic_client) - client._last_call_id_name = ("call_editor1", "text_editor_code_execution") # Create mock text editor result with error mock_content = MagicMock() @@ -3345,7 +3622,6 @@ def test_parse_text_editor_result_error(mock_anthropic_client: MagicMock) -> Non def test_parse_text_editor_result_view(mock_anthropic_client: MagicMock) -> None: """Test parsing text editor view result.""" client = create_test_anthropic_client(mock_anthropic_client) - client._last_call_id_name = ("call_editor2", "text_editor_code_execution") # Create mock text editor view result mock_content = MagicMock() @@ -3368,7 +3644,6 @@ def test_parse_text_editor_result_view(mock_anthropic_client: MagicMock) -> None def test_parse_text_editor_result_str_replace(mock_anthropic_client: MagicMock) -> None: """Test parsing text editor string replace result.""" client = create_test_anthropic_client(mock_anthropic_client) - client._last_call_id_name = ("call_editor3", "text_editor_code_execution") # Create mock text editor str_replace result mock_content = MagicMock() @@ -3393,7 +3668,6 @@ def test_parse_text_editor_result_str_replace(mock_anthropic_client: MagicMock) def test_parse_text_editor_result_file_create(mock_anthropic_client: MagicMock) -> None: """Test parsing text editor file create result.""" client = create_test_anthropic_client(mock_anthropic_client) - client._last_call_id_name = ("call_editor4", "text_editor_code_execution") # Create mock text editor create result mock_content = MagicMock()