Python: Defer provider-injected tool approvals to harness in-run execution

python/packages/ag-ui/agent_framework_ag_ui/_agent_run.py

@@ -37,6 +37,7 @@
 from agent_framework._tools import (
     _ALREADY_APPROVED_APPROVAL_REQUEST_GROUPS_KEY,  # type: ignore
     _collect_approval_responses,  # type: ignore
+    _get_tool_map,  # type: ignore
     _replace_approval_contents_with_results,  # type: ignore
     _TOOL_APPROVAL_STATE_KEY,  # type: ignore
     _try_execute_function_calls,  # type: ignore
@@ -1304,8 +1305,21 @@ async def _resolve_approval_responses(
 
     approved_function_results: list[Any] = []
 
-    # Execute approved tool calls
-    if approved_responses and tools:
+    # Partition approved responses into static (execute now) and deferred (execute during run)
+    tool_map = _get_tool_map(tools) if tools else {}
+    static_approved = []
+    deferred_approval_ids = set()
+
+    for approval in approved_responses:
+        tool_name = approval.function_call.name if approval.function_call else None
+        if tool_name not in tool_map:
+            # Provider-injected tool — defer to harness for execution during run
+            deferred_approval_ids.add(approval.id or "")
+        else:
+            static_approved.append(approval)
+
+    # Execute only statically-available approved tool calls
+    if static_approved and tools:
         client = getattr(agent, "client", None)
         config = normalize_function_invocation_configuration(getattr(client, "function_invocation_configuration", None))
         middleware_pipeline = FunctionMiddlewarePipeline(
@@ -1318,7 +1332,7 @@ async def _resolve_approval_responses(
             results, _ = await _try_execute_function_calls(
                 custom_args=tool_kwargs,
                 attempt_idx=0,
-                function_calls=approved_responses,
+                function_calls=static_approved,
                 tools=tools,
                 middleware_pipeline=middleware_pipeline,
                 config=config,
@@ -1328,9 +1342,10 @@ async def _resolve_approval_responses(
             logger.exception("Failed to execute approved tool calls; injecting error results: %s", e)
             approved_function_results = []
 
-    # Build results for approved responses (used for TOOL_CALL_RESULT event emission)
+    # Build results for executed (static) approved responses (used for TOOL_CALL_RESULT event emission)
+    # Deferred provider-injected approvals are left in messages for ToolApprovalMiddleware to process.
     approved_results: list[Content] = []
-    for idx, approval in enumerate(approved_responses):
+    for idx, approval in enumerate(static_approved):
         if (
             idx < len(approved_function_results)
             and getattr(approved_function_results[idx], "type", None) == "function_result"

python/packages/ag-ui/tests/ag_ui/test_endpoint.py

@@ -4038,3 +4038,106 @@ def factory(thread_id: str) -> Any:
 
     runner.clear_thread_workflow("thread-1")
     assert runner._resolve_workflow("thread-1", "tenant-b") is not workflow_b
+
+
+async def test_endpoint_agent_approval_defers_provider_injected_tools() -> None:
+    """Approving provider-injected tools should not produce rejection errors.
+
+    When approval responses include tools not in the static tool map,
+    they should be deferred (not executed by transport) rather than
+    producing "Tool call invocation was rejected" errors.
+
+    This prevents the scenario where approval mode loops with tool invocation failures.
+    Fixes issue #7043.
+    """
+    executed_tools: list[str] = []
+
+    def static_tool() -> str:
+        executed_tools.append("static")
+        return "static result"
+
+    # Create approval responses: one for static tool, one for provider tool (not in static list)
+    static_approval = Content.from_function_approval_response(
+        id="static_id",
+        approved=True,
+        function_call=Content.from_function_call(
+            call_id="call_static",
+            name="static_tool",
+            arguments="{}",
+        ),
+    )
+    provider_approval = Content.from_function_approval_response(
+        id="provider_id",
+        approved=True,
+        function_call=Content.from_function_call(
+            call_id="call_provider",
+            name="provider_file_access",  # Not in agent's tool list
+            arguments="{}",
+        ),
+    )
+
+    # User message containing both approval responses
+    message = Message(
+        role="user",
+        contents=[static_approval, provider_approval],
+    )
+
+    # Stub agent that returns the approval responses embedded in a message
+    agent = StubAgent(
+        updates=[
+            AgentResponseUpdate(contents=[static_approval, provider_approval], role="user"),
+        ]
+    )
+
+    app = FastAPI()
+    wrapped_agent = AgentFrameworkAgent(agent=agent, require_confirmation=False)
+    add_agent_framework_fastapi_endpoint(app, wrapped_agent, path="/approval")
+    client = TestClient(app)
+
+    # Submit approval responses (static and provider-injected)
+    response = client.post(
+        "/approval",
+        json={
+            "runId": "run-approval",
+            "threadId": "thread-approval",
+            "messages": [
+                {
+                    "role": "user",
+                    "content": [
+                        {
+                            "type": "function_approval_response",
+                            "id": "static_id",
+                            "approved": True,
+                            "function_call": {
+                                "call_id": "call_static",
+                                "name": "static_tool",
+                                "arguments": "{}",
+                            },
+                        },
+                        {
+                            "type": "function_approval_response",
+                            "id": "provider_id",
+                            "approved": True,
+                            "function_call": {
+                                "call_id": "call_provider",
+                                "name": "provider_file_access",
+                                "arguments": "{}",
+                            },
+                        },
+                    ],
+                }
+            ],
+        },
+    )
+
+    assert response.status_code == 200
+    events = _decode_sse_events(response)
+    response_text = json.dumps(events)
+
+    # Verify no rejection error for provider tool
+    assert "Tool call invocation was rejected" not in response_text, (
+        "Deferred provider tool should not produce rejection error"
+    )
+
+
+