diff --git a/docs/specs/004-python-function-calling-loop.md b/docs/specs/004-python-function-calling-loop.md index b7ad58ceb90..b35a4fe23d9 100644 --- a/docs/specs/004-python-function-calling-loop.md +++ b/docs/specs/004-python-function-calling-loop.md @@ -616,6 +616,7 @@ that manually replay messages own the equivalent rule: do not resend an approval | Declaration-only mixed batch | Known local siblings are never converted into Host pauses. With an authoritative session they remain deferred until every Host pause occurrence is answered, survive session serialization with a process-portable logical-run budget (including a reclassified approval continuation), reclassify against the current tool policy, execute exactly once using the configured concurrent or sequential invocation mode, replay results in original model order, and remain inert on replay; without one they execute on the same turn under the same configured invocation mode. Streaming updates and finalization preserve one Host pause and one result per local sibling. | `test_mixed_declaration_only_batch_defers_siblings_until_all_host_results`, `test_mixed_declaration_only_deferred_call_keeps_ordinary_middleware_semantics`, `test_mixed_declaration_only_deferred_call_reclassifies_changed_approval_policy`, `test_mixed_declaration_only_budget_survives_reclassification_to_approval`, `test_mixed_declaration_only_deferred_call_does_not_execute_removed_tool`, `test_mixed_declaration_only_budget_is_process_portable_without_approval_middleware`, `test_sessionless_mixed_declaration_only_batch_executes_siblings_sequentially`, `test_sessionless_mixed_declaration_only_batch_executes_siblings_concurrently`, `test_streaming_sessionless_mixed_declaration_only_batch_returns_pause_and_result` | | Function invocation disabled | The client bypasses the invocation loop without losing invocation kwargs. | `test_function_invocation_config_enabled_false`, `test_function_invocation_config_enabled_false_preserves_invocation_kwargs`, `test_streaming_function_invocation_config_enabled_false` | | Runtime tool changes | Added tools become available on the next iteration and retain approval behavior. | `test_add_tools_available_next_iteration`, `test_add_tools_with_approval_required_tool` | +| Anthropic pause-turn continuation | `pause_turn` replays the returned assistant content with the same tools until Anthropic completes the server-side turn in streaming and non-streaming modes when every native content block can be reconstructed faithfully. Intermediate messages and usage remain caller-visible and retain provider message boundaries. Unsupported parsed content, native blocks dropped during parsing (including redacted reasoning), streamed MCP calls whose argument deltas are not retained, and bounded-continuation exhaustion keep `finish_reason="pause_turn"` instead of sending a lossy transcript or reporting normal completion. | `packages/anthropic/tests/test_anthropic_client.py::test_agent_continues_anthropic_pause_turn`, `test_streaming_hosted_tool_pause_turn_continues`, `test_streaming_pause_turn_preserves_message_boundaries_for_structured_output`, `test_non_streaming_code_execution_pause_turn_is_not_auto_continued`, `test_streaming_code_execution_pause_turn_is_not_auto_continued`, `test_streaming_mcp_pause_turn_is_not_auto_continued`, `test_non_streaming_redacted_thinking_pause_turn_is_not_auto_continued`, `test_streaming_redacted_thinking_pause_turn_is_not_auto_continued`, `test_anthropic_pause_turn_limit_preserves_finish_reason` | | Direct delegated session propagation | A custom loop must explicitly identify parent provider-owned state before sharing a non-empty session; missing ownership fails before the child runs, while declared parent keys remain isolated and application state still propagates. | `packages/core/tests/core/test_agents.py::test_chat_agent_as_tool_direct_propagation_requires_parent_provider_ownership`, `test_chat_agent_as_tool_direct_propagation_uses_explicit_parent_provider_ownership` | | In-run compaction summaries | Summaries inserted for tool groups excluded by in-run compaction are returned in the final non-streaming response transcript before the group they replace, so history loaded with `skip_excluded` keeps the summarized content; summaries of caller-owned input messages are not added. | `packages/core/tests/core/test_clients.py::test_function_loop_returns_compaction_summaries_in_final_response`, `test_function_loop_returns_compaction_summaries_when_iteration_budget_exhausted`, `test_function_loop_reconciles_nested_compaction_summaries`, `test_function_loop_returns_compacted_transcript_on_early_terminal_exit`, `packages/core/tests/core/test_agents.py::test_agent_run_returns_and_persists_compaction_summaries` | | Per-service-call provider context | When framework-managed history is loaded around each model call, tools and instructions contributed by `HistoryProvider.before_run` remain available to the model and function loop without duplication in streaming or non-streaming execution. | `packages/core/tests/core/test_agents.py::test_vector_history_search_tool_is_available_with_per_service_call_persistence` | diff --git a/python/packages/anthropic/README.md b/python/packages/anthropic/README.md index f3288c38688..ad73de6650d 100644 --- a/python/packages/anthropic/README.md +++ b/python/packages/anthropic/README.md @@ -44,3 +44,19 @@ response = await client.get_response("Hello", options={"instructions": system_bl Instructions contributed later in a run — by a context provider such as `SkillsProvider`, or by per-run `options` — are appended as an additional text block after the configured blocks. The blocks you supply keep their structure and their position, so a `cache_control` breakpoint stays valid. + +### Persisted server-tool history + +Anthropic server-tool calls and results use `Content.additional_properties["anthropic"]["server_tool_block"]` +internally to preserve the native block type, structured result content, and optional caller metadata across +message serialization. Result metadata can contain provider-encrypted fields such as `encrypted_content`; the +adapter stores those plain serialized values unchanged because Anthropic requires them for replay. Applications +should treat this namespace as adapter-owned replay data and must not modify it. + +### Paused server-tool turns + +When Anthropic returns `pause_turn`, the client automatically replays the paused assistant content with the same +tools when every content block can be reconstructed faithfully. It performs at most 10 continuation requests. +If the paused response contains unsupported native tool content, includes a native block dropped during parsing +(such as redacted reasoning), contains a streamed MCP call whose argument deltas are not retained, or remains paused +after that limit, the response retains `finish_reason="pause_turn"` so the caller can decide whether to continue later. diff --git a/python/packages/anthropic/agent_framework_anthropic/_chat_client.py b/python/packages/anthropic/agent_framework_anthropic/_chat_client.py index dd894913609..82fc86206a1 100644 --- a/python/packages/anthropic/agent_framework_anthropic/_chat_client.py +++ b/python/packages/anthropic/agent_framework_anthropic/_chat_client.py @@ -28,6 +28,7 @@ TextSpanRegion, UsageDetails, add_usage_details, + get_namespaced_value, tool, ) from agent_framework._settings import SecretString, load_settings @@ -39,6 +40,7 @@ ChatClientException, ChatClientInvalidAuthException, ChatClientInvalidRequestException, + ContentError, ResponseInvalidatedException, ) from agent_framework.observability import ChatTelemetryLayer @@ -97,6 +99,18 @@ ANTHROPIC_DEFAULT_MAX_TOKENS: Final[int] = 1024 BETA_FLAGS: Final[list[str]] = ["mcp-client-2025-04-04"] UNSUPPORTED_SAMPLING_OPTIONS: Final[tuple[str, ...]] = ("temperature", "top_p", "top_k") +_ANTHROPIC_ADDITIONAL_PROPS_NAMESPACE: Final[str] = "anthropic" +_SERVER_TOOL_BLOCK_KEY: Final[str] = "server_tool_block" +_SERVER_TOOL_USE_ID_PREFIX: Final[str] = "srvtoolu_" +_SERVER_TOOL_RESULT_TYPES: Final[frozenset[str]] = frozenset({ + "web_fetch_tool_result", + "web_search_tool_result", +}) +_MAX_PAUSE_TURN_CONTINUATIONS: Final[int] = 10 +_PAUSE_TURN_REPLAYABLE_CONTENT_TYPES: Final[frozenset[str]] = frozenset({ + "text", + "text_reasoning", +}) ResponseModelT = TypeVar("ResponseModelT", bound=BaseModel | None, default=None) AnthropicAsyncClient = AsyncAnthropic | AsyncAnthropicBedrock | AsyncAnthropicFoundry | AsyncAnthropicVertex @@ -119,6 +133,30 @@ def _wrap_anthropic_error(ex: Exception) -> ChatClientException: return ChatClientException(f"Anthropic chat request failed: {ex}", inner_exception=ex) +def _serialize_anthropic_payload(value: Any) -> Any: + """Convert Anthropic SDK payloads into plain persisted data.""" + if isinstance(value, BaseModel): + return value.model_dump(mode="json", exclude_none=True) + if isinstance(value, Mapping): + mapping = cast(Mapping[Any, Any], value) + return {str(key): _serialize_anthropic_payload(item) for key, item in mapping.items()} + if isinstance(value, Sequence) and not isinstance(value, (str, bytes, bytearray)): + sequence = cast(Sequence[Any], value) + return [_serialize_anthropic_payload(item) for item in sequence] + return value + + +def _server_tool_block_metadata(content_block: Any, *, include_content: bool = False) -> dict[str, Any]: + """Build persisted metadata needed to reconstruct an Anthropic server-tool block.""" + block: dict[str, Any] = {"type": content_block.type} + if include_content: + block["content"] = _serialize_anthropic_payload(content_block.content) + caller = getattr(content_block, "caller", None) + if isinstance(caller, (BaseModel, Mapping)): + block["caller"] = _serialize_anthropic_payload(caller) + return {_ANTHROPIC_ADDITIONAL_PROPS_NAMESPACE: {_SERVER_TOOL_BLOCK_KEY: block}} + + # region Anthropic Chat Options TypedDict @@ -262,7 +300,6 @@ def _drop_unsupported_sampling_options(options: dict[str, Any]) -> None: "tool_use": "tool_calls", "end_turn": "stop", "refusal": "content_filter", - "pause_turn": "stop", } @@ -284,6 +321,9 @@ class _AnthropicRequestState: active_call_id: str | None = None active_call_content_type: str | None = None + active_call_is_function_call: bool = False + dropped_content_types: set[str] = field(default_factory=set[str]) + stream_stop_reason: str | None = None tool_name_aliases: dict[str, str] = field(default_factory=dict[str, str]) @@ -616,19 +656,24 @@ def _inner_get_response( if stream: # Streaming mode - async def _stream() -> AsyncIterable[ChatResponseUpdate]: + + async def _stream_turn( + turn_request_state: _AnthropicRequestState, + ) -> AsyncIterable[ChatResponseUpdate]: # Anthropic streams cumulative usage snapshots (message_start seeds it, # each message_delta carries the running total), so thread a per-stream # accumulator to _process_stream_event to emit increments instead. emitted_usage: dict[str, int] = {} local_function_call_started = False open_local_function_call_blocks: set[int] = set() - latest_stop_reason: str | None = None message_stop_received = False - 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, request_state=request_state) + parsed_chunk = self._process_stream_event( + chunk, + emitted_usage, + request_state=turn_request_state, + ) if chunk.type == "content_block_start" and parsed_chunk is not None: if any( content.type == "function_call" and not content.informational_only @@ -639,11 +684,11 @@ async def _stream() -> AsyncIterable[ChatResponseUpdate]: elif chunk.type == "content_block_stop": open_local_function_call_blocks.discard(chunk.index) elif chunk.type == "message_delta" and chunk.delta.stop_reason is not None: - latest_stop_reason = chunk.delta.stop_reason + turn_request_state.stream_stop_reason = chunk.delta.stop_reason elif chunk.type == "message_stop": message_stop_received = True if local_function_call_started and ( - open_local_function_call_blocks or latest_stop_reason != "tool_use" + open_local_function_call_blocks or turn_request_state.stream_stop_reason != "tool_use" ): raise ResponseInvalidatedException( "Anthropic invalidated partial response output; " @@ -671,23 +716,125 @@ async def _stream() -> AsyncIterable[ChatResponseUpdate]: raise raise _wrap_anthropic_error(ex) from ex + async def _stream() -> AsyncIterable[ChatResponseUpdate]: + mark_feature_used(FeatureIndex.ANTHROPIC) + for continuation_index in range(_MAX_PAUSE_TURN_CONTINUATIONS + 1): + turn_request_state = ( + request_state + if continuation_index == 0 + else _AnthropicRequestState(tool_name_aliases=dict(request_state.tool_name_aliases)) + ) + turn_updates: list[ChatResponseUpdate] = [] + async for update in _stream_turn(turn_request_state): + turn_updates.append(update) + yield update + turn_response = self._finalize_response_updates( + turn_updates, + response_format=options.get("response_format"), + ) + if ( + turn_response.finish_reason != "pause_turn" + or continuation_index == _MAX_PAUSE_TURN_CONTINUATIONS + ): + return + if not self._append_pause_turn_messages_for_anthropic( + run_options, + turn_response.messages, + turn_request_state, + ): + return + return self._build_response_stream(_stream(), response_format=options.get("response_format")) # Non-streaming mode async def _get_response() -> ChatResponse: mark_feature_used(FeatureIndex.ANTHROPIC) - try: - message = await self.anthropic_client.beta.messages.create(**run_options, stream=False) - except AgentFrameworkException: - raise - except Exception as ex: - raise _wrap_anthropic_error(ex) from ex - return self._process_message(message, options, request_state=request_state) + accumulated_messages: list[Message] = [] + aggregated_usage: UsageDetails | None = None + raw_messages: list[BetaMessage] = [] + for continuation_index in range(_MAX_PAUSE_TURN_CONTINUATIONS + 1): + turn_request_state = ( + request_state + if continuation_index == 0 + else _AnthropicRequestState(tool_name_aliases=dict(request_state.tool_name_aliases)) + ) + try: + message = await self.anthropic_client.beta.messages.create(**run_options, stream=False) + except AgentFrameworkException: + raise + except Exception as ex: + raise _wrap_anthropic_error(ex) from ex + response = self._process_message(message, options, request_state=turn_request_state) + accumulated_messages.extend(response.messages) + aggregated_usage = add_usage_details(aggregated_usage, response.usage_details) + raw_messages.append(message) + should_continue = ( + message.stop_reason == "pause_turn" + and continuation_index < _MAX_PAUSE_TURN_CONTINUATIONS + and self._append_pause_turn_messages_for_anthropic( + run_options, + response.messages, + turn_request_state, + ) + ) + if should_continue: + continue + response.messages = accumulated_messages + response.usage_details = aggregated_usage + response.raw_representation = raw_messages if len(raw_messages) > 1 else message + return response + raise RuntimeError("Anthropic pause-turn continuation loop exited unexpectedly.") return _get_response() # region Prep methods + def _append_pause_turn_messages_for_anthropic( + self, + run_options: dict[str, Any], + messages: Sequence[Message], + request_state: _AnthropicRequestState, + ) -> bool: + """Append faithfully replayable paused assistant content without a synthetic user turn.""" + unsupported_types = set(request_state.dropped_content_types) + unsupported_types.update( + content.type + for message in messages + for content in message.contents + if not self._is_pause_turn_content_replayable(content) + ) + if unsupported_types: + # Sorting is only for deterministic diagnostics; replay behavior does not depend on type order. + sorted_unsupported_types = sorted(unsupported_types) + logger.warning( + "Anthropic pause_turn content cannot be replayed faithfully; returning the paused response. " + "Unsupported content types: %s", + ", ".join(sorted_unsupported_types), + ) + return False + + prepared_messages = cast(list[dict[str, Any]], run_options["messages"]) + for message in messages: + prepared_messages.extend(self._prepare_message_groups_for_anthropic(message)) + return True + + def _is_pause_turn_content_replayable(self, content: Content) -> bool: + """Return whether the Anthropic serializer preserves this paused content item.""" + if content.type in _PAUSE_TURN_REPLAYABLE_CONTENT_TYPES: + return True + if content.type == "function_call": + try: + return self._prepare_server_tool_call_for_anthropic(content) is not None + except ContentError: + return False + if content.type == "function_result": + try: + block = self._server_tool_block(content) + except ContentError: + return False + return bool(block is not None and block.get("type") in _SERVER_TOOL_RESULT_TYPES and "content" in block) + return False + def _prepare_options( self, messages: Sequence[Message], @@ -941,7 +1088,11 @@ def _prepare_message_groups_for_anthropic(self, message: Message) -> list[dict[s def _role_for_anthropic_content_block(self, content_block: Mapping[str, Any], default_role: str) -> str: """Return the Anthropic message role required for a content block.""" match content_block: - case {"type": "tool_use" | "mcp_tool_use" | "server_tool_use"}: + case { + "type": ( + "tool_use" | "mcp_tool_use" | "server_tool_use" | "web_fetch_tool_result" | "web_search_tool_result" + ) + }: return "assistant" case {"type": "tool_result" | "mcp_tool_result"}: return "user" @@ -964,6 +1115,72 @@ def _message_has_tool_use(self, message: dict[str, Any]) -> bool: return False + @staticmethod + def _server_tool_block(content: Content) -> Mapping[str, Any] | None: + """Return validated persisted Anthropic server-tool metadata.""" + block = get_namespaced_value( + content.additional_properties, + _ANTHROPIC_ADDITIONAL_PROPS_NAMESPACE, + _SERVER_TOOL_BLOCK_KEY, + ) + if block is None: + return None + if not isinstance(block, Mapping): + raise ContentError("Anthropic server-tool replay metadata must be a mapping.") + return cast(Mapping[str, Any], block) + + def _prepare_server_tool_call_for_anthropic(self, content: Content) -> dict[str, Any] | None: + """Reconstruct an Anthropic server_tool_use block when this is a hosted call.""" + block = self._server_tool_block(content) + is_server_tool = bool(content.call_id and content.call_id.startswith(_SERVER_TOOL_USE_ID_PREFIX)) + if block is not None: + block_type = block.get("type") + if block_type != "server_tool_use": + raise ContentError(f"Unsupported Anthropic server-tool call metadata type: {block_type!r}.") + is_server_tool = True + if not is_server_tool: + return None + + result: dict[str, Any] = { + "type": "server_tool_use", + "id": content.call_id, + "name": content.name, + "input": content.parse_arguments(), + } + if block is not None and (caller := block.get("caller")) is not None: + if not isinstance(caller, Mapping): + raise ContentError("Anthropic server-tool caller metadata must be a mapping.") + result["caller"] = dict(cast(Mapping[str, Any], caller)) + return result + + def _prepare_server_tool_result_for_anthropic(self, content: Content) -> dict[str, Any] | None: + """Reconstruct an Anthropic server-tool result block from persisted metadata.""" + block = self._server_tool_block(content) + if block is None: + if content.call_id and content.call_id.startswith(_SERVER_TOOL_USE_ID_PREFIX): + raise ContentError( + f"Anthropic server-tool result {content.call_id!r} cannot be replayed because its provider " + "metadata is missing." + ) + return None + + block_type = block.get("type") + if block_type not in _SERVER_TOOL_RESULT_TYPES: + raise ContentError(f"Unsupported Anthropic server-tool result metadata type: {block_type!r}.") + if "content" not in block: + raise ContentError("Anthropic server-tool result metadata is missing content.") + + result: dict[str, Any] = { + "type": block_type, + "tool_use_id": content.call_id, + "content": block["content"], + } + if (caller := block.get("caller")) is not None: + if not isinstance(caller, Mapping): + raise ContentError("Anthropic server-tool caller metadata must be a mapping.") + result["caller"] = dict(cast(Mapping[str, Any], caller)) + return result + def _prepare_message_for_anthropic(self, message: Message) -> dict[str, Any]: """Prepare a Message for the Anthropic client. @@ -1001,14 +1218,19 @@ 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": - a_content.append({ - "type": "tool_use", - "id": content.call_id, - "name": content.name, - "input": content.parse_arguments(), - }) + if server_tool_call := self._prepare_server_tool_call_for_anthropic(content): + a_content.append(server_tool_call) + else: + a_content.append({ + "type": "tool_use", + "id": content.call_id, + "name": content.name, + "input": content.parse_arguments(), + }) case "function_result": - if content.items: + if server_tool_result := self._prepare_server_tool_result_for_anthropic(content): + a_content.append(server_tool_result) + elif content.items: tool_content: list[dict[str, Any]] = [] for item in content.items: if item.type == "text": @@ -1239,6 +1461,7 @@ def _process_message( Message( role="assistant", contents=contents, + message_id=message.id, raw_representation=message, ) ], @@ -1282,6 +1505,7 @@ def _process_stream_event( return ChatResponseUpdate( role="assistant", response_id=event.message.id, + message_id=event.message.id, contents=[ *self._parse_contents_from_anthropic( event.message.content, @@ -1415,6 +1639,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_function_call = False if content_block.type == "mcp_tool_use": contents.append( Content.from_mcp_server_tool_call( @@ -1440,6 +1665,7 @@ def _parse_contents_from_anthropic( ) ) else: + request_state.active_call_is_function_call = True resolved_tool_name = request_state.tool_name_aliases.get(content_block.name, content_block.name) contents.append( Content.from_function_call( @@ -1449,6 +1675,11 @@ def _parse_contents_from_anthropic( informational_only=content_block.type == "server_tool_use", hosted=content_block.type == "server_tool_use", merge_mode="append" if streaming else None, + additional_properties=( + _server_tool_block_metadata(content_block) + if content_block.type == "server_tool_use" + else None + ), raw_representation=content_block, ) ) @@ -1486,6 +1717,7 @@ def _parse_contents_from_anthropic( Content.from_function_result( call_id=content_block.tool_use_id, result=content_block.content, + additional_properties=_server_tool_block_metadata(content_block, include_content=True), raw_representation=content_block, ) ) @@ -1686,16 +1918,18 @@ 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"): + if not request_state.active_call_is_function_call: pass else: call_id = request_state.active_call_id or "" + hosted = request_state.active_call_content_type == "server_tool_use" contents.append( Content.from_function_call( call_id=call_id, name="", arguments=content_block.partial_json, + informational_only=hosted, + hosted=hosted, merge_mode="append" if streaming else None, raw_representation=content_block, ) @@ -1719,6 +1953,7 @@ def _parse_contents_from_anthropic( ) ) case _: + request_state.dropped_content_types.add(content_block.type) logger.debug(f"Ignoring unsupported content type: {content_block.type} for now") return contents diff --git a/python/packages/anthropic/tests/test_anthropic_client.py b/python/packages/anthropic/tests/test_anthropic_client.py index 7822da91600..102b0c9c23c 100644 --- a/python/packages/anthropic/tests/test_anthropic_client.py +++ b/python/packages/anthropic/tests/test_anthropic_client.py @@ -17,6 +17,7 @@ ChatResponse, ChatResponseUpdate, Content, + FileHistoryProvider, FunctionInvocationLayer, FunctionTool, InlineSkill, @@ -34,14 +35,19 @@ ChatClientException, ChatClientInvalidAuthException, ChatClientInvalidRequestException, + ContentError, ) from agent_framework.observability import ChatTelemetryLayer from anthropic.types.beta import ( BetaMessage, BetaMessageDeltaUsage, + BetaRedactedThinkingBlock, + BetaServerToolUseBlock, BetaTextBlock, BetaToolUseBlock, BetaUsage, + BetaWebSearchResultBlock, + BetaWebSearchToolResultBlock, ) from pydantic import BaseModel, Field @@ -346,6 +352,106 @@ def test_prepare_message_for_anthropic_function_call( assert result["content"][0]["input"] == {"location": "San Francisco"} +def test_prepare_message_for_anthropic_server_tool_call( + mock_anthropic_client: MagicMock, +) -> None: + """Hosted Anthropic calls must replay as server_tool_use blocks.""" + client = create_test_anthropic_client(mock_anthropic_client) + message = Message( + role="assistant", + contents=[ + Content.from_function_call( + call_id="srvtoolu_123", + name="web_search", + arguments={"query": "Agent Framework"}, + informational_only=True, + hosted=True, + additional_properties={ + "anthropic": { + "server_tool_block": { + "type": "server_tool_use", + "caller": {"type": "direct"}, + } + } + }, + ) + ], + ) + + result = client._prepare_message_for_anthropic(message) + + assert result == { + "role": "assistant", + "content": [ + { + "type": "server_tool_use", + "id": "srvtoolu_123", + "name": "web_search", + "input": {"query": "Agent Framework"}, + "caller": {"type": "direct"}, + } + ], + } + + +def test_prepare_message_for_anthropic_legacy_server_tool_call_uses_id_prefix( + mock_anthropic_client: MagicMock, +) -> None: + """Legacy history can recover server_tool_use from Anthropic's reserved ID prefix.""" + client = create_test_anthropic_client(mock_anthropic_client) + message = Message( + role="assistant", + contents=[ + Content.from_function_call( + call_id="srvtoolu_legacy", + name="web_search", + arguments={"query": "Agent Framework"}, + ) + ], + ) + + result = client._prepare_message_for_anthropic(message) + + assert result["content"] == [ + { + "type": "server_tool_use", + "id": "srvtoolu_legacy", + "name": "web_search", + "input": {"query": "Agent Framework"}, + } + ] + + +def test_prepare_message_for_anthropic_does_not_infer_server_tool_from_hosted( + mock_anthropic_client: MagicMock, +) -> None: + """Provider-neutral hosted calls remain ordinary tool_use without Anthropic provenance.""" + client = create_test_anthropic_client(mock_anthropic_client) + message = Message( + role="assistant", + contents=[ + Content.from_function_call( + call_id="hosted-call", + name="external_hosted_tool", + arguments={"value": "test"}, + informational_only=True, + hosted=True, + ) + ], + ) + + result = client._prepare_message_for_anthropic(message) + + assert result["content"] == [ + { + "type": "tool_use", + "id": "hosted-call", + "name": "external_hosted_tool", + "input": {"value": "test"}, + } + ] + + def test_prepare_message_for_anthropic_function_result( mock_anthropic_client: MagicMock, ) -> None: @@ -376,6 +482,20 @@ def test_prepare_message_for_anthropic_function_result( assert result["content"][0]["is_error"] is False +def test_prepare_message_for_anthropic_rejects_unrecoverable_legacy_server_tool_result( + mock_anthropic_client: MagicMock, +) -> None: + """Legacy server results without structured provider metadata must fail before the request.""" + client = create_test_anthropic_client(mock_anthropic_client) + message = Message( + role="assistant", + contents=[Content.from_function_result(call_id="srvtoolu_legacy", result="lost structured result")], + ) + + with pytest.raises(ContentError, match="provider metadata is missing"): + client._prepare_message_for_anthropic(message) + + def test_prepare_message_for_anthropic_preserves_redacted_error_status( mock_anthropic_client: MagicMock, ) -> None: @@ -878,6 +998,151 @@ def test_prepare_messages_for_anthropic_splits_assistant_embedded_tool_results( assert result[1]["content"][0]["tool_use_id"] == "call_1" +@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 remains server-owned while the local tool result is returned.""" + client = create_test_anthropic_client(mock_anthropic_client) + server_call = MagicMock() + server_call.type = "server_tool_use" + server_call.id = "srvtoolu_search" + server_call.name = "web_search" + server_call.input = {"query": "Agent Framework"} + server_call.caller = None + local_call = MagicMock() + local_call.type = "tool_use" + local_call.id = "toolu_local" + local_call.name = "local_value" + local_call.input = {} + + messages = [ + Message(role="user", contents=["Run both tools"]), + Message(role="assistant", contents=client._parse_contents_from_anthropic([server_call, local_call])), + 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 [block["type"] for block in result[1]["content"]] == ["server_tool_use", "tool_use"] + assert result[1]["content"][0]["id"] == "srvtoolu_search" + assert result[2]["content"][0]["type"] == "tool_result" + assert result[2]["content"][0]["tool_use_id"] == "toolu_local" + + +@pytest.mark.parametrize("reload_history", [False, True], ids=["in_memory", "reloaded"]) +def test_prepare_messages_preserves_reused_server_tool_call_ids( + mock_anthropic_client: MagicMock, + reload_history: bool, +) -> None: + """A completed server call does not change a later occurrence that reuses its call ID.""" + client = create_test_anthropic_client(mock_anthropic_client) + first_call = Content.from_function_call( + call_id="srvtoolu_search", + name="web_search", + arguments={"query": "first"}, + informational_only=True, + hosted=True, + additional_properties={"anthropic": {"server_tool_block": {"type": "server_tool_use"}}}, + ) + first_result = Content.from_function_result( + call_id="srvtoolu_search", + result="first results", + additional_properties={ + "anthropic": { + "server_tool_block": { + "type": "web_search_tool_result", + "content": [ + { + "type": "web_search_result", + "url": "https://example.com/first", + "title": "First", + "encrypted_content": "first-state", + } + ], + } + } + }, + ) + second_call = Content.from_function_call( + call_id="srvtoolu_search", + name="web_search", + arguments={"query": "second"}, + informational_only=True, + hosted=True, + additional_properties={"anthropic": {"server_tool_block": {"type": "server_tool_use"}}}, + ) + messages = [ + Message(role="user", contents=["Search once."]), + Message(role="assistant", contents=[first_call]), + Message(role="assistant", contents=[first_result, Content.from_text("First search done.")]), + Message(role="user", contents=["Search again."]), + Message(role="assistant", contents=[second_call]), + ] + if reload_history: + messages = [Message.from_dict(message.to_dict()) for message in messages] + + result = client._prepare_messages_for_anthropic(messages) + blocks = [ + block + for message in result + if isinstance(message["content"], list) + for block in message["content"] + if block["type"] != "text" + ] + + assert [(block["type"], block.get("id") or block.get("tool_use_id")) for block in blocks] == [ + ("server_tool_use", "srvtoolu_search"), + ("web_search_tool_result", "srvtoolu_search"), + ("server_tool_use", "srvtoolu_search"), + ] + assert blocks[0]["input"] == {"query": "first"} + assert blocks[1]["content"][0]["encrypted_content"] == "first-state" + assert blocks[2]["input"] == {"query": "second"} + + +@pytest.mark.parametrize( + "content", + [ + Content.from_code_interpreter_tool_call(call_id="code-call"), + Content.from_code_interpreter_tool_result(call_id="code-call"), + Content.from_shell_tool_result(call_id="shell-call"), + Content.from_mcp_server_tool_call( + call_id="mcp-call", + tool_name="search_docs", + server_name="docs", + arguments={}, + ), + Content.from_function_result(call_id="srvtoolu_editor", result="editor output"), + Content.from_mcp_server_tool_result(call_id="mcp-call", output=[Content.from_text("mcp output")]), + ], + ids=["code_call", "code_result", "shell_result", "mcp_call", "native_function_result", "mcp_result"], +) +def test_pause_turn_replay_rejects_unsupported_native_tool_content( + mock_anthropic_client: MagicMock, + content: Content, +) -> None: + """Automatic continuation must not silently drop native tool content.""" + client = create_test_anthropic_client(mock_anthropic_client) + run_options: dict[str, Any] = {"messages": []} + + appended = client._append_pause_turn_messages_for_anthropic( + run_options, + [Message(role="assistant", contents=[content])], + _AnthropicRequestState(), + ) + + assert appended is False + assert run_options["messages"] == [] + + # Tool Conversion Tests @@ -1691,6 +1956,7 @@ def test_process_message_basic(mock_anthropic_client: MagicMock) -> None: assert response.model == "claude-3-5-sonnet-20241022" assert len(response.messages) == 1 assert response.messages[0].role == "assistant" + assert response.messages[0].message_id == "msg_123" assert len(response.messages[0].contents) == 1 assert response.messages[0].contents[0].type == "text" assert response.messages[0].contents[0].text == "Hello there!" @@ -1753,7 +2019,7 @@ def test_process_message_with_tool_use(mock_anthropic_client: MagicMock) -> None [ ("end_turn", "stop"), ("stop_sequence", "stop"), - ("pause_turn", "stop"), + ("pause_turn", "pause_turn"), ("max_tokens", "length"), ("tool_use", "tool_calls"), ("refusal", "content_filter"), @@ -1849,6 +2115,22 @@ def test_parse_contents_from_anthropic_text(mock_anthropic_client: MagicMock) -> assert result[0].text == "Hello!" +def test_parse_contents_records_dropped_redacted_thinking( + mock_anthropic_client: MagicMock, +) -> None: + """Dropped native blocks remain visible to the pause-turn replay guard.""" + client = create_test_anthropic_client(mock_anthropic_client) + request_state = _AnthropicRequestState() + + result = client._parse_contents_from_anthropic( + [BetaRedactedThinkingBlock(type="redacted_thinking", data="opaque")], + request_state, + ) + + assert result == [] + assert request_state.dropped_content_types == {"redacted_thinking"} + + def test_parse_contents_from_anthropic_tool_use( mock_anthropic_client: MagicMock, ) -> None: @@ -1921,50 +2203,90 @@ 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_fragment( mock_anthropic_client: MagicMock, ) -> None: - """Regression test: input_json_delta events are ignored after a server_tool_use block. - - 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. - """ + """Server-tool argument deltas remain hosted and survive finalization and persistence.""" client = create_test_anthropic_client(mock_anthropic_client) request_state = _AnthropicRequestState() - # 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.id = "srvtoolu_abc" server_tool_content.name = "web_search" server_tool_content.input = {} - 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 result[0].hosted is True + opening = client._parse_contents_from_anthropic([server_tool_content], request_state, streaming=True) + assert len(opening) == 1 + assert opening[0].type == "function_call" + assert opening[0].informational_only is True + assert opening[0].hosted is True + assert opening[0].additional_properties == {"anthropic": {"server_tool_block": {"type": "server_tool_use"}}} assert request_state.active_call_content_type == "server_tool_use" + assert request_state.active_call_is_function_call is True + + first_delta_content = MagicMock() + first_delta_content.type = "input_json_delta" + first_delta_content.partial_json = '{"query": "latest' + + first_delta = client._parse_contents_from_anthropic([first_delta_content], request_state, streaming=True) + assert len(first_delta) == 1 + assert first_delta[0].call_id == "srvtoolu_abc" + assert first_delta[0].informational_only is True + assert first_delta[0].hosted is True + assert first_delta[0].arguments == '{"query": "latest' + + second_delta_content = MagicMock() + second_delta_content.type = "input_json_delta" + second_delta_content.partial_json = ' news"}' + + second_delta = client._parse_contents_from_anthropic([second_delta_content], request_state, streaming=True) + response = ChatResponse.from_updates([ + ChatResponseUpdate(role="assistant", contents=opening), + ChatResponseUpdate(contents=first_delta), + ChatResponseUpdate(contents=second_delta), + ]) - # 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"}' + assert len(response.messages[0].contents) == 1 + finalized_call = response.messages[0].contents[0] + assert finalized_call.parse_arguments() == {"query": "latest news"} + assert finalized_call.informational_only is True + assert finalized_call.hosted is True + + restored = Message.from_dict(response.messages[0].to_dict()) + prepared = client._prepare_messages_for_anthropic([restored]) + assert prepared[0]["content"] == [ + { + "type": "server_tool_use", + "id": "srvtoolu_abc", + "name": "web_search", + "input": {"query": "latest news"}, + } + ] - 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 - delta_content_2 = MagicMock() - delta_content_2.type = "input_json_delta" - delta_content_2.partial_json = '{"extra": true}' +def test_parse_contents_mcp_and_code_execution_input_json_delta_ignored( + mock_anthropic_client: MagicMock, +) -> None: + """MCP and code-execution argument deltas do not become generic function calls.""" + client = create_test_anthropic_client(mock_anthropic_client) - 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 - ) + 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) + + delta_content = MagicMock() + delta_content.type = "input_json_delta" + delta_content.partial_json = '{"query": "latest news"}' + + 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 @@ -2199,6 +2521,7 @@ def _tool_start_event(*, block_type: str, call_id: str, name: str, index: int = event.content_block.id = call_id event.content_block.name = name event.content_block.input = {} + event.content_block.caller = None return event @@ -2294,9 +2617,12 @@ def test_non_streaming_hosted_tool_pause_turn_is_not_invalidated( response = client._process_message(message, {}) - assert response.finish_reason == "stop" + assert response.finish_reason == "pause_turn" assert response.messages[0].contents[0].type == "function_call" assert response.messages[0].contents[0].informational_only is True + restored = Message.from_dict(response.messages[0].to_dict()) + prepared = client._prepare_messages_for_anthropic([restored]) + assert prepared[0]["content"][0]["type"] == "server_tool_use" async def test_valid_streaming_local_tool_call_executes_and_continues( @@ -2355,6 +2681,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 mixed streamed turn replays the pending server call with its complete input.""" + 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"), [ @@ -2453,24 +2840,46 @@ async def events() -> Any: pass -async def test_streaming_hosted_tool_pause_turn_is_not_invalidated( +async def test_streaming_hosted_tool_pause_turn_continues( mock_anthropic_client: MagicMock, ) -> None: - """Hosted-only calls keep pause_turn behavior even when their name matches a local tool.""" + """Streaming replays a paused hosted call with its finalized input.""" @tool(name="local_tool", approval_mode="never_require") def local_tool(value: str) -> str: return value - async def events() -> Any: + async def paused_events() -> Any: yield _message_start_event("hosted-response") yield _tool_start_event(block_type="server_tool_use", call_id="hosted-call", name="local_tool") + yield _argument_delta_event('{"value":') + yield _argument_delta_event('"a"}') yield _content_block_stop_event() yield _message_delta_event("pause_turn") yield _message_stop_event() + async def final_events() -> Any: + yield _message_start_event("final-response") + text_start = MagicMock() + text_start.type = "content_block_start" + text_start.index = 0 + text_start.content_block.type = "text" + text_start.content_block.text = "done" + text_start.content_block.citations = None + yield text_start + yield _content_block_stop_event() + yield _message_delta_event("end_turn") + yield _message_stop_event() + + requests: list[dict[str, Any]] = [] + streams = [paused_events(), final_events()] + + async def create(**kwargs: Any) -> Any: + requests.append(kwargs) + return streams[len(requests) - 1] + client = create_test_anthropic_client(mock_anthropic_client) - mock_anthropic_client.beta.messages.create.return_value = events() + mock_anthropic_client.beta.messages.create.side_effect = create stream = client._inner_get_response( # type: ignore[attr-defined] messages=[Message(role="user", contents=["run"])], options={"tools": [local_tool], "max_tokens": 64}, @@ -2481,10 +2890,32 @@ async def events() -> Any: updates = [update async for update in stream] response = await stream.get_final_response() - hosted_calls = [content for update in updates for content in update.contents if content.type == "function_call"] + assert len(requests) == 2 + assert [message["role"] for message in requests[1]["messages"]] == ["user", "assistant"] + assert requests[1]["messages"][1]["content"][0] == { + "type": "server_tool_use", + "id": "hosted-call", + "name": "local_tool", + "input": {"value": "a"}, + } + assert all( + content.informational_only and content.hosted + for update in updates + for content in update.contents + if content.type == "function_call" + ) + hosted_calls = [ + content for message in response.messages for content in message.contents if content.type == "function_call" + ] assert len(hosted_calls) == 1 assert hosted_calls[0].informational_only is True + assert hosted_calls[0].parse_arguments() == {"value": "a"} assert response.finish_reason == "stop" + assert response.messages[-1].text == "done" + restored = Message.from_dict(response.messages[0].to_dict()) + prepared = client._prepare_messages_for_anthropic([restored]) + assert prepared[0]["content"][0]["type"] == "server_tool_use" + assert prepared[0]["content"][0]["input"] == {"value": "a"} async def test_concurrent_streams_isolate_tool_state_when_peer_fails( @@ -2577,11 +3008,11 @@ async def consume_primary() -> tuple[list[ChatResponseUpdate], ChatResponse]: 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 [(content.name, content.call_id, content.arguments) for content in hosted_calls] == [ + ("web_search", "shared-call", {}), + ("", "shared-call", '{"query":"framework"}'), + ] + assert all(content.informational_only and content.hosted for content in hosted_calls) async def test_concurrent_streams_keep_approval_requests_request_local( @@ -2948,6 +3379,530 @@ def test_process_stream_event_message_start_without_role_reproduces_bug() -> Non assert has_function_call, "Expected bug: function_call leaked into tool message" +async def test_agent_continues_anthropic_pause_turn() -> None: + """A paused server turn is replayed with the same tools until Anthropic completes it.""" + requests: list[dict[str, Any]] = [] + responses = [ + BetaMessage( + id="msg_paused", + content=[BetaTextBlock(type="text", text="Still researching.")], + model="claude-test", + role="assistant", + stop_reason="pause_turn", + type="message", + usage=BetaUsage(input_tokens=10, output_tokens=2), + ), + BetaMessage( + id="msg_finished", + content=[BetaTextBlock(type="text", text="Finished.")], + model="claude-test", + role="assistant", + stop_reason="end_turn", + type="message", + usage=BetaUsage(input_tokens=12, output_tokens=3), + ), + ] + + async def create(**kwargs: Any) -> BetaMessage: + requests.append(kwargs) + return responses[len(requests) - 1] + + transport = MagicMock() + transport.base_url = "https://example.invalid" + transport.beta.messages.create = create + client = AnthropicClient(anthropic_client=transport, model="claude-test") + agent = client.as_agent( + tools=[client.get_web_search_tool()], + default_options={"max_tokens": 64}, + ) + + async with agent: + response = await agent.run("Research this topic.") + + assert len(requests) == 2 + assert requests[1]["tools"] == requests[0]["tools"] + assert [message["role"] for message in requests[1]["messages"]] == ["user", "assistant"] + assert requests[1]["messages"][1]["content"] == [{"type": "text", "text": "Still researching."}] + assert response.finish_reason == "stop" + assert [message.text for message in response.messages] == ["Still researching.", "Finished."] + assert response.usage_details == { + "input_token_count": 22, + "output_token_count": 5, + } + + +async def test_streaming_pause_turn_preserves_message_boundaries_for_structured_output( + mock_anthropic_client: MagicMock, +) -> None: + """Paused text and the final structured answer remain separate provider messages.""" + + class StructuredPauseResult(BaseModel): + answer: str + + async def turn_events(response_id: str, text: str, stop_reason: str) -> Any: + yield _message_start_event(response_id) + text_start = MagicMock() + text_start.type = "content_block_start" + text_start.index = 0 + text_start.content_block.type = "text" + text_start.content_block.text = text + text_start.content_block.citations = None + yield text_start + yield _content_block_stop_event() + yield _message_delta_event(stop_reason) + yield _message_stop_event() + + streams = [ + turn_events("msg_paused", "Still researching.", "pause_turn"), + turn_events("msg_finished", '{"answer": "done"}', "end_turn"), + ] + mock_anthropic_client.beta.messages.create.side_effect = lambda **_: streams.pop(0) + client = create_test_anthropic_client(mock_anthropic_client) + stream = client._inner_get_response( # type: ignore[attr-defined] + messages=[Message(role="user", contents=["Research this topic."])], + options=ChatOptions[StructuredPauseResult]( + max_tokens=64, + response_format=StructuredPauseResult, + ), + stream=True, + ) + assert isinstance(stream, ResponseStream) + + _ = [update async for update in stream] + response = await stream.get_final_response() + + assert [message.message_id for message in response.messages] == ["msg_paused", "msg_finished"] + assert [message.text for message in response.messages] == [ + "Still researching.", + '{"answer": "done"}', + ] + assert response.value == StructuredPauseResult(answer="done") + + +async def test_streaming_pause_turn_separates_reused_server_tool_call_ids( + mock_anthropic_client: MagicMock, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Reused call IDs in separate paused turns do not merge their arguments.""" + monkeypatch.setattr("agent_framework_anthropic._chat_client._MAX_PAUSE_TURN_CONTINUATIONS", 1) + + async def turn_events(response_id: str, query: str) -> Any: + yield _message_start_event(response_id) + yield _tool_start_event( + block_type="server_tool_use", + call_id="srvtoolu_reused", + name="web_search", + ) + yield _argument_delta_event(f'{{"query": "{query}"}}') + yield _content_block_stop_event() + yield _message_delta_event("pause_turn") + yield _message_stop_event() + + streams = [ + turn_events("msg_first", "first query"), + turn_events("msg_second", "second query"), + ] + mock_anthropic_client.beta.messages.create.side_effect = lambda **_: streams.pop(0) + client = create_test_anthropic_client(mock_anthropic_client) + stream = client._inner_get_response( # type: ignore[attr-defined] + messages=[Message(role="user", contents=["Search twice."])], + options={"tools": [client.get_web_search_tool()], "max_tokens": 64}, + stream=True, + ) + assert isinstance(stream, ResponseStream) + + _ = [update async for update in stream] + response = await stream.get_final_response() + + assert [message.message_id for message in response.messages] == ["msg_first", "msg_second"] + calls = [message.contents[0] for message in response.messages] + assert [call.call_id for call in calls] == ["srvtoolu_reused", "srvtoolu_reused"] + assert [call.parse_arguments() for call in calls] == [ + {"query": "first query"}, + {"query": "second query"}, + ] + + +async def test_non_streaming_code_execution_pause_turn_is_not_auto_continued( + mock_anthropic_client: MagicMock, +) -> None: + """Code-execution content returns pause_turn when native replay is unavailable.""" + code_call = MagicMock() + code_call.type = "server_tool_use" + code_call.id = "srvtoolu_code" + code_call.name = "code_execution" + code_call.input = {"code": "print('done')"} + code_result_content = MagicMock() + code_result_content.stdout = "done" + code_result_content.stderr = None + code_result_content.content = [] + code_result = MagicMock() + code_result.type = "code_execution_tool_result" + code_result.tool_use_id = "srvtoolu_code" + code_result.content = code_result_content + message = MagicMock(spec=BetaMessage) + message.id = "msg_code_pause" + message.model = "claude-test" + message.content = [code_call, code_result] + message.usage = BetaUsage(input_tokens=1, output_tokens=1) + message.stop_reason = "pause_turn" + mock_anthropic_client.beta.messages.create.return_value = message + client = create_test_anthropic_client(mock_anthropic_client) + + response = await client._inner_get_response( # type: ignore[attr-defined] + messages=[Message(role="user", contents=["Run code."])], + options={"tools": [client.get_code_interpreter_tool()], "max_tokens": 64}, + ) + + assert mock_anthropic_client.beta.messages.create.call_count == 1 + assert response.finish_reason == "pause_turn" + assert [content.type for content in response.messages[0].contents] == [ + "code_interpreter_tool_call", + "code_interpreter_tool_result", + ] + + +async def test_streaming_code_execution_pause_turn_is_not_auto_continued( + mock_anthropic_client: MagicMock, +) -> None: + """Streaming code-execution pauses remain caller-visible without lossy replay.""" + + async def events() -> Any: + yield _message_start_event("msg_code_pause") + start = _tool_start_event( + block_type="server_tool_use", + call_id="srvtoolu_code", + name="code_execution", + ) + start.content_block.input = {"code": "print('done')"} + yield start + yield _content_block_stop_event() + yield _message_delta_event("pause_turn") + yield _message_stop_event() + + mock_anthropic_client.beta.messages.create.return_value = events() + client = create_test_anthropic_client(mock_anthropic_client) + stream = client._inner_get_response( # type: ignore[attr-defined] + messages=[Message(role="user", contents=["Run code."])], + options={"tools": [client.get_code_interpreter_tool()], "max_tokens": 64}, + stream=True, + ) + assert isinstance(stream, ResponseStream) + + _ = [update async for update in stream] + response = await stream.get_final_response() + + assert mock_anthropic_client.beta.messages.create.call_count == 1 + assert response.finish_reason == "pause_turn" + assert response.messages[0].message_id == "msg_code_pause" + assert response.messages[0].contents[0].type == "code_interpreter_tool_call" + + +async def test_streaming_mcp_pause_turn_is_not_auto_continued( + mock_anthropic_client: MagicMock, +) -> None: + """Streamed MCP arguments are not replayed until their deltas are preserved.""" + + async def events() -> Any: + yield _message_start_event("msg_mcp_pause") + start = _tool_start_event( + block_type="mcp_tool_use", + call_id="mcp_call", + name="search_docs", + ) + start.content_block.server_name = "docs" + yield start + yield _argument_delta_event('{"query": "latest docs"}') + yield _content_block_stop_event() + yield _message_delta_event("pause_turn") + yield _message_stop_event() + + mock_anthropic_client.beta.messages.create.return_value = events() + client = create_test_anthropic_client(mock_anthropic_client) + stream = client._inner_get_response( # type: ignore[attr-defined] + messages=[Message(role="user", contents=["Search docs."])], + options={ + "tools": [client.get_mcp_tool(name="docs", url="https://example.com/mcp")], + "max_tokens": 64, + }, + stream=True, + ) + assert isinstance(stream, ResponseStream) + + _ = [update async for update in stream] + response = await stream.get_final_response() + + assert mock_anthropic_client.beta.messages.create.call_count == 1 + assert response.finish_reason == "pause_turn" + assert response.messages[0].contents[0].type == "mcp_server_tool_call" + assert response.messages[0].contents[0].parse_arguments() == {} + + +async def test_non_streaming_redacted_thinking_pause_turn_is_not_auto_continued( + mock_anthropic_client: MagicMock, +) -> None: + """Dropped redacted reasoning prevents an incomplete continuation transcript.""" + message = BetaMessage( + id="msg_redacted_pause", + content=[ + BetaRedactedThinkingBlock(type="redacted_thinking", data="opaque"), + BetaTextBlock(type="text", text="Still working."), + ], + model="claude-test", + role="assistant", + stop_reason="pause_turn", + type="message", + usage=BetaUsage(input_tokens=1, output_tokens=1), + ) + mock_anthropic_client.beta.messages.create.return_value = message + client = create_test_anthropic_client(mock_anthropic_client) + + response = await client._inner_get_response( # type: ignore[attr-defined] + messages=[Message(role="user", contents=["Think carefully."])], + options={"max_tokens": 64}, + ) + + assert mock_anthropic_client.beta.messages.create.call_count == 1 + assert response.finish_reason == "pause_turn" + assert response.text == "Still working." + + +async def test_streaming_redacted_thinking_pause_turn_is_not_auto_continued( + mock_anthropic_client: MagicMock, +) -> None: + """Streaming also retains pause_turn when redacted reasoning was dropped.""" + + async def events() -> Any: + yield _message_start_event("msg_redacted_pause") + redacted_start = MagicMock() + redacted_start.type = "content_block_start" + redacted_start.index = 0 + redacted_start.content_block = BetaRedactedThinkingBlock(type="redacted_thinking", data="opaque") + yield redacted_start + yield _content_block_stop_event(index=0) + text_start = MagicMock() + text_start.type = "content_block_start" + text_start.index = 1 + text_start.content_block.type = "text" + text_start.content_block.text = "Still working." + text_start.content_block.citations = None + yield text_start + yield _content_block_stop_event(index=1) + yield _message_delta_event("pause_turn") + yield _message_stop_event() + + mock_anthropic_client.beta.messages.create.return_value = events() + client = create_test_anthropic_client(mock_anthropic_client) + stream = client._inner_get_response( # type: ignore[attr-defined] + messages=[Message(role="user", contents=["Think carefully."])], + options={"max_tokens": 64}, + stream=True, + ) + assert isinstance(stream, ResponseStream) + + _ = [update async for update in stream] + response = await stream.get_final_response() + + assert mock_anthropic_client.beta.messages.create.call_count == 1 + assert response.finish_reason == "pause_turn" + assert response.text == "Still working." + + +async def test_anthropic_pause_turn_limit_preserves_finish_reason(monkeypatch: pytest.MonkeyPatch) -> None: + """A bounded continuation returns pause_turn instead of reporting successful completion.""" + monkeypatch.setattr("agent_framework_anthropic._chat_client._MAX_PAUSE_TURN_CONTINUATIONS", 1) + requests: list[dict[str, Any]] = [] + + async def create(**kwargs: Any) -> BetaMessage: + requests.append(kwargs) + return BetaMessage( + id=f"msg_paused_{len(requests)}", + content=[BetaTextBlock(type="text", text=f"Paused {len(requests)}.")], + model="claude-test", + role="assistant", + stop_reason="pause_turn", + type="message", + usage=BetaUsage(input_tokens=1, output_tokens=1), + ) + + transport = MagicMock() + transport.base_url = "https://example.invalid" + transport.beta.messages.create = create + client = AnthropicClient(anthropic_client=transport, model="claude-test") + agent = client.as_agent( + tools=[client.get_web_search_tool()], + default_options={"max_tokens": 64}, + ) + + async with agent: + response = await agent.run("Research this topic.") + + assert len(requests) == 2 + assert response.finish_reason == "pause_turn" + assert [message.text for message in response.messages] == ["Paused 1.", "Paused 2."] + + +@pytest.mark.filterwarnings("ignore::agent_framework._feature_stage.ExperimentalWarning") +async def test_agent_file_history_replays_server_tool_transcript_end_to_end(tmp_path: Path) -> None: + """Public agent APIs preserve a mixed server/local tool transcript through file history.""" + first_requests: list[dict[str, Any]] = [] + local_executions = 0 + + @tool(name="local_value", approval_mode="never_require") + def local_value() -> str: + nonlocal local_executions + local_executions += 1 + return "local-ok" + + first_responses = [ + BetaMessage( + id="msg_mixed_tools", + content=[ + BetaServerToolUseBlock( + type="server_tool_use", + id="srvtoolu_search", + name="web_search", + input={"query": "Microsoft Agent Framework"}, + ), + BetaToolUseBlock( + type="tool_use", + id="toolu_local", + name="local_value", + input={}, + ), + ], + model="claude-test", + role="assistant", + stop_reason="tool_use", + type="message", + usage=BetaUsage(input_tokens=1, output_tokens=1), + ), + BetaMessage( + id="msg_server_result", + content=[ + BetaWebSearchToolResultBlock( + type="web_search_tool_result", + tool_use_id="srvtoolu_search", + content=[ + BetaWebSearchResultBlock( + type="web_search_result", + url="https://github.com/microsoft/agent-framework", + title="Microsoft Agent Framework", + encrypted_content="opaque-search-state", + page_age=None, + ) + ], + ), + BetaTextBlock(type="text", text="first turn complete"), + ], + model="claude-test", + role="assistant", + stop_reason="end_turn", + type="message", + usage=BetaUsage(input_tokens=1, output_tokens=1), + ), + ] + + async def create_first_response(**kwargs: Any) -> BetaMessage: + first_requests.append(kwargs) + return first_responses[len(first_requests) - 1] + + first_transport = MagicMock() + first_transport.base_url = "https://example.invalid" + first_transport.beta.messages.create = create_first_response + first_history = FileHistoryProvider(tmp_path) + first_client = AnthropicClient(anthropic_client=first_transport, model="claude-test") + first_agent = first_client.as_agent( + tools=[first_client.get_web_search_tool(), local_value], + context_providers=[first_history], + default_options={"max_tokens": 64}, + ) + session_id = "anthropic-server-tool-history" + + async with first_agent: + first_response = await first_agent.run( + "Search for Agent Framework and call local_value.", + session=first_agent.create_session(session_id=session_id), + ) + + assert "first turn complete" in first_response.text + assert local_executions == 1 + assert len(first_requests) == 2 + continuation_messages = first_requests[1]["messages"] + assert [message["role"] for message in continuation_messages] == ["user", "assistant", "user"] + assert [block["type"] for block in continuation_messages[1]["content"]] == [ + "server_tool_use", + "tool_use", + ] + assert continuation_messages[2]["content"][0]["tool_use_id"] == "toolu_local" + + persisted = await first_history.get_messages(session_id) + persisted_types = [content.type for message in persisted for content in message.contents] + assert persisted_types.count("function_call") == 2 + assert "function_result" in persisted_types + + replay_requests: list[dict[str, Any]] = [] + + async def create_replay_response(**kwargs: Any) -> BetaMessage: + replay_requests.append(kwargs) + return BetaMessage( + id="msg_replay", + content=[BetaTextBlock(type="text", text="history replay complete")], + model="claude-test", + role="assistant", + stop_reason="end_turn", + type="message", + usage=BetaUsage(input_tokens=1, output_tokens=1), + ) + + replay_transport = MagicMock() + replay_transport.base_url = "https://example.invalid" + replay_transport.beta.messages.create = create_replay_response + replay_history = FileHistoryProvider(tmp_path) + replay_client = AnthropicClient(anthropic_client=replay_transport, model="claude-test") + replay_agent = replay_client.as_agent( + tools=[replay_client.get_web_search_tool(), local_value], + context_providers=[replay_history], + default_options={"max_tokens": 64}, + ) + + async with replay_agent: + replay_response = await replay_agent.run( + "Use the prior result and reply briefly.", + session=replay_agent.create_session(session_id=session_id), + ) + + assert replay_response.text == "history replay complete" + assert local_executions == 1 + assert len(replay_requests) == 1 + replayed_messages = replay_requests[0]["messages"] + server_call_message = next( + message + for message in replayed_messages + if isinstance(message["content"], list) + and any(block["type"] == "server_tool_use" for block in message["content"]) + ) + server_result_message = next( + message + for message in replayed_messages + if isinstance(message["content"], list) + and any(block["type"] == "web_search_tool_result" for block in message["content"]) + ) + local_result_message = next( + message + for message in replayed_messages + if isinstance(message["content"], list) and any(block["type"] == "tool_result" for block in message["content"]) + ) + + assert server_call_message["role"] == "assistant" + assert server_result_message["role"] == "assistant" + assert local_result_message["role"] == "user" + search_result = next( + block for block in server_result_message["content"] if block["type"] == "web_search_tool_result" + ) + assert search_result["content"][0]["encrypted_content"] == "opaque-search-state" + + # Integration Tests @@ -3051,6 +4006,61 @@ async def test_anthropic_client_integration_hosted_tools() -> None: assert response.text is not None +@pytest.mark.flaky +@pytest.mark.integration +@pytest.mark.filterwarnings("ignore::agent_framework._feature_stage.ExperimentalWarning") +@skip_if_anthropic_integration_tests_disabled +async def test_anthropic_client_integration_file_history_replays_web_search(tmp_path: Path) -> None: + """A fresh real-service agent can replay a persisted Anthropic web-search transcript.""" + session_id = "anthropic-web-search-history-integration" + tools = [AnthropicClient.get_web_search_tool()] + first_history = FileHistoryProvider(tmp_path) + first_client = AnthropicClient() + first_agent = first_client.as_agent( + tools=tools, + context_providers=[first_history], + default_options={ + "max_tokens": 512, + "tool_choice": {"mode": "required", "required_function_name": "web_search"}, + }, + ) + + async with first_agent: + first_response = await first_agent.run( + "Use web_search exactly once to find the official GitHub repository for Microsoft Agent Framework. " + "Return the repository URL.", + session=first_agent.create_session(session_id=session_id), + ) + + assert first_response.text + persisted = await first_history.get_messages(session_id) + server_blocks = [ + content.additional_properties.get("anthropic", {}).get("server_tool_block", {}) + for message in persisted + for content in message.contents + if content.type in {"function_call", "function_result"} + ] + assert any(block.get("type") == "server_tool_use" for block in server_blocks) + assert any(block.get("type") == "web_search_tool_result" for block in server_blocks) + + replay_history = FileHistoryProvider(tmp_path) + replay_client = AnthropicClient() + replay_agent = replay_client.as_agent( + tools=tools, + context_providers=[replay_history], + default_options={"max_tokens": 128}, + ) + + async with replay_agent: + replay_response = await replay_agent.run( + "Do not search again. From the prior result, reply with only the repository URL.", + session=replay_agent.create_session(session_id=session_id), + options={"tool_choice": "none"}, + ) + + assert "github.com/microsoft/agent-framework" in replay_response.text.lower() + + @pytest.mark.flaky @pytest.mark.integration @skip_if_anthropic_integration_tests_disabled @@ -3488,38 +4498,96 @@ def test_parse_contents_mcp_tool_result_object_content( assert result[0].type == "mcp_server_tool_result" -def test_parse_contents_web_search_tool_result( +@pytest.mark.parametrize( + ("block_type", "call_id", "block_content", "expected_content"), + [ + ( + "web_search_tool_result", + "srvtoolu_search", + [ + BetaWebSearchResultBlock( + type="web_search_result", + url="https://example.com", + title="Example", + encrypted_content="opaque", + page_age="today", + ) + ], + [ + { + "encrypted_content": "opaque", + "page_age": "today", + "title": "Example", + "type": "web_search_result", + "url": "https://example.com", + } + ], + ), + ( + "web_fetch_tool_result", + "srvtoolu_fetch", + { + "type": "web_fetch_result", + "url": "https://example.com", + "content": { + "type": "document", + "source": { + "type": "text", + "media_type": "text/plain", + "data": "Example", + }, + }, + }, + { + "type": "web_fetch_result", + "url": "https://example.com", + "content": { + "type": "document", + "source": { + "type": "text", + "media_type": "text/plain", + "data": "Example", + }, + }, + }, + ), + ], +) +def test_server_tool_result_survives_message_round_trip( mock_anthropic_client: MagicMock, + block_type: str, + call_id: str, + block_content: Any, + expected_content: Any, ) -> None: - """Test parsing web search tool result.""" - client = create_test_anthropic_client(mock_anthropic_client) - - # Create mock web search tool result - mock_block = MagicMock() - mock_block.type = "web_search_tool_result" - mock_block.tool_use_id = "call_789" - mock_block.content = "Search results" - - result = client._parse_contents_from_anthropic([mock_block]) - - assert len(result) == 1 - assert result[0].type == "function_result" - - -def test_parse_contents_web_fetch_tool_result(mock_anthropic_client: MagicMock) -> None: - """Test parsing web fetch tool result.""" + """Server-tool results retain their native type and structured content after persistence.""" client = create_test_anthropic_client(mock_anthropic_client) - # Create mock web fetch tool result mock_block = MagicMock() - mock_block.type = "web_fetch_tool_result" - mock_block.tool_use_id = "call_101" - mock_block.content = "Fetched content" - - result = client._parse_contents_from_anthropic([mock_block]) - - assert len(result) == 1 - assert result[0].type == "function_result" + mock_block.type = block_type + mock_block.tool_use_id = call_id + mock_block.content = block_content + mock_block.caller = {"type": "direct"} + + parsed = client._parse_contents_from_anthropic([mock_block]) + restored = Message.from_dict(Message(role="assistant", contents=parsed).to_dict()) + replay = client._prepare_message_groups_for_anthropic(restored) + + assert len(parsed) == 1 + assert parsed[0].type == "function_result" + assert replay == [ + { + "role": "assistant", + "content": [ + { + "type": block_type, + "tool_use_id": call_id, + "content": expected_content, + "caller": {"type": "direct"}, + } + ], + } + ] # MCP Tool Configuration Tests