Python: Integrate tool approval into the harness

python/packages/core/agent_framework/_harness/_agent.py

@@ -24,6 +24,7 @@
 from ._memory import MemoryContextProvider, MemoryStore
 from ._mode import AgentModeProvider
 from ._todo import TodoProvider
+from ._tool_approval import ToolApprovalMiddleware
 
 if TYPE_CHECKING:
     from collections.abc import Mapping
@@ -34,6 +35,7 @@
     from .._compaction import CompactionStrategy, TokenizerProtocol
     from .._middleware import MiddlewareTypes
     from .._tools import ToolTypes
+    from ._tool_approval import ToolApprovalRuleCallback
 
 logger = logging.getLogger(__name__)
 
@@ -250,6 +252,8 @@ def create_harness_agent(
     shell_executor: ShellExecutor | None = None,
     shell_environment_provider_options: ShellEnvironmentProviderOptions | None = None,
     disable_web_search: bool = False,
+    disable_tool_auto_approval: bool = False,
+    auto_approval_rules: Sequence[ToolApprovalRuleCallback] | None = None,
     otel_provider_name: str | None = None,
     context_providers: Sequence[ContextProvider] | None = None,
     middleware: Sequence[MiddlewareTypes] | None = None,
@@ -267,6 +271,8 @@ def create_harness_agent(
     - **MemoryContextProvider** — file-based durable memory (when ``memory_store`` provided)
     - **SkillsProvider** — skill discovery and progressive loading
     - **BackgroundAgentsProvider** — delegate work to background sub-agents
+    - **Tool approval** — "don't ask again" standing approval rules plus heuristic
+      auto-approval callbacks
     - **OpenTelemetry** — observability via ``AgentTelemetryLayer``
 
     Each feature can be disabled or customized via keyword arguments.
@@ -364,6 +370,16 @@ def create_harness_agent(
             When False (default), the web search tool is automatically added if the
             client implements SupportsWebSearchTool. A warning is logged if the client
             does not support web search.
+        disable_tool_auto_approval: When True, do not wire the tool auto-approval middleware.
+            When False (default), a :class:`~agent_framework.ToolApprovalMiddleware` is added
+            (outermost) to coordinate "don't ask again" standing approval rules and queued
+            approval prompts; callers must pass an :class:`~agent_framework.AgentSession` to
+            :meth:`~agent_framework.Agent.run` when enabled.
+        auto_approval_rules: Optional heuristic callbacks that can auto-approve a function call
+            that would otherwise require approval. Each callback receives the ``function_call``
+            content and returns ``True`` to approve it. Rules are evaluated after standing rules
+            (derived from prior user approvals) but before prompting the user. Only used when
+            ``disable_tool_auto_approval`` is False.
         otel_provider_name: Custom OpenTelemetry provider/source name for telemetry.
         context_providers: Additional context providers to include after the built-in ones.
         middleware: Additional middleware to include.
@@ -455,6 +471,16 @@ def create_harness_agent(
     if max_output_tokens is not None:
         default_opts.setdefault("max_tokens", max_output_tokens)
 
+    # Assemble middleware. Tool approval is enabled by default (like the .NET harness) and is
+    # placed first so it sits outermost: it intercepts inbound "always approve" responses and
+    # outbound approval requests at the caller boundary, and its re-invocation loop re-runs any
+    # user-supplied middleware. ToolApprovalMiddleware requires an AgentSession at run time.
+    assembled_middleware: list[MiddlewareTypes] = []
+    if not disable_tool_auto_approval:
+        assembled_middleware.append(ToolApprovalMiddleware(auto_approval_rules=auto_approval_rules))
+    if middleware:
+        assembled_middleware.extend(middleware)
+
     agent = Agent(
         client,
         instructions,
@@ -464,7 +490,7 @@ def create_harness_agent(
         tools=final_tools,
         default_options=default_opts,  # type: ignore[arg-type]
         context_providers=assembled_providers,
-        middleware=list(middleware) if middleware else None,
+        middleware=assembled_middleware or None,
         require_per_service_call_history_persistence=True,
     )
 

python/packages/core/tests/core/test_harness_agent.py

@@ -667,3 +667,104 @@ class _BadExecutor:
             disable_web_search=True,
             shell_executor=_BadExecutor(),
         )
+
+
+# --- Tool Approval Tests ---
+
+
+def _find_tool_approval_middleware(agent: Any) -> Any:
+    from agent_framework import ToolApprovalMiddleware
+
+    for mw in agent.middleware or []:
+        if isinstance(mw, ToolApprovalMiddleware):
+            return mw
+    return None
+
+
+def test_create_harness_agent_adds_tool_approval_by_default() -> None:
+    """Tool approval middleware should be wired in by default."""
+    agent = create_harness_agent(
+        client=_FakeChatClient(),  # type: ignore[arg-type]
+        max_context_window_tokens=128_000,
+        max_output_tokens=16_384,
+    )
+    assert _find_tool_approval_middleware(agent) is not None
+
+
+def test_create_harness_agent_disable_tool_auto_approval() -> None:
+    """disable_tool_auto_approval=True should omit the tool approval middleware."""
+    agent = create_harness_agent(
+        client=_FakeChatClient(),  # type: ignore[arg-type]
+        max_context_window_tokens=128_000,
+        max_output_tokens=16_384,
+        disable_tool_auto_approval=True,
+    )
+    assert _find_tool_approval_middleware(agent) is None
+
+
+def test_create_harness_agent_passes_auto_approval_rules() -> None:
+    """auto_approval_rules should be forwarded to the tool approval middleware."""
+
+    def _rule(content: Any) -> bool:
+        return True
+
+    agent = create_harness_agent(
+        client=_FakeChatClient(),  # type: ignore[arg-type]
+        max_context_window_tokens=128_000,
+        max_output_tokens=16_384,
+        auto_approval_rules=[_rule],
+    )
+    middleware = _find_tool_approval_middleware(agent)
+    assert middleware is not None
+    assert _rule in middleware.auto_approval_rules
+
+
+def test_create_harness_agent_tool_approval_outermost_with_user_middleware() -> None:
+    """Tool approval middleware should be placed first (outermost) ahead of user middleware."""
+    from agent_framework import AgentMiddleware, ToolApprovalMiddleware
+
+    class _CustomMiddleware(AgentMiddleware):
+        async def process(self, context: Any, call_next: Any) -> None:
+            await call_next()
+
+    custom = _CustomMiddleware()
+    agent = create_harness_agent(
+        client=_FakeChatClient(),  # type: ignore[arg-type]
+        max_context_window_tokens=128_000,
+        max_output_tokens=16_384,
+        middleware=[custom],
+    )
+    assert agent.middleware is not None
+    assert isinstance(agent.middleware[0], ToolApprovalMiddleware)
+    assert custom in agent.middleware
+    assert agent.middleware.index(custom) > 0
+
+
+def test_create_harness_agent_disable_tool_auto_approval_preserves_user_middleware() -> None:
+    """When tool approval is disabled, only user-supplied middleware should remain."""
+    from agent_framework import AgentMiddleware
+
+    class _CustomMiddleware(AgentMiddleware):
+        async def process(self, context: Any, call_next: Any) -> None:
+            await call_next()
+
+    custom = _CustomMiddleware()
+    agent = create_harness_agent(
+        client=_FakeChatClient(),  # type: ignore[arg-type]
+        max_context_window_tokens=128_000,
+        max_output_tokens=16_384,
+        disable_tool_auto_approval=True,
+        middleware=[custom],
+    )
+    assert agent.middleware == [custom]
+
+
+def test_create_harness_agent_no_middleware_when_tool_approval_disabled_and_none() -> None:
+    """No middleware should be installed when tool approval is disabled and none is supplied."""
+    agent = create_harness_agent(
+        client=_FakeChatClient(),  # type: ignore[arg-type]
+        max_context_window_tokens=128_000,
+        max_output_tokens=16_384,
+        disable_tool_auto_approval=True,
+    )
+    assert agent.middleware is None

python/samples/02-agents/harness/README.md

@@ -18,6 +18,7 @@ from a chat client.
 | MemoryContextProvider | File-based durable memory (when `memory_store` provided) |
 | SkillsProvider | File-based skill discovery and progressive loading |
 | Shell tool | Shell command execution + environment probing (when `shell_executor` provided) |
+| Tool approval | "Don't ask again" standing rules + heuristic auto-approval (enabled by default) |
 | OpenTelemetry | Built-in observability |
 
 Each feature can be disabled or customized via keyword arguments.

python/samples/02-agents/harness/console/formatters.py

@@ -182,6 +182,8 @@ def format_detail(self, call: Content) -> str | None:
                 args_dict = json.loads(call.arguments)
             except (json.JSONDecodeError, TypeError):
                 return None
+            if not isinstance(args_dict, dict):
+                return None
         elif isinstance(call.arguments, dict):
             args_dict = call.arguments
         else:

python/samples/02-agents/harness/console/observers/tool_approval.py

@@ -84,23 +84,38 @@ def _build_approval_question(self, request: Content) -> ChoiceFollowUpQuestion:
         tool_name = self._format_tool_name(request)
         prompt = f"🔐 Tool approval: {tool_name}"
 
-        # TODO(westey-m): Add "Always approve" options when the framework supports
-        # CreateAlwaysApproveToolResponse / CreateAlwaysApproveToolWithArgumentsResponse.
-        choices = [
-            "Approve this call",
-            "Deny",
-        ]
+        approve_once = "Approve this call"
+        always_tool = "Always approve this tool (any arguments)"
+        always_tool_args = "Always approve this tool with these arguments"
+        deny = "Deny"
+        choices = [approve_once, always_tool, always_tool_args, deny]
 
         async def continuation(
             selection: str,
             ux: IUXStateDriver,
         ) -> Message | None:
-            from agent_framework import Message
+            from agent_framework import (
+                Message,
+                create_always_approve_tool_response,
+                create_always_approve_tool_with_arguments_response,
+            )
 
-            if selection == "Deny":
+            if selection == deny:
                 response_content = request.to_function_approval_response(approved=False)
                 action_label = "❌ Denied"
                 color = "red"
+            elif selection == always_tool:
+                response_content = create_always_approve_tool_response(
+                    request, reason="User chose to always approve this tool"
+                )
+                action_label = "✅ Always approved (any args)"
+                color = "green"
+            elif selection == always_tool_args:
+                response_content = create_always_approve_tool_with_arguments_response(
+                    request, reason="User chose to always approve this tool with these arguments"
+                )
+                action_label = "✅ Always approved (these args)"
+                color = "green"
             else:
                 response_content = request.to_function_approval_response(approved=True)
                 action_label = "✅ Approved"