From 3f489672f75b36c7110e78b7f6e0f4c0cf142a84 Mon Sep 17 00:00:00 2001
From: Weijian He <230282609+Madin-H23@users.noreply.github.com>
Date: Wed, 21 Oct 2026 16:18:10 +0800
Subject: [PATCH 1/5] Python: fix(tools): answer a function call whose
 arguments are not a JSON object

Closes #9562

Content.parse_arguments() wraps a payload that is not a JSON object under a
single "raw" key. That is the right answer for a caller that wants to see
what the model sent, but the function-calling loop used it as the arguments
of the call, so a model that answered with something that is not a JSON
object had its tool invoked with an argument the tool never declared, and
the model was never told that its call was malformed.

The loop now parses the payload strictly and answers a call whose arguments
are not a JSON object the way it already answers a missing required
property: with an argument-validation error result that names the cause, so
the model can retry. parse_arguments() keeps its documented behaviour for
direct callers.
---
 .../packages/core/agent_framework/_tools.py   | 29 +++++++-
 python/packages/core/tests/core/test_tools.py | 66 +++++++++++++++++++
 2 files changed, 94 insertions(+), 1 deletion(-)

diff --git a/python/packages/core/agent_framework/_tools.py b/python/packages/core/agent_framework/_tools.py
index eb40fec7ee7..e58c08c523d 100644
--- a/python/packages/core/agent_framework/_tools.py
+++ b/python/packages/core/agent_framework/_tools.py
@@ -2164,6 +2164,30 @@ def _finalize_function_result(
     return function_result
 
 
+def _parse_function_call_arguments(function_call: Content, tool_name: str) -> dict[str, Any]:
+    """Return the arguments of a function call, rejecting a payload that is not a JSON object.
+
+    `Content.parse_arguments()` wraps a payload that is not a JSON object under a
+    single "raw" key. That is the right answer for a caller that wants to see
+    what the model sent, but not for a function call: the tool would run with an
+    argument it never declared, and the model would not learn that its call was
+    malformed. Such a call is answered like a missing required argument, so the
+    model can retry it.
+    """
+    arguments = function_call.arguments
+    if isinstance(arguments, str) and arguments.strip():
+        try:
+            decoded = json.loads(arguments)
+        except (json.JSONDecodeError, TypeError):
+            decoded = None
+        if not isinstance(decoded, dict):
+            raise _FunctionArgumentValidationError(
+                f"Invalid arguments for '{tool_name}': arguments are not a JSON object",
+                redacted_message=f"Invalid arguments for '{tool_name}'.",
+            )
+    return dict(function_call.parse_arguments() or {})
+
+
 async def _auto_invoke_function(
     function_call_content: Content,
     custom_args: dict[str, Any] | None = None,
@@ -2239,7 +2263,10 @@ async def _auto_invoke_function(
         approval_response = function_call_content
         function_call_content = approved_function_call
 
-    parsed_args: dict[str, Any] = dict(function_call_content.parse_arguments() or {})
+    try:
+        parsed_args = _parse_function_call_arguments(function_call_content, tool.name)
+    except _FunctionArgumentValidationError as exc:
+        return _function_argument_validation_error_result(function_call_content, exc, config)
 
     # Filter out internal framework kwargs before passing to tools.
     # conversation_id is an internal tracking ID that should not be forwarded to tools.
diff --git a/python/packages/core/tests/core/test_tools.py b/python/packages/core/tests/core/test_tools.py
index 8dd1da6f053..45b2f8ff551 100644
--- a/python/packages/core/tests/core/test_tools.py
+++ b/python/packages/core/tests/core/test_tools.py
@@ -431,6 +431,72 @@ def search(query: str) -> str:
         await search.invoke(arguments={})
 
 
+@pytest.mark.parametrize("arguments", ["not {json", "[1, 2]", '"text"'])
+async def test_function_call_arguments_that_are_not_a_json_object_are_answered_with_an_error(
+    arguments: str,
+) -> None:
+    """A payload that is not a JSON object is not arguments, and must not be called as one.
+
+    Wrapping it under "raw" would call the tool with an argument it never
+    declared, and the model would not learn that its call was malformed.
+    """
+    calls: list[dict[str, Any]] = []
+
+    async def list_accounts(**kwargs: Any) -> str:
+        calls.append(kwargs)
+        return "ok"
+
+    account_tool = FunctionTool(
+        name="list_accounts",
+        description="List the accounts",
+        func=list_accounts,
+        input_model={"type": "object", "properties": {}},
+    )
+
+    result_groups, should_terminate = await _try_execute_function_call_groups(
+        custom_args={},
+        function_calls=[
+            Content.from_function_call(call_id="call-1", name="list_accounts", arguments=arguments),
+        ],
+        tools=[account_tool],
+        config={},
+    )
+
+    assert not should_terminate
+    assert calls == []
+    result = result_groups[0][0]
+    assert "not a JSON object" in str(result.exception)
+
+
+async def test_function_call_with_object_arguments_still_runs_the_tool() -> None:
+    """A well-formed call is unaffected by the guard."""
+    calls: list[dict[str, Any]] = []
+
+    async def list_accounts(**kwargs: Any) -> str:
+        calls.append(kwargs)
+        return "ok"
+
+    account_tool = FunctionTool(
+        name="list_accounts",
+        description="List the accounts",
+        func=list_accounts,
+        input_model={"type": "object", "properties": {}},
+    )
+
+    result_groups, should_terminate = await _try_execute_function_call_groups(
+        custom_args={},
+        function_calls=[
+            Content.from_function_call(call_id="call-1", name="list_accounts", arguments='{"limit": 5}'),
+        ],
+        tools=[account_tool],
+        config={},
+    )
+
+    assert not should_terminate
+    assert calls == [{"limit": 5}]
+    assert result_groups[0][0].result == "ok"
+
+
 async def test_invoke_preserves_explicit_null_argument():
     """A required nullable argument the model sets to null must reach the function.
 

From 84de7d5226601d4518a5c9d70820f3a0c36dc9a5 Mon Sep 17 00:00:00 2001
From: Weijian He <230282609+Madin-H23@users.noreply.github.com>
Date: Thu, 22 Oct 2026 10:35:08 +0800
Subject: [PATCH 2/5] Python: fix(tools): reject a whitespace-only function
 call payload

The strict parse skipped a payload that was only whitespace, so such a call
still reached the tool as a bogus "raw" argument. A payload that is not empty
has to be a JSON object, and whitespace is not one.
---
 python/packages/core/agent_framework/_tools.py | 2 +-
 python/packages/core/tests/core/test_tools.py  | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/python/packages/core/agent_framework/_tools.py b/python/packages/core/agent_framework/_tools.py
index e58c08c523d..06a7a09fd23 100644
--- a/python/packages/core/agent_framework/_tools.py
+++ b/python/packages/core/agent_framework/_tools.py
@@ -2175,7 +2175,7 @@ def _parse_function_call_arguments(function_call: Content, tool_name: str) -> di
     model can retry it.
     """
     arguments = function_call.arguments
-    if isinstance(arguments, str) and arguments.strip():
+    if isinstance(arguments, str) and arguments:
         try:
             decoded = json.loads(arguments)
         except (json.JSONDecodeError, TypeError):
diff --git a/python/packages/core/tests/core/test_tools.py b/python/packages/core/tests/core/test_tools.py
index 45b2f8ff551..e734351d05e 100644
--- a/python/packages/core/tests/core/test_tools.py
+++ b/python/packages/core/tests/core/test_tools.py
@@ -431,7 +431,7 @@ def search(query: str) -> str:
         await search.invoke(arguments={})
 
 
-@pytest.mark.parametrize("arguments", ["not {json", "[1, 2]", '"text"'])
+@pytest.mark.parametrize("arguments", ["not {json", "[1, 2]", '"text"', "   "])
 async def test_function_call_arguments_that_are_not_a_json_object_are_answered_with_an_error(
     arguments: str,
 ) -> None:

From a08f451c456a7151527abfebdbc0f3ffb974d517 Mon Sep 17 00:00:00 2001
From: Weijian He <230282609+Madin-H23@users.noreply.github.com>
Date: Thu, 22 Oct 2026 16:43:28 +0800
Subject: [PATCH 3/5] Python: fix(tools): answer a malformed call before it is
 sent for approval

A payload that is not arguments cannot become a working call by approving it,
so the model is now told what went wrong instead of a human being asked to
approve a call that can only fail. The guard is the one the invocation path
already uses.

The guard's docstring now records that it runs before the middleware pipeline,
so middleware does not see a malformed payload as a tool call.
---
 .../packages/core/agent_framework/_tools.py   | 33 ++++++++++++++---
 python/packages/core/tests/core/test_tools.py | 35 +++++++++++++++++++
 2 files changed, 63 insertions(+), 5 deletions(-)

diff --git a/python/packages/core/agent_framework/_tools.py b/python/packages/core/agent_framework/_tools.py
index 06a7a09fd23..3df695cbeeb 100644
--- a/python/packages/core/agent_framework/_tools.py
+++ b/python/packages/core/agent_framework/_tools.py
@@ -2173,6 +2173,10 @@ def _parse_function_call_arguments(function_call: Content, tool_name: str) -> di
     argument it never declared, and the model would not learn that its call was
     malformed. Such a call is answered like a missing required argument, so the
     model can retry it.
+
+    The check runs before the middleware pipeline, so middleware never sees a
+    malformed payload as a tool call: a payload that is not arguments is not a
+    call, and it is answered instead of being invoked.
     """
     arguments = function_call.arguments
     if isinstance(arguments, str) and arguments:
@@ -2715,6 +2719,7 @@ async def execute_local_calls(function_calls_to_execute: Sequence[Content]) -> l
             if function_call.type == "function_call"
         ]
         pause_groups: list[list[Content]] = []
+        surfaced_groups: list[list[Content]] = []
         for function_call in function_calls:
             if function_call.type != "function_call":
                 continue
@@ -2724,24 +2729,42 @@ async def execute_local_calls(function_calls_to_execute: Sequence[Content]) -> l
                 and tool_name not in approval_tool_names
                 and (tool_name in declaration_only_tool_names or tool_name in additional_tool_names)
             ):
-                pause_groups.append([_as_user_input_pause(function_call)])
+                group = [_as_user_input_pause(function_call)]
+                pause_groups.append(group)
+                surfaced_groups.append(group)
                 continue
+            if tool_name is not None:
+                # A payload that is not arguments cannot become a working call by
+                # approving it, so answer it here instead of asking a human to
+                # approve a call that can only fail. The same guard runs again in
+                # _auto_invoke_function.
+                try:
+                    _parse_function_call_arguments(function_call, tool_name)
+                except _FunctionArgumentValidationError as exc:
+                    surfaced_groups.append([_function_argument_validation_error_result(function_call, exc, config)])
+                    continue
             approval_request = Content.from_function_approval_request(
                 id=function_call.id or function_call.call_id,  # type: ignore[arg-type]
                 function_call=function_call,
             )
             if tool_name is None:
                 visible_requests.append(approval_request)
-                pause_groups.append([approval_request])
+                group = [approval_request]
+                pause_groups.append(group)
+                surfaced_groups.append(group)
                 continue
             tool = tool_map.get(tool_name)
             if tool_name in approval_tool_names or tool is None:
                 visible_requests.append(approval_request)
-                pause_groups.append([approval_request])
+                group = [approval_request]
+                pause_groups.append(group)
+                surfaced_groups.append(group)
                 continue
             if not _has_authoritative_approval_session(approval_session):
                 visible_requests.append(approval_request)
-                pause_groups.append([approval_request])
+                group = [approval_request]
+                pause_groups.append(group)
+                surfaced_groups.append(group)
                 continue
             already_approved_requests.append(approval_request)
         _store_already_approved_approval_requests(
@@ -2752,7 +2775,7 @@ async def execute_local_calls(function_calls_to_execute: Sequence[Content]) -> l
         )
         _store_pending_approval_requests(approval_session, visible_requests)
         _store_pending_mixed_pause_batch(approval_session, pause_groups)
-        return pause_groups, False
+        return surfaced_groups, False
     if has_declaration_only_call:
         # Host-owned calls pause for caller input, but known local siblings must
         # still execute. A caller-owned session can defer those siblings until
diff --git a/python/packages/core/tests/core/test_tools.py b/python/packages/core/tests/core/test_tools.py
index e734351d05e..47aaa6b13c4 100644
--- a/python/packages/core/tests/core/test_tools.py
+++ b/python/packages/core/tests/core/test_tools.py
@@ -497,6 +497,41 @@ async def list_accounts(**kwargs: Any) -> str:
     assert result_groups[0][0].result == "ok"
 
 
+@pytest.mark.parametrize("arguments", ["not {json", "[1, 2]"])
+async def test_malformed_arguments_are_answered_before_the_call_is_sent_for_approval(
+    arguments: str,
+) -> None:
+    """A payload that is not arguments is answered instead of being sent for approval.
+
+    Approving a call whose arguments are not arguments cannot make that call work, so
+    the model is told what went wrong rather than a human being asked to approve a
+    call that can only fail.
+    """
+    calls: list[dict[str, Any]] = []
+
+    @tool(name="guarded_write", approval_mode="always_require")
+    def guarded_write(value: str = "") -> str:
+        calls.append({"value": value})
+        return "ok"
+
+    result_groups, should_terminate = await _try_execute_function_call_groups(
+        custom_args={},
+        function_calls=[
+            Content.from_function_call(call_id="call-1", name="guarded_write", arguments=arguments),
+        ],
+        tools=[guarded_write],
+        config={},
+    )
+
+    assert not should_terminate
+    assert calls == []
+    assert len(result_groups) == 1
+    result = result_groups[0][0]
+    assert result.type == "function_result"
+    assert "not a JSON object" in str(result.exception)
+    assert all(content.type != "function_approval_request" for group in result_groups for content in group)
+
+
 async def test_invoke_preserves_explicit_null_argument():
     """A required nullable argument the model sets to null must reach the function.
 

From 130f0f98408decc96214ace657e7e48cae338d37 Mon Sep 17 00:00:00 2001
From: Weijian He <230282609+Madin-H23@users.noreply.github.com>
Date: Fri, 23 Oct 2026 12:50:25 +0800
Subject: [PATCH 4/5] Python: fix(tools): validate malformed arguments after
 middleware

Address review feedback: the function-loop contract lets middleware
receive the raw parsed mapping and repair it before any final
validation, so the not-a-JSON-object answer moves into the innermost
handler.

- A payload that is not a JSON object keeps reaching middleware as the
  raw parsed mapping; argument preparation is skipped for it, since the
  model would silently drop the wrapper as an extra field.
- The innermost handler answers a call whose arguments are still the
  wrapped payload, immediately before the tool would run; a repaired
  call executes like any other.
- A malformed call is no longer answered before approval: middleware
  may still repair it once the call is approved.
---
 .../packages/core/agent_framework/_tools.py   | 105 ++++++++---------
 python/packages/core/tests/core/test_tools.py | 111 ++++++++++++++++--
 2 files changed, 149 insertions(+), 67 deletions(-)

diff --git a/python/packages/core/agent_framework/_tools.py b/python/packages/core/agent_framework/_tools.py
index 3df695cbeeb..ee5d1c75c5f 100644
--- a/python/packages/core/agent_framework/_tools.py
+++ b/python/packages/core/agent_framework/_tools.py
@@ -2164,19 +2164,16 @@ def _finalize_function_result(
     return function_result
 
 
-def _parse_function_call_arguments(function_call: Content, tool_name: str) -> dict[str, Any]:
-    """Return the arguments of a function call, rejecting a payload that is not a JSON object.
-
-    `Content.parse_arguments()` wraps a payload that is not a JSON object under a
-    single "raw" key. That is the right answer for a caller that wants to see
-    what the model sent, but not for a function call: the tool would run with an
-    argument it never declared, and the model would not learn that its call was
-    malformed. Such a call is answered like a missing required argument, so the
-    model can retry it.
-
-    The check runs before the middleware pipeline, so middleware never sees a
-    malformed payload as a tool call: a payload that is not arguments is not a
-    call, and it is answered instead of being invoked.
+def _function_call_payload_is_not_an_object(function_call: Content) -> bool:
+    """Return whether the model sent a payload that is not a JSON object.
+
+    `Content.parse_arguments()` wraps such a payload under a single "raw" key.
+    That is the right answer for a caller that wants to see what the model sent,
+    but not for a function call: the tool would run with an argument it never
+    declared, and the model would not learn that its call was malformed. The
+    wrapped mapping still reaches the function middleware, which can repair the
+    arguments the way it repairs any other invalid arguments; when it stays
+    unrepaired, the innermost handler answers the call instead of running it.
     """
     arguments = function_call.arguments
     if isinstance(arguments, str) and arguments:
@@ -2184,12 +2181,8 @@ def _parse_function_call_arguments(function_call: Content, tool_name: str) -> di
             decoded = json.loads(arguments)
         except (json.JSONDecodeError, TypeError):
             decoded = None
-        if not isinstance(decoded, dict):
-            raise _FunctionArgumentValidationError(
-                f"Invalid arguments for '{tool_name}': arguments are not a JSON object",
-                redacted_message=f"Invalid arguments for '{tool_name}'.",
-            )
-    return dict(function_call.parse_arguments() or {})
+        return not isinstance(decoded, dict)
+    return False
 
 
 async def _auto_invoke_function(
@@ -2267,10 +2260,12 @@ async def _auto_invoke_function(
         approval_response = function_call_content
         function_call_content = approved_function_call
 
-    try:
-        parsed_args = _parse_function_call_arguments(function_call_content, tool.name)
-    except _FunctionArgumentValidationError as exc:
-        return _function_argument_validation_error_result(function_call_content, exc, config)
+    # A payload that is not a JSON object still reaches middleware as the raw
+    # parsed mapping it wraps, so middleware has the same chance to repair it
+    # that it has for any other invalid arguments. If it is still unrepaired,
+    # the innermost handler answers it before the tool runs.
+    payload_is_not_an_object = _function_call_payload_is_not_an_object(function_call_content)
+    parsed_args = dict(function_call_content.parse_arguments() or {})
 
     # Filter out internal framework kwargs before passing to tools.
     # conversation_id is an internal tracking ID that should not be forwarded to tools.
@@ -2309,6 +2304,13 @@ async def _auto_invoke_function(
         # No middleware - execute directly
         direct_context = None
         try:
+            if payload_is_not_an_object:
+                # Without middleware there is nothing that could repair the
+                # wrapped payload, so answer it before the tool runs.
+                raise _FunctionArgumentValidationError(
+                    f"Invalid arguments for '{tool.name}': arguments are not a JSON object",
+                    redacted_message=f"Invalid arguments for '{tool.name}'.",
+                )
             if getattr(tool, "_context_parameter_name", None):
                 direct_context = FunctionInvocationContext(
                     function=tool,
@@ -2343,13 +2345,17 @@ async def _auto_invoke_function(
             return _function_execution_error_result(function_call_content, tool.name, exc, config, direct_context)
     # Execute through middleware pipeline if available
     arguments_prepared = False
-    try:
-        args = tool._prepare_arguments(args)  # pyright: ignore[reportPrivateUsage]
-        arguments_prepared = True
-    except _FunctionArgumentValidationError:
-        # Invalid provider arguments are intentionally exposed to middleware so
-        # it has a supported opportunity to repair them before final validation.
-        pass
+    if not payload_is_not_an_object:
+        try:
+            args = tool._prepare_arguments(args)  # pyright: ignore[reportPrivateUsage]
+            arguments_prepared = True
+        except _FunctionArgumentValidationError:
+            # Invalid provider arguments are intentionally exposed to middleware so
+            # it has a supported opportunity to repair them before final validation.
+            pass
+    # A wrapped malformed payload skips preparation: the model would silently
+    # drop the wrapper as an extra field, and middleware would never see what
+    # the model actually sent.
 
     middleware_context = FunctionInvocationContext(
         function=tool,
@@ -2388,10 +2394,20 @@ async def _auto_invoke_function(
         )
 
     final_handler_started = False
+    unrepaired_payload_token = _argument_comparison_token(args) if payload_is_not_an_object else None
 
     async def final_function_handler(context_obj: Any) -> Any:
         nonlocal final_handler_started
         final_handler_started = True
+        if unrepaired_payload_token is not None and _argument_comparison_token(context_obj.arguments) == (
+            unrepaired_payload_token
+        ):
+            # Middleware left the wrapped payload untouched, so the arguments are
+            # still not arguments: answer it here, immediately before the tool runs.
+            raise _FunctionArgumentValidationError(
+                f"Invalid arguments for '{tool.name}': arguments are not a JSON object",
+                redacted_message=f"Invalid arguments for '{tool.name}'.",
+            )
         return await tool.invoke(
             arguments=context_obj.arguments,
             context=context_obj,
@@ -2719,7 +2735,6 @@ async def execute_local_calls(function_calls_to_execute: Sequence[Content]) -> l
             if function_call.type == "function_call"
         ]
         pause_groups: list[list[Content]] = []
-        surfaced_groups: list[list[Content]] = []
         for function_call in function_calls:
             if function_call.type != "function_call":
                 continue
@@ -2729,42 +2744,24 @@ async def execute_local_calls(function_calls_to_execute: Sequence[Content]) -> l
                 and tool_name not in approval_tool_names
                 and (tool_name in declaration_only_tool_names or tool_name in additional_tool_names)
             ):
-                group = [_as_user_input_pause(function_call)]
-                pause_groups.append(group)
-                surfaced_groups.append(group)
+                pause_groups.append([_as_user_input_pause(function_call)])
                 continue
-            if tool_name is not None:
-                # A payload that is not arguments cannot become a working call by
-                # approving it, so answer it here instead of asking a human to
-                # approve a call that can only fail. The same guard runs again in
-                # _auto_invoke_function.
-                try:
-                    _parse_function_call_arguments(function_call, tool_name)
-                except _FunctionArgumentValidationError as exc:
-                    surfaced_groups.append([_function_argument_validation_error_result(function_call, exc, config)])
-                    continue
             approval_request = Content.from_function_approval_request(
                 id=function_call.id or function_call.call_id,  # type: ignore[arg-type]
                 function_call=function_call,
             )
             if tool_name is None:
                 visible_requests.append(approval_request)
-                group = [approval_request]
-                pause_groups.append(group)
-                surfaced_groups.append(group)
+                pause_groups.append([approval_request])
                 continue
             tool = tool_map.get(tool_name)
             if tool_name in approval_tool_names or tool is None:
                 visible_requests.append(approval_request)
-                group = [approval_request]
-                pause_groups.append(group)
-                surfaced_groups.append(group)
+                pause_groups.append([approval_request])
                 continue
             if not _has_authoritative_approval_session(approval_session):
                 visible_requests.append(approval_request)
-                group = [approval_request]
-                pause_groups.append(group)
-                surfaced_groups.append(group)
+                pause_groups.append([approval_request])
                 continue
             already_approved_requests.append(approval_request)
         _store_already_approved_approval_requests(
@@ -2775,7 +2772,7 @@ async def execute_local_calls(function_calls_to_execute: Sequence[Content]) -> l
         )
         _store_pending_approval_requests(approval_session, visible_requests)
         _store_pending_mixed_pause_batch(approval_session, pause_groups)
-        return surfaced_groups, False
+        return pause_groups, False
     if has_declaration_only_call:
         # Host-owned calls pause for caller input, but known local siblings must
         # still execute. A caller-owned session can defer those siblings until
diff --git a/python/packages/core/tests/core/test_tools.py b/python/packages/core/tests/core/test_tools.py
index 47aaa6b13c4..b2a74a198ea 100644
--- a/python/packages/core/tests/core/test_tools.py
+++ b/python/packages/core/tests/core/test_tools.py
@@ -3,9 +3,10 @@
 import copy
 import logging
 import threading
+from collections.abc import Awaitable, Callable
 from contextvars import ContextVar
 from datetime import date, datetime
-from typing import Annotated, Any, Literal, get_args, get_origin
+from typing import Annotated, Any, Literal, cast, get_args, get_origin
 from unittest.mock import Mock
 
 import pytest
@@ -20,7 +21,11 @@
     FunctionTool,
     tool,
 )
-from agent_framework._middleware import FunctionInvocationContext
+from agent_framework._middleware import (
+    FunctionInvocationContext,
+    FunctionMiddleware,
+    FunctionMiddlewarePipeline,
+)
 from agent_framework._tools import (
     _auto_invoke_function,
     _format_tool_parameters,
@@ -497,15 +502,12 @@ async def list_accounts(**kwargs: Any) -> str:
     assert result_groups[0][0].result == "ok"
 
 
-@pytest.mark.parametrize("arguments", ["not {json", "[1, 2]"])
-async def test_malformed_arguments_are_answered_before_the_call_is_sent_for_approval(
-    arguments: str,
-) -> None:
-    """A payload that is not arguments is answered instead of being sent for approval.
+async def test_malformed_arguments_reach_approval_like_any_other_call() -> None:
+    """A malformed call is not answered before approval.
 
-    Approving a call whose arguments are not arguments cannot make that call work, so
-    the model is told what went wrong rather than a human being asked to approve a
-    call that can only fail.
+    Middleware may still repair the arguments once the call is approved, so the
+    request goes to approval the way every other call does; the innermost
+    handler answers the call only if the payload is still unrepaired there.
     """
     calls: list[dict[str, Any]] = []
 
@@ -517,7 +519,7 @@ def guarded_write(value: str = "") -> str:
     result_groups, should_terminate = await _try_execute_function_call_groups(
         custom_args={},
         function_calls=[
-            Content.from_function_call(call_id="call-1", name="guarded_write", arguments=arguments),
+            Content.from_function_call(call_id="call-1", name="guarded_write", arguments="not {json"),
         ],
         tools=[guarded_write],
         config={},
@@ -526,10 +528,93 @@ def guarded_write(value: str = "") -> str:
     assert not should_terminate
     assert calls == []
     assert len(result_groups) == 1
+    assert result_groups[0][0].type == "function_approval_request"
+
+
+async def test_middleware_can_repair_a_payload_that_is_not_a_json_object() -> None:
+    """Middleware sees the raw parsed mapping and may repair it into real arguments.
+
+    This is the function-loop contract: middleware receives the arguments before
+    any final validation, and a repaired call executes like any other.
+    """
+
+    calls: list[dict[str, Any]] = []
+
+    @tool(name="list_accounts")
+    def list_accounts(limit: int = 0) -> str:
+        calls.append({"limit": limit})
+        return "ok"
+
+    class RepairRawArguments(FunctionMiddleware):
+        async def process(
+            self,
+            context: FunctionInvocationContext,
+            call_next: Callable[[], Awaitable[None]],
+        ) -> None:
+            arguments = cast("dict[str, Any]", context.arguments)
+            if "raw" in arguments:
+                context.arguments = {"limit": 5}
+            await call_next()
+
+    result_groups, should_terminate = await _try_execute_function_call_groups(
+        custom_args={},
+        function_calls=[
+            Content.from_function_call(call_id="call-1", name="list_accounts", arguments="[1, 2]"),
+        ],
+        tools=[list_accounts],
+        config={},
+        middleware_pipeline=FunctionMiddlewarePipeline(RepairRawArguments()),
+    )
+
+    assert not should_terminate
+    assert calls == [{"limit": 5}]
+    assert result_groups[0][0].result == "ok"
+
+
+@pytest.mark.parametrize("arguments", ["not {json", "[1, 2]"])
+async def test_malformed_arguments_left_unrepaired_are_answered_before_the_tool_runs(
+    arguments: str,
+) -> None:
+    """When middleware does not repair the wrapped payload, the call is answered.
+
+    The error is the argument-validation result, raised immediately before the
+    tool would have executed, so the tool never runs on undeclared arguments.
+    """
+    calls: list[dict[str, Any]] = []
+
+    async def list_accounts(**kwargs: Any) -> str:
+        calls.append(kwargs)
+        return "ok"
+
+    account_tool = FunctionTool(
+        name="list_accounts",
+        description="List the accounts",
+        func=list_accounts,
+        input_model={"type": "object", "properties": {}},
+    )
+
+    class Passthrough(FunctionMiddleware):
+        async def process(
+            self,
+            context: FunctionInvocationContext,
+            call_next: Callable[[], Awaitable[None]],
+        ) -> None:
+            await call_next()
+
+    result_groups, should_terminate = await _try_execute_function_call_groups(
+        custom_args={},
+        function_calls=[
+            Content.from_function_call(call_id="call-1", name="list_accounts", arguments=arguments),
+        ],
+        tools=[account_tool],
+        config={},
+        middleware_pipeline=FunctionMiddlewarePipeline(Passthrough()),
+    )
+
+    assert not should_terminate
+    assert calls == []
     result = result_groups[0][0]
-    assert result.type == "function_result"
     assert "not a JSON object" in str(result.exception)
-    assert all(content.type != "function_approval_request" for group in result_groups for content in group)
 
 
 async def test_invoke_preserves_explicit_null_argument():

From 8c52ac40b720668b6f6e01ad659fdfaa85d060d8 Mon Sep 17 00:00:00 2001
From: H23 <madin23@agent.qq.com>
Date: Sat, 24 Oct 2026 22:37:44 +0800
Subject: [PATCH 5/5] Python: fix(tools): recognize an injected-arguments
 middleware as leaving the payload unrepaired

Address review follow-up: the repaired/unrepaired decision compared the
whole arguments mapping, so a middleware that appends a key to every call
(a user or trace id) preserved the wrapped payload while counting as a
repair, and the tool still ran with the undeclared raw argument.

The wrapper now counts as repaired only when its original raw entry is
gone or rewritten. A raw entry the tool itself declares is ordinary
arguments and never counts as the wrapper, which also restores the
pre-guard behaviour for tools that accept raw by name.
---
 .../packages/core/agent_framework/_tools.py   |  43 ++++++--
 python/packages/core/tests/core/test_tools.py | 104 ++++++++++++++++++
 2 files changed, 139 insertions(+), 8 deletions(-)

diff --git a/python/packages/core/agent_framework/_tools.py b/python/packages/core/agent_framework/_tools.py
index ee5d1c75c5f..68386d0902c 100644
--- a/python/packages/core/agent_framework/_tools.py
+++ b/python/packages/core/agent_framework/_tools.py
@@ -2164,6 +2164,19 @@ def _finalize_function_result(
     return function_result
 
 
+def _tool_declares_parameter(tool: FunctionTool, name: str) -> bool:
+    """Return whether the tool's own schema declares a parameter with this name.
+
+    Schema-supplied tools declare through their JSON ``properties``; decorated
+    tools through their resolved input model. A ``**kwargs`` catch-all does not
+    declare anything: the model never agreed to such a parameter by name.
+    """
+    if tool._schema_supplied:  # pyright: ignore[reportPrivateUsage]
+        properties = tool._input_schema_cached.get("properties") if tool._input_schema_cached else None  # pyright: ignore[reportPrivateUsage]
+        return isinstance(properties, Mapping) and name in properties
+    return tool.input_model is not None and name in tool.input_model.model_fields
+
+
 def _function_call_payload_is_not_an_object(function_call: Content) -> bool:
     """Return whether the model sent a payload that is not a JSON object.
 
@@ -2266,6 +2279,19 @@ async def _auto_invoke_function(
     # the innermost handler answers it before the tool runs.
     payload_is_not_an_object = _function_call_payload_is_not_an_object(function_call_content)
     parsed_args = dict(function_call_content.parse_arguments() or {})
+    wrapped_raw_value = parsed_args.get("raw") if payload_is_not_an_object and set(parsed_args) == {"raw"} else None
+
+    def _unrepaired_wrapper(arguments: Any) -> bool:
+        """Return whether the arguments are still the wrapped malformed payload.
+
+        The wrapper counts as repaired only when middleware replaced it. Keeping
+        the original ``raw`` entry - which an argument-injecting middleware does
+        - leaves the call malformed, while a ``raw`` entry the tool itself
+        declares is ordinary arguments and never counts as the wrapper.
+        """
+        if wrapped_raw_value is None or _tool_declares_parameter(tool, "raw"):
+            return False
+        return isinstance(arguments, Mapping) and cast("Mapping[str, Any]", arguments).get("raw") == wrapped_raw_value
 
     # Filter out internal framework kwargs before passing to tools.
     # conversation_id is an internal tracking ID that should not be forwarded to tools.
@@ -2304,9 +2330,11 @@ async def _auto_invoke_function(
         # No middleware - execute directly
         direct_context = None
         try:
-            if payload_is_not_an_object:
+            if _unrepaired_wrapper(args):
                 # Without middleware there is nothing that could repair the
-                # wrapped payload, so answer it before the tool runs.
+                # wrapped payload, so answer it before the tool runs. A tool
+                # that declares ``raw`` receives the payload as ordinary
+                # arguments instead.
                 raise _FunctionArgumentValidationError(
                     f"Invalid arguments for '{tool.name}': arguments are not a JSON object",
                     redacted_message=f"Invalid arguments for '{tool.name}'.",
@@ -2394,16 +2422,15 @@ async def _auto_invoke_function(
         )
 
     final_handler_started = False
-    unrepaired_payload_token = _argument_comparison_token(args) if payload_is_not_an_object else None
 
     async def final_function_handler(context_obj: Any) -> Any:
         nonlocal final_handler_started
         final_handler_started = True
-        if unrepaired_payload_token is not None and _argument_comparison_token(context_obj.arguments) == (
-            unrepaired_payload_token
-        ):
-            # Middleware left the wrapped payload untouched, so the arguments are
-            # still not arguments: answer it here, immediately before the tool runs.
+        if _unrepaired_wrapper(context_obj.arguments):
+            # Middleware left the wrapped payload in place - an argument-injecting
+            # middleware keeps its ``raw`` entry with the original value - so the
+            # arguments are still not arguments: answer it here, immediately
+            # before the tool runs.
             raise _FunctionArgumentValidationError(
                 f"Invalid arguments for '{tool.name}': arguments are not a JSON object",
                 redacted_message=f"Invalid arguments for '{tool.name}'.",
diff --git a/python/packages/core/tests/core/test_tools.py b/python/packages/core/tests/core/test_tools.py
index b2a74a198ea..c9d4e702a15 100644
--- a/python/packages/core/tests/core/test_tools.py
+++ b/python/packages/core/tests/core/test_tools.py
@@ -617,6 +617,110 @@ async def process(
     assert "not a JSON object" in str(result.exception)
 
 
+class _InjectUser(FunctionMiddleware):
+    """Middleware that appends a key to every call's arguments."""
+
+    async def process(
+        self,
+        context: FunctionInvocationContext,
+        call_next: Callable[[], Awaitable[None]],
+    ) -> None:
+        context.arguments = {**cast("dict[str, Any]", context.arguments), "user": "u1"}
+        await call_next()
+
+
+async def test_injecting_middleware_does_not_make_a_malformed_payload_count_as_repaired() -> None:
+    """A middleware that appends a key must not un-mark the wrapped payload.
+
+    The injection preserves the original ``raw`` entry, so the arguments are
+    still the wrapper plus an undeclared key - the tool must not run.
+    """
+    calls: list[dict[str, Any]] = []
+
+    async def list_accounts(**kwargs: Any) -> str:
+        calls.append(kwargs)
+        return "ok"
+
+    account_tool = FunctionTool(
+        name="list_accounts",
+        description="List the accounts",
+        func=list_accounts,
+        input_model={"type": "object", "properties": {}},
+    )
+
+    result_groups, should_terminate = await _try_execute_function_call_groups(
+        custom_args={},
+        function_calls=[
+            Content.from_function_call(call_id="call-1", name="list_accounts", arguments="[1, 2]"),
+        ],
+        tools=[account_tool],
+        config={},
+        middleware_pipeline=FunctionMiddlewarePipeline(_InjectUser()),
+    )
+
+    assert not should_terminate
+    assert calls == []
+    result = result_groups[0][0]
+    assert "not a JSON object" in str(result.exception)
+
+
+async def test_injecting_middleware_does_not_silence_a_malformed_payload_for_a_typed_tool() -> None:
+    """The same holds for a typed tool: the wrapper must not reach it either.
+
+    The input model would silently drop the undeclared ``raw`` key, so the tool
+    would run on injected values alone without the model ever learning that its
+    call was malformed.
+    """
+    calls: list[dict[str, Any]] = []
+
+    @tool(name="typed_target")
+    def typed_target(user: str = "", limit: int = 0) -> str:
+        calls.append({"user": user, "limit": limit})
+        return "ok"
+
+    result_groups, should_terminate = await _try_execute_function_call_groups(
+        custom_args={},
+        function_calls=[
+            Content.from_function_call(call_id="call-1", name="typed_target", arguments="[1, 2]"),
+        ],
+        tools=[typed_target],
+        config={},
+        middleware_pipeline=FunctionMiddlewarePipeline(_InjectUser()),
+    )
+
+    assert not should_terminate
+    assert calls == []
+    result = result_groups[0][0]
+    assert "not a JSON object" in str(result.exception)
+
+
+async def test_a_tool_that_declares_raw_receives_a_non_object_payload_as_arguments() -> None:
+    """A tool that declares ``raw`` gets the wrapped payload as ordinary arguments.
+
+    For such a tool the wrapper coincides with a valid call, which is the
+    behaviour the loop had before the malformed-argument guard existed.
+    """
+    calls: list[dict[str, Any]] = []
+
+    @tool(name="raw_writer")
+    def raw_writer(raw: str = "") -> str:
+        calls.append({"raw": raw})
+        return "ok"
+
+    result_groups, should_terminate = await _try_execute_function_call_groups(
+        custom_args={},
+        function_calls=[
+            Content.from_function_call(call_id="call-1", name="raw_writer", arguments="not {json"),
+        ],
+        tools=[raw_writer],
+        config={},
+    )
+
+    assert not should_terminate
+    assert calls == [{"raw": "not {json"}]
+    assert result_groups[0][0].result == "ok"
+
+
 async def test_invoke_preserves_explicit_null_argument():
     """A required nullable argument the model sets to null must reach the function.