diff --git a/python/packages/core/agent_framework/_harness/_agent.py b/python/packages/core/agent_framework/_harness/_agent.py index 1a6178b54d3..4b5518690bb 100644 --- a/python/packages/core/agent_framework/_harness/_agent.py +++ b/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, ) diff --git a/python/packages/core/tests/core/test_harness_agent.py b/python/packages/core/tests/core/test_harness_agent.py index 0a280e87ed8..400ecf18335 100644 --- a/python/packages/core/tests/core/test_harness_agent.py +++ b/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 diff --git a/python/samples/02-agents/harness/README.md b/python/samples/02-agents/harness/README.md index e162c1ff45e..13fa2aa7f5d 100644 --- a/python/samples/02-agents/harness/README.md +++ b/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. diff --git a/python/samples/02-agents/harness/console/formatters.py b/python/samples/02-agents/harness/console/formatters.py index 47327c6637f..13fc8a261e8 100644 --- a/python/samples/02-agents/harness/console/formatters.py +++ b/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: diff --git a/python/samples/02-agents/harness/console/observers/tool_approval.py b/python/samples/02-agents/harness/console/observers/tool_approval.py index 1fc9533a0d8..e06bebc2aa6 100644 --- a/python/samples/02-agents/harness/console/observers/tool_approval.py +++ b/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"