Python: Mark hosted tool calls informational-only

python/packages/anthropic/agent_framework_anthropic/_chat_client.py

@@ -1188,6 +1188,7 @@ def _parse_contents_from_anthropic(
                                 call_id=content_block.id,
                                 name=resolved_tool_name,
                                 arguments=content_block.input,
+                                informational_only=content_block.type == "server_tool_use",
                                 raw_representation=content_block,
                             )
                         )

python/packages/anthropic/tests/test_anthropic_client.py

@@ -1452,9 +1452,10 @@ def test_parse_contents_server_tool_use_input_json_delta_ignored(
     server_tool_content.input = {}
 
     result = client._parse_contents_from_anthropic([server_tool_content])
-    # server_tool_use falls through to function_call (not mcp_tool_use / code_execution)
+    # 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]
 
     # input_json_delta events after server_tool_use must be silently ignored

python/packages/core/agent_framework/_tools.py

@@ -1626,6 +1626,10 @@ def _get_tool_map(
     return tool_list
 
 
+def _is_actionable_function_call(content: Content) -> bool:
+    return content.type == "function_call" and not content.informational_only
+
+
 async def _try_execute_function_calls(
     custom_args: dict[str, Any],
     attempt_idx: int,
@@ -1655,6 +1659,14 @@ async def _try_execute_function_calls(
     """
     from ._types import Content
 
+    function_calls = [
+        function_call
+        for function_call in function_calls
+        if function_call.type == "function_approval_response" or _is_actionable_function_call(function_call)
+    ]
+    if not function_calls:
+        return ([], False)
+
     tool_map = _get_tool_map(tools)
     # The live tools list (when tools is the run-local list) is exposed on the
     # FunctionInvocationContext so tools can add/remove tools during the run.
@@ -1680,14 +1692,18 @@ async def _try_execute_function_calls(
             fcc_name,
             fcc_name in approval_tools,
         )
-        if fcc.type == "function_call" and fcc.name in approval_tools:
+        if _is_actionable_function_call(fcc) and fcc.name in approval_tools:
             logger.debug("Approval needed for function: %s", fcc.name)
             approval_needed = True
             break
-        if fcc.type == "function_call" and (fcc.name in declaration_only or fcc.name in additional_tool_names):
+        if _is_actionable_function_call(fcc) and (fcc.name in declaration_only or fcc.name in additional_tool_names):
             declaration_only_flag = True
             break
-        if config.get("terminate_on_unknown_calls", False) and fcc.type == "function_call" and fcc.name not in tool_map:
+        if (
+            config.get("terminate_on_unknown_calls", False)
+            and _is_actionable_function_call(fcc)
+            and fcc.name not in tool_map
+        ):
             raise KeyError(f'Error: Requested function "{fcc.name}" not found.')
     if approval_needed:
         # approval can only be needed for Function Call Content, not Approval Responses.
@@ -2182,7 +2198,7 @@ def _extract_function_calls(response: ChatResponse) -> list[Content]:
     function_calls: list[Content] = []
     for message in response.messages:
         for item in message.contents:
-            if item.type != "function_call":
+            if not _is_actionable_function_call(item):
                 continue
             if item.call_id and item.call_id in function_results:
                 continue
@@ -2793,7 +2809,7 @@ async def _stream() -> AsyncIterable[ChatResponseUpdate]:
                 )
 
                 if not any(
-                    item.type in ("function_call", "function_approval_request")
+                    item.type == "function_approval_request" or _is_actionable_function_call(item)
                     for msg in response.messages
                     for item in msg.contents
                 ):

python/packages/core/agent_framework/_types.py

@@ -496,6 +496,7 @@ def __init__(
         name: str | None = None,
         arguments: str | Mapping[str, Any] | None = None,
         exception: str | None = None,
+        informational_only: bool = False,
         result: Any = None,
         items: Sequence[Content] | None = None,
         # Hosted file/vector store fields
@@ -556,6 +557,7 @@ def __init__(
         self.name = name
         self.arguments = arguments
         self.exception = exception
+        self.informational_only = informational_only or type == "mcp_server_tool_call"
         self.result = result
         self.items = items
         self.file_id = file_id
@@ -805,17 +807,39 @@ def from_function_call(
         *,
         arguments: str | Mapping[str, Any] | None = None,
         exception: str | None = None,
+        informational_only: bool = False,
         annotations: Sequence[Annotation] | None = None,
         additional_properties: MutableMapping[str, Any] | None = None,
         raw_representation: Any = None,
     ) -> ContentT:
-        """Create function call content."""
+        """Create function call content.
+
+        Args:
+            call_id: The model- or service-provided identifier for this function call. Function results use the
+                same ID to indicate which call they answer.
+            name: The function name requested by the model or service.
+
+        Keyword Args:
+            arguments: The arguments for the requested function call. May be a JSON string, a mapping that can be
+                serialized as arguments, or None when no arguments were provided.
+            exception: Error information associated with the function call, if the provider returned the call in an
+                error state.
+            informational_only: Whether the function call is present only for transcript fidelity and should not be
+                executed by Agent Framework function invocation.
+            annotations: Optional annotations attached to this content item.
+            additional_properties: Extra provider-specific properties to preserve with the content item.
+            raw_representation: The original provider-specific object or payload this content item was created from.
+
+        Returns:
+            Function call content.
+        """
         return cls(
             "function_call",
             call_id=call_id,
             name=name,
             arguments=arguments,
             exception=exception,
+            informational_only=informational_only,
             annotations=annotations,
             additional_properties=additional_properties,
             raw_representation=raw_representation,
@@ -1184,13 +1208,36 @@ def from_mcp_server_tool_call(
         additional_properties: MutableMapping[str, Any] | None = None,
         raw_representation: Any = None,
     ) -> ContentT:
-        """Create MCP server tool call content."""
+        """Create MCP server tool call content.
+
+        MCP server tool calls are provider-hosted tool calls that the model/service
+        already routed to a remote MCP server. They are recorded for transcript
+        fidelity and for matching provider-returned MCP tool results, but they are
+        not local function invocation requests. The returned content is always
+        marked ``informational_only=True``.
+
+        Args:
+            call_id: The model- or service-provided identifier for this MCP tool call.
+            tool_name: The remote MCP tool name that was called.
+
+        Keyword Args:
+            server_name: The remote MCP server label or name, when provided by the service.
+            arguments: The arguments sent to the remote MCP tool. May be a JSON string,
+                a mapping, or None when no arguments were provided.
+            annotations: Optional annotations attached to this content item.
+            additional_properties: Extra provider-specific properties to preserve with the content item.
+            raw_representation: The original provider-specific object or payload this content item was created from.
+
+        Returns:
+            MCP server tool call content.
+        """
         return cls(
             "mcp_server_tool_call",
             call_id=call_id,
             tool_name=tool_name,
             server_name=server_name,
             arguments=arguments,
+            informational_only=True,
             annotations=annotations,
             additional_properties=additional_properties,
             raw_representation=raw_representation,
@@ -1323,6 +1370,7 @@ def to_dict(self, *, exclude_none: bool = True, exclude: set[str] | None = None)
             "name",
             "arguments",
             "exception",
+            "informational_only",
             "result",
             "items",
             "file_id",
@@ -1356,6 +1404,8 @@ def to_dict(self, *, exclude_none: bool = True, exclude: set[str] | None = None)
             value = getattr(self, field, None)
             if field in exclude:
                 continue
+            if field == "informational_only" and (self.type != "function_call" or not value):
+                continue
             if exclude_none and value is None:
                 continue
             result[field] = _serialize_value(value, exclude_none)
@@ -1499,6 +1549,8 @@ def _add_function_call_content(self, other: Content) -> Content:
             name=getattr(self, "name", getattr(other, "name", None)),
             arguments=arguments,
             exception=getattr(self, "exception", None) or getattr(other, "exception", None),
+            informational_only=getattr(self, "informational_only", False)
+            or getattr(other, "informational_only", False),
             additional_properties=_combine_additional_props(self.additional_properties, other.additional_properties),
             raw_representation=_combine_raw_representations(self.raw_representation, other.raw_representation),
         )

python/packages/core/tests/core/test_function_invocation_logic.py

@@ -767,6 +767,122 @@ def func_with_approval(arg1: str) -> str:
             assert exec_counter == 0  # Neither function executed yet
 
 
+async def test_informational_only_function_call_is_not_invoked(chat_client_base: SupportsChatGetResponse):
+    exec_counter = 0
+
+    @tool(name="no_approval_func", approval_mode="never_require")
+    def func_no_approval(arg1: str) -> str:
+        nonlocal exec_counter
+        exec_counter += 1
+        return f"Processed {arg1}"
+
+    informational_call = Content.from_function_call(
+        call_id="1",
+        name="no_approval_func",
+        arguments='{"arg1": "value1"}',
+        informational_only=True,
+    )
+    chat_client_base.run_responses = [  # type: ignore[attr-defined]  # ty: ignore[unresolved-attribute]
+        ChatResponse(messages=Message(role="assistant", contents=[informational_call]))
+    ]
+
+    response = await chat_client_base.get_response(
+        [Message(role="user", contents=["hello"])],
+        options={"tool_choice": "auto", "tools": [func_no_approval]},
+    )
+
+    assert exec_counter == 0
+    assert len(response.messages) == 1
+    assert response.messages[0].contents == [informational_call]
+
+
+async def test_only_actionable_function_calls_are_invoked(chat_client_base: SupportsChatGetResponse):
+    exec_counter = 0
+
+    @tool(name="no_approval_func", approval_mode="never_require")
+    def func_no_approval(arg1: str) -> str:
+        nonlocal exec_counter
+        exec_counter += 1
+        return f"Processed {arg1}"
+
+    informational_call = Content.from_function_call(
+        call_id="1",
+        name="no_approval_func",
+        arguments='{"arg1": "value1"}',
+        informational_only=True,
+    )
+    text_content = Content.from_text("Provider transcript text")
+    chat_client_base.run_responses = [  # type: ignore[attr-defined]  # ty: ignore[unresolved-attribute]
+        ChatResponse(messages=Message(role="assistant", contents=[informational_call, text_content]))
+    ]
+
+    response = await chat_client_base.get_response(
+        [Message(role="user", contents=["hello"])],
+        options={"tool_choice": "auto", "tools": [func_no_approval]},
+    )
+
+    assert exec_counter == 0
+    assert len(response.messages) == 1
+    assert response.messages[0].contents == [informational_call, text_content]
+
+
+async def test_informational_only_function_call_does_not_request_approval(chat_client_base: SupportsChatGetResponse):
+    @tool(name="approval_func", approval_mode="always_require")
+    def func_with_approval(arg1: str) -> str:
+        return f"Approved {arg1}"
+
+    informational_call = Content.from_function_call(
+        call_id="1",
+        name="approval_func",
+        arguments='{"arg1": "value1"}',
+        informational_only=True,
+    )
+    chat_client_base.run_responses = [  # type: ignore[attr-defined]  # ty: ignore[unresolved-attribute]
+        ChatResponse(messages=Message(role="assistant", contents=[informational_call]))
+    ]
+
+    response = await chat_client_base.get_response(
+        [Message(role="user", contents=["hello"])],
+        options={"tool_choice": "auto", "tools": [func_with_approval]},
+    )
+
+    assert response.messages[0].contents == [informational_call]
+    assert not any(content.type == "function_approval_request" for content in response.messages[0].contents)
+
+
+async def test_streaming_informational_only_function_call_is_not_invoked(chat_client_base: SupportsChatGetResponse):
+    exec_counter = 0
+
+    @tool(name="no_approval_func", approval_mode="never_require")
+    def func_no_approval(arg1: str) -> str:
+        nonlocal exec_counter
+        exec_counter += 1
+        return f"Processed {arg1}"
+
+    informational_call = Content.from_function_call(
+        call_id="1",
+        name="no_approval_func",
+        arguments='{"arg1": "value1"}',
+        informational_only=True,
+    )
+    chat_client_base.streaming_responses = [  # type: ignore[attr-defined]  # ty: ignore[unresolved-attribute]
+        [ChatResponseUpdate(contents=[informational_call], role="assistant")]
+    ]
+
+    updates = [
+        update
+        async for update in chat_client_base.get_response(
+            [Message(role="user", contents=["hello"])],
+            options={"tool_choice": "auto", "tools": [func_no_approval]},
+            stream=True,
+        )
+    ]
+
+    assert exec_counter == 0
+    assert len(updates) == 1
+    assert updates[0].contents == [informational_call]
+
+
 async def test_rejected_approval(chat_client_base: SupportsChatGetResponse):
     """Test that rejecting an approval alongside an approved one is handled correctly."""